diff --git a/packages/daemon/src/domain/queue-repository.ts b/packages/daemon/src/domain/queue-repository.ts index f57b019956..7d43d84658 100644 --- a/packages/daemon/src/domain/queue-repository.ts +++ b/packages/daemon/src/domain/queue-repository.ts @@ -4614,6 +4614,17 @@ export class QueueRepository { } catch { return false; } } + /** #514 — a pending explicit informational update addressed to a currently registered human + * (external or canonical alias, regardless of terminal transport) is correspondence for reading, + * not an obligation to claim. Registry unavailable/unregistered and agent destinations read false. */ + isRegisteredHumanUpdate(row: Pick): boolean { + if (row.state !== "pending" || row.humanIntent !== "update") return false; + try { + const registry = this.loadHumanRegistryFn(); + return registry.ok && resolveRegisteredHumanAddress(row.destinationSession, registry.entities) !== null; + } catch { return false; } + } + classifyDestinationOf(destinationSession: string): DestinationClass { return classifyDestination(destinationSession, { entities: (() => { @@ -5125,15 +5136,22 @@ export class QueueRepository { // episode (same predicate as runStuckSweep), so the view must not promise that sweep. const posted = view.state === "pending" && this.humanNotificationPostedThisEpisode(qitemId, listed); const onSteering = !posted && view.state === "pending" && this.humanAskWithoutAlert(qitemId, listed); - const row = !posted && !onSteering ? this.db.prepare(`SELECT state, ${this.hasHumanIntentColumn ? "human_intent" : "NULL AS human_intent"}, destination_session, last_nudge_result FROM queue_items WHERE qitem_id = ?`).get(qitemId) as { state: QueueItem["state"]; human_intent: QueueItem["humanIntent"]; destination_session: string; last_nudge_result: string | null } | undefined : undefined; - const noticeOutcome = listed ? listed.ledger?.outcome : this.deliveryOutcomeFor(qitemId)?.outcome; - const localOnly = !!row && noticeOutcome !== "never-posted" && noticeOutcome !== "transport-failed" + const row = view.state === "pending" && !onSteering + ? this.db.prepare(`SELECT state, ${this.hasHumanIntentColumn ? "human_intent" : "NULL AS human_intent"}, destination_session, last_nudge_result FROM queue_items WHERE qitem_id = ?`).get(qitemId) as { state: QueueItem["state"]; human_intent: QueueItem["humanIntent"]; destination_session: string; last_nudge_result: string | null } | undefined + : undefined; + const humanUpdate = !!row && this.isRegisteredHumanUpdate({ state: row.state, humanIntent: row.human_intent, destinationSession: row.destination_session }); + const noticeOutcome = !posted && !onSteering ? (listed ? listed.ledger?.outcome : this.deliveryOutcomeFor(qitemId)?.outcome) : undefined; + const localOnly = !!row && !posted && !humanUpdate && noticeOutcome !== "never-posted" && noticeOutcome !== "transport-failed" && this.isLocalHumanDecisionWithoutGateway({ state: row.state, humanIntent: row.human_intent, destinationSession: row.destination_session, lastNudgeResult: row.last_nudge_result }); const awaitingHuman = posted || onSteering || localOnly; - if (awaitingHuman) { + if (awaitingHuman || humanUpdate) { view.nextBackstop = { owner: view.owner, mechanism: localOnly ? "local attention feed (no registered human gateway)" : onSteering ? "none (on Steering for the human, with no Slack alert; awaiting their decision)" - : "none (posted to the human; awaiting their decision)", dueAt: null, intervalSeconds: null }; + : humanUpdate + ? posted + ? "none (posted to the human; informational correspondence requires no claim)" + : "none (informational correspondence to the human; no claim required)" + : "none (posted to the human; awaiting their decision)", dueAt: null, intervalSeconds: null }; } // Quiet hours: a held ask is owned by its window's release job until it posts. const quiet = awaitingHuman ? null : quietCustodyBackstop(view.owner, quietHoldCustody(this.db, qitemId)); @@ -5143,7 +5161,7 @@ export class QueueRepository { } const recovery = readWakeLadderBackstop(this.db, qitemId, target => this.acceptsDestination(target)); if (recovery && !localOnly) { - if (!awaitingHuman) view.laterBackstop = { ...view.nextBackstop, note: "Conditional safety net; current delivery/recovery ownership is evaluated first." }; + if (!awaitingHuman && !humanUpdate) view.laterBackstop = { ...view.nextBackstop, note: "Conditional safety net; current delivery/recovery ownership is evaluated first." }; view.nextBackstop = recovery; } else if (!awaitingHuman && view.state === "pending" && (this.db.prepare("SELECT last_nudge_result FROM queue_items WHERE qitem_id = ?") .get(qitemId) as { last_nudge_result: string | null } | undefined)?.last_nudge_result === "retained:typing_guard") { diff --git a/packages/daemon/src/domain/queue-stuck-sweep.ts b/packages/daemon/src/domain/queue-stuck-sweep.ts index 71681f412b..b5526836da 100644 --- a/packages/daemon/src/domain/queue-stuck-sweep.ts +++ b/packages/daemon/src/domain/queue-stuck-sweep.ts @@ -449,6 +449,9 @@ function deliveryResolutionEvidence( if (quiet && !quiet.unreadable && (quiet.state === "held" || quiet.state === "released")) { return `source held by quiet-release job for window ${quiet.windowId}`; } + if (kind === "unclaimed-obligation" && !quiet && deps.queueRepo.isRegisteredHumanUpdate(row) && !deps.queueRepo.deliveryOutcomeFor(qitemId)) { + return "informational update to a registered human requires no claim"; + } if (kind !== "undelivered-human-notice" && row.lastNudgeResult === "retained:typing_guard") { return "source wake held by the seat typing guard"; } @@ -727,6 +730,9 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise kind === "stalled-after-claim" || kind === "overdue-claim"; - if (currentKind && currentKind !== c.kind && pickupOrDeadline(currentKind) && pickupOrDeadline(c.kind)) { + const humanUpdateNotice = currentKind === "unclaimed-obligation" + && c.kind === "undelivered-human-notice" + && deps.queueRepo.isRegisteredHumanUpdate(c.row); + if (currentKind && currentKind !== c.kind && ((pickupOrDeadline(currentKind) && pickupOrDeadline(c.kind)) || humanUpdateNotice)) { const marker = "\n\n--- Original finding (historical) ---\n"; const original = sharedRow.body.includes(marker) ? sharedRow.body.split(marker, 2)[1]! : sharedRow.body; const body = evidenceBody(deps.db, c) + marker + original; diff --git a/packages/daemon/test/queue-stuck-sweep.test.ts b/packages/daemon/test/queue-stuck-sweep.test.ts index 7ff396d34c..1a83f2676c 100644 --- a/packages/daemon/test/queue-stuck-sweep.test.ts +++ b/packages/daemon/test/queue-stuck-sweep.test.ts @@ -2077,15 +2077,24 @@ describe("#514 unclaimed sweep and a pending ask posted to a registered human", const BETA = human("human-beta"); // A registered human whose id has no human- prefix: its canonical alias (owner@kernel) cannot carry an intent. const PLAIN = human("owner"); + // Feed-only correspondence owes no Slack post; a missing delivery must not obscure the unclaimed net. + const READER: HumanFragment = { entityId: "human-reader", class: "human", displayName: "Reader", + address: "human-reader@external", connectorBindings: [], prefs: {} }; let db: Database.Database; let repo: QueueRepository; + let registeredHumans: HumanFragment[]; + let registryAvailable: boolean; beforeEach(() => { db = new Database(":memory:"); migrate(db, ALL_MIGRATIONS); + registeredHumans = [OWNER, ALPHA, BETA, PLAIN, READER]; + registryAvailable = true; repo = new QueueRepository(db, new EventBus(db), { validateRig: () => true, - loadHumanRegistry: () => ({ ok: true as const, entities: [OWNER, ALPHA, BETA, PLAIN] }), + loadHumanRegistry: () => registryAvailable + ? { ok: true as const, entities: registeredHumans } + : { ok: false as const, error: "fixture registry outage" }, transport: { send: async () => ({ ok: true, verified: true }) }, }); }); @@ -2244,6 +2253,343 @@ describe("#514 unclaimed sweep and a pending ask posted to a registered human", const unclaimedFinding = async (qitemId: string) => (await findingsFor(qitemId)) .filter((f) => (f.tags ?? []).includes(`stuck-sweep:unclaimed-obligation:${qitemId}`)); + function bindHumanPane(destination: string): void { + // Isolated topology fixture, as in seedLeadSeat: no native session or pane is contacted. + const member = destination.split("@")[0]!; + db.prepare("INSERT INTO rigs (id, name) VALUES ('rig-reader', 'kernel')").run(); + db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime, cwd) VALUES ('node-reader', 'rig-reader', ?, 'terminal', '/tmp')").run(member); + db.prepare("INSERT INTO sessions (id, node_id, session_name, status, created_at) VALUES ('sess-reader', 'node-reader', ?, 'running', '2026-11-05 00:00:00')").run(destination); + db.prepare("INSERT INTO bindings (id, node_id, attachment_type, tmux_session, tmux_pane) VALUES ('bind-reader', 'node-reader', 'tmux', ?, '%1')").run(destination); + // Distinguish explicit fixture binding from the empty-topology legacy fallback. + expect(repo.classifyDestinationOf("unbound-fixture@kernel").class).toBe("unroutable"); + expect(repo.classifyDestinationOf(destination).class).toBe("pane-bound"); + } + + function bindAskingAgentPane(): void { + db.prepare("INSERT INTO rigs (id, name) VALUES ('rig-asker', 'r')").run(); + db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime, cwd) VALUES ('node-asker', 'rig-asker', 'asker', 'codex', '/tmp')").run(); + db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES ('sess-asker', 'node-asker', 'asker@r', 'running')").run(); + db.prepare("INSERT INTO bindings (id, node_id, attachment_type, tmux_session, tmux_pane) VALUES ('bind-asker', 'node-asker', 'tmux', 'asker@r', '%2')").run(); + expect(repo.classifyDestinationOf("asker@r").class).toBe("pane-bound"); + } + + const recoveriesFor = (id: string) => repo.list({ limit: 500 }) + .filter((item) => item.tags?.includes(`recovery-for:${id}`)); + + it.each([ + { destination: "human-reader@external", transport: "gateway-routable" }, + { destination: "human-reader@kernel", transport: "pane-bound" }, + ])("keeps an aged informational update readable at $destination without creating recovery work", async ({ destination, transport }) => { + if (transport === "pane-bound") bindHumanPane(destination); + expect(repo.classifyDestinationOf(READER.address)).toMatchObject({ class: "gateway-routable", resolvedHuman: READER.address }); + expect(repo.classifyDestinationOf(destination).class).toBe(transport); + const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination, + body: "The requested build finished. No action or decision is needed.", summary: "Build finished", + evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false }); + age(row.qitemId); + const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(before).toBeDefined(); + // This person uses the Feed, so there is no missing Slack delivery to recover. + expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull(); + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before); + expect(repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId)) + .toMatchObject({ body: row.body, state: "pending", humanIntent: "update" }); + expect(await findingsFor(row.qitemId)).toEqual([]); + }); + + it.each([ + { destination: "human-reader@external", interruption: "removed" }, + { destination: "human-reader@kernel", interruption: "removed" }, + { destination: "human-reader@external", interruption: "outage" }, + { destination: "human-reader@kernel", interruption: "outage" }, + ])("PR5-F1 closes only obsolete claim work after $interruption registration recovers at $destination", async ({ destination, interruption }) => { + if (destination.endsWith("@kernel")) bindHumanPane(destination); + const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination, + body: "Build finished; no action needed.", summary: "Build status", evidenceRef: "https://example.test/build", + humanIntent: "update", nudge: false }); + age(row.qitemId); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(stored).toBeDefined(); + if (interruption === "removed") registeredHumans = registeredHumans.filter((person) => person.entityId !== READER.entityId); + else registryAvailable = false; + const initial = await sweep(); + expect(initial.outcome, initial.error).not.toBe("failed"); + const [finding] = await unclaimedFinding(row.qitemId); + expect(finding).toMatchObject({ state: "pending" }); + // Reading an unresolved or subsequently registered source cannot close recovery work. + repo.waitingView(row.qitemId); + expect(repo.getById(finding!.qitemId)?.state).toBe("pending"); + registeredHumans = [OWNER, ALPHA, BETA, PLAIN, READER]; + registryAvailable = true; + const beforeRead = db.serialize(); + expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", humanIntent: "update" }); + expect(repo.list({ destinationSession: destination }).map((item) => item.qitemId)).toContain(row.qitemId); + repo.waitingView(row.qitemId); + expect(db.serialize()).toEqual(beforeRead); + expect(repo.getById(finding!.qitemId)?.state).toBe("pending"); + const second = await sweep(); + expect(second.outcome, second.error).not.toBe("failed"); + expect.soft(second.findings, "F1: sweep closes obsolete unclaimed finding").toContainEqual( + expect.objectContaining({ kind: "unclaimed-obligation", findingQitemId: finding!.qitemId, action: "closed" })); + expect.soft(repo.getById(finding!.qitemId), "F1: recovery is terminal without claiming correspondence") + .toMatchObject({ state: "done", closureReason: "no-follow-on" }); + const resolution = repo.listTransitions(finding!.qitemId).at(-1)?.transitionNote ?? ""; + expect.soft(resolution, "F1: resolution names informational correspondence").toMatch(/informational|update/i); + expect.soft(resolution).not.toMatch(/source claimed|human (ask|notification) posted|wake verified/i); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + expect(repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId)) + .toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null }); + expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull(); + }); + + it.each([ + { destination: "human-owner@external", interruption: "removed", outcome: "never-posted" }, + { destination: "human-owner@kernel", interruption: "removed", outcome: "never-posted" }, + { destination: "human-owner@external", interruption: "outage", outcome: "transport-failed" }, + { destination: "human-owner@kernel", interruption: "outage", outcome: "transport-failed" }, + ] as const)("PR2586 represents $outcome delivery after $interruption registration returns at $destination", async ({ destination, interruption, outcome }) => { + bindAskingAgentPane(); + if (destination.endsWith("@kernel")) bindHumanPane(destination); + const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination, + body: "Build finished; no decision needed.", summary: "Build status", evidenceRef: "https://example.test/build", + humanIntent: "update", nudge: false }); + const notification = repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!; + expect(notification).toMatchObject({ ownerNotificationKind: "human-update", ownerNotificationLevel: "NOTICE" }); + const key = currentKey(row.qitemId); + if (outcome === "transport-failed") await receipt(row.qitemId, "transport-failed", key); + // An unrelated receipt is not evidence that this current episode reached its person. + const staleKey = `${row.qitemId}:0`; + expect(staleKey).not.toBe(key); + await receipt(row.qitemId, "posted", staleKey); + age(row.qitemId); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome); + if (interruption === "removed") registeredHumans = registeredHumans.filter((person) => person.entityId !== OWNER.entityId); + else registryAvailable = false; + expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull(); + const first = await sweep(); + expect(first.outcome, first.error).not.toBe("failed"); + const initial = recoveriesFor(row.qitemId); + expect(initial).toHaveLength(1); + const prior = initial[0]!; + expect(prior).toMatchObject({ state: "pending", tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) }); + const originalHistory = repo.listTransitions(prior.qitemId); + registeredHumans = [OWNER, ALPHA, BETA, PLAIN, READER]; + registryAvailable = true; + expect(currentKey(row.qitemId)).toBe(key); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(stored).toBeDefined(); + let current: QueueItem | undefined; + for (const pass of [1, 2]) { + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + const open = recoveriesFor(row.qitemId).filter((item) => ["pending", "in-progress", "blocked"].includes(item.state)); + expect(open, "one live recovery still owns the unresolved delivery").toHaveLength(1); + const next = open[0]!; + if (pass === 1) expect(next).toMatchObject({ qitemId: prior.qitemId, + destinationSession: prior.destinationSession, state: prior.state }); + if (current) expect(next).toMatchObject({ qitemId: current.qitemId, state: current.state, + destinationSession: current.destinationSession, body: current.body, summary: current.summary }); + current = next; + expect.soft(current.summary, `delivery pass ${pass}: current condition is visible`).toContain("undelivered-human-notice"); + expect.soft(current.body, `delivery pass ${pass}: current recipient and failure are visible`) + .toContain(`human notice ${outcome} to ${destination}`); + expect(repo.getById(row.qitemId)!.waiting!.nextBackstop).toMatchObject({ owner: current.destinationSession, + mechanism: "queue-recovery:delegated", recovery: { qitemId: current.qitemId, state: current.state } }); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + expect(repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId)) + .toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null }); + const history = repo.getById(prior.qitemId)!; + if (history.body !== prior.body) expect(history.body).toContain(prior.body); + expect(repo.listTransitions(prior.qitemId).slice(0, originalHistory.length)).toEqual(originalHistory); + } + console.info("PR2586_DELIVERY_TRANSITION", { destination, interruption, outcome, priorOwner: prior.destinationSession, + currentOwner: current!.destinationSession, currentSummary: current!.summary, currentState: current!.state }); + // Only a real current-episode post ends the notice; registration and stale receipts did not. + await receipt(row.qitemId, "posted", key); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted"); + const posted = await sweep(); + expect(posted.outcome, posted.error).not.toBe("failed"); + expect(recoveriesFor(row.qitemId).filter((item) => ["pending", "in-progress", "blocked"].includes(item.state))).toEqual([]); + expect(repo.getById(prior.qitemId)).toMatchObject({ state: "done" }); + expect(repo.listTransitions(prior.qitemId).at(-1)).toMatchObject({ state: "done", + transitionNote: expect.stringContaining("current human ask posted") }); + expect(repo.getById(row.qitemId)).toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null }); + }); + + it.each([ + { origin: "sweep", state: "pending" }, + { origin: "sweep", state: "in-progress" }, + { origin: "sweep", state: "blocked" }, + { origin: "authored", state: "pending" }, + { origin: "authored", state: "in-progress" }, + { origin: "authored", state: "blocked" }, + ] as const)("PR2586 preserves $origin recovery custody in $state during the delivery transition", async ({ origin, state }) => { + bindAskingAgentPane(); + const row = await repo.create({ sourceSession: "asker@r", destinationSession: OWNER.address, + body: "Build status", summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false }); + age(row.qitemId); + registryAvailable = false; + if (origin === "authored") await repo.create({ sourceSession: "asker@r", destinationSession: "recovery-owner@r", + body: "Keep this investigation and its author evidence.", summary: "Investigate delivery", nudge: false, + tags: [`recovery-for:${row.qitemId}`] }); + const initial = await sweep(); + expect(initial.outcome, initial.error).not.toBe("failed"); + expect(recoveriesFor(row.qitemId)).toHaveLength(1); + const prior = recoveriesFor(row.qitemId)[0]!; + if (origin === "sweep") expect(prior.tags).toContain(`stuck-sweep:unclaimed-obligation:${row.qitemId}`); + else expect(prior.tags).not.toContain("stuck-sweep-finding"); + if (state !== "pending") repo.claim({ qitemId: prior.qitemId, destinationSession: prior.destinationSession }); + if (state === "blocked") await repo.update({ qitemId: prior.qitemId, actorSession: prior.destinationSession, + state: "blocked", blockedOn: "external:delivery-investigation", transitionNote: "Waiting for the transport investigation" }); + const owned = repo.getById(prior.qitemId)!; + expect(owned.state).toBe(state); + const history = repo.listTransitions(prior.qitemId); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + registryAvailable = true; + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("never-posted"); + for (const pass of [1, 2]) { + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + const open = recoveriesFor(row.qitemId).filter((item) => ["pending", "in-progress", "blocked"].includes(item.state)); + expect(open, `custody pass ${pass}`).toHaveLength(1); + expect(open[0]).toMatchObject({ qitemId: owned.qitemId, sourceSession: owned.sourceSession, + destinationSession: owned.destinationSession, state: owned.state, claimedAt: owned.claimedAt, blockedOn: owned.blockedOn }); + if (origin === "authored") expect(open[0]).toMatchObject({ body: owned.body, summary: owned.summary }); + else if (open[0]!.body !== owned.body) expect(open[0]!.body).toContain(owned.body); + expect(repo.listTransitions(prior.qitemId).slice(0, history.length)).toEqual(history); + expect(repo.getById(row.qitemId)!.waiting!.nextBackstop).toMatchObject({ owner: owned.destinationSession, + recovery: { qitemId: owned.qitemId, state: owned.state } }); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + } + }); + + it.each([ + { destination: "human-reader@external", delivery: "feed-only" }, + { destination: "human-reader@kernel", delivery: "feed-only" }, + { destination: "human-owner@external", delivery: "posted" }, + { destination: "human-owner@kernel", delivery: "posted" }, + ])("PR5-F2 shows $delivery informational correspondence at $destination without claim or decision promises", async ({ destination, delivery }) => { + if (destination.endsWith("@kernel")) bindHumanPane(destination); + const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination, + body: "Build finished; no decision requested.", summary: "Build status", evidenceRef: "https://example.test/build", + humanIntent: "update", nudge: false }); + if (delivery === "posted") receipt(row.qitemId, "posted"); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome ?? null).toBe(delivery === "posted" ? "posted" : null); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(stored).toBeDefined(); + for (const phase of ["before sweep", "after sweep"]) { + if (phase === "after sweep") { + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + } + const beforeRead = db.serialize(); + const shown = repo.getById(row.qitemId)!; + const listed = repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId)!; + for (const [surface, view] of [["show", shown.waiting], ["list", listed.waiting], ["waitingView", repo.waitingView(row.qitemId)]] as const) { + expect(view, `${phase} ${surface}`).toBeDefined(); + expect.soft(view!.nextBackstop, `F2: ${phase} ${surface} has no claim deadline`) + .toMatchObject({ dueAt: null, intervalSeconds: null }); + expect.soft(view!.nextBackstop.mechanism, `F2: ${phase} ${surface} is correspondence`) + .not.toMatch(/unclaimed|decision/i); + expect.soft(view!.laterBackstop?.mechanism ?? "", `F2: ${phase} ${surface} has no later claim promise`).not.toMatch(/unclaimed|decision/i); + } + expect(db.serialize()).toEqual(beforeRead); + expect(shown).toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null }); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + expect(await findingsFor(row.qitemId)).toEqual([]); + } + }); + + it("keeps a registered informational update visibly held by its canonical seat's typing guard", async () => { + bindHumanPane("human-reader@kernel"); + const row = await repo.create({ sourceSession: "asker@r", destinationSession: "human-reader@kernel", + body: "Build status", summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false }); + repo.recordNudgeAttempt(row.qitemId, "retained:typing_guard"); + age(row.qitemId); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(repo.getById(row.qitemId)!.waiting!.nextBackstop).toMatchObject({ owner: "human-reader@kernel", held: true, + mechanism: "held by the seat's typing guard", dueAt: null }); + await sweep(); + expect(await findingsFor(row.qitemId)).toEqual([]); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + }); + + it.each(["decision", undefined] as const)("keeps an unclaimed human request eligible with intent %s", async (humanIntent) => { + const row = await repo.create({ sourceSession: "asker@r", destinationSession: READER.address, + body: "Choose option A or B.", summary: "A or B?", evidenceRef: "https://example.test/decision", + humanIntent, humanAlert: true, nudge: false }); + age(row.qitemId); + const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(before).toBeDefined(); + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: READER.address, + tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) })]); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before); + }); + + it.each([false, true])("keeps an agent obligation eligible (rerouted informational intent: %s)", async (rerouted) => { + const row = rerouted + ? await repo.create({ sourceSession: "asker@r", destinationSession: READER.address, body: "Build status", + summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false }) + : await repo.create({ sourceSession: "asker@r", destinationSession: "worker@r", body: "Review the build", nudge: false }); + // Direct creates reject humanIntent on agents. Exercise the supported reroute that retains the original intent. + if (rerouted) repo.routeToFallback(row.qitemId, "worker@r", "Agent follow-up required"); + age(row.qitemId); + const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(before).toBeDefined(); + expect(repo.getById(row.qitemId)).toMatchObject({ destinationSession: "worker@r", humanIntent: rerouted ? "update" : null }); + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: "worker@r", + tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) })]); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before); + }); + + it("does not exempt an update merely because its former human address still looks human", async () => { + const row = await repo.create({ sourceSession: "asker@r", destinationSession: READER.address, body: "Build status", + summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false }); + registeredHumans = registeredHumans.filter((person) => person.entityId !== READER.entityId); + age(row.qitemId); + const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(before).toBeDefined(); + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: READER.address, + tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) })]); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before); + }); + + it.each(["never-posted", "transport-failed"] as const)("still reports a %s informational human notice to its asking agent", async (outcome) => { + const row = await repo.create({ sourceSession: "asker@r", destinationSession: OWNER.address, body: "Build status", + summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false }); + if (outcome === "transport-failed") await repo.update({ qitemId: row.qitemId, actorSession: "daemon@kernel", + transitionNote: `slack-owner-notification-transport-failed notification_key=${currentKey(row.qitemId)} level=NOTICE detail=fixture-http-500` }); + // Five minutes exercises delivery recovery without introducing the sixty-minute unclaimed condition. + age(row.qitemId, 5); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome); + const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + expect(before).toBeDefined(); + const result = await sweep(); + expect(result.outcome, result.error).not.toBe("failed"); + expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: "asker@r", + tags: expect.arrayContaining([`stuck-sweep:undelivered-human-notice:${row.qitemId}`]), + body: expect.stringContaining(`human notice ${outcome}`) })]); + const [recovery] = await findingsFor(row.qitemId); + const view = repo.getById(row.qitemId)!.waiting!; + expect(view.nextBackstop).toMatchObject({ owner: "asker@r", mechanism: "queue-recovery:delegated", + recovery: { qitemId: recovery!.qitemId, state: "pending" } }); + expect.soft(view.laterBackstop?.mechanism ?? "", "F2: delivery recovery must not retain a later unclaimed promise").not.toContain("unclaimed"); + await sweep(); + expect(repo.getById(recovery!.qitemId)?.state).toBe("pending"); + expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before); + }); + it("a row whose wake the seat's level or hold held isn't reported as unclaimed: it's held at the seat, not stuck", async () => { const heldRow = await repo.create({ sourceSession: "asker@r", destinationSession: "worker@r", body: "held at a quiet seat", nudge: false }); const draftRow = await repo.create({ sourceSession: "asker@r", destinationSession: "worker@r", body: "held by a draft", nudge: false }); diff --git a/packages/daemon/test/quiet-hold-custody.test.ts b/packages/daemon/test/quiet-hold-custody.test.ts index f2880cfb49..5b467e1615 100644 --- a/packages/daemon/test/quiet-hold-custody.test.ts +++ b/packages/daemon/test/quiet-hold-custody.test.ts @@ -77,15 +77,17 @@ describe("custody of asks held for quiet hours", () => { return w; } let n = 0; - const ask = () => repo.create({ sourceSession: "asker@r", destinationSession: "person@external", body: `Question ${++n}: pick one`, - summary: `Pick ${n}`, evidenceRef: `/proof/${n}.md`, humanIntent: "decision", humanAlert: true, nudge: false } as never) as Promise; + const ask = (humanIntent: "decision" | "update" = "decision") => repo.create({ sourceSession: "asker@r", destinationSession: "person@external", + body: humanIntent === "decision" ? `Question ${++n}: pick one` : `Build ${++n} finished; no decision needed`, + summary: humanIntent === "decision" ? `Pick ${n}` : `Build ${n} finished`, evidenceRef: `/proof/${n}.md`, + humanIntent, humanAlert: humanIntent === "decision", nudge: false } as never) as Promise; const alertFor = async (row: QueueItem) => (await makeQueuePorts(repo, { loadHumanRegistry: registry } as never).listHumanAlerts({ minimumLevel: "RECORD" })).find((a) => a.qitemId === row.qitemId)!; const quietOn = () => { const now = new Date(); quiet.setUntil("person", { until: new Date(now.getTime() + 24 * 3_600_000), timezone: "UTC", setBy: "t", now }); }; const releaseJobs = () => new WatchdogJobsRepository(db).listActive().filter((j) => j.policy === QUIET_RELEASE_POLICY); - async function held(): Promise { + async function held(humanIntent: "decision" | "update" = "decision"): Promise { quietOn(); - const row = await ask(); + const row = await ask(humanIntent); const w = wire(); const alert = await alertFor(row); w.dispatcher.dispatch(OUTBOUND_OP, alert.destinationSession!, toPayload(alert)); @@ -106,12 +108,15 @@ describe("custody of asks held for quiet hours", () => { resolveOrchestrator: () => "orch@r", attemptWake: async () => "verified", log: () => {} }); const ladderTouched = (row: QueueItem) => repo.listTransitions(row.qitemId).some((t) => t.actorSession === "wake-ladder@system"); - it("a healthy hold aged past the thresholds is held: no stuck finding, and the waiting view says held by quiet hours until the boundary", async () => { - const row = await held(); + it.each(["decision", "update"] as const)("a healthy %s hold aged past the thresholds is held: no stuck finding, and the waiting view says held by quiet hours until the boundary", async (humanIntent) => { + const row = await held(humanIntent); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); expect(findingFor(await sweep(), row)).toEqual([]); const view = repo.waitingView(row.qitemId)!; expect(view.nextBackstop).toMatchObject({ mechanism: expect.stringMatching(/^quiet-release:/), dueAt: quiet.quietStateFor("person", new Date()).nextBoundary!.at }); expect(String((view.nextBackstop as { note?: string }).note)).toMatch(/^Held by quiet hours until /); + expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", humanIntent, claimedAt: null }); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); }); it("a healthy hold isn't counted as picked up, and the wake ladder leaves it to its release job", async () => { @@ -146,17 +151,50 @@ describe("custody of asks held for quiet hours", () => { } }); - it("a hold whose release job was stopped from outside (the only way one loses its job) is found, by the sweep and by the ladder", async () => { - const row = await held(); + it.each(["decision", "update"] as const)("a %s hold whose release job was stopped from outside (the only way one loses its job) is found, by the sweep and by the ladder", async (humanIntent) => { + const row = await held(humanIntent); asBaton(row); new WatchdogJobsRepository(db).markTerminal(releaseJobs()[0]!.jobId, "test: the job went away"); + expect(quietHoldCustody(db, row.qitemId)).toMatchObject({ state: "orphaned" }); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); const r = await sweep(); - expect(findingFor(r, row).length).toBeGreaterThan(0); - expect(findingWhy(row)).toMatch(/held for quiet hours .* no release job is live/); + expect(r.outcome, r.error).not.toBe("failed"); + expect.soft(findingFor(r, row).length).toBeGreaterThan(0); + expect.soft(findingWhy(row)).toMatch(/held for quiet hours .* no release job is live/); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + expect(posts).toHaveLength(0); await ladder(); expect(ladderTouched(row), "an orphaned hold isn't pickup, so the ladder takes it up").toBe(true); }); + it("an orphaned update hold keeps its existing unclaimed finding open without informational resolution", async () => { + const row = await held("update"); + new WatchdogJobsRepository(db).markTerminal(releaseJobs()[0]!.jobId, "test: the job went away"); + expect(quietHoldCustody(db, row.qitemId)).toMatchObject({ state: "orphaned" }); + expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull(); + const finding = await repo.create({ sourceSession: "asker@r", destinationSession: "orch@r", body: "prior unclaimed hold evidence", nudge: false, + tags: ["stuck-sweep-finding", `stuck-sweep:unclaimed-obligation:${row.qitemId}`, `recovery-for:${row.qitemId}`] }); + expect(finding.state).toBe("pending"); + const history = repo.listTransitions(finding.qitemId); + const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId); + // The finding predates a larger reporting window; falling outside it isn't resolution. + const now = LATER(); + expect(Date.parse(row.tsCreated)).toBeGreaterThan(now.getTime() - 180 * 60_000); + const r = await runStuckSweep({ db, queueRepo: repo, unclaimedAgeMinutes: 180, now, + resolveOrchestrator: () => "orch@r", log: () => {} }); + expect(r.outcome, r.error).not.toBe("failed"); + expect(findingFor(r, row).filter((f) => f.action !== "closed")).toEqual([]); + expect.soft(repo.getById(finding.qitemId)?.state).toBe("pending"); + expect.soft(repo.listTransitions(finding.qitemId).slice(history.length) + .some((t) => t.transitionNote?.includes("informational update to a registered human requires no claim"))).toBe(false); + expect(repo.getById(finding.qitemId)?.body).toContain(finding.body); + expect(repo.listTransitions(finding.qitemId).slice(0, history.length)).toEqual(history); + expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored); + expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", humanIntent: "update", claimedAt: null }); + expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull(); + expect(posts).toHaveLength(0); + }); + it("a release dispatched but never posted whose job is gone is orphaned and found, not left waiting for transport", async () => { const row = await held(); quiet.end("person", { endedBy: "t", now: new Date() });