diff --git a/docs/as-built/architecture/coordination-primitive.md b/docs/as-built/architecture/coordination-primitive.md index 44d18e3281..5929b2b1ec 100644 --- a/docs/as-built/architecture/coordination-primitive.md +++ b/docs/as-built/architecture/coordination-primitive.md @@ -146,7 +146,7 @@ that gateway side too. For the pane side, read: `indeterminate:…` reads `claimed` (`presentedNudgeResult`). Unclaiming shows the stored result again. - `claimedAt`: when the destination claimed the row. Closing doesn't clear it. -- `pickup`: the derived pickup receipt (`unclaimed`, `working`, +- `pickup`: the derived pickup receipt (`unclaimed`, `working`, `needs-input`, `stalled-after-claim`, `parked`, `terminal`), computed at read time. A claim isn't proof that the body was read or that the terminal render was @@ -159,6 +159,20 @@ has claimed the item (claimed)."; for any claimed item it adds "A claim shows the seat took the item; it does not prove the body was read." (`terminalDeliveryResult`, `packages/cli/src/commands/queue.ts`). +Pickup receipts are derived by `queue-pickup.ts` for queue reads, the pickup +view and the execution view. A claimed item in progress whose owner's activity +oracle reports outstanding input reads `needs-input`, with the reason in +`pickup.evidence`, regardless of working/idle activity or claim age. This does +not change the stored queue state or count as task progress. The pickup stall +check resumes when the oracle clears the input; unanswered prompts do not age +out. The waiting view suspends the pickup backstop while retaining any explicit +closure deadline, and attention surfaces still show the outstanding prompt. +Unclaimed, blocked and terminal items retain their existing receipt semantics. +The sweep retires an existing stall finding with a specific input-wait receipt; +that automatic closure does not suppress a later genuine stall. Manual recovery +dispositions remain authoritative. New findings use the retirement transition +as part of their deterministic ID, without adding progress to the work item. + The standing detector in `queue-stuck-sweep.ts` creates findings through the queue repository, with `evidenceRef: rig queue show ` (`queue-stuck-sweep.ts:605`) pointing to the underlying durable work row. This diff --git a/packages/daemon/src/domain/execution-view.ts b/packages/daemon/src/domain/execution-view.ts index e0d34116e9..8b404df326 100644 --- a/packages/daemon/src/domain/execution-view.ts +++ b/packages/daemon/src/domain/execution-view.ts @@ -834,6 +834,7 @@ export async function buildExecutionView(deps: ExecutionViewDeps, opts?: { missi state: r.state, lastMeaningfulAt: lastMeaningfulTransition(deps.db, r.qitem_id)?.at, activity: arbitrated?.activity, needsInput: arbitrated?.needsInput.count, + needsInputReason: arbitrated?.needsInput.reason, claimedAt: r.claimed_at, lastHeartbeat: r.last_heartbeat, postClaimMotionCount: Number(r.post_claim_motion ?? 0), @@ -1167,6 +1168,7 @@ export async function buildExecutionView(deps: ExecutionViewDeps, opts?: { missi const pickup = derivePickup({ lastMeaningfulAt: lastMeaningfulTransition(deps.db, r.qitem_id)?.at, activity: activity?.activity, needsInput: activity?.needsInput.count, + needsInputReason: activity?.needsInput.reason, state: r.state, claimedAt: r.claimed_at, lastHeartbeat: r.last_heartbeat, diff --git a/packages/daemon/src/domain/queue-pickup.ts b/packages/daemon/src/domain/queue-pickup.ts index 23ec14fffb..a2017382f7 100644 --- a/packages/daemon/src/domain/queue-pickup.ts +++ b/packages/daemon/src/domain/queue-pickup.ts @@ -6,6 +6,7 @@ // (S02 owns the standing sweep; this module exports the INPUT contract it consumes). // // Working activity is positive liveness evidence, not proof of task progress. +// Outstanding input is an explicit wait, even when working/idle activity is unknown. // Otherwise the grace follows the latest meaningful queue change; an old note // cannot keep the row working forever. Stalled-after-claim names this evidence // without inferring idle/dead from age. Blocked remains parked; wake health is @@ -23,8 +24,8 @@ export const PICKUP_STALL_THRESHOLD_KEY = "queue.pickup_stall_threshold_minutes" export const DEFAULT_PICKUP_STALL_THRESHOLD_MINUTES = 3; export interface PickupReceipt { - state: "unclaimed" | "working" | "stalled-after-claim" | "parked" | "terminal"; - /** Present iff stalled or terminal: the named evidence replacing the manual cross-surface join. */ + state: "unclaimed" | "working" | "needs-input" | "stalled-after-claim" | "parked" | "terminal"; + /** Present when waiting for input, stalled or terminal: the evidence behind the receipt. */ evidence?: string; } @@ -55,6 +56,7 @@ export interface PickupFacts { lastMeaningfulAt?: string; activity?: string; needsInput?: number; + needsInputReason?: string | null; now?: Date; thresholdMinutes?: number; } @@ -67,6 +69,9 @@ export function derivePickup(facts: PickupFacts): PickupReceipt { return { state: "terminal", evidence: `closed (${facts.state})` }; } if (!facts.claimedAt) return { state: "unclaimed" }; + if ((facts.needsInput ?? 0) > 0) { + return { state: "needs-input", evidence: facts.needsInputReason ? `owner needs input: ${facts.needsInputReason}` : "owner needs input" }; + } const now = facts.now ?? new Date(); const claimedMs = Date.parse(facts.claimedAt); // Keep this null arm for the 0.5.7 mechanized-pull turn-end hook that knows the in-flight row; diff --git a/packages/daemon/src/domain/queue-recovery.ts b/packages/daemon/src/domain/queue-recovery.ts index 6260d38597..54e028b27c 100644 --- a/packages/daemon/src/domain/queue-recovery.ts +++ b/packages/daemon/src/domain/queue-recovery.ts @@ -5,6 +5,24 @@ import { lastMeaningfulTransition } from "./queue-waiting.js"; /** Common provenance on existing diagnostic queue rows, never a second ledger. * Both deadline and delivery detectors consult the same recovery disposition. */ export const recoveryTag = (qitemId: string): string => `recovery-for:${qitemId}`; +/** Machine-authored retirement receipt: this false stall is resolved, not the work. */ +export const INPUT_WAIT_RESOLVED_NOTE = "stuck-sweep resolved: owner needs input"; + +/** Only the latest transition can retire a finding this way. A subsequent manual + * disposition still owns recovery. Read the archive too: human waits can outlast retention. */ +export function inputWaitResolution(db: Database.Database, qitemId: string): number | null { + const sources = db.prepare("SELECT name FROM sqlite_master WHERE name IN ('queue_transitions', 'queue_transitions_archive')").all() as Array<{ name: string }>; + if (sources.length === 0) return null; + const fields = "transition_id, state, transition_note, actor_session"; + const transition = db.prepare(`SELECT t.*, q.source_session FROM ( + ${sources.map(({ name }) => `SELECT ${fields} FROM ${name} WHERE qitem_id = ?`).join(" UNION ALL ")} + ) t JOIN queue_items q ON q.qitem_id = ? AND q.state = 'done' ORDER BY transition_id DESC LIMIT 1`) + .get(...sources.map(() => qitemId), qitemId) as { + transition_id: number; state: string; transition_note: string | null; actor_session: string; source_session: string; + } | undefined; + return transition?.state === "done" && transition.actor_session === transition.source_session + && transition.transition_note === INPUT_WAIT_RESOLVED_NOTE ? transition.transition_id : null; +} function failedAttempt(db: Database.Database, qitemId: string): string | null { // A typing-guard refusal (retained:) is newer evidence about the original // obligation exactly like a failed nudge: a wake WAS offered after the @@ -15,7 +33,13 @@ function failedAttempt(db: Database.Database, qitemId: string): string | null { return (db.prepare("SELECT last_nudge_attempt AS at FROM queue_items WHERE qitem_id = ? AND (last_nudge_result LIKE 'failed:%' OR last_nudge_result LIKE 'retained:%')").get(qitemId) as { at: string | null } | undefined)?.at ?? null; } export function recoveryId(db: Database.Database, qitemId: string): string { - return `qitem-recovery-${createHash("sha256").update(JSON.stringify([qitemId, lastMeaningfulTransition(db, qitemId)?.id ?? null, failedAttempt(db, qitemId)])).digest("hex").slice(0, 16)}`; + const previous = latestRecovery(db, qitemId, true); + const retired = previous ? inputWaitResolution(db, previous.qitem_id) : null; + const evidence = [qitemId, lastMeaningfulTransition(db, qitemId)?.id ?? null, failedAttempt(db, qitemId)]; + // A later stall must get a new, deterministic ID even when answering the prompt + // wrote no queue transition. Concurrent sweeps use the same retirement receipt. + if (retired !== null) evidence.push(`input-wait-resolved:${retired}`); + return `qitem-recovery-${createHash("sha256").update(JSON.stringify(evidence)).digest("hex").slice(0, 16)}`; } /** Rows whose tag text could hold a recovery tag: valid JSON stores a string verbatim or with * backslash escapes, so this superset loses no exact match. It repeats the first group of @@ -23,14 +47,19 @@ export function recoveryId(db: Database.Database, qitemId: string): string { * planner read that small index instead of scanning and parsing every row. */ const RECOVERY_TAG_ROWS = `(tags LIKE '%"recovery-for:%' OR tags LIKE '%\\%')`; const RECOVERY_ORDER = "ORDER BY CASE WHEN state IN ('pending','in-progress','blocked') THEN 0 ELSE 1 END, ts_updated DESC, qitem_id DESC LIMIT 1"; -export function findQueueRecovery(db: Database.Database, qitemId: string): { qitemId: string; state: string } | null { +function latestRecovery(db: Database.Database, qitemId: string, terminalOnly = false) { const tag = recoveryTag(qitemId); - const row = db.prepare(`SELECT qitem_id, state, ts_updated FROM queue_items + return db.prepare(`SELECT qitem_id, state, ts_updated FROM queue_items WHERE ${RECOVERY_TAG_ROWS} AND json_valid(tags) AND EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?) + ${terminalOnly ? "AND state NOT IN ('pending','in-progress','blocked')" : ""} ${RECOVERY_ORDER}`) .get(tag) as { qitem_id: string; state: string; ts_updated: string } | undefined; +} +export function findQueueRecovery(db: Database.Database, qitemId: string): { qitemId: string; state: string } | null { + const row = latestRecovery(db, qitemId); if (!row) return null; if (!["pending", "in-progress", "blocked"].includes(row.state)) { + if (inputWaitResolution(db, row.qitem_id) !== null) return null; const changed = lastMeaningfulTransition(db, qitemId); const disposition = lastMeaningfulTransition(db, row.qitem_id); if ((changed && disposition && changed.id > disposition.id) || (changed && Date.parse(changed.at) > Date.parse(row.ts_updated)) || Date.parse(failedAttempt(db, qitemId) ?? "") > Date.parse(row.ts_updated)) return null; diff --git a/packages/daemon/src/domain/queue-repository.ts b/packages/daemon/src/domain/queue-repository.ts index e44c682dda..115659cee1 100644 --- a/packages/daemon/src/domain/queue-repository.ts +++ b/packages/daemon/src/domain/queue-repository.ts @@ -162,9 +162,9 @@ export interface QueueItem { deliveryFailureDetail?: string; tags: string[] | null; blockedOn: string | null; - /** S04 — the DERIVED pickup receipt (unclaimed/working/stalled-after-claim/parked/terminal). + /** S04 — the DERIVED pickup receipt (unclaimed/working/needs-input/stalled-after-claim/parked/terminal). * Never stored: computed at projection time from state + claimed_at + the transition log - * + heartbeat; a closed row is terminal, never stalled. */ + * + heartbeat and owner activity; a closed row is terminal, never stalled. */ pickup?: PickupReceipt; waiting?: WaitingView; handedOffTo: string | null; @@ -4048,6 +4048,7 @@ export class QueueRepository { lastMeaningfulAt: meaningful?.at, activity: activity?.activity, needsInput: activity?.needsInput.count, + needsInputReason: activity?.needsInput.reason, claimedAt: row.claimed_at, lastHeartbeat: row.last_heartbeat, postClaimMotionCount: 0, // this reader supplies the current meaningful timestamp diff --git a/packages/daemon/src/domain/queue-stuck-sweep.ts b/packages/daemon/src/domain/queue-stuck-sweep.ts index 53b25b7ab8..4f4c4ae7b0 100644 --- a/packages/daemon/src/domain/queue-stuck-sweep.ts +++ b/packages/daemon/src/domain/queue-stuck-sweep.ts @@ -1,4 +1,4 @@ -import { findQueueRecovery, recoveryId, recoveryTag } from "./queue-recovery.js"; +import { findQueueRecovery, inputWaitResolution, INPUT_WAIT_RESOLVED_NOTE, recoveryId, recoveryTag } from "./queue-recovery.js"; import { queueWaitNotice } from "./queue-wait-backoff.js"; import { lastMeaningfulTransition, pendingSince } from "./queue-waiting.js"; import { resolvePickupThresholdMinutes } from "./queue-pickup.js"; @@ -607,7 +607,7 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise t.startsWith("stuck-sweep:")); if (!dedupTag || liveDedupTags.has(dedupTag)) continue; const [, kind, stuckId] = dedupTag.match(/^stuck-sweep:([a-z-]+):(.+)$/) ?? []; + const waitingForInput = kind === "stalled-after-claim" && stuckId + && deps.queueRepo.getById(stuckId)?.pickup?.state === "needs-input"; await deps.queueRepo.update({ qitemId: f.qitem_id, actorSession: f.source_session, state: "done", closureReason: "no-follow-on", - transitionNote: `stuck-sweep resolved: ${kind ?? "finding"} on ${stuckId ?? "row"} no longer detected`, + transitionNote: waitingForInput ? INPUT_WAIT_RESOLVED_NOTE + : `stuck-sweep resolved: ${kind ?? "finding"} on ${stuckId ?? "row"} no longer detected`, }); if (kind && stuckId) { findings.push({ diff --git a/packages/daemon/src/domain/queue-waiting.ts b/packages/daemon/src/domain/queue-waiting.ts index 135f6a1d0e..e240dfeda2 100644 --- a/packages/daemon/src/domain/queue-waiting.ts +++ b/packages/daemon/src/domain/queue-waiting.ts @@ -63,9 +63,9 @@ export interface WaitingView { } export function readWaitingView(db: Database.Database, id: string, readActivity?: WaitingActivityReader): WaitingView | null { - const row = db.prepare(`SELECT qitem_id, source_session, destination_session, state, blocked_on, closure_required_at + const row = db.prepare(`SELECT qitem_id, source_session, destination_session, state, claimed_at, blocked_on, closure_required_at FROM queue_items WHERE qitem_id = ?`).get(id) as { - qitem_id: string; source_session: string; destination_session: string; state: string; blocked_on: string | null; closure_required_at: string | null; + qitem_id: string; source_session: string; destination_session: string; state: string; claimed_at: string | null; blocked_on: string | null; closure_required_at: string | null; } | undefined; if (!row) return null; const blocker = row.blocked_on?.startsWith("qitem-") ? db.prepare( @@ -93,6 +93,16 @@ export function readWaitingView(db: Database.Database, id: string, readActivity? view.nextBackstop.dueAt = base ? new Date(Date.parse(base) + delay * 1000).toISOString() : null; view.nextBackstop.mechanism = row.state === "pending" ? "queue-stuck-sweep:unclaimed" : "queue-stuck-sweep:pickup (activity checked at detection)"; } + if (row.state === "in-progress" && row.claimed_at && (observed?.needsInput.count ?? 0) > 0) { + // Input waits suspend the pickup-age finding, never an explicit closure deadline. + view.nextBackstop = { + owner: row.destination_session, + mechanism: row.closure_required_at ? "queue-stuck-sweep:overdue" : "none (owner needs input)", + dueAt: row.closure_required_at, + intervalSeconds: row.closure_required_at ? sweepInterval : null, + note: observed?.needsInput.reason ?? "Waiting for owner input to clear.", + }; + } if (row.state === "blocked") { const hasTimers = db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transition_wakes'").get(); const timer = hasTimers ? db.prepare(`SELECT j.job_id, j.policy, j.last_evaluation_at, j.registered_at, j.interval_seconds, j.scan_interval_seconds, j.spec_yaml diff --git a/packages/daemon/src/domain/view-projector.ts b/packages/daemon/src/domain/view-projector.ts index 3c9bb5c801..13d3b7a635 100644 --- a/packages/daemon/src/domain/view-projector.ts +++ b/packages/daemon/src/domain/view-projector.ts @@ -179,7 +179,7 @@ export class ViewProjector { // S04 — the pickup lens post-processes through the ONE shared derivation rule // (derivePickup), never a second SQL copy of it: claimed live rows, oldest claim first, - // each carrying pickup_state and (when stalled) the named pickup_evidence. + // each carrying pickup_state and the evidence for a stall or input wait. if (name === "pickup") { const rows = this.db .prepare( @@ -202,6 +202,7 @@ export class ViewProjector { const receipt = derivePickup({ lastMeaningfulAt: lastMeaningfulTransition(this.db, String(r.qitem_id))?.at, activity: activity?.activity, needsInput: activity?.needsInput.count, + needsInputReason: activity?.needsInput.reason, state: String(r.state), claimedAt: (r.claimed_at as string | null) ?? null, lastHeartbeat: (r.last_heartbeat as string | null) ?? null, diff --git a/packages/daemon/test/queue-input-wait.test.ts b/packages/daemon/test/queue-input-wait.test.ts new file mode 100644 index 0000000000..7055368953 --- /dev/null +++ b/packages/daemon/test/queue-input-wait.test.ts @@ -0,0 +1,241 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import Database from "better-sqlite3"; +import { readClaudeSelfReportEvidence } from "../src/adapters/claude-code-adapter.js"; +import { ALL_MIGRATIONS } from "../src/db/all-migrations.js"; +import { migrate } from "../src/db/migrate.js"; +import { CLAUDE_ACTIVITY_RUNG_INVENTORY, CODEX_ACTIVITY_RUNG_INVENTORY } from "../src/domain/activity-taxonomy.js"; +import { AgentActivityStore } from "../src/domain/agent-activity-store.js"; +import { EventBus } from "../src/domain/event-bus.js"; +import { buildExecutionView } from "../src/domain/execution-view.js"; +import { PsProjectionService } from "../src/domain/ps-projection.js"; +import { derivePickup } from "../src/domain/queue-pickup.js"; +import { QueueRepository } from "../src/domain/queue-repository.js"; +import { archiveAgedTerminalTransitions } from "../src/domain/queue-retention.js"; +import { runStuckSweep, STUCK_SWEEP_FINDING_TAG } from "../src/domain/queue-stuck-sweep.js"; +import { SeatActivityService } from "../src/domain/seat-activity-service.js"; +import { ViewProjector } from "../src/domain/view-projector.js"; +import { evidenceFromHookActivity } from "../src/routes/activity.js"; + +const SESSION = "worker@input-wait"; +const NODE = "node-input-wait"; +const advance = (seconds: number) => vi.setSystemTime(Date.now() + seconds * 1000); + +// The stub script can emit turn boundaries but has no permission/elicitation step. +// Exercise the real hook normalization, activity arbitration, queue projections and +// sweep over SQLite instead, with a clock that can outlast an unanswered prompt. +describe("claimed work waiting on human input (#331)", () => { + let db: Database.Database; + let queue: QueueRepository; + let activity: SeatActivityService; + let hooks: AgentActivityStore; + let projector: ViewProjector; + let ps: PsProjectionService; + let seq: number; + + beforeEach(() => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime("2026-10-06T12:00:00Z"); + vi.stubEnv("OPENRIG_QUEUE_PICKUP_STALL_THRESHOLD_MINUTES", "3"); + db = new Database(":memory:"); + migrate(db, ALL_MIGRATIONS); + db.prepare("INSERT INTO rigs (id, name) VALUES ('rig-input-wait', 'input-wait')").run(); + db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime) VALUES (?, 'rig-input-wait', 'worker', 'claude-code')").run(NODE); + db.prepare("INSERT INTO sessions (id, node_id, session_name, status, startup_status, created_at) VALUES ('session-input-wait', ?, ?, 'running', 'ready', ?)") + .run(NODE, SESSION, new Date().toISOString()); + const bus = new EventBus(db); + queue = new QueueRepository(db, bus, { validateRig: () => true }); + hooks = new AgentActivityStore({ db, eventBus: bus }); + activity = new SeatActivityService({ tmux: { readPaneLastActivity: async () => null }, defaultWindowSeconds: 3 }); + activity.declareRungInventory({ seatNodeId: NODE, sessionName: SESSION }, CLAUDE_ACTIVITY_RUNG_INVENTORY); + queue.attachActivityReader(session => activity.getSeatStateBySession(session)); + projector = new ViewProjector(db, bus); + projector.setExecutionDeps({ db, slicesRoot: () => null, seatActivity: activity }); + ps = new PsProjectionService({ db, seatActivity: activity, agentActivity: hooks }); + seq = 0; + }); + + afterEach(() => { db.close(); vi.useRealTimers(); vi.unstubAllEnvs(); }); + + function hook(hookEvent: string, subtype: string | null = null, runtime = "claude-code") { + const recorded = hooks.recordHookEvent({ runtime, sessionName: SESSION, nodeId: NODE, hookEvent, subtype, occurredAt: new Date().toISOString() }); + if (!recorded.ok) throw new Error(recorded.code); + const evidence = evidenceFromHookActivity({ seatNodeId: NODE, sessionName: SESSION, runtime, activity: recorded.activity, seq: ++seq }); + if (evidence) activity.reportEvidence(evidence); + } + + async function claimed(tags?: string[]) { + const row = await queue.create({ sourceSession: "sender@input-wait", destinationSession: SESSION, body: "Work awaiting approval", nudge: false, tags }); + queue.claim({ qitemId: row.qitemId, destinationSession: SESSION }); + advance(240); + return row.qitemId; + } + + const sweep = () => runStuckSweep({ db, queueRepo: queue, now: new Date(), unclaimedAgeMinutes: 60, resolveOrchestrator: () => null, isRegisteredHost: () => false }); + const findings = () => queue.list({ limit: 100 }).filter(row => row.tags?.includes(STUCK_SWEEP_FINDING_TAG)); + const attention = () => ps.getEntries().find(entry => entry.rigId === "rig-input-wait")!.attentionCount; + + it.each(["working", "idle-at-prompt", "unknown"])("pending input takes precedence over %s activity and old claim age", ownerActivity => { + const old = new Date(Date.now() - 3_600_000).toISOString(); + expect(derivePickup({ state: "in-progress", claimedAt: old, lastMeaningfulAt: old, lastHeartbeat: null, postClaimMotionCount: 0, activity: ownerActivity, needsInput: 1 }).state).toBe("needs-input"); + }); + + it("keeps a long unanswered prompt visible across queue, pickup, execution and attention reads without a stall finding", async () => { + const id = await claimed(["slice:OPR.9.9.331"]); + const transitions = queue.transitionLog.listForQitem(id); + hook("Notification", "permission_prompt"); + advance(24 * 3600); + const receipt = queue.getByIdOrThrow(id).pickup!; + expect(receipt.state).toBe("needs-input"); + const reason = activity.getSeatStateBySession(SESSION)!.needsInput.reason!; + expect(receipt.evidence).toContain(reason); + expect(queue.list({ destinationSession: SESSION }).find(row => row.qitemId === id)?.pickup).toEqual(receipt); + expect(projector.show("pickup").rows.find(row => row.qitem_id === id)).toMatchObject({ pickup_state: "needs-input", pickup_evidence: receipt.evidence }); + const execution = await buildExecutionView({ db, slicesRoot: () => null, seatActivity: activity }); + expect(execution.q1_lanes).toContainEqual(expect.objectContaining({ qitem_id: id, pickup: receipt })); + expect(execution.q5_park).toContainEqual(expect.objectContaining({ qitem_id: id, pickup_state: "needs-input", pickup_evidence: receipt.evidence, park_kind: "indeterminate" })); + expect(queue.waitingView(id)!.nextBackstop).toMatchObject({ dueAt: null, intervalSeconds: null }); + expect(attention()).toBe(1); + expect((await sweep()).outcome).toBe("clean"); + expect(findings()).toHaveLength(0); + expect(queue.transitionLog.listForQitem(id)).toEqual(transitions); + }); + + it("honors a Claude waiting self-report without any lifecycle hook", async () => { + const id = await claimed(); + let status = "waiting"; + const report = () => activity.reportEvidence(readClaudeSelfReportEvidence({ + seatNodeId: NODE, sessionName: SESSION, sessionsDir: "/unused", + read: { listFiles: () => ["worker.json"], readFile: () => JSON.stringify({ name: SESSION, status, statusUpdatedAt: Date.now() }) }, + })!); + report(); + expect(queue.getByIdOrThrow(id).pickup).toEqual({ state: "needs-input", evidence: "owner needs input: dialog owns input" }); + expect((await sweep()).outcome).toBe("clean"); + advance(1); + status = "busy"; + report(); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("working"); + }); + + it("keeps a confirmed Codex request out of pickup stalls until a fresh pane clear", async () => { + const id = await claimed(); + db.prepare("UPDATE nodes SET runtime = 'codex' WHERE id = ?").run(NODE); + let visible = true; + activity = new SeatActivityService({ tmux: { readPaneLastActivity: async () => null }, defaultWindowSeconds: 3, + permissionPromptReader: async () => visible }); + activity.declareRungInventory({ seatNodeId: NODE, sessionName: SESSION }, CODEX_ACTIVITY_RUNG_INVENTORY); + ps = new PsProjectionService({ db, seatActivity: activity, agentActivity: hooks }); + hook("PermissionRequest", null, "codex"); + advance(4); // Codex's confirmation probe waits for automatic review to settle. + await activity.observeSeat(SESSION); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("needs-input"); + expect(attention()).toBe(1); + expect((await sweep()).outcome).toBe("clean"); + + visible = false; + advance(1); + await activity.observeSeat(SESSION); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("stalled-after-claim"); + expect(attention()).toBe(0); + expect((await sweep()).findings).toContainEqual(expect.objectContaining({ kind: "stalled-after-claim", qitemId: id, action: "created" })); + }); + + it.each([false, true])("closes a false stall and detects recurrence without fabricated queue progress (archived=%s)", async archived => { + const id = await claimed(); + expect((await sweep()).findings).toContainEqual(expect.objectContaining({ kind: "stalled-after-claim", qitemId: id, action: "created" })); + const original = findings()[0]!; + const transitions = queue.transitionLog.listForQitem(id); + advance(1); + hook("Notification", "permission_prompt"); + expect((await sweep()).findings).toContainEqual(expect.objectContaining({ findingQitemId: original.qitemId, action: "closed" })); + expect(queue.getByIdOrThrow(original.qitemId).state).toBe("done"); + expect(queue.waitingView(id)!.nextBackstop.dueAt).toBeNull(); + expect(attention()).toBe(1); + + if (archived) { + advance(31 * 24 * 3600); + expect(archiveAgedTerminalTransitions(db, { nowIso: new Date().toISOString() }).archivedRows).toBeGreaterThan(0); + } + advance(1); + hook("UserPromptSubmit"); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("working"); + expect(attention()).toBe(0); + expect((await sweep()).outcome).toBe("clean"); + advance(240); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("stalled-after-claim"); + expect((await sweep()).findings).toContainEqual(expect.objectContaining({ kind: "stalled-after-claim", qitemId: id, action: "created" })); + await Promise.all([sweep(), sweep()]); + expect(findings()).toHaveLength(2); + expect(findings().filter(row => row.state === "pending")).toHaveLength(1); + expect(queue.transitionLog.listForQitem(id)).toEqual(transitions); + + // Another wait/clear cycle cannot resurrect either retired finding ID. + advance(1); + hook("Notification", "permission_prompt"); + await sweep(); + advance(1); + hook("Stop"); + advance(240); + await Promise.all([sweep(), sweep()]); + expect(findings()).toHaveLength(3); + expect(findings().filter(row => row.state === "pending")).toHaveLength(1); + }); + + it("an observed clear without working evidence restores stall detection immediately", async () => { + const id = await claimed(); + hook("Notification", "elicitation_dialog"); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("needs-input"); + expect((await sweep()).outcome).toBe("clean"); + advance(1); + activity.reportEvidence({ seatNodeId: NODE, sessionName: SESSION, rung: "needs-input-chrome", sourceId: "claude:chrome", seq: 1, observedAt: new Date().toISOString(), needsInput: { count: 0, reason: null } }); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("stalled-after-claim"); + expect(attention()).toBe(0); + expect((await sweep()).findings).toContainEqual(expect.objectContaining({ kind: "stalled-after-claim", qitemId: id, action: "created" })); + }); + + it("does not let a pending prompt hide an explicit closure deadline", async () => { + const id = await claimed(); + hook("Notification", "permission_prompt"); + const deadline = new Date(Date.now() - 60_000).toISOString(); + db.prepare("UPDATE queue_items SET closure_required_at = ? WHERE qitem_id = ?").run(deadline, id); + expect(queue.waitingView(id)?.deadlineAt).toBe(deadline); + expect(queue.waitingView(id)?.nextBackstop).toMatchObject({ mechanism: "queue-stuck-sweep:overdue", dueAt: deadline }); + expect((await sweep()).findings).toContainEqual(expect.objectContaining({ kind: "overdue-claim", qitemId: id, action: "created" })); + expect(findings()).toHaveLength(1); + }); + + it.each([false, true])("keeps a manual disposition authoritative across input events (reopened=%s)", async reopened => { + const id = await claimed(); + await sweep(); + const finding = findings()[0]!; + if (reopened) { + advance(1); + hook("Notification", "permission_prompt"); + await sweep(); + await queue.update({ qitemId: finding.qitemId, actorSession: SESSION, state: "pending", reopen: true, transitionNote: "Review this recovery" }); + } + await queue.update({ qitemId: finding.qitemId, actorSession: SESSION, state: "done", closureReason: "no-follow-on", transitionNote: "Reviewed; no further recovery needed" }); + advance(1); + hook("Notification", "permission_prompt"); + await sweep(); + advance(1); + hook("Stop"); + advance(240); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("stalled-after-claim"); + expect((await sweep()).outcome).toBe("clean"); + expect(findings()).toHaveLength(1); + }); + + it("a replacement occupant does not inherit the previous occupant's input wait", async () => { + const id = await claimed(); + hook("Notification", "permission_prompt"); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("needs-input"); + activity.declareOccupantSwap(NODE, "replacement"); + expect(queue.getByIdOrThrow(id).pickup?.state).toBe("stalled-after-claim"); + }); + + it.each(["pending", "blocked", "done", "failed", "denied", "canceled", "handed-off"])("input on the owner does not replace %s obligation semantics", state => { + const old = new Date(Date.now() - 3_600_000).toISOString(); + const receipt = derivePickup({ state, claimedAt: state === "pending" ? null : old, lastHeartbeat: null, postClaimMotionCount: 0, needsInput: 1 }); + expect(receipt.state).toBe(state === "pending" ? "unclaimed" : state === "blocked" ? "parked" : "terminal"); + }); +}); diff --git a/packages/daemon/test/queue-pickup.test.ts b/packages/daemon/test/queue-pickup.test.ts index 74c3979bf7..4e26218459 100644 --- a/packages/daemon/test/queue-pickup.test.ts +++ b/packages/daemon/test/queue-pickup.test.ts @@ -5,6 +5,7 @@ // unclaimed — no claim yet; // working — claimed AND (substantive motion since the claim OR still inside // the threshold — the anti-noise direction); +// needs-input — claimed, with outstanding input reported by the owner oracle; // stalled-after-claim — claimed, past threshold, zero substantive motion; evidence named; // parked — state=blocked only. Wake health is a separate `rig parked` // diagnosis; a wakeless blocked row still projects as parked; @@ -24,7 +25,7 @@ import { SettingsStore } from "../src/domain/user-settings/settings-store.js"; const THRESHOLD_ENV = "OPENRIG_QUEUE_PICKUP_STALL_THRESHOLD_MINUTES"; interface PickupShape { - state: "unclaimed" | "working" | "stalled-after-claim" | "parked" | "terminal"; + state: "unclaimed" | "working" | "needs-input" | "stalled-after-claim" | "parked" | "terminal"; evidence?: string; } const pickupOf = (item: QueueItem): PickupShape | undefined => diff --git a/packages/daemon/test/seat-activity-retirement.test.ts b/packages/daemon/test/seat-activity-retirement.test.ts index 09f46d037a..5a8cd9b82c 100644 --- a/packages/daemon/test/seat-activity-retirement.test.ts +++ b/packages/daemon/test/seat-activity-retirement.test.ts @@ -23,7 +23,7 @@ async function fixture() { CREATE TABLE sessions(id INTEGER, node_id TEXT, session_name TEXT, status TEXT); CREATE TABLE bindings(node_id TEXT, attachment_type TEXT); CREATE TABLE queue_items(qitem_id TEXT PRIMARY KEY, source_session TEXT, destination_session TEXT, - state TEXT, blocked_on TEXT, closure_required_at TEXT, ts_created TEXT);`); + state TEXT, blocked_on TEXT, closure_required_at TEXT, ts_created TEXT, claimed_at TEXT);`); const adapter = new TmuxAdapter(async () => { throw new Error("unexpected shell path"); }, undefined, async argv => tmux(...argv.slice(1))); // A reader clock ten seconds ahead makes genuine native window timestamps idle @@ -40,7 +40,7 @@ async function fixture() { await tmux("new-session", "-d", "-s", session, "sleep 60"); } function waiting(session: string) { - db.prepare("INSERT OR REPLACE INTO queue_items VALUES ('q1', 'sender@fixture', ?, 'pending', NULL, NULL, ?)") + db.prepare("INSERT OR REPLACE INTO queue_items VALUES ('q1', 'sender@fixture', ?, 'pending', NULL, NULL, ?, NULL)") .run(session, new Date().toISOString()); // The startup activity-reader SQL seam resolves the durable node; the real // service and queue projection supply all activity/confidence values.