fix(queue): distinguish input waits from stalled claims

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 <source-qitem-id>`
 (`queue-stuck-sweep.ts:605`) pointing to the underlying durable work row. This

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,

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;

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,22 +33,33 @@ 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
  * idx_queue_items_recovery_tags' WHERE clause (migration 099) byte for byte, which is what lets the
  * 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;

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

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<StuckSweepRes
       if (existing && existingIsOpen) {
         // Age is derived at read time; an unchanged scan is not a transition.
         findings.push({ kind: c.kind, qitemId: c.row.qitemId, findingQitemId: existing.qitem_id, action: "refreshed" });
-      } else if (!existing || evidenceIsNewer(c.evidenceAt, existing.ts_updated)) {
+      } else if (!existing || evidenceIsNewer(c.evidenceAt, existing.ts_updated) || inputWaitResolution(deps.db, existing.qitem_id) !== null) {
         const created = await deps.queueRepo.create({
           qitemId: recoveryId(deps.db, c.row.qitemId),
           // The detector is machinery, not a seat: the obligation's own creator is the
@@ -642,12 +642,15 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
       const dedupTag = tags.find((t) => 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({

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

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,