fix(queue): make notes state-preserving

This commit is contained in:
v-openrig-build authored and mvschwarz committed 2026-08-26 03:10:45 -07:00
1 parent ae19100f53
commit b7c2a2ac6f
5 files changed
+293 -8

No files matched your search

+6 -3
View File
@@ -477,9 +477,10 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
cmd
.command("update <qitemId>")
.description("Mutate qitem state. state=done REQUIRES --closure-reason (one of: handed_off_to, blocked_on, denied, canceled, no-follow-on, escalation). Closure ≠ acceptance: handed_off_to records delivery to the next stage; acceptance is the next stage's verdict on its own qitem, not this closure.")
.description("Append a note and/or mutate qitem state. A note without --state never changes row state. state=done REQUIRES --closure-reason (one of: handed_off_to, blocked_on, denied, canceled, no-follow-on, escalation). Closure ≠ acceptance: handed_off_to records delivery to the next stage; acceptance is the next stage's verdict on its own qitem, not this closure.")
.option("--actor <session>", "(deprecated, ignored) the actor is derived from the seat env (X-OpenRig-Session); P21 I3 made the update route derive it from the transport header")
.requiredOption("--state <state>", "New state: pending | in-progress | done | blocked | failed | denied | canceled | handed-off")
.option("--state <state>", "New state: pending | in-progress | done | blocked | failed | denied | canceled | handed-off")
.option("--reopen", "Explicitly acknowledge a deliberate terminal-to-active repair; requires --state and --note")
.option("--closure-reason <reason>", "Required for state=done; also 'superseded' on state=canceled (with --closure-target = the successor) records a supersession, distinct from an abandoned cancel")
.option("--closure-target <target>", "Required for handed_off_to, blocked_on, escalation, and superseded")
.option("--blocked-on <blocker>", "For state=blocked: the blocker — a qitem id (must exist and be live), a human seat (FR-6 park; requires summary + evidence_ref), or a typed non-qitem gate 'fold:<what>' / 'auth:<what>' / 'external:<what>'")
@@ -489,7 +490,8 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
.option("--json", "JSON output for agents")
.action(async (qitemId: string, opts: {
actor?: string;
state: string;
state?: string;
reopen?: boolean;
closureReason?: string;
closureTarget?: string;
blockedOn?: string;
@@ -506,6 +508,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
await withClient(deps, async (client) => {
const res = await client.post<unknown>(`/api/queue/${encodeURIComponent(qitemId)}/update`, {
state: opts.state,
reopen: opts.reopen,
closureReason: opts.closureReason,
closureTarget: opts.closureTarget,
blockedOn: opts.blockedOn,
+43
View File
@@ -571,6 +571,49 @@ describe("rig queue CLI", () => {
expect(out).toContain("validReasons");
});
it("update accepts a note without state", async () => {
const { deps, calls } = makeDeps();
const program = createProgram({ queueDeps: deps });
program.exitOverride();
await program.parseAsync([
"node", "rig", "queue", "update", "qitem-note",
"--note", "audit bytes",
"--json",
]);
expect(calls.find((c) => c.path === "/api/queue/qitem-note/update")?.body).toEqual({
state: undefined,
reopen: undefined,
closureReason: undefined,
closureTarget: undefined,
blockedOn: undefined,
summary: undefined,
evidenceRef: undefined,
transitionNote: "audit bytes",
});
});
it("update forwards explicit reopen acknowledgment", async () => {
const { deps, calls } = makeDeps();
const program = createProgram({ queueDeps: deps });
program.exitOverride();
await program.parseAsync([
"node", "rig", "queue", "update", "qitem-reopen",
"--state", "pending",
"--note", "repair",
"--reopen",
"--json",
]);
expect(calls.find((c) => c.path === "/api/queue/qitem-reopen/update")?.body).toMatchObject({
state: "pending",
reopen: true,
transitionNote: "repair",
});
});
it("handoff-and-complete sends NO body fromSession — the handing-off seat derives from the transport header (X-OpenRig-Session); --from is dropped, --to stays the target", async () => {
const saved = process.env["OPENRIG_SESSION_NAME"];
process.env["OPENRIG_SESSION_NAME"] = "bob@rig"; // the seat env == the X-OpenRig-Session the DaemonClient stamps
+81 -4
View File
@@ -225,7 +225,9 @@ export interface QueueCreateInput {
export interface QueueUpdateInput {
qitemId: string;
actorSession: string;
state: QueueState;
state?: QueueState;
/** Explicit acknowledgment for a deliberate terminal → active repair. */
reopen?: boolean;
/**
* OPR.0.4.6.WF3 FR-6 — set ONLY by the workflow domain's own write
* paths (projector close, route close): they hold the frontier
@@ -1798,6 +1800,51 @@ export class QueueRepository {
`qitem ${input.qitemId} not found`
);
}
const hasNote = typeof input.transitionNote === "string" && input.transitionNote.trim().length > 0;
const isGuardedTerminal = (["done", "canceled", "handed-off"] as const).includes(
qitem.state as "done" | "canceled" | "handed-off",
);
const isStatePreservingAppend = input.state === undefined || (isGuardedTerminal && input.state === qitem.state);
if (isStatePreservingAppend) {
if (input.state === undefined && !hasNote) {
throw new QueueRepositoryError(
"state_or_note_required",
"queue update requires --state or a non-empty --note; nothing was written",
);
}
const disallowed = input.reopen === true
|| input.closureReason != null
|| input.closureTarget != null
|| input.handedOffTo != null
|| input.blockedOn != null
|| input.summary != null
|| input.evidenceRef != null;
if (disallowed) {
throw new QueueRepositoryError(
"note_append_fields_not_admitted",
"a state-preserving note append accepts only --note (and an optional same --state); state-write fields were supplied, so nothing was written",
);
}
this.transitionLog.append({
qitemId: input.qitemId,
state: qitem.state,
actorSession: input.actorSession,
transitionNote: input.transitionNote,
identityProvenance: input.identityProvenance ?? null,
});
return this.eventBus.persistWithinTransaction({
type: "queue.updated",
qitemId: input.qitemId,
fromState: qitem.state,
toState: qitem.state,
closureReason: null,
closureTarget: null,
actorSession: input.actorSession,
summary: qitem.summary ?? null,
});
}
if (!isQueueState(input.state)) {
throw new QueueRepositoryError(
"invalid_state",
@@ -1805,6 +1852,36 @@ export class QueueRepository {
);
}
const isReopen = isGuardedTerminal && input.state !== qitem.state;
if (isReopen && !input.reopen) {
throw new QueueRepositoryError(
"terminal_reopen_requires_ack",
`qitem ${input.qitemId} is currently '${qitem.state}'; state='${input.state}' would reopen a terminal row. Re-run deliberately with --reopen --note <reason>.`,
{ currentState: qitem.state, requestedState: input.state },
);
}
if (isReopen && !isBlockerLive(input.state)) {
throw new QueueRepositoryError(
"terminal_reopen_target_invalid",
`qitem ${input.qitemId} is currently '${qitem.state}'; --reopen requires an active target state (pending, in-progress, or blocked), not '${input.state}'.`,
{ currentState: qitem.state, requestedState: input.state },
);
}
if (isReopen && !hasNote) {
throw new QueueRepositoryError(
"reopen_note_required",
`qitem ${input.qitemId} is currently '${qitem.state}'; deliberate reopen requires --note <reason> so the repair is auditable.`,
{ currentState: qitem.state, requestedState: input.state },
);
}
if (input.reopen && !isReopen) {
throw new QueueRepositoryError(
"reopen_not_applicable",
`--reopen applies only when moving a terminal row to an active state; qitem ${input.qitemId} is currently '${qitem.state}'.`,
{ currentState: qitem.state, requestedState: input.state },
);
}
const validation = validateClosure({
state: input.state,
closureReason: input.closureReason ?? null,
@@ -2001,7 +2078,7 @@ export class QueueRepository {
qitemId: input.qitemId,
state: input.state,
actorSession: input.actorSession,
transitionNote: input.transitionNote,
transitionNote: isReopen ? `reopen acknowledged: ${input.transitionNote}` : input.transitionNote,
closureReason: validation.closureReason ?? undefined,
closureTarget: validation.closureTarget ?? undefined,
identityProvenance: input.identityProvenance ?? null, // P21 §4 era-stamp
@@ -2448,6 +2525,6 @@ export class QueueRepository {
}
}
function isQueueState(value: string): value is QueueState {
return (QUEUE_STATES as readonly string[]).includes(value);
function isQueueState(value: unknown): value is QueueState {
return typeof value === "string" && (QUEUE_STATES as readonly string[]).includes(value);
}
+8 -1
View File
@@ -139,6 +139,12 @@ export function queueRoutes(): Hono {
: err.code === "invalid_closure_reason" ? 400
: err.code === "missing_closure_target" ? 400
: err.code === "invalid_state" ? 400
: err.code === "state_or_note_required" ? 400
: err.code === "note_append_fields_not_admitted" ? 400
: err.code === "terminal_reopen_requires_ack" ? 409
: err.code === "terminal_reopen_target_invalid" ? 400
: err.code === "reopen_note_required" ? 400
: err.code === "reopen_not_applicable" ? 400
: err.code === "claim_destination_mismatch" ? 403
: err.code === "qitem_not_claimable" ? 409
: err.code === "qitem_not_in_progress" ? 409
@@ -536,6 +542,7 @@ export function queueRoutes(): Hono {
const body = await c.req.json<{
actorSession?: string;
state?: QueueState;
reopen?: boolean;
transitionNote?: string;
closureReason?: string;
closureTarget?: string;
@@ -550,13 +557,13 @@ export function queueRoutes(): Hono {
const identity = requireSenderIdentity(c, { verb: "queue update", bodyClaim: body.actorSession });
if (!identity.ok) return identity.response;
const actorSession = identity.session;
if (!body.state) return c.json({ error: "state is required" }, 400);
try {
const item = getRepo(c).update({
qitemId,
actorSession,
state: body.state,
reopen: body.reopen,
transitionNote: body.transitionNote,
closureReason: body.closureReason,
closureTarget: body.closureTarget,
@@ -0,0 +1,155 @@
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { Hono } from "hono";
import type Database from "better-sqlite3";
import { createDb } from "../src/db/connection.js";
import { migrate } from "../src/db/migrate.js";
import { coreSchema } from "../src/db/migrations/001_core_schema.js";
import { eventsSchema } from "../src/db/migrations/003_events.js";
import { streamItemsSchema } from "../src/db/migrations/023_stream_items.js";
import { queueItemsSchema } from "../src/db/migrations/024_queue_items.js";
import { queueTransitionsSchema } from "../src/db/migrations/025_queue_transitions.js";
import { inboxEntriesSchema } from "../src/db/migrations/026_inbox_entries.js";
import { outboxEntriesSchema } from "../src/db/migrations/027_outbox_entries.js";
import { i3IdentityProvenanceSchema } from "../src/db/migrations/067_i3_identity_provenance.js";
import { EventBus } from "../src/domain/event-bus.js";
import { QueueRepository, type QueueState } from "../src/domain/queue-repository.js";
import { queueRoutes } from "../src/routes/queue.js";
describe("S8a — a note is not a state write", () => {
let db: Database.Database;
let repo: QueueRepository;
let app: Hono;
beforeEach(() => {
db = createDb();
migrate(db, [
coreSchema,
eventsSchema,
streamItemsSchema,
queueItemsSchema,
queueTransitionsSchema,
inboxEntriesSchema,
outboxEntriesSchema,
i3IdentityProvenanceSchema,
]);
const eventBus = new EventBus(db);
repo = new QueueRepository(db, eventBus);
app = new Hono();
app.use("*", async (c, next) => {
c.set("eventBus" as never, eventBus);
c.set("queueRepo" as never, repo);
await next();
});
app.route("/api/queue", queueRoutes());
});
afterEach(() => db.close());
async function createTerminal(state: "done" | "canceled" | "handed-off") {
const item = await repo.create({ sourceSession: "author@rig", destinationSession: "owner@rig", body: "work" });
repo.update({
qitemId: item.qitemId,
actorSession: "owner@rig",
state,
...(state === "done" ? { closureReason: "no-follow-on" as const } : {}),
...(state === "handed-off"
? { closureReason: "handed_off_to" as const, closureTarget: "review@rig" }
: {}),
});
return item.qitemId;
}
function update(qitemId: string, body: Record<string, unknown>) {
return app.request(`/api/queue/${qitemId}/update`, {
method: "POST",
headers: { "content-type": "application/json", "X-OpenRig-Session": "auditor@rig" },
body: JSON.stringify(body),
});
}
it.each([
"same-state specimen note",
"testimony specimen note",
])("appends '%s' on a handed-off row without changing one byte of the row", async (note) => {
const qitemId = await createTerminal("handed-off");
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(qitemId);
const response = await update(qitemId, { transitionNote: note });
expect(response.status).toBe(200);
const after = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(qitemId);
expect(after).toEqual(before);
const transitions = repo.transitionLog.listForQitem(qitemId);
expect(transitions.at(-1)).toMatchObject({
state: "handed-off",
actorSession: "auditor@rig",
transitionNote: note,
identityProvenance: "transport:v1",
});
});
it.each(["done", "canceled", "handed-off"] as const)(
"refuses an accidental %s -> pending reopen and leaves the row unchanged",
async (terminalState) => {
const qitemId = await createTerminal(terminalState);
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(qitemId);
const transitionCount = repo.transitionLog.listForQitem(qitemId).length;
const response = await update(qitemId, { state: "pending", transitionNote: "audit" });
expect(response.status).toBe(409);
await expect(response.json()).resolves.toMatchObject({
error: "terminal_reopen_requires_ack",
currentState: terminalState,
requestedState: "pending",
});
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(qitemId)).toEqual(before);
expect(repo.transitionLog.listForQitem(qitemId)).toHaveLength(transitionCount);
},
);
it("reopens explicitly exactly once and records the acknowledgment, actor, and note", async () => {
const qitemId = await createTerminal("handed-off");
const transitionCount = repo.transitionLog.listForQitem(qitemId).length;
const response = await update(qitemId, {
state: "pending",
transitionNote: "repair the mistaken handoff",
reopen: true,
});
expect(response.status).toBe(200);
expect(repo.getById(qitemId)?.state).toBe("pending");
const transitions = repo.transitionLog.listForQitem(qitemId);
expect(transitions).toHaveLength(transitionCount + 1);
expect(transitions.at(-1)).toMatchObject({
state: "pending",
actorSession: "auditor@rig",
transitionNote: "reopen acknowledged: repair the mistaken handoff",
identityProvenance: "transport:v1",
});
});
it("errors when a note cannot be written instead of reporting silent success", async () => {
const response = await update("qitem-missing", { transitionNote: "must persist" });
expect(response.status).toBe(404);
await expect(response.json()).resolves.toMatchObject({ error: "qitem_not_found" });
});
it("same-state reassertion on a terminal row is an append, not a row write", async () => {
const qitemId = await createTerminal("canceled");
db.prepare("UPDATE queue_items SET ts_updated = ? WHERE qitem_id = ?").run("2000-01-01T00:00:00.000Z", qitemId);
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(qitemId);
const result = repo.update({
qitemId,
actorSession: "auditor@rig",
state: "canceled" as QueueState,
transitionNote: "confirm cancellation",
});
expect(result.state).toBe("canceled");
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(qitemId)).toEqual(before);
expect(repo.transitionLog.listForQitem(qitemId).at(-1)?.transitionNote).toBe("confirm cancellation");
});
});