From 4e55452ba719d4827b1d941c968744118399179e Mon Sep 17 00:00:00 2001 From: dev-driver Date: Tue, 3 Nov 2026 13:28:36 +0000 Subject: [PATCH 1/6] test(mission-control): an opt-in decision revision on the settling verbs Tests only (#2446). A client reads an ask's decisionRevision from the single-row queue read, and an answer, resolve, approve or deny sent with it acts only on that decision: - an unchanged revision acts and returns the saved revision and row; - a stale one (changed question text under the same ids, another person's partial answer, a re-park) conflicts with nothing written; - of two concurrent answers on one revision, one commits and one conflicts; - a retry of the same committed operation is reconciled from its record, and another verb or payload under the same operation id is not; - a revision on a verb that doesn't settle a decision is refused; - a summary, a wake or a claim leaves the revision alone; - the route passes the revision to the write contract, and a registered-host forward carries it to the origin. A caller that sends no revision keeps today's behaviour and result fields. --- .../daemon/test/decision-revision.test.ts | 186 ++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 packages/daemon/test/decision-revision.test.ts diff --git a/packages/daemon/test/decision-revision.test.ts b/packages/daemon/test/decision-revision.test.ts new file mode 100644 index 0000000000..04a31a05fb --- /dev/null +++ b/packages/daemon/test/decision-revision.test.ts @@ -0,0 +1,186 @@ +// #2446: an opt-in decision revision. A client reads the decision with its revision, and a settling act sent with that +// revision runs only on that decision. Real SQLite (all migrations), the real queue, the Mission Control route and write +// contract, and the queue's single-row read; the route-wiring cells stub the contract and the remote fetch. Nothing native. +import { afterEach, describe, expect, it } from "vitest"; +import Database from "better-sqlite3"; +import { Hono } from "hono"; +import { migrate } from "../src/db/migrate.js"; +import { ALL_MIGRATIONS } from "../src/db/all-migrations.js"; +import { EventBus } from "../src/domain/event-bus.js"; +import { QueueRepository } from "../src/domain/queue-repository.js"; +import { MissionControlWriteContract } from "../src/domain/mission-control/mission-control-write-contract.js"; +import { MissionControlActionLog } from "../src/domain/mission-control/mission-control-action-log.js"; +import { missionControlRoutes } from "../src/routes/mission-control.js"; +import { queueRoutes } from "../src/routes/queue.js"; +import type { HostRegistry } from "../src/domain/hosts/hosts-registry-reader.js"; + +const PERSON = "human-test@external", OTHER = "human-other@external", ASKER = "requester@test-rig"; +const person = (entityId: string, address: string) => ({ entityId, class: "human" as const, displayName: entityId, address, + connectorBindings: [], prefs: { deliveryClass: "A" as const } }); +const registry = { ok: true as const, entities: [person("human-test", PERSON), person("human-other", OTHER)] }; +const QUESTIONS = [ + { id: "q1", question: "Ship it?", options: [{ id: "yes", label: "Yes" }, { id: "no", label: "No" }] }, + { id: "q2", question: "Tell the team?", options: [{ id: "yes", label: "Yes" }, { id: "no", label: "No" }] }, +]; +// The fields an existing action result has: a caller that sends no revision keeps exactly these. +const ANSWER_RESULT = ["actionId", "answer", "auditedAt", "closedQitem", "createdQitemId", "notifyAttempted", "notifyResult", "qitemId", "verb"]; +const ACT_RESULT = ANSWER_RESULT.filter(key => key !== "answer"); + +const closes: Array<() => void> = []; +afterEach(() => { for (const close of closes.splice(0)) close(); }); + +function fixture() { + const db = new Database(":memory:"); migrate(db, ALL_MIGRATIONS); const bus = new EventBus(db); + const repo = new QueueRepository(db, bus, { validateRig: () => true, loadHumanRegistry: () => registry, + transport: { send: async () => ({ ok: true, verified: true, sent: true }) } }); + const contract = new MissionControlWriteContract({ db, eventBus: bus, queueRepo: repo, actionLog: new MissionControlActionLog(db) }); + const app = new Hono(); + app.use("*", async (c, next) => { + c.set("queueRepo" as never, repo as never); c.set("eventBus" as never, bus as never); + c.set("missionControlWriteContract" as never, contract as never); await next(); + }); + app.route("/api/mission-control", missionControlRoutes()); + app.route("/api/queue", queueRoutes()); + closes.push(() => db.close()); + const ask = async () => (await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Synthetic two-question ask", + summary: "Ship?", humanIntent: "decision", humanQuestions: QUESTIONS, nudge: false })).qitemId; + const park = async () => { + const row = await repo.create({ sourceSession: ASKER, destinationSession: "owner@test-rig", body: "Work waiting on a person", nudge: false }); + repo.claim({ qitemId: row.qitemId, destinationSession: "owner@test-rig" }); + repo.update({ qitemId: row.qitemId, actorSession: "owner@test-rig", state: "blocked", blockedOn: PERSON }); + return row.qitemId; + }; + const revision = async (id: string) => (await (await app.request(`/api/queue/${id}`)).json() as { decisionRevision?: unknown }).decisionRevision; + const act = async (body: Record) => { + const response = await app.request("/api/mission-control/action", { method: "POST", + headers: { "Content-Type": "application/json", "X-OpenRig-Session": PERSON }, body: JSON.stringify({ actorSession: PERSON, ...body }) }); + return { status: response.status, body: await response.json() as Record }; + }; + const count = (sql: string, id: string) => (db.prepare(sql).get(id) as { n: number }).n; + const written = (id: string) => ({ row: repo.getById(id), transitions: repo.listTransitions(id).length, + replies: count("SELECT count(*) n FROM queue_items WHERE reply_to = ?", id), + actions: count("SELECT count(*) n FROM mission_control_actions WHERE qitem_id = ?", id) }); + return { db, repo, ask, park, revision, act, written }; +} + +describe("an opt-in decision revision on the settling verbs (#2446)", () => { + it("an unchanged revision acts and returns the saved revision and state", async () => { + const f = fixture(); const id = await f.ask(); + const read = await f.revision(id); + expect(read).toEqual(expect.stringMatching(/\S/)); + const result = await f.act({ verb: "answer", qitemId: id, questionId: "q1", optionId: "yes", expectedRevision: read }); + expect(result).toMatchObject({ status: 200, body: { answer: { outcome: "recorded" }, closedQitem: { humanAnswers: { q1: "yes" } } } }); + expect(result.body.decisionRevision).toBe(await f.revision(id)); + expect(result.body.decisionRevision).not.toBe(read); + }); + + it.each([ + ["answer", "question text changed under the same ids"], + ["approve", "another person recorded a partial answer"], + ["deny", "question text changed under the same ids"], + ["resolve", "re-parked on another person"], + ] as const)("%s on a stale revision (%s) conflicts and writes nothing", async (verb, change) => { + const f = fixture(); const id = verb === "resolve" ? await f.park() : await f.ask(); + const read = await f.revision(id); + if (change === "question text changed under the same ids") { + f.db.prepare("UPDATE queue_items SET human_questions = ? WHERE qitem_id = ?").run(JSON.stringify([{ ...QUESTIONS[0], question: "Ship it to everyone?" }, QUESTIONS[1]]), id); + } else if (change === "another person recorded a partial answer") { + expect(f.repo.recordHumanAnswer({ qitemId: id, actorSession: OTHER, questionId: "q1", optionId: "no" })).toMatchObject({ status: "recorded" }); + } else { + f.db.prepare("UPDATE queue_items SET blocked_on = ? WHERE qitem_id = ?").run(OTHER, id); + } + const before = f.written(id); + const result = await f.act({ verb, qitemId: id, expectedRevision: read, + ...(verb === "answer" ? { questionId: "q2", optionId: "yes" } : verb === "resolve" ? { decision: "Go ahead" } : { reason: "reviewed" }) }); + expect(result).toMatchObject({ status: 409, body: { error: "decision_revision_changed", currentRevision: await f.revision(id) } }); + expect(f.written(id)).toEqual(before); + }); + + it("concurrent answers on one revision: one commits, the other conflicts with nothing of its own written", async () => { + const f = fixture(); const id = await f.ask(); + const read = await f.revision(id); + const results = await Promise.all(["q1", "q2"].map(questionId => f.act({ verb: "answer", qitemId: id, questionId, optionId: "yes", expectedRevision: read }))); + expect(results.map(r => r.status).sort()).toEqual([200, 409]); + expect(results.find(r => r.status === 409)!.body.error).toBe("decision_revision_changed"); + expect(Object.keys(f.repo.getById(id)!.humanAnswers ?? {})).toHaveLength(1); + expect(f.written(id)).toMatchObject({ actions: 1, replies: 0 }); + }); + + it("a caller that sends no revision acts exactly as before, with the same result fields", async () => { + const f = fixture(); const answered = await f.ask(), approved = await f.ask(); + f.db.prepare("UPDATE queue_items SET human_questions = ? WHERE qitem_id IN (?, ?)").run(JSON.stringify([{ ...QUESTIONS[0], question: "Changed" }, QUESTIONS[1]]), answered, approved); + const answer = await f.act({ verb: "answer", qitemId: answered, questionId: "q1", optionId: "yes" }); + expect(answer.status).toBe(200); + expect(Object.keys(answer.body).sort()).toEqual(ANSWER_RESULT); + const approve = await f.act({ verb: "approve", qitemId: approved }); + expect(approve).toMatchObject({ status: 200, body: { closedQitem: { state: "done" } } }); + expect(Object.keys(approve.body).sort()).toEqual(ACT_RESULT); + }); + + it("a retry of the same committed operation is reconciled from the record; another verb or payload under its ID is not", async () => { + const f = fixture(); const id = await f.ask(); + const read = await f.revision(id); + const request = { verb: "approve", qitemId: id, reason: "reviewed", expectedRevision: read, auditNotes: { operationId: "op-1" } }; + const first = await f.act(request); + expect(first).toMatchObject({ status: 200, body: { closedQitem: { state: "done" } } }); + const retry = await f.act(request); + expect(retry).toMatchObject({ status: 200, body: { reconciled: true, actionId: first.body.actionId, decisionRevision: await f.revision(id) } }); + expect((await f.act({ ...request, verb: "deny" })).body.error).toBe("decision_revision_changed"); + expect((await f.act({ ...request, reason: "a different reason" })).body.error).toBe("decision_revision_changed"); + expect(f.written(id).actions).toBe(1); + }); + + it("a revision sent with a verb that doesn't settle a decision is refused, not silently ignored", async () => { + const f = fixture(); const id = await f.ask(); + const before = f.written(id); + expect(await f.act({ verb: "hold", qitemId: id, reason: "later", expectedRevision: await f.revision(id) })) + .toMatchObject({ status: 400, body: { error: "decision_revision_unsupported" } }); + expect(f.written(id)).toEqual(before); + }); + + it("a summary, a wake or a claim leaves the revision alone", async () => { + const f = fixture(); const id = await f.ask(); + const read = await f.revision(id); + expect(read).toEqual(expect.stringMatching(/\S/)); + f.db.prepare("UPDATE queue_items SET summary = 'Ship? (relabelled)', ts_updated = '2099-01-01T00:00:00.000Z', last_nudge_attempt = '2099-01-01T00:00:00.000Z', last_nudge_result = 'delivered', state = 'in-progress', claimed_at = '2099-01-01T00:00:00.000Z' WHERE qitem_id = ?").run(id); + expect(await f.revision(id)).toBe(read); + expect(await f.act({ verb: "approve", qitemId: id, expectedRevision: read })).toMatchObject({ status: 200, body: { closedQitem: { state: "done" } } }); + }); +}); + +describe("the route carries a client's revision to the write that compares it (#2446)", () => { + const REGISTRY: HostRegistry = { hosts: [{ id: "vps-b", transport: "http", url: "http://vps-b:7433", bearer_env: "B" }] }; + function wiring() { + const acts: Array> = [], forwarded: Array> = []; + const app = new Hono(); + app.use("*", async (c, next) => { + const set = c.set.bind(c) as (key: string, value: unknown) => void; + set("missionControlWriteContract", { act: async (input: Record) => { acts.push(input); return { actionId: "local-1" }; } }); + set("hostRegistryLoader", () => ({ ok: true, registry: REGISTRY })); + set("remoteFetchImpl", (async (_url: string, init: RequestInit) => { + forwarded.push(JSON.parse(String(init.body))); + return new Response(JSON.stringify({ actionId: "remote-1" }), { status: 200, headers: { "Content-Type": "application/json" } }); + }) as typeof fetch); + set("eventBus", { emit: () => {} }); + await next(); + }); + app.route("/api/mission-control", missionControlRoutes({ bearerToken: null })); + const post = (body: Record) => app.request("/api/mission-control/action", { method: "POST", + headers: { "Content-Type": "application/json", "X-OpenRig-Session": PERSON }, body: JSON.stringify({ actorSession: PERSON, ...body }) }); + return { acts, forwarded, post }; + } + process.env["B"] = "remote-token"; + + it("passes it to the local write contract", async () => { + const w = wiring(); + expect((await w.post({ verb: "resolve", qitemId: "qitem-1", decision: "Go", expectedRevision: "rev-1" })).status).toBe(200); + expect(w.acts[0]).toMatchObject({ expectedRevision: "rev-1" }); + }); + + it("forwards it to the origin host, which compares it there", async () => { + const w = wiring(); + expect((await w.post({ verb: "resolve", qitemId: "qitem-1", decision: "Go", expectedRevision: "rev-1", hostId: "vps-b" })).status).toBe(200); + expect(w.acts).toEqual([]); + expect(w.forwarded[0]).toMatchObject({ verb: "resolve", expectedRevision: "rev-1" }); + }); +}); From 52ca1cd288688f6cb009a15a986c407ced7c54d3 Mon Sep 17 00:00:00 2001 From: dev-driver Date: Tue, 3 Nov 2026 14:31:05 +0000 Subject: [PATCH 2/6] feat(mission-control): an opt-in decision revision on the settling verbs Refs #2446. GET /api/queue/:qitemId carries decisionRevision, an opaque hash of what a person decides on: the body, intent and questions, and the decision's state (open, blocked or terminal; who it's addressed to or parked on; its answers, a decision made on their behalf, the current park and the replies returned). A summary, a wake or a claim doesn't change it. POST /api/mission-control/action takes an optional expectedRevision on answer, resolve, approve and deny. The compare runs at the start of the act, before any branch that writes, with no await before the first write. A stale revision is 409 decision_revision_changed with nothing written, and the current revision, state and latest action. A retry of the caller's own committed operation (the same operationId in auditNotes, verb, sender and payload) is answered from its record. On other verbs it's 400 decision_revision_unsupported. Without it, every act and result is as before. A registered-host forward carries it to the origin. --- .../mission-control-write-contract.ts | 44 +++++++++++++++++++ .../daemon/src/domain/queue-repository.ts | 12 +++++ packages/daemon/src/routes/mission-control.ts | 6 +++ packages/daemon/src/routes/queue.ts | 2 + 4 files changed, 64 insertions(+) diff --git a/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts b/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts index 05d4fde3a0..15f4e738d5 100644 --- a/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts +++ b/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts @@ -39,6 +39,7 @@ // closure_target+handed_off_to=; // creates new qitem at destination (4-step canonical) +import { createHash } from "node:crypto"; import type Database from "better-sqlite3"; import type { EventBus } from "../event-bus.js"; import { isTypedGateBlocker, UNATTRIBUTED_ACTOR, type QueueRepository, type QueueItem } from "../queue-repository.js"; @@ -100,6 +101,9 @@ export interface MissionControlActionInput { * the default person or the claim; the audit notes say whose ask it was. Only a request with no transport identity * carries it. A surface that knows its person sends them without it. */ authority?: "primary-user"; + /** #2446, opt-in: the `decisionRevision` the caller read. A settling act then runs only on that decision; omitted, the + * act and its result are exactly as before. An `operationId` in `auditNotes` lets a retry be answered from the record. */ + expectedRevision?: string; /** * For handoff: opt-in best-effort wake. Default true (PL-004 Phase A R1 * pattern: durable + waking by default; operators opt out for cold queues). @@ -119,6 +123,10 @@ export interface MissionControlActionResult { auditedAt: string; /** For `answer`: recorded with questions still open, or answered (returned to the asker, and the ask closed). */ answer?: { outcome: "recorded" | "answered"; unanswered: string[]; message?: string }; + /** With expectedRevision only: the decision's revision as saved. */ + decisionRevision?: string; + /** With expectedRevision only: this retry is the caller's own committed operation, answered from its record. */ + reconciled?: true; } interface WriteContractDeps { @@ -150,6 +158,34 @@ export class MissionControlWriteContract { * notify subscribers + opt-in transport wake (handoff only). */ async act(input: MissionControlActionInput): Promise { + if (input.expectedRevision === undefined) return this.actOn(input); + if (!SETTLING_VERBS.has(input.verb)) { + throw new MissionControlWriteContractError("decision_revision_unsupported", + `verb=${input.verb} doesn't take expectedRevision; answer, resolve, approve and deny do. Nothing was written.`, { verb: input.verb }); + } + // #2446: the act runs only on the decision its caller read. No await separates this check from the act's first + // write, so no other request in this daemon changes the decision in between. + const operation = typeof input.auditNotes?.operationId === "string" ? operationFingerprint(input) : null; + const item = this.queueRepo.getById(input.qitemId); + const current = item ? this.queueRepo.decisionRevision(item) : null; + if (item && current !== input.expectedRevision) { + const last = this.actionLog.listForQitem(input.qitemId, 1)[0]; + // The caller's own committed operation retried after an uncertain result: answered from its record, nothing written. + if (operation && last?.auditNotes?.operation === operation) { + return { actionId: last.actionId, verb: input.verb, qitemId: input.qitemId, closedQitem: item, createdQitemId: null, + notifyAttempted: false, notifyResult: null, auditedAt: last.actedAt, reconciled: true, decisionRevision: current! }; + } + throw new MissionControlWriteContractError("decision_revision_changed", + `qitem ${input.qitemId} changed since this revision was read, so nothing was written. Read it again for the current decision.`, + { qitemId: input.qitemId, currentRevision: current, state: item.state, humanAnswers: item.humanAnswers ?? null, + lastAction: last ? { actionId: last.actionId, verb: last.actionVerb, actorSession: last.actorSession, actedAt: last.actedAt, auditNotes: last.auditNotes } : null }); + } + const result = await this.actOn(operation ? { ...input, auditNotes: { ...input.auditNotes, operation } } : input); + const saved = this.queueRepo.getById(input.qitemId); + return saved ? { ...result, decisionRevision: this.queueRepo.decisionRevision(saved) } : result; + } + + private async actOn(input: MissionControlActionInput): Promise { if (input.verb === "annotate") { return this.annotateOnly(input); } @@ -720,6 +756,14 @@ interface ClosureMapping { blockedOn?: string; } +/** One logical operation (#2446): the caller's operationId with the verb, who sent it and what it carries, so a retry + * matches only itself, never another verb or payload sent under the same ID. */ +function operationFingerprint(input: MissionControlActionInput): string { + const sent = [input.auditNotes?.operationId, input.verb, input.actorSession, input.authority ?? null, input.questionId ?? null, + input.optionId ?? null, input.decision ?? null, input.reason ?? null, input.destinationSession ?? null, input.body ?? null, input.annotation ?? null]; + return createHash("sha256").update(JSON.stringify(sent)).digest("hex").slice(0, 24); +} + function verbToClosure(input: MissionControlActionInput): ClosureMapping { switch (input.verb) { case "approve": diff --git a/packages/daemon/src/domain/queue-repository.ts b/packages/daemon/src/domain/queue-repository.ts index 19b78c7948..42d716d324 100644 --- a/packages/daemon/src/domain/queue-repository.ts +++ b/packages/daemon/src/domain/queue-repository.ts @@ -3959,6 +3959,18 @@ export class QueueRepository { } /** A park's current question: its latest human-required transition, and that notification's final post receipt, if any. */ + /** The opaque revision of what a person decides on (#2446): the question (body, intent, questions) and its decision state + * (open, blocked or terminal; who it's addressed to or parked on; its answers, a decision made for them, the current park, + * and the replies returned). A label, a wake or a claim doesn't change it; a new park does. */ + decisionRevision(item: QueueItem): string { + const phase = item.state === "blocked" ? "blocked" : item.state === "pending" || item.state === "in-progress" ? "open" : item.state; + const replies = this.hasReplyToColumn ? (this.db.prepare("SELECT qitem_id FROM queue_items WHERE reply_to = ? ORDER BY ts_created, rowid") + .all(item.qitemId) as Array<{ qitem_id: string }>).map((row) => row.qitem_id) : []; + const decision = [item.body, item.humanIntent ?? null, item.humanQuestions ?? null, phase, item.destinationSession, item.blockedOn, + item.humanAnswers ?? null, item.delegatedDecision ?? null, this.currentHumanRequiredEpisode(item.qitemId)?.key ?? null, replies]; + return `dr1-${createHash("sha256").update(JSON.stringify(decision)).digest("hex").slice(0, 24)}`; + } + currentHumanRequiredEpisode(qitemId: string): { key: string; receipt: ReturnType } | null { const episode = this.transitionLog.listForQitem(qitemId).filter((transition) => transition.ownerNotificationKind === "human-required").at(-1); if (!episode) return null; diff --git a/packages/daemon/src/routes/mission-control.ts b/packages/daemon/src/routes/mission-control.ts index 6b5d1e65ce..bc250ad173 100644 --- a/packages/daemon/src/routes/mission-control.ts +++ b/packages/daemon/src/routes/mission-control.ts @@ -247,6 +247,8 @@ export function missionControlRoutes(opts?: MissionControlRoutesOpts): Hono { : err.code === "qitem_not_leg1_parked" ? 409 : err.code === "answer_not_closed" ? 409 : err.code === "answer_not_recorded" ? 400 + : err.code === "decision_revision_changed" ? 409 + : err.code === "decision_revision_unsupported" ? 400 : 500; return c.json( { error: err.code, message: err.message, ...(err.details ?? {}) }, @@ -330,9 +332,11 @@ export function missionControlRoutes(opts?: MissionControlRoutesOpts): Hono { auditNotes?: Record; authority?: string; hostId?: string; + expectedRevision?: unknown; }>() .catch(() => ({} as never)); if (!body.verb) return c.json({ error: "verb is required" }, 400); + if (body.expectedRevision !== undefined && typeof body.expectedRevision !== "string") return c.json({ error: "expectedRevision must be a string" }, 400); if (!MISSION_CONTROL_VERBS.includes(body.verb)) { return c.json( { @@ -434,6 +438,8 @@ export function missionControlRoutes(opts?: MissionControlRoutesOpts): Hono { decision: body.decision, ...(typeof body.questionId === "string" ? { questionId: body.questionId } : {}), ...(typeof body.optionId === "string" ? { optionId: body.optionId } : {}), + // #2446: opt-in; a request without it acts exactly as before. A forward carries it to the origin, which compares it. + ...(typeof body.expectedRevision === "string" ? { expectedRevision: body.expectedRevision } : {}), notify: body.notify, auditNotes: body.auditNotes, // The person at the terminal (the TUI sends no seat header): the actor is the primary user, whoever was asked. diff --git a/packages/daemon/src/routes/queue.ts b/packages/daemon/src/routes/queue.ts index bfa20f1b3c..04c55c96f0 100644 --- a/packages/daemon/src/routes/queue.ts +++ b/packages/daemon/src/routes/queue.ts @@ -1338,6 +1338,8 @@ export function queueRoutes(): Hono { const closing = repo.closingTransition(qitemId, item.state); return c.json({ ...item, + // #2446: what a settling Mission Control act can be guarded with (expectedRevision). + decisionRevision: repo.decisionRevision(item), // A closed row's outcome (its note and reason) lives only in the transition log; omitted while the row is open. ...(closing ? { closingTransition: { transitionId: closing.transitionId, From c183c8016e7d3e031a271c17fe9e08359b1d6622 Mon Sep 17 00:00:00 2001 From: dev-driver Date: Tue, 3 Nov 2026 15:06:05 +0000 Subject: [PATCH 3/6] test(mission-control): a runner ask's typed answers race, a re-parked gate, and an unchecked revision Tests only, #2448's delta: - two concurrent typed answers to a runner's permission ask on one revision: one closes it and the other conflicts, with one audit row; - a typed-gate park resolved and parked again on the same gate conflicts with the old revision, while a note on the park leaves the revision unchanged; - a revision sent with hold isn't checked: the hold runs, and the result says so (replacing the earlier refusal). --- .../daemon/test/decision-revision.test.ts | 37 +++++++++++++++---- 1 file changed, 29 insertions(+), 8 deletions(-) diff --git a/packages/daemon/test/decision-revision.test.ts b/packages/daemon/test/decision-revision.test.ts index 04a31a05fb..72e7582cb3 100644 --- a/packages/daemon/test/decision-revision.test.ts +++ b/packages/daemon/test/decision-revision.test.ts @@ -44,12 +44,15 @@ function fixture() { closes.push(() => db.close()); const ask = async () => (await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Synthetic two-question ask", summary: "Ship?", humanIntent: "decision", humanQuestions: QUESTIONS, nudge: false })).qitemId; - const park = async () => { + const park = async (blockedOn = PERSON) => { const row = await repo.create({ sourceSession: ASKER, destinationSession: "owner@test-rig", body: "Work waiting on a person", nudge: false }); repo.claim({ qitemId: row.qitemId, destinationSession: "owner@test-rig" }); - repo.update({ qitemId: row.qitemId, actorSession: "owner@test-rig", state: "blocked", blockedOn: PERSON }); + repo.update({ qitemId: row.qitemId, actorSession: "owner@test-rig", state: "blocked", blockedOn }); return row.qitemId; }; + // A runner's permission request: the person's typed answer is recorded on the ask itself, with no reply row. + const permissionAsk = async () => (await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Allow the runner to write?", + summary: "Allow?", humanIntent: "decision", tags: ["runner-permission", "permission-node:node-1", "permission-generation:gen-1"], nudge: false })).qitemId; const revision = async (id: string) => (await (await app.request(`/api/queue/${id}`)).json() as { decisionRevision?: unknown }).decisionRevision; const act = async (body: Record) => { const response = await app.request("/api/mission-control/action", { method: "POST", @@ -60,7 +63,7 @@ function fixture() { const written = (id: string) => ({ row: repo.getById(id), transitions: repo.listTransitions(id).length, replies: count("SELECT count(*) n FROM queue_items WHERE reply_to = ?", id), actions: count("SELECT count(*) n FROM mission_control_actions WHERE qitem_id = ?", id) }); - return { db, repo, ask, park, revision, act, written }; + return { db, repo, ask, park, permissionAsk, revision, act, written }; } describe("an opt-in decision revision on the settling verbs (#2446)", () => { @@ -106,6 +109,26 @@ describe("an opt-in decision revision on the settling verbs (#2446)", () => { expect(f.written(id)).toMatchObject({ actions: 1, replies: 0 }); }); + it("concurrent typed answers to a runner's permission ask on one revision: one closes it, the other conflicts", async () => { + const f = fixture(); const id = await f.permissionAsk(); + const read = await f.revision(id); + const results = await Promise.all(["Allow it", "Deny it"].map(decision => f.act({ verb: "answer", qitemId: id, decision, expectedRevision: read }))); + expect(results.map(r => r.status).sort()).toEqual([200, 409]); + expect(results.find(r => r.status === 409)!.body.error).toBe("decision_revision_changed"); + expect(f.written(id)).toMatchObject({ row: { state: "done" }, actions: 1 }); + }); + + it("a typed-gate park resolved and parked again on the same gate is a new decision; a note on a park isn't", async () => { + const f = fixture(); const id = await f.park("external:qa-pass"); + const first = await f.revision(id); + f.repo.update({ qitemId: id, actorSession: "owner@test-rig", state: "blocked", blockedOn: "external:qa-pass", transitionNote: "still waiting" }); + expect(await f.revision(id)).toBe(first); + expect(await f.act({ verb: "resolve", qitemId: id, decision: "QA passed" })).toMatchObject({ status: 200 }); + f.repo.update({ qitemId: id, actorSession: "owner@test-rig", state: "blocked", blockedOn: "external:qa-pass" }); + expect(await f.act({ verb: "resolve", qitemId: id, decision: "QA passed", expectedRevision: first })) + .toMatchObject({ status: 409, body: { error: "decision_revision_changed" } }); + }); + it("a caller that sends no revision acts exactly as before, with the same result fields", async () => { const f = fixture(); const answered = await f.ask(), approved = await f.ask(); f.db.prepare("UPDATE queue_items SET human_questions = ? WHERE qitem_id IN (?, ?)").run(JSON.stringify([{ ...QUESTIONS[0], question: "Changed" }, QUESTIONS[1]]), answered, approved); @@ -130,12 +153,10 @@ describe("an opt-in decision revision on the settling verbs (#2446)", () => { expect(f.written(id).actions).toBe(1); }); - it("a revision sent with a verb that doesn't settle a decision is refused, not silently ignored", async () => { + it("a revision sent with a verb that doesn't settle a decision isn't checked: the act runs as today and says so", async () => { const f = fixture(); const id = await f.ask(); - const before = f.written(id); - expect(await f.act({ verb: "hold", qitemId: id, reason: "later", expectedRevision: await f.revision(id) })) - .toMatchObject({ status: 400, body: { error: "decision_revision_unsupported" } }); - expect(f.written(id)).toEqual(before); + expect(await f.act({ verb: "hold", qitemId: id, reason: "later", expectedRevision: "dr1-not-this-row" })) + .toMatchObject({ status: 200, body: { closedQitem: { state: "blocked" }, revisionNote: expect.stringContaining("isn't checked for hold") } }); }); it("a summary, a wake or a claim leaves the revision alone", async () => { From 3ab5afddee23a47c9d2d1d81445e21aa6d25e1d9 Mon Sep 17 00:00:00 2001 From: dev-driver Date: Tue, 3 Nov 2026 15:48:25 +0000 Subject: [PATCH 4/6] fix(mission-control): re-compare before a runner ask's typed close, a new park is a new revision, and an unchecked revision is noted #2448's delta (#2446): - A typed answer to a runner's permission ask writes nothing until its close, after an await, so a guarded answer compares its revision again just before that close, with no await in between. - The revision includes the transition that entered the current park, so a typed-gate park resolved and parked again on the same gate is a new revision. A note on a parked row doesn't change it. - A revision sent with a verb that doesn't settle a decision is no longer refused: the act runs as today, and the result's revisionNote says it wasn't checked. --- .../mission-control-write-contract.ts | 30 ++++++++++++++----- .../daemon/src/domain/queue-repository.ts | 10 +++++-- packages/daemon/src/routes/mission-control.ts | 1 - 3 files changed, 31 insertions(+), 10 deletions(-) diff --git a/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts b/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts index 15f4e738d5..8b48585981 100644 --- a/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts +++ b/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts @@ -127,6 +127,8 @@ export interface MissionControlActionResult { decisionRevision?: string; /** With expectedRevision only: this retry is the caller's own committed operation, answered from its record. */ reconciled?: true; + /** With expectedRevision on a verb that doesn't settle a decision: it wasn't checked, and the act ran as without it. */ + revisionNote?: string; } interface WriteContractDeps { @@ -159,12 +161,13 @@ export class MissionControlWriteContract { */ async act(input: MissionControlActionInput): Promise { if (input.expectedRevision === undefined) return this.actOn(input); + // Only the settling verbs compare it. Another verb acts as today, and says the revision wasn't checked. if (!SETTLING_VERBS.has(input.verb)) { - throw new MissionControlWriteContractError("decision_revision_unsupported", - `verb=${input.verb} doesn't take expectedRevision; answer, resolve, approve and deny do. Nothing was written.`, { verb: input.verb }); + return { ...await this.actOn(input), revisionNote: `expectedRevision isn't checked for ${input.verb}; only answer, resolve, approve and deny compare it.` }; } // #2446: the act runs only on the decision its caller read. No await separates this check from the act's first - // write, so no other request in this daemon changes the decision in between. + // write, so no other request in this daemon changes the decision in between. The one path whose first write follows + // an await, a typed answer to a runner's permission ask, compares again before it (answerAsk). const operation = typeof input.auditNotes?.operationId === "string" ? operationFingerprint(input) : null; const item = this.queueRepo.getById(input.qitemId); const current = item ? this.queueRepo.decisionRevision(item) : null; @@ -175,16 +178,22 @@ export class MissionControlWriteContract { return { actionId: last.actionId, verb: input.verb, qitemId: input.qitemId, closedQitem: item, createdQitemId: null, notifyAttempted: false, notifyResult: null, auditedAt: last.actedAt, reconciled: true, decisionRevision: current! }; } - throw new MissionControlWriteContractError("decision_revision_changed", - `qitem ${input.qitemId} changed since this revision was read, so nothing was written. Read it again for the current decision.`, - { qitemId: input.qitemId, currentRevision: current, state: item.state, humanAnswers: item.humanAnswers ?? null, - lastAction: last ? { actionId: last.actionId, verb: last.actionVerb, actorSession: last.actorSession, actedAt: last.actedAt, auditNotes: last.auditNotes } : null }); + throw this.revisionChanged(item, current!); } const result = await this.actOn(operation ? { ...input, auditNotes: { ...input.auditNotes, operation } } : input); const saved = this.queueRepo.getById(input.qitemId); return saved ? { ...result, decisionRevision: this.queueRepo.decisionRevision(saved) } : result; } + /** A guarded act's decision changed since its revision was read: nothing was written, and the caller gets what's saved. */ + private revisionChanged(item: QueueItem, current: string): MissionControlWriteContractError { + const last = this.actionLog.listForQitem(item.qitemId, 1)[0]; + return new MissionControlWriteContractError("decision_revision_changed", + `qitem ${item.qitemId} changed since this revision was read, so nothing was written. Read it again for the current decision.`, + { qitemId: item.qitemId, currentRevision: current, state: item.state, humanAnswers: item.humanAnswers ?? null, + lastAction: last ? { actionId: last.actionId, verb: last.actionVerb, actorSession: last.actorSession, actedAt: last.actedAt, auditNotes: last.auditNotes } : null }); + } + private async actOn(input: MissionControlActionInput): Promise { if (input.verb === "annotate") { return this.annotateOnly(input); @@ -660,6 +669,13 @@ export class MissionControlWriteContract { const decision = chose || resume ? formatHumanAnswers(questions, this.queueRepo.getById(input.qitemId)?.humanAnswers ?? {}).join("; ") : returned.qitemId === input.qitemId ? words : humanAnswerText(returned.body); + // #2446: typed words to a runner's permission ask have written nothing yet (no reply row, unlike a choice), and the + // await above let other requests in, so a guarded answer compares its revision again here, with no await before the close. + if (input.expectedRevision !== undefined && replyId === input.qitemId && !chose) { + const now = this.queueRepo.getById(input.qitemId); + const revision = now ? this.queueRepo.decisionRevision(now) : null; + if (now && revision !== input.expectedRevision) throw this.revisionChanged(now, revision!); + } if (!this.queueRepo.resolveDirectHumanReply({ qitemId: input.qitemId, actorSession: input.actorSession, decision, primaryUser }) && this.queueRepo.getById(input.qitemId)?.state !== "done") { throw new MissionControlWriteContractError("answer_not_closed", diff --git a/packages/daemon/src/domain/queue-repository.ts b/packages/daemon/src/domain/queue-repository.ts index 42d716d324..fa18e1be12 100644 --- a/packages/daemon/src/domain/queue-repository.ts +++ b/packages/daemon/src/domain/queue-repository.ts @@ -3961,13 +3961,19 @@ export class QueueRepository { /** A park's current question: its latest human-required transition, and that notification's final post receipt, if any. */ /** The opaque revision of what a person decides on (#2446): the question (body, intent, questions) and its decision state * (open, blocked or terminal; who it's addressed to or parked on; its answers, a decision made for them, the current park, - * and the replies returned). A label, a wake or a claim doesn't change it; a new park does. */ + * and the replies returned). A label, a wake, a claim or a note on a park doesn't change it; a new park does, on any blocker. */ decisionRevision(item: QueueItem): string { const phase = item.state === "blocked" ? "blocked" : item.state === "pending" || item.state === "in-progress" ? "open" : item.state; const replies = this.hasReplyToColumn ? (this.db.prepare("SELECT qitem_id FROM queue_items WHERE reply_to = ? ORDER BY ts_created, rowid") .all(item.qitemId) as Array<{ qitem_id: string }>).map((row) => row.qitem_id) : []; + // The park this is: the transition into blocked from another state (a later blocked-to-blocked note stays this park). + const transitions = item.state === "blocked" ? this.transitionLog.listForQitem(item.qitemId) : []; + let parkEntry: number | null = null; + for (let i = transitions.length - 1; i >= 0 && parkEntry === null; i--) { + if (transitions[i]!.state === "blocked" && transitions[i - 1]?.state !== "blocked") parkEntry = transitions[i]!.transitionId; + } const decision = [item.body, item.humanIntent ?? null, item.humanQuestions ?? null, phase, item.destinationSession, item.blockedOn, - item.humanAnswers ?? null, item.delegatedDecision ?? null, this.currentHumanRequiredEpisode(item.qitemId)?.key ?? null, replies]; + item.humanAnswers ?? null, item.delegatedDecision ?? null, this.currentHumanRequiredEpisode(item.qitemId)?.key ?? null, parkEntry, replies]; return `dr1-${createHash("sha256").update(JSON.stringify(decision)).digest("hex").slice(0, 24)}`; } diff --git a/packages/daemon/src/routes/mission-control.ts b/packages/daemon/src/routes/mission-control.ts index bc250ad173..69c8b6d9a9 100644 --- a/packages/daemon/src/routes/mission-control.ts +++ b/packages/daemon/src/routes/mission-control.ts @@ -248,7 +248,6 @@ export function missionControlRoutes(opts?: MissionControlRoutesOpts): Hono { : err.code === "answer_not_closed" ? 409 : err.code === "answer_not_recorded" ? 400 : err.code === "decision_revision_changed" ? 409 - : err.code === "decision_revision_unsupported" ? 400 : 500; return c.json( { error: err.code, message: err.message, ...(err.details ?? {}) }, From 4796147f39e858ada56bacf9226726d3029cb8ea Mon Sep 17 00:00:00 2001 From: dev-driver Date: Tue, 3 Nov 2026 16:25:00 +0000 Subject: [PATCH 5/6] test(mission-control): a runner ask's already-complete choice conflicts when a deny lands during its return Tests only, #2448 round 3. A runner's permission ask still pending with its answers already recorded (an interrupted return): a guarded choice writes nothing before its return, a deny lands during it, and the choice must conflict with only the deny recorded. A control pins that a choice which records its answer still closes the ask. --- .../daemon/test/decision-revision.test.ts | 38 +++++++++++++++++-- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/packages/daemon/test/decision-revision.test.ts b/packages/daemon/test/decision-revision.test.ts index 72e7582cb3..8c6d59fdea 100644 --- a/packages/daemon/test/decision-revision.test.ts +++ b/packages/daemon/test/decision-revision.test.ts @@ -1,7 +1,7 @@ // #2446: an opt-in decision revision. A client reads the decision with its revision, and a settling act sent with that // revision runs only on that decision. Real SQLite (all migrations), the real queue, the Mission Control route and write // contract, and the queue's single-row read; the route-wiring cells stub the contract and the remote fetch. Nothing native. -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import Database from "better-sqlite3"; import { Hono } from "hono"; import { migrate } from "../src/db/migrate.js"; @@ -51,8 +51,17 @@ function fixture() { return row.qitemId; }; // A runner's permission request: the person's typed answer is recorded on the ask itself, with no reply row. - const permissionAsk = async () => (await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Allow the runner to write?", - summary: "Allow?", humanIntent: "decision", tags: ["runner-permission", "permission-node:node-1", "permission-generation:gen-1"], nudge: false })).qitemId; + const permissionAsk = async (opts: { question?: boolean } = {}) => { + if (opts.question) { + // The runner's own seat, at the generation the request names, so the ask isn't closed as stale. + db.exec(`INSERT OR IGNORE INTO rigs(id,name) VALUES ('rig-1','test-rig'); INSERT OR IGNORE INTO nodes(id,rig_id,logical_id,runtime) VALUES ('node-1','rig-1','dev.runner','codex'); + INSERT OR IGNORE INTO occupant_tenures(id,node_id,generation_ordinal,generation_uuid,kind) VALUES ('t-1','node-1',1,'gen-1','fresh');`); + } + return (await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Allow the runner to write?", + summary: "Allow?", humanIntent: "decision", tags: ["runner-permission", "permission-node:node-1", "permission-generation:gen-1"], + ...(opts.question ? { humanQuestions: [{ id: "permission", question: "Allow?", options: [{ id: "allow", label: "Allow" }, { id: "deny", label: "Deny" }] }] } : {}), + nudge: false })).qitemId; + }; const revision = async (id: string) => (await (await app.request(`/api/queue/${id}`)).json() as { decisionRevision?: unknown }).decisionRevision; const act = async (body: Record) => { const response = await app.request("/api/mission-control/action", { method: "POST", @@ -118,6 +127,29 @@ describe("an opt-in decision revision on the settling verbs (#2446)", () => { expect(f.written(id)).toMatchObject({ row: { state: "done" }, actions: 1 }); }); + it("a choice on a runner's permission ask whose answers were already complete conflicts when a deny lands during its return", async () => { + const f = fixture(); const id = await f.permissionAsk({ question: true }); + // An interrupted return: the complete set is recorded and the ask is still pending, so this choice writes nothing first. + f.db.prepare("UPDATE queue_items SET human_answers = ? WHERE qitem_id = ?").run(JSON.stringify({ permission: "allow" }), id); + const read = await f.revision(id); + const original = f.repo.createHumanAnswer.bind(f.repo); + vi.spyOn(f.repo, "createHumanAnswer").mockImplementationOnce(async input => { + const returned = await original(input); + expect(await f.act({ verb: "deny", qitemId: id, reason: "not now" })).toMatchObject({ status: 200 }); + return returned; + }); + const result = await f.act({ verb: "answer", qitemId: id, questionId: "permission", optionId: "allow", expectedRevision: read }); + expect(result).toMatchObject({ status: 409, body: { error: "decision_revision_changed" } }); + expect(f.db.prepare("SELECT action_verb FROM mission_control_actions WHERE qitem_id = ?").all(id)).toEqual([{ action_verb: "deny" }]); + }); + + it("a choice on a runner's permission ask that records its answer closes it (the writing choice, compared before its write)", async () => { + const f = fixture(); const id = await f.permissionAsk({ question: true }); + const read = await f.revision(id); + expect(await f.act({ verb: "answer", qitemId: id, questionId: "permission", optionId: "allow", expectedRevision: read })) + .toMatchObject({ status: 200, body: { answer: { outcome: "answered" }, closedQitem: { state: "done", humanAnswers: { permission: "allow" } } } }); + }); + it("a typed-gate park resolved and parked again on the same gate is a new decision; a note on a park isn't", async () => { const f = fixture(); const id = await f.park("external:qa-pass"); const first = await f.revision(id); From aebb57c6e5be676fc039e3ecc422c190c5d2f45f Mon Sep 17 00:00:00 2001 From: dev-driver Date: Tue, 3 Nov 2026 17:19:37 +0000 Subject: [PATCH 6/6] fix(mission-control): re-compare before a runner ask's close whenever the call wrote nothing first #2448 round 3. On a runner's permission ask (no reply row), the re-compare before the close now keys on whether this call changed the ask before its await, not on choice versus typed. A choice whose answer set was already complete (an interrupted return) writes nothing, so it compares again like typed words. A choice that recorded its answer was compared before that write and keeps its behaviour, so the nothing-was-written conflict stays true. --- .../mission-control/mission-control-write-contract.ts | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts b/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts index 8b48585981..56090f8a25 100644 --- a/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts +++ b/packages/daemon/src/domain/mission-control/mission-control-write-contract.ts @@ -618,6 +618,7 @@ export class MissionControlWriteContract { const resume = !chose && !words && questions.length > 0 && unansweredQuestions(questions, ask.humanAnswers ?? {}).length === 0; const answersReply = (answers: HumanAnswers) => `${formatHumanAnswers(questions, answers).join("\n")}\n\n---\nAnswers (question id → option id): ${JSON.stringify(answers)}\nIn reply to: ${input.qitemId} (its humanAnswers field holds the same)`; let reply: string; + let wrote = false; // whether this call changed the ask before the return below awaits if (chose) { const recorded = this.queueRepo.recordHumanAnswer({ qitemId: input.qitemId, actorSession: input.actorSession, questionId: input.questionId ?? "", optionId: input.optionId ?? "", primaryUser }); @@ -625,6 +626,8 @@ export class MissionControlWriteContract { throw new MissionControlWriteContractError("answer_not_recorded", `The answer wasn't recorded: ${recorded.reason}.`, { qitemId: input.qitemId, reason: recorded.reason }); } + // An already-complete set (an interrupted return) comes back without a write. + wrote = JSON.stringify(recorded.answers) !== JSON.stringify(ask.humanAnswers ?? {}); const [just] = formatHumanAnswers(recorded.questions.filter((q) => q.id === input.questionId), recorded.answers); if (!recorded.complete) { return this.recordAnswer(input, before, just ?? `answered ${input.questionId}`, { @@ -669,9 +672,11 @@ export class MissionControlWriteContract { const decision = chose || resume ? formatHumanAnswers(questions, this.queueRepo.getById(input.qitemId)?.humanAnswers ?? {}).join("; ") : returned.qitemId === input.qitemId ? words : humanAnswerText(returned.body); - // #2446: typed words to a runner's permission ask have written nothing yet (no reply row, unlike a choice), and the - // await above let other requests in, so a guarded answer compares its revision again here, with no await before the close. - if (input.expectedRevision !== undefined && replyId === input.qitemId && !chose) { + // #2446: on a runner's permission ask (no reply row), a call that changed nothing before the await above (typed words, + // or a choice whose set was already complete) still has its first write ahead, and the await let other requests in, so + // a guarded answer compares its revision again here, with no await before the close. A call that wrote was compared + // before that write. + if (input.expectedRevision !== undefined && replyId === input.qitemId && !wrote) { const now = this.queueRepo.getById(input.qitemId); const revision = now ? this.queueRepo.decisionRevision(now) : null; if (now && revision !== input.expectedRevision) throw this.revisionChanged(now, revision!);