diff --git a/BUILD-LOG.md b/BUILD-LOG.md index d4390ce93..7d6633d47 100644 --- a/BUILD-LOG.md +++ b/BUILD-LOG.md @@ -6011,3 +6011,4 @@ Owner directed static-only fallback after actual nativepriority drift: no new pr 2026-11-03 | GPT-6.1 Terra implementation sub-agent | #797 / Gitea #1839 lifecycle Task 7a | Added a bounded reviewed-artifact reader for committed regular blobs in trusted bare working repositories. Uses the existing bounded Git reader and safe environment, literal paths, disabled replacement refs, full commit IDs and a 1 MiB ceiling; rejects non-regular entries and checks size before reading bytes. RED: missing reader module; GREEN: artifact 9/9 and seat 232/232, zero skips; logs /mnt/storage/scratch/tmp/0G9g1p/. Queue next refused linked-worktree use; controller owns queue and integration. Local commit only; no delivery, broker, Console, push or deployment. 2026-11-04T03:48:47Z | GPT-6.1 Sol implementation sub-agent, t3/pm-intake-tools | #797 / Gitea #1839 lifecycle Task 6a before/after | Assigned base 6dd4f92b0; harness had no intake tools and staff.request required a task. Added intake_open/request.open and intake_propose/request.propose only for PM holding message.send, plus intake_status/request.get and intake_projects/request.projects in PM and CTO pre-task manifests, independent of tracker/checkouts. Existing broker authority is unchanged: its request reads still refuse CTO (request-access-refused); tested and retained rather than widening authority. intake_question omitted because no backing request.question verb exists. staff.request/request_staff takes exactly one of request or task_ref with roles/note/decision unchanged, PM-only message.send and chief routing preserved; original request provenance/reachability uses the existing host check without a fabricated task or opened journal. Exported the existing runner toolServer for acceptance tests against production dispatch, including MCP; no new imports or sandbox/process mounts needed (import closure 1/1). PM duty prose and the generated PM staffing prompt describe pre-task intake and staffing. New tests cover actual PM/CTO/coder/reviewer/assistant bundles for pi and claude-code, assistant -> chief -> PM relays, exact user-mode decision with null task/requirement, duplicate open, wrong roles, invalid/unreachable input, exclusive staff references and assistant coordination restrictions. RED on assigned parent: 6 pass/12 fail (dispatch not exported); after exporting only the existing dispatch seam 7 pass/11 fail for missing tools/staff intake. First GREEN attempt 17/18 exposed a test fixture treating message.receive's array as an object, corrected without weakening assertions. Existing required tests first 53/55 and full harness first 222/223 failed only outdated manifest/schema/prompt expectations introduced by this slice; original parent reruns required 55/55 and bundle 22/22 passed. A further generated staffing prompt RED 21/22 proved the stale task-only instruction before updating it. Final gates: request-tools 18/18, required command 55/55, full harness 223/223, bus 1477/1477, business 305/305, sandbox import closure 1/1, all zero failures/skips/cancellations. Full logs, baseline extraction and red evidence: /mnt/storage/scratch/tmp/JHTDYm/; every run used that TMPDIR with pipefail. Local commit only; no push, no role JSON/contract/migration/schema/config/credential changes, no runtime/bootstrap execution, no continuation/PRD/advance/recovery work. Controller owns review, integration and queue/tracker coordination; linked-worktree queue command refused as designed. +2026-11-04T04:06:34Z | GPT-6.1 Sol implementation sub-agent, t3/lifecycle-task-5a-create-recovery | #797 / Gitea #1839 lifecycle Task 5a before/after | Before: 6dd4f92b0 rejects operation keys and cannot recover a lost tracker create response. After: optional op reserves an immutable action.allowed tracker.create.prepared event, complete POST body/digest, target board and original PM/bot identity before any create-path network I/O; optional project_key and requirement_snapshot are opaque evidence and visible marker fields. Reusing an op cannot POST again, even after a complete negative scan or a restart before POST. Sync-only id-keyed full scans include closed tasks and require one marker, unchanged create fields, original board and creator; edited/missing/duplicate markers and partial/failed scans retain a deduplicated action.refused tracker.create.wait reconciliation receipt. Host-only recordRecoveredCreate validates the prepared receipt, body, creator and board bucket, atomically records the missing self snapshot/task.created once with original attribution, and has no socket route. Adapter exposes reconcileCreates and inspectProject, startup recovers; live PM retries resume labels/relations through read-before-write and current authority checks, assignment uses the existing separate verb. Legacy calls without new fields preserve their existing behavior; lifecycle callers must reuse a stable op for safe retries. Scope-only calls receive a generated op recorded in evidence. No migration, schema-pin edits, role/contract changes or runtime.mjs edits; packages/tasks/requirements.json and its consistency test unchanged. RED: new recovery suite 0/16 on parent; final regressions 0/21 on parent, expected invalid-request for op. Authority and forged-bucket checks also observed RED before their fixes. Corrections: event kind/subject guards required existing action event kinds instead of inventing task/ tracker kinds; restart fixtures needed historical launch cleanup before a successor claim; initial npm workspace invocation refused before tests and was replaced with per-package npm test. First full bus 1473/1477 exposed the added broker import outside the sandbox source closure, including three managed-launch failures; moved pure identity helpers into already-mounted broker.mjs, with parent project-runtime 3/3 and model-hire-ownership 18/18, then focused 42/42. Second bus 1476/1477 encountered unchanged row 280 sshd fixture cleanup ENOTEMPTY and hung; terminated only that run's owned host-run test and sshd, then waited for completion. Parent host-run was 25/26 with T27 failure (row 280 passed); local isolated host-run 26/26. Final GREEN: recovery 21/21, required verbs/adapter/tree/requirements 61/61, full tasks npm test 121/121, full bus npm test 1477/1477, zero failures/skips, diff check clean. Full logs and parent archive: /mnt/storage/scratch/tmp/CkjqAm/ (gate-recovery.log, gate-required.log, gate-tasks.log, gate-bus.log plus diagnostic history). Queue command refused this linked worktree, so no queue mutation; lead owns integration/queue. Task 5b catalogue validation, requirementsFor, built-in binding, task scope filtering and push product binding remain deferred as directed, pending Task 4. One local commit only; no push or live tracker/network/instance changes. diff --git a/docs/SESSIONS.md b/docs/SESSIONS.md index 861b92fde..7fe422443 100644 --- a/docs/SESSIONS.md +++ b/docs/SESSIONS.md @@ -1999,3 +1999,5 @@ are never rewritten or removed; corrections are new entries. 2026-11-04 | Lead (Claude Opus 5.5, T3 c5c1e8ba) | #797 lifecycle Task 2 review, fix round and landing | R1 needs fixes, fix 883003041, R2 left I1/I2/N1 residuals as a pre-.53 hardening follow-up; landed on refactor 2026-11-03 | GPT-6.1 Terra implementation sub-agent, t3/lifecycle-task-7a-artifact-reader | #797 / Gitea #1839 lifecycle Task 7a | Bounded committed-artifact reader and real Git regressions; artifact 9/9, seat 232/232, zero skips; logs /mnt/storage/scratch/tmp/0G9g1p/; queue refused linked worktree, controller owns integration; local commit only, no push. 2026-11-04 | GPT-6.1 Sol implementation sub-agent, t3/pm-intake-tools | #797 / Gitea #1839 lifecycle Task 6a | Four backed typed intake tools, pre-task PM staff.request with original request provenance, actual pi/claude-code role manifests and production runner/MCP acceptance tests; PM duty and generated staff prose updated. CTO read manifests added while existing broker read refusal remains; intake_question absent because no backing verb. Gates request-tools 18/18, required 55/55, harness 223/223, bus 1477/1477, business 305/305, sandbox imports 1/1, zero skips; logs /mnt/storage/scratch/tmp/JHTDYm/. Parent owns review/integration/queue; local commit only, no push or authority/schema/contract/migration changes. + +2026-11-03 | GPT-6.1 Sol implementation sub-agent, t3/lifecycle-task-5a-create-recovery | #797 / Gitea #1839 lifecycle Task 5a | Optional journaled create operations, sync-only complete-scan adoption, host-only recovery with original attribution and PM follow-up resume; no migration. Recovery 21/21, required 61/61, tasks 121/121, bus 1477/1477, zero skips; diagnostics and parent reruns in /mnt/storage/scratch/tmp/CkjqAm/. Task 5b deferred, queue left to lead after linked-worktree refusal; one local commit only, no push or live changes. diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index d50c12def..67a9bca7a 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -231,6 +231,27 @@ const fail = (code) => { throw new BusError(code); }; const object = (x) => x !== null && typeof x === 'object' && !Array.isArray(x); +// Lifecycle Task 5a: keep these pure helpers here because broker.mjs is also loaded by the +// sandbox client. A new host-only import would widen its mounted source closure. +const createJson = value => { + if (Array.isArray(value)) return `[${value.map(createJson).join(',')}]`; + if (object(value)) return `{${Object.keys(value).sort().map(k => `${JSON.stringify(k)}:${createJson(value[k])}`).join(',')}}`; + return JSON.stringify(value); +}; +export const createDigest = value => sha256(createJson(value)); +export function createMarker({ op, request, project_key, requirement_snapshot }) { + return `Mosaic operation: ${op}; request: ${request}` + + (project_key === undefined ? '' : `; scope: ${project_key}`) + + (requirement_snapshot === undefined ? '' : `; snapshot: ${requirement_snapshot}`); +} +export function matchesCreate(prepared, fields, creator) { + const body = prepared.body; + return prepared.body_digest === createDigest(body) && + fields.project_id === prepared.project && creator === prepared.creator && + fields.title === body.title && fields.description === body.description && + fields.due_date === (body.due_date ?? null) && fields.priority === (body.priority ?? 0); +} + // C2-06 (design 5.2, B7): the page store keeps a session's last 20 broker fetches, 200 KB of text each. const PAGE_COUNT = 20; const PAGE_BYTES = 200 * 1024; @@ -4800,6 +4821,79 @@ export class Broker { return this.#event(s, kind, body, subject); }); } + // Host-only create journal. A prepared event is also the one-shot POST reservation. These + // methods are deliberately absent from request()/the socket vocabulary. No live cap is fabricated + // when startup recovers evidence for an ended session. + prepareTaskCreate(cap, prepared) { + const s = this.#session(cap); + this.#secretCheck(prepared); + return this.#store.transaction(() => { + this.#agent(s); + const role = this.#business(s.business).roles[s.role]; + if (role?.definition !== 'pm' || role.tracker?.botId !== prepared.creator) fail('field-writer'); + if (typeof prepared.op !== 'string' || !prepared.op || !Number.isSafeInteger(prepared.project) || prepared.project < 1 || + prepared.body_digest !== createDigest(prepared.body) || !prepared.body.description.endsWith(createMarker(prepared))) fail('invalid-request'); + const old = this.taskCreateReceipts(s.business).find(r => r.body.op === prepared.op); + if (old) { + if (old.actor_role !== s.role || createDigest(old.body) !== createDigest(prepared)) fail('task-create-conflict'); + this.#checkAuthority(s, 'task.create', { decision: prepared.decision, target: null }); + return { receipt: old, fresh: false }; + } + this.#consumeAuthority(s, 'task.create', { decision: prepared.decision, target: null }); + const id = this.#event(s, 'action.allowed', { operation: 'tracker.create.prepared', ...prepared }); + return { receipt: { id, business: s.business, actor_role: s.role, actor_run: s.run, body: prepared }, fresh: true }; + }); + } + taskCreateReceipts(business) { + this.#business(business); + return this.#store.all("SELECT id,business,actor_role,actor_run,body FROM events WHERE business=? AND kind='action.allowed' AND json_extract(body,'$.operation')='tracker.create.prepared' ORDER BY seq", business) + .map(r => { const { operation, ...body } = JSON.parse(r.body); return { ...r, body }; }); + } + #preparedCreate(receipt) { + if (!object(receipt) || !id(receipt.id)) fail('invalid-request'); + const stored = this.taskCreateReceipts(receipt.business).find(r => r.id === receipt.id); + if (!stored || createDigest(stored) !== createDigest(receipt) || stored.body.body_digest !== createDigest(stored.body.body)) fail('task-create-conflict'); + return stored; + } + taskCreateResult(receipt) { + const r = this.#preparedCreate(receipt); + return this.#store.get("SELECT subject FROM events WHERE business=? AND kind='task.created' AND json_extract(body,'$.create_receipt')=? ORDER BY seq LIMIT 1", r.business, r.id)?.subject ?? null; + } + recordCreateWait({ receipt, code }) { + return this.#store.transaction(() => { + const r = this.#preparedCreate(receipt); + if (code !== 'task-create-uncertain') fail('invalid-request'); + const last = this.#store.get("SELECT body FROM events WHERE business=? AND kind='action.refused' AND json_extract(body,'$.operation')='tracker.create.wait' AND json_extract(body,'$.receipt')=? ORDER BY seq DESC LIMIT 1", r.business, r.id); + if (last) return; + this.#event({ business: r.business, role: r.actor_role, run: r.actor_run }, 'action.refused', { + operation: 'tracker.create.wait', receipt: r.id, op: r.body.op, request: r.body.request, code, wait: 'reconciliation', + }); + }); + } + recordRecoveredCreate({ receipt, task_ref, snapshot }) { + this.#secretCheck({ receipt, task_ref, snapshot }); + return this.#store.transaction(() => { + const r = this.#preparedCreate(receipt), p = r.body; + if (!taskRef(task_ref) || !task_ref.startsWith(`vikunja:${p.project}/`) || snapshot?.task_ref !== task_ref || + !object(snapshot.fields) || !Object.values(p.view.ids).includes(snapshot.fields.bucket) || + snapshot.digest !== createDigest(snapshot.fields) || !matchesCreate(p, snapshot.fields, snapshot.creator)) fail('task-create-conflict'); + const old = this.taskCreateResult(r); + if (old) { + if (old !== task_ref) fail('task-create-conflict'); + return { task_ref: old }; + } + const s = { business: r.business, role: r.actor_role, run: r.actor_run }; + const { creator, ...kept } = snapshot; + this.#snapshot(s, kept); + this.#event(s, 'task.created', { + request: p.request, requirement: p.requirement, op: p.op, create_receipt: r.id, create_body_digest: p.body_digest, + ...(p.project_key === undefined ? {} : { project_key: p.project_key }), + ...(p.requirement_snapshot === undefined ? {} : { requirement_snapshot: p.requirement_snapshot }), + }, task_ref); + return { task_ref }; + }); + } + // Trusted S3 adapter only. With a cap it records a role's own completed write: the external call // already happened, so this records and does not re-authorize. Without one it records sync reads. recordTask({ cap = null, business, snapshots = [], events = [], commentCursors = [] }) { diff --git a/packages/tasks/README.md b/packages/tasks/README.md index 0e19ebc91..6bb87b15f 100644 --- a/packages/tasks/README.md +++ b/packages/tasks/README.md @@ -147,8 +147,9 @@ loads it for its `requirements` read. Neither starts on a bad file. A write that answers 5xx or doesn't answer is never retried. The adapter reads the task again. If the write took effect, it continues. If not, or the re-read fails too, the verb refuses with `write-uncertain`. A -create can't be checked that way, so an uncertain create is reported as -`write-uncertain` and a person checks the board. +legacy create without an operation key can't be checked that way, so it reports +`write-uncertain` and a person checks the board. Creates with `op` use the +durable recovery protocol below. On success the self snapshot holds the fields the verb set, not what the final read saw. If a person edits the task between the last write and @@ -170,7 +171,7 @@ them, so it isn't passed on: | Verb | Who | Arguments | Event | |---|---|---|---| -| `task.create` | pm | `title`, `request`, `requirement`; optional `project`, `description`, `due_date`, `priority`, `labels`, `relations` | `task.created` | +| `task.create` | pm | `title`, `request`, `requirement`; optional `project`, `description`, `due_date`, `priority`, `labels`, `relations`, `op`, `project_key`, `requirement_snapshot` | `task.created` | | `task.assign` | pm | `task_ref`, `role` | `task.assigned` | | `task.reassign` | pm | `task_ref`, `role` | `task.assigned` | | `task.schedule` | pm | `task_ref`; one or more of `due_date`, `labels.{add,remove}`, `relations.{add,remove}` | snapshot only | @@ -235,6 +236,54 @@ Details that aren't obvious from the table: Verbs for one business run one at a time, in the same queue as the poll, so a poll never reads between two writes of a verb. +## Recoverable creates (lifecycle Task 5a) + +A lifecycle caller supplies a stable `op` with `task.create` and reuses it for +all retries of that exact request and body. Existing calls without the new +fields keep their previous behavior. Optional `project_key` and +`requirement_snapshot` are opaque strings carried into evidence; Task 5b will +validate their catalogue binding. If either is supplied without `op`, the +adapter generates a key, recorded in `task.created` evidence. Supply your own +key to retry safely after a lost response. + +The adapter adds this visible suffix to the description, omitting absent fields: + +```text +Mosaic operation: ; request: ; scope: ; snapshot: +``` + +Before any create-path network I/O, the broker atomically checks authority, +consumes any approval once and appends an `action.allowed` event with +`operation: tracker.create.prepared`. It stores the original role/run, bot, +request, requirement, complete POST body and SHA-256 digest, target project and +installed board, labels and relations. The prepared event is a one-shot POST +reservation. A repeated `op` with a changed body or target refuses with +`task-create-conflict`. Even a restart before the POST leaves the operation +reserved; a negative scan cannot prove that no create happened. + +An uncertain answer records an `action.refused` reconciliation wait with +`operation: tracker.create.wait` and code `task-create-uncertain`. Recovery walks +the target project's full task list by increasing ID, including completed tasks. +Only one matching marker, the original creator bot, unchanged create fields and +the original installed board permit adoption. Missing, edited, duplicate or +foreign markers, changed fields and incomplete/failed scans stay uncertain. +There is no replacement POST. The wait is deduplicated; a later `task.created` +receipt resolves it without rewriting history. + +The adapter's host-only `reconcileCreates(business)` runs at startup and can be +called for recovery. It only reads the tracker with sync authority. The +host-only `inspectProject(business, project)` exposes the same installation +checks startup uses. `Broker.recordRecoveredCreate({receipt, task_ref, +snapshot})` verifies the stored preparation, body digest, target and creator, +then atomically writes the missing self snapshot and `task.created` event once +under the original role/run. These are host methods, never socket verbs. + +A live PM retry resumes labels and relations using the existing compare-read +and uncertain-write checks, rechecking current authority before effects. Startup +never writes these as sync. Assignment remains the existing separate +`task.assign` operation on the adopted reference. A closed task can be adopted; +recovery does not reopen it or add follow-up writes to it. + ## Sync Every `pollSeconds` the adapter reads the whole board, then the task list @@ -385,6 +434,8 @@ on a live read with the sync key, and a switch is a follow-up row. | `tracker-invalid` | Vikunja answered 400 or 422 | | `tracker-rate-limited` | 429 | | `tracker-unavailable` | 5xx or no answer on a read | +| `task-create-uncertain` | reserved operation requires reconciliation; never authorizes another POST | +| `task-create-conflict` | an operation key or recovery receipt has a different identity, body or target | | `write-uncertain` | 5xx or no answer on a write, not settled by a re-read | | `tracker-shape`, `tracker-paging`, `tracker-unexpected` | Vikunja sent something the probes didn't show | | `task-moved`, `task-placement`, `task-project` | the task is in another project, or on no bucket | diff --git a/packages/tasks/src/adapter.mjs b/packages/tasks/src/adapter.mjs index 101c5df3e..75b8da824 100644 --- a/packages/tasks/src/adapter.mjs +++ b/packages/tasks/src/adapter.mjs @@ -1,7 +1,8 @@ import { lstatSync } from 'node:fs'; import { BusError } from '../../bus/src/broker.mjs'; import { client, apiRoot } from './vikunja.mjs'; -import { startup } from './startup.mjs'; +import { reconcileCreate } from './create-recovery.mjs'; +import { startup, inspect } from './startup.mjs'; import { tick, reconcile } from './sync.mjs'; import { verbs } from './verbs.mjs'; // S3's adapter for startBroker({tasks}). `trackers` is the host's plain-data boot config: @@ -146,10 +147,17 @@ export function tasksAdapter({ if (stamp(f.file) !== f.stamp) tell('credential.changed', `${ctx.business}/${instance}`, null); if (events.length) broker.recordTask({ business: ctx.business, events }); } + async function reconcileCreates(ctx) { + const results = []; + for (const receipt of broker.taskCreateReceipts(ctx.business)) + results.push(await reconcileCreate({ ctx, receipt })); + return results; + } async function start(ctx) { try { await startup(ctx); await reconcile(ctx); + await reconcileCreates(ctx); ctx.state = 'ready'; ctx.refused = null; if (ctx.untested) log(`tasks ${ctx.business} startup untested-version`); @@ -174,6 +182,9 @@ export function tasksAdapter({ if (ctx.state !== 'ready') throw new BusError(ctx.state === 'starting' ? 'tracker-starting' : ctx.refused); return serial(ctx, () => ctx.handle(cap, verb, args)); }, + // Trusted host surface only, never a socket verb. Startup recovery only reads the tracker. + reconcileCreates: (business) => serial(contexts.get(business), () => reconcileCreates(contexts.get(business))), + inspectProject: (business, project) => serial(contexts.get(business), () => inspect(contexts.get(business), project)), // For tests and the host's status verb. Each returns once the operation has run. start: (business) => serial(contexts.get(business), () => start(contexts.get(business))), tick: (business) => serial(contexts.get(business), () => tick(contexts.get(business))), diff --git a/packages/tasks/src/create-recovery.mjs b/packages/tasks/src/create-recovery.mjs new file mode 100644 index 000000000..9b21a00e8 --- /dev/null +++ b/packages/tasks/src/create-recovery.mjs @@ -0,0 +1,43 @@ +import { board, walk, normal } from './sync.mjs'; +import { checkTask, ref, taskFields, digest } from './digest.mjs'; +import { outcome } from './vikunja.mjs'; +import { createMarker, matchesCreate, createDigest } from '../../bus/src/broker.mjs'; + +// Host/adapter only. All tracker I/O here uses the read-only sync authority. A complete negative +// scan is evidence of uncertainty, never permission to issue a replacement create. +export async function reconcileCreate({ ctx, receipt }) { + const p = receipt.body; + const wait = () => { + ctx.broker.recordCreateWait({ receipt, code: 'task-create-uncertain' }); + return { state: 'uncertain', code: 'task-create-uncertain' }; + }; + const committed = ctx.broker.taskCreateResult(receipt); + if (committed) return { state: 'committed', task_ref: committed }; + try { + const view = ctx.tree?.get(p.project); + if (!view || createDigest(view) !== createDigest(p.view)) return wait(); + // An id-keyed complete walk includes done tasks and cannot accept a truncated page set. + const tasks = await walk(ctx, '', {}, p.project); + const marker = createMarker(p); + // Count all occurrences of this operation, including edited suffixes and foreign creators. + // A duplicate with changed fields must not make an otherwise valid candidate unique. + const operation = `Mosaic operation: ${p.op}`; + const hits = tasks.filter(t => (t.description ?? '').split('\n').some(line => line === operation || line.startsWith(operation + ';'))); + if (hits.length !== 1 || !hits[0].description.endsWith(marker)) return wait(); + const r = await ctx.sync('GET', `/tasks/${hits[0].id}`); + if (outcome(r) !== 'ok') return wait(); + const task = r.json; + checkTask(task); + if (task.id !== hits[0].id || task.project_id !== p.project) return wait(); + const bucket = task.done ? view.done : (await board(ctx, p.project, view)).get(task.id)?.bucket; + const fields = taskFields(task, bucket); + if (!matchesCreate(p, fields, task.created_by?.id)) return wait(); + const task_ref = ref(p.project, task.id); + ctx.broker.recordRecoveredCreate({ receipt, task_ref, snapshot: { + task_ref, updated: normal(task.updated), etag: r.etag ?? null, digest: digest(fields), fields, creator: task.created_by.id, + } }); + return { state: 'committed', task_ref }; + } catch { + return wait(); + } +} diff --git a/packages/tasks/src/startup.mjs b/packages/tasks/src/startup.mjs index 8eaeccb00..731e71bca 100644 --- a/packages/tasks/src/startup.mjs +++ b/packages/tasks/src/startup.mjs @@ -64,7 +64,7 @@ export async function startup(ctx) { } // Install checks: one manual kanban view, the five buckets once each, done and default wired. // Returns the view, or refuses tracker-install. A failed read refuses tracker-unavailable. -async function inspect(ctx, project) { +export async function inspect(ctx, project) { const views = await all(ctx.syncCall, null, `/projects/${project}/views`, {}, ctx.per); if (!views.ok) refuse('tracker-unavailable'); const kanban = views.items.filter((x) => x?.view_kind === 'kanban'); diff --git a/packages/tasks/src/verbs.mjs b/packages/tasks/src/verbs.mjs index fc3191535..8d1ac10d4 100644 --- a/packages/tasks/src/verbs.mjs +++ b/packages/tasks/src/verbs.mjs @@ -1,4 +1,6 @@ -import { BusError } from '../../bus/src/broker.mjs'; +import { randomUUID } from 'node:crypto'; +import { reconcileCreate } from './create-recovery.mjs'; +import { BusError, createMarker, createDigest } from '../../bus/src/broker.mjs'; import { outcome, refusal } from './vikunja.mjs'; import { ref, parseRef, checkTask, taskFields, digest } from './digest.mjs'; import { board, normal } from './sync.mjs'; @@ -200,7 +202,7 @@ export function verbs(ctx) { if (args?.expect !== undefined && !/^[0-9a-f]{64}$/.test(args.expect)) fail('invalid-request'); switch (verb) { case 'task.create': { - keys(args, ['title', 'description', 'due_date', 'priority', 'labels', 'relations', 'request', 'requirement', 'decision', 'project'], ['title', 'request', 'requirement']); + keys(args, ['title', 'description', 'due_date', 'priority', 'labels', 'relations', 'request', 'requirement', 'decision', 'project', 'op', 'project_key', 'requirement_snapshot'], ['title', 'request', 'requirement']); // Mos's ruling: an optional project, which must be an installed project of the tree. Without // it the task goes to the root. const project = args.project === undefined ? ctx.project : args.project; @@ -221,6 +223,17 @@ export function verbs(ctx) { plan.requirement = text(args.requirement, 64); if (!/^REQ-[A-Z]+-[1-9][0-9]*$/.test(plan.requirement)) fail('invalid-request'); if (!requirements.has(plan.requirement)) fail('unknown-requirement'); + for (const field of ['op', 'project_key', 'requirement_snapshot']) + if (args[field] !== undefined) { + plan[field] = text(args[field], 256, 1); + if (/[;\r\n]/.test(plan[field])) fail('invalid-request'); + } + // Existing unscoped calls retain their established body and evidence. The new lifecycle + // opts into recovery with an op; scope-only callers get a generated one. + if (plan.op !== undefined || plan.project_key !== undefined || plan.requirement_snapshot !== undefined) { + plan.op ??= randomUUID(); + plan.body.description = (body.description ? body.description + '\n\n' : '') + createMarker(plan); + } return plan; } case 'task.assign': @@ -386,8 +399,67 @@ export function verbs(ctx) { fail('unknown-verb'); } } + async function recoverableCreate(cap, me, plan) { + // Reservation is durable before even a relation's compare-read. A kill anywhere beyond this + // point can only reconcile, including a kill before the POST actually left the process. + const prepared = { + op: plan.op, request: plan.request, requirement: plan.requirement, decision: plan.decision, + project: plan.at.project, view: plan.at.view, creator: ctx.roles[me.role].botId, + body: plan.body, body_digest: createDigest(plan.body), labels: plan.labels, relations: plan.relations, + ...(plan.project_key === undefined ? {} : { project_key: plan.project_key }), + ...(plan.requirement_snapshot === undefined ? {} : { requirement_snapshot: plan.requirement_snapshot }), + }; + const { receipt, fresh } = ctx.broker.prepareTaskCreate(cap, prepared); + let taskRef; + if (fresh) { + await located(plan.others); + ctx.broker.prepareTaskCreate(cap, prepared); + const r = await ctx.as(me, 'POST', `/projects/${plan.at.project}/tasks`, { body: plan.body }); + if (outcome(r) !== 'ok' || ![200, 201].includes(r.status)) { + ctx.broker.recordCreateWait({ receipt, code: 'task-create-uncertain' }); + fail('task-create-uncertain'); + } + try { + checkTask(r.json); + taskRef = ref(plan.at.project, r.json.id); + const fields = taskFields(r.json, plan.at.view.todo); + ctx.broker.recordRecoveredCreate({ receipt, task_ref: taskRef, snapshot: { + task_ref: taskRef, fields, digest: digest(fields), updated: normal(r.json.updated), etag: null, creator: r.json.created_by?.id, + } }); + } catch { + ctx.broker.recordCreateWait({ receipt, code: 'task-create-uncertain' }); + fail('task-create-uncertain'); + } + } else { + const result = await reconcileCreate({ ctx, receipt }); + if (result.state !== 'committed') fail(result.code); + taskRef = result.task_ref; + } + const at = target(taskRef); + let after = await read(at); + // The read side may recover a closed task; only an acting PM can finish writes on an open one. + if (!after.task.done) { + await located(plan.others); + for (const l of plan.labels) if (!after.fields.labels.includes(l)) { + ctx.broker.prepareTaskCreate(cap, prepared); + await step(me, 'POST', `/tasks/${at.id}/labels`, { label_id: l }, at, a => a.fields.labels.includes(l)); + after = await read(at); + } + for (const x of plan.relations) if (!relatedTo(after.task, x.kind, x.other)) { + ctx.broker.prepareTaskCreate(cap, prepared); + await step(me, 'POST', `/tasks/${at.id}/relations`, { other_task_id: x.other, relation_kind: x.kind }, at, + a => relatedTo(a.task, x.kind, x.other)); + after = await read(at); + } + } + ctx.broker.recordTask({ cap, business: ctx.business, snapshots: [{ + task_ref: taskRef, fields: after.fields, digest: after.digest, updated: after.updated, etag: after.etag, + }] }); + return { task_ref: taskRef, digest: after.digest, updated: after.updated }; + } async function create(cap, me, plan) { if (!ctx.broker.taskView(ctx.business, 'input', plan.request)) fail('request-not-found'); + if (plan.op !== undefined) return recoverableCreate(cap, me, plan); await located(plan.others); ctx.broker.authorize(cap, 'task.create', { decision: plan.decision, target: null }); // No re-read can find a create whose answer was lost, so an uncertain create is reported as such. diff --git a/packages/tasks/tests/create-recovery.test.mjs b/packages/tasks/tests/create-recovery.test.mjs new file mode 100644 index 000000000..f5631050d --- /dev/null +++ b/packages/tasks/tests/create-recovery.test.mjs @@ -0,0 +1,285 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { world } from './world.mjs'; +import { FakeVikunja } from '../src/fake.mjs'; +import { digest } from '../src/digest.mjs'; + +const journal = (w, operation) => w.events(operation.endsWith('.wait') ? 'action.refused' : 'action.allowed').filter(e => e.body.operation === operation); +const resumePm = w => { + w.broker.recoverLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi' }); + const cap = w.broker.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-resumed', harness: 'pi' }); + w.broker.request(cap, { verb: 'role.claim' }); + return (w.caps.pm = cap); +}; +const uncertain = e => e.code === 'task-create-uncertain'; +async function fixture(t, edit = {}) { + const w = await world(t, edit); + await w.adapter.start('demo'); + const pm = w.agent('pm'); + const args = { title: 'recover me', description: 'original body', request: w.instruction(), requirement: 'REQ-TASK-1', op: 'create-5a' }; + const create = (extra = {}, cap = pm) => w.call(cap, 'task.create', { ...args, ...extra }); + const posts = () => w.fake.requests.filter(r => r.method === 'POST' && /^\/projects\/\d+\/tasks$/.test(r.path)).length; + return { w, pm, args, create, posts }; +} + +test('lost response after tracker create is adopted once with original attribution and optional scope', async t => { + const { w, args, create, posts } = await fixture(t); + w.fake.fault('POST', /^\/projects\/\d+\/tasks$/, 0, { apply: true }); + const scope = { project_key: 'demo/product', requirement_snapshot: 'snapshot-v7' }; + await assert.rejects(create(scope), uncertain); + assert.equal(posts(), 1); + assert.equal(w.events('task.created').length, 0); + assert.equal(w.fake.task(1).description, `original body\n\nMosaic operation: create-5a; request: ${args.request}; scope: demo/product; snapshot: snapshot-v7`); + await w.stop(); + await w.restart(); + const [created] = w.events('task.created'); + assert.equal(created.subject, `vikunja:${w.project}/1`); + assert.equal(created.actor_role, 'pm'); + assert.equal(created.actor_run, 'pm-run'); + assert.equal(created.body.project_key, 'demo/product'); + assert.equal(created.body.requirement_snapshot, 'snapshot-v7'); + assert.equal(created.body.op, args.op); + await w.adapter.reconcileCreates('demo'); + const made = await create(scope, resumePm(w)); + assert.equal(made.task_ref, created.subject); + assert.equal(w.events('task.created').length, 1); + assert.equal(posts(), 1); +}); + +test('restart before create receipt survives a real Store close and reopen', async t => { + const { w, create, posts } = await fixture(t); + w.store.run("CREATE TRIGGER crash_receipt BEFORE INSERT ON events WHEN NEW.kind='task.created' BEGIN SELECT RAISE(ABORT,'crash before receipt'); END"); + await assert.rejects(create()); + assert.equal(posts(), 1); + assert.equal(journal(w, 'tracker.create.prepared').length, 1); + assert.equal(w.events('task.created').length, 0); + w.store.run('DROP TRIGGER crash_receipt'); + await w.stop(); + await w.restart(); + assert.equal(w.events('task.created').length, 1); + assert.equal(posts(), 1); +}); + +test('duplicate PM calls share one create and changed operation bodies are refused', async t => { + const { w, create, posts } = await fixture(t); + const [a, b] = await Promise.all([create(), create()]); + assert.equal(a.task_ref, b.task_ref); + assert.equal(posts(), 1); + assert.equal(w.events('task.created').length, 1); + await assert.rejects(create({ title: 'changed' }), e => e.code === 'task-create-conflict'); + await assert.rejects(create({ labels: [w.label] }), e => e.code === 'task-create-conflict'); + assert.equal(posts(), 1); +}); + +test('label and relation follow-up failures resume by read-before-write without another create', async t => { + const { w, create, posts } = await fixture(t); + const other = await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'related' }); + assert.equal(other.status, 201); + const extras = { labels: [w.label], relations: [{ kind: 'blocking', task_ref: `vikunja:${w.project}/1` }] }; + w.fake.fault('POST', /^\/tasks\/2\/labels$/, 500); + await assert.rejects(create(extras)); + assert.equal(w.events('task.created').length, 1); + await w.stop(); + await w.restart(); + w.fake.fault('POST', /^\/tasks\/2\/relations$/, 500); + const pm = resumePm(w); + await assert.rejects(create(extras, pm)); + const labels = () => w.fake.requests.filter(r => r.method === 'POST' && r.path === '/tasks/2/labels').length; + assert.equal(labels(), 2); + const r = await create(extras, pm); + assert.equal(r.task_ref, `vikunja:${w.project}/2`); + assert.equal(labels(), 2); + assert.deepEqual(w.fake.task(2).relations, [{ kind: 'blocking', other: 1 }]); + assert.equal(posts(), 2); // one UI task and one Mosaic task + assert.equal(w.events('task.created').length, 1); +}); + +test('assignment failure after recovery can resume on the adopted task', async t => { + const { w, create, posts } = await fixture(t); + w.fake.fault('POST', /^\/projects\/\d+\/tasks$/, 502, { apply: true }); + await assert.rejects(create(), uncertain); + await w.adapter.reconcileCreates('demo'); + const { task_ref } = await create(); + w.fake.fault('POST', /^\/tasks\/1\/assignees$/, 500); + await assert.rejects(w.call(w.caps.pm, 'task.assign', { task_ref, role: 'coder' })); + await w.stop(); + await w.restart(); + const same = await create({}, resumePm(w)); + await w.call(w.caps.pm, 'task.assign', { task_ref: same.task_ref, role: 'coder' }); + assert.deepEqual(w.fake.task(1).assignees, [w.bots.coder]); + assert.equal(posts(), 1); +}); + +test('closed matching task is recovered across all pages including completed tasks', async t => { + const { w, create, posts } = await fixture(t); + for (let i = 0; i < 51; i++) await w.ui('POST', `/projects/${w.project}/tasks`, { title: `padding ${i}` }); + w.fake.fault('POST', /^\/projects\/\d+\/tasks$/, 0, { apply: true }); + await assert.rejects(create(), uncertain); + await w.ui('PATCH', '/tasks/52', { done: true }); + await w.stop(); + await w.restart(); + assert.equal(w.events('task.created')[0].subject, `vikunja:${w.project}/52`); + assert.equal(w.snapshots(`vikunja:${w.project}/52`).at(-1).fields.done, true); + assert.equal(posts(), 52); +}); + +for (const change of ['edited marker', 'changed body', 'wrong creator', 'moved board', 'duplicate marker', 'missing marker', 'complete scan failure', 'partial scan failure']) { + test(`${change} stays uncertain with a durable reconciliation wait and never posts again`, async t => { + const { w, create, posts } = await fixture(t); + w.fake.fault('POST', /^\/projects\/\d+\/tasks$/, 0, { apply: true }); + await assert.rejects(create(), uncertain); + if (change === 'edited marker') await w.ui('PATCH', '/tasks/1', { description: w.fake.task(1).description.replace('create-5a', 'edited') }); + if (change === 'changed body') await w.ui('PATCH', '/tasks/1', { title: 'edited title' }); + if (change === 'wrong creator') w.fake.task(1).by = w.owner; + if (change === 'moved board') { + const p = w.fake.project('Elsewhere', w.owner); + w.fake.moveToProject(1, p); + } + if (change === 'duplicate marker') await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'recover me', description: w.fake.task(1).description }); + if (change === 'missing marker') await w.ui('PATCH', '/tasks/1', { description: 'removed' }); + if (change === 'partial scan failure') { + for (let i = 0; i < 50; i++) await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'padding' }); + } + const before = posts(); + if (change.includes('scan failure')) { + // Recovery's scan is id-keyed. Fail every scan, including after the first page. + if (change === 'partial scan failure') { + const use = w.credentials.use.bind(w.credentials); + w.credentials.use = (instance, service, fn) => use(instance, service, async token => { + const result = await fn(token); + if (instance === 'demo/@sync' && result.json?.items?.length === 50 && result.json.items[0]?.title === 'recover me') + w.fake.fault('GET', /^\/projects\/\d+\/tasks$/, 500); + return result; + }); + } else w.fake.fault('GET', /^\/projects\/\d+\/tasks$/, 500); + } + const results = await w.adapter.reconcileCreates('demo'); + assert.equal(results[0].state, 'uncertain'); + if (change === 'complete scan failure') w.fake.fault('GET', /^\/projects\/\d+\/tasks$/, 500); + await assert.rejects(create(), uncertain); + assert.equal(w.events('task.created').length, 0); + assert.equal(posts(), before); + const waits = journal(w, 'tracker.create.wait'); + assert.ok(waits.length > 0); + assert.equal(waits.at(-1).body.code, 'task-create-uncertain'); + }); +} + +test('negative scan after a prepared create never authorizes a second POST', async t => { + const { w, create, posts } = await fixture(t); + w.fake.fault('POST', /^\/projects\/\d+\/tasks$/, 0); + await assert.rejects(create(), uncertain); + await w.stop(); + await w.restart(); + await assert.rejects(create({}, resumePm(w)), uncertain); + assert.equal(posts(), 1); + assert.equal(w.fake.task(1), undefined); +}); + +test('host recovery verifies prepared identity body and target and has no socket verb', async t => { + const { w, pm, create } = await fixture(t); + const made = await create(); + const [receipt] = w.broker.taskCreateReceipts('demo'); + const snapshot = { ...w.snapshots(made.task_ref).at(-1), creator: w.bots.pm }; + const kept = Object.fromEntries(['task_ref', 'updated', 'etag', 'digest', 'fields', 'creator'].map(k => [k, snapshot[k]])); + const recover = (r = receipt, s = kept, task_ref = made.task_ref) => w.broker.recordRecoveredCreate({ receipt: r, task_ref, snapshot: s }); + assert.doesNotThrow(() => recover()); + assert.equal(w.events('task.created').length, 1); + assert.throws(() => recover({ ...receipt, actor_run: 'invented' })); + assert.throws(() => recover(receipt, { ...kept, creator: w.owner })); + assert.throws(() => recover(receipt, { ...kept, fields: { ...kept.fields, title: 'changed' } })); + assert.throws(() => recover(receipt, kept, 'vikunja:999/1')); + for (const change of [{ title: 'changed' }, { description: 'changed' }, { priority: 5 }, { due_date: '2026-12-01T00:00:00.000Z' }, { bucket: 999 }]) { + const fields = { ...kept.fields, ...change }; + assert.throws(() => recover(receipt, { ...kept, fields, digest: digest(fields) })); + } + assert.throws(() => recover({ ...receipt, body: { ...receipt.body, body_digest: '0'.repeat(64) } })); + assert.throws(() => w.broker.request(pm, { verb: 'recordRecoveredCreate', args: {} }), e => e.code === 'unknown-verb'); + const inspection = await w.adapter.inspectProject('demo', w.project); + assert.equal(inspection.id, w.installed.view); +}); + + +test('prepared body digest and target are durable before the first network read or POST', async t => { + const fake = new FakeVikunja(); + let observe = () => {}; + const { w, args, create, posts } = await fixture(t, { fake, fetch: (url, options) => { + observe(url, options); + return fake.fetch(url, options); + } }); + let reads = 0; + const other = await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'other' }); + const extras = { relations: [{ kind: 'blocking', task_ref: `vikunja:${w.project}/${other.json.id}` }] }; + observe = (url, options) => { + const [event] = journal(w, 'tracker.create.prepared'); + assert.ok(event, 'reservation precedes every create-path network effect'); + assert.equal(event.actor_run, 'pm-run'); + assert.equal(event.body.project, w.project); + assert.match(event.body.body_digest, /^[a-f0-9]{64}$/); + assert.equal(event.body.body.title, args.title); + if (options.method === 'GET') reads++; + }; + await create(extras); + assert.ok(reads > 0); + assert.equal(posts(), 2); +}); + +test('restart after reservation but before POST only reconciles and never creates', async t => { + const { w, create, posts } = await fixture(t); + const other = await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'other' }); + const extras = { relations: [{ kind: 'blocking', task_ref: `vikunja:${w.project}/${other.json.id}` }] }; + w.fake.fault('GET', /^\/tasks\/1$/, 500); + await assert.rejects(create(extras)); + assert.equal(journal(w, 'tracker.create.prepared').length, 1); + assert.equal(posts(), 1); + await w.stop(); + await w.restart(); + await assert.rejects(create(extras, resumePm(w)), uncertain); + assert.equal(posts(), 1); +}); + +test('lost follow-up responses use existing reads to avoid duplicate label and relation writes', async t => { + const { w, create, posts } = await fixture(t); + await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'other' }); + const extras = { labels: [w.label], relations: [{ kind: 'blocking', task_ref: `vikunja:${w.project}/1` }] }; + w.fake.fault('POST', /^\/tasks\/2\/labels$/, 0, { apply: true }); + w.fake.fault('POST', /^\/tasks\/2\/relations$/, 0, { apply: true }); + await create(extras); + await create(extras); + assert.equal(w.fake.requests.filter(r => r.method === 'POST' && r.path === '/tasks/2/labels').length, 1); + assert.equal(w.fake.requests.filter(r => r.method === 'POST' && r.path === '/tasks/2/relations').length, 1); + assert.equal(posts(), 2); +}); + +test('recovery works after the original PM session has ended and only reads as sync', async t => { + const fake = new FakeVikunja(); + const requests = []; + const { w, create, posts } = await fixture(t, { fake, fetch: (url, options) => { + requests.push({ url, method: options.method, authority: options.headers.Authorization }); + return fake.fetch(url, options); + } }); + fake.fault('POST', /^\/projects\/\d+\/tasks$/, 0, { apply: true }); + await assert.rejects(create(), uncertain); + w.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exited', exitCode: 1 }); + const before = requests.length; + await w.adapter.reconcileCreates('demo'); + const recovery = requests.slice(before); + assert.ok(recovery.length > 0); + assert.ok(recovery.every(r => r.method === 'GET' && r.authority === `Bearer ${w.tokens.sync}`)); + assert.equal(w.events('task.created')[0].actor_run, 'pm-run'); + assert.equal(posts(), 1); +}); + + +test('a PM retry rechecks current task.create authority before resuming follow-up writes', async t => { + const { w, create } = await fixture(t); + const extras = { labels: [w.label] }; + w.fake.fault('POST', /^\/tasks\/1\/labels$/, 500); + await assert.rejects(create(extras)); + w.businesses.demo.roles.pm.authority.withinRole = w.businesses.demo.roles.pm.authority.withinRole.filter(a => a !== 'task.create'); + w.broker.reloadBusiness('demo', w.businesses.demo); + const writes = () => w.fake.requests.filter(r => r.method === 'POST').length; + const before = writes(); + await assert.rejects(create(extras), e => e.code === 'decision-required'); + assert.equal(writes(), before); +});