Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
// closure_target+handed_off_to=<destination>;
// 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";
Expand Down Expand Up @@ -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).
Expand All @@ -119,6 +123,12 @@ 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;
/** 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 {
Expand Down Expand Up @@ -150,6 +160,41 @@ export class MissionControlWriteContract {
* notify subscribers + opt-in transport wake (handoff only).
*/
async act(input: MissionControlActionInput): Promise<MissionControlActionResult> {
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)) {
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. 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;
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 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<MissionControlActionResult> {
if (input.verb === "annotate") {
return this.annotateOnly(input);
}
Expand Down Expand Up @@ -573,13 +618,16 @@ 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 });
if (recorded.status !== "recorded") {
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}`, {
Expand Down Expand Up @@ -624,6 +672,15 @@ 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: 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!);
}
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",
Expand Down Expand Up @@ -720,6 +777,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":
Expand Down
18 changes: 18 additions & 0 deletions packages/daemon/src/domain/queue-repository.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3959,6 +3959,24 @@ 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, 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, parkEntry, replies];
return `dr1-${createHash("sha256").update(JSON.stringify(decision)).digest("hex").slice(0, 24)}`;
}

currentHumanRequiredEpisode(qitemId: string): { key: string; receipt: ReturnType<QueueTransitionLog["getOwnerNotificationReceipt"]> } | null {
const episode = this.transitionLog.listForQitem(qitemId).filter((transition) => transition.ownerNotificationKind === "human-required").at(-1);
if (!episode) return null;
Expand Down
5 changes: 5 additions & 0 deletions packages/daemon/src/routes/mission-control.ts
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,7 @@ 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
: 500;
return c.json(
{ error: err.code, message: err.message, ...(err.details ?? {}) },
Expand Down Expand Up @@ -330,9 +331,11 @@ export function missionControlRoutes(opts?: MissionControlRoutesOpts): Hono {
auditNotes?: Record<string, unknown>;
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(
{
Expand Down Expand Up @@ -434,6 +437,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.
Expand Down
2 changes: 2 additions & 0 deletions packages/daemon/src/routes/queue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading
Loading