fix(daemon): the stuck sweep's reason note isn't progress on its recovery

The reason note an open recovery gets when the sweep's reason changes is the
sweep's own bookkeeping, but lastMeaningfulTransition counted it as progress.
The wake ladder reads only the markers after the last progress, so a claimed
recovery whose parked-owner wake failed and whose ladder was exhausted started
retrying again on each reason change.

Add stuck-sweep@system to the machine actors whose same-state transitions
aren't progress. The sweep's only other write as that actor is its resolution
close, a state change, which still counts.

Refs #2533. Found by review-r2.

packages/daemon/src/domain/queue-stuck-sweep.ts

@@ -491,6 +491,31 @@ export function custodyHeadline(rowId: string, targets: string[], ageMinutes: nu
     : `Custody not checked: no successor of ${rowId} in this store (${ageMinutes} min)`;
 }
 
+function findingHeadline(c: Candidate): string {
+  return c.verificationTargets
+    ? custodyHeadline(c.row.qitemId, c.verificationTargets, c.ageMinutes)
+    : `Stuck sweep: ${c.kind} on ${c.row.qitemId} (${c.ageMinutes} min)`;
+}
+
+const REASON_NOTE = "stuck-sweep reason: ";
+
+/** An open recovery opened for another finding kind keeps custody of the row, but its headline and body give that
+ *  kind's reason. When the reason last recorded on it isn't this candidate's, record this one: its headline, and its
+ *  evidence as a note. The body stays as written, the reason it was opened for, so the headline names that too. */
+async function recordCurrentReason(deps: StuckSweepDeps, recovery: QueueItem, c: Candidate): Promise<void> {
+  const openedFor = (recovery.tags ?? []).find((tag) => tag.startsWith("stuck-sweep:"))?.match(/^stuck-sweep:([a-z-]+):/)?.[1];
+  const latest = deps.queueRepo.transitionLog.listForQitem(recovery.qitemId)
+    .map((t) => t.transitionNote ?? "").reverse().find((note) => note.startsWith(REASON_NOTE));
+  const recorded = latest?.slice(REASON_NOTE.length).match(/^([a-z-]+) /)?.[1] ?? openedFor;
+  if (!recorded || recorded === c.kind) return;
+  await deps.queueRepo.update({
+    qitemId: recovery.qitemId,
+    actorSession: "stuck-sweep@system",
+    summary: c.kind === openedFor ? findingHeadline(c) : `${findingHeadline(c)}, opened for ${openedFor}`,
+    transitionNote: `${REASON_NOTE}${c.kind} on ${c.row.qitemId}; the sweep reported ${recorded} before\n${evidenceBody(deps.db, c)}`,
+  });
+}
+
 function evidenceBody(db: Database.Database, c: Candidate): string {
   if (c.verificationTargets?.length) {
     const checks = c.verificationTargets
@@ -944,7 +969,16 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
       if (shared && !renewedDeadline) {
         const existingTags = sharedRow?.tags ?? [];
         for (const tag of existingTags) if (tag.startsWith("stuck-sweep:")) liveDedupTags.add(tag);
-        if (["pending", "in-progress", "blocked"].includes(shared.state)) findings.push({ kind: c.kind, qitemId: c.row.qitemId, findingQitemId: shared.qitemId, action: "refreshed" });
+        if (["pending", "in-progress", "blocked"].includes(shared.state)) {
+          if (sharedRow) {
+            try {
+              await recordCurrentReason(deps, sharedRow, c);
+            } catch (err) {
+              log(`[stuck-sweep] ${c.kind} reason for ${c.row.qitemId} not recorded on ${shared.qitemId}: ${err instanceof Error ? err.message : String(err)}`);
+            }
+          }
+          findings.push({ kind: c.kind, qitemId: c.row.qitemId, findingQitemId: shared.qitemId, action: "refreshed" });
+        }
         continue;
       }
       liveDedupTags.add(dedupTag);
@@ -973,9 +1007,7 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
           sourceSession: c.row.sourceSession,
           destinationSession: c.route,
           body: evidenceBody(deps.db, c) + (routeNotes.has(c.row.qitemId) ? `\nescalation route: ${routeNotes.get(c.row.qitemId)}` : ""),
-          summary: c.verificationTargets
-            ? custodyHeadline(c.row.qitemId, c.verificationTargets, c.ageMinutes)
-            : `Stuck sweep: ${c.kind} on ${c.row.qitemId} (${c.ageMinutes} min)`,
+          summary: findingHeadline(c),
           evidenceRef: `rig queue show ${c.row.qitemId}`,
           tags: [STUCK_SWEEP_FINDING_TAG, dedupTag, recoveryTag(c.row.qitemId),
             ...(routed.has(c.row.qitemId) || c.row.tags?.includes(ESCALATION_HOP_TAG) ? [ESCALATION_HOP_TAG] : [])],

packages/daemon/src/domain/queue-waiting.ts

@@ -12,13 +12,15 @@ export type WaitingActivityReader = (session: string) => Pick<ArbitratedSeatStat
  * actors and typed wake receipts are bookkeeping, never task progress. Historical
  * untyped author notes remain author testimony; their text is not classified.
  * Delivery bookkeeping uses the reserved daemon actors and must not reset the
- * ladder before it can join its dispatch marker to the delivery receipt. */
+ * ladder before it can join its dispatch marker to the delivery receipt. The
+ * stuck sweep's same-state notes on its recoveries are its bookkeeping too; its
+ * resolution close still counts, as every state change does. */
 export function lastMeaningfulTransition(db: Database.Database, id: string): { id: number; at: string } | null {
   if (!db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transitions'").get()) return null;
   const hasWakes = db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transition_wakes'").get();
   const row = db.prepare(`SELECT t.transition_id, t.ts FROM (
       SELECT *, LAG(state) OVER (ORDER BY transition_id) AS previous_state FROM queue_transitions WHERE qitem_id = ?
-    ) t WHERE (t.actor_session NOT IN ('watchdog@system', 'wake-ladder@system', 'daemon@kernel', 'daemon@system') OR t.state IS NOT t.previous_state)
+    ) t WHERE (t.actor_session NOT IN ('watchdog@system', 'wake-ladder@system', 'stuck-sweep@system', 'daemon@kernel', 'daemon@system') OR t.state IS NOT t.previous_state)
     ${hasWakes ? "AND NOT EXISTS (SELECT 1 FROM queue_transition_wakes w WHERE w.transition_id = t.transition_id AND w.phase = 'fired')" : ""}
     ORDER BY t.transition_id DESC LIMIT 1`).get(id) as { transition_id: number; ts: string } | undefined;
   return row ? { id: row.transition_id, at: row.ts } : null;

packages/daemon/test/queue-stuck-sweep.test.ts

@@ -353,6 +353,79 @@ describe("S02 standing stuck sweep — both halves, routed findings, quiet-but-o
     });
   });
 
+  // #2533: progress clears the stall while the row is still overdue. The open recovery keeps custody of the row, and
+  // what its recipient reads names the reason that remains, once.
+  it("an open stall recovery that now stands for an overdue claim says so, once, and keeps custody", async () => {
+    const row = await mkRow();
+    repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
+    ageClaim(row.qitemId, 60);
+    await runSweep();
+    const prior = (await findingsFor(row.qitemId))[0]!;
+    expect(prior.body).toContain("stalled-after-claim");
+
+    makeOverdue(row.qitemId);
+    await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", transitionNote: "resumed work" });
+    expect(repo.getById(row.qitemId)!.pickup?.state).toBe("working");
+    const after = await runSweep();
+    expect(after.result.findings).toContainEqual(expect.objectContaining({
+      kind: "overdue-claim", action: "refreshed", findingQitemId: prior.qitemId,
+    }));
+    const recovery = repo.getById(prior.qitemId)!;
+    expect(recovery.state).toBe("pending");
+    expect(recovery.summary).toMatch(new RegExp(`^Stuck sweep: overdue-claim on ${row.qitemId} \\(\\d+ min\\), opened for stalled-after-claim$`));
+    const reasonNotes = () => repo.transitionLog.listForQitem(prior.qitemId)
+      .map((t) => t.transitionNote ?? "").filter((note) => note.startsWith("stuck-sweep reason: "));
+    expect(reasonNotes()).toHaveLength(1);
+    expect(reasonNotes()[0]).toContain(`stuck-sweep reason: overdue-claim on ${row.qitemId}; the sweep reported stalled-after-claim before`);
+    expect(reasonNotes()[0]).toContain("STUCK SWEEP FINDING (overdue-claim)");
+
+    await runSweep();
+    expect(reasonNotes()).toHaveLength(1);
+    expect(await findingsFor(row.qitemId)).toHaveLength(1);
+
+    // The worker goes quiet again while the row is still overdue: the stall is the sweep's reason again, and the headline
+    // is the stall's own once more.
+    db.prepare("UPDATE queue_transitions SET ts = ? WHERE qitem_id = ? AND transition_note = 'resumed work'")
+      .run(new Date(Date.now() - 45 * 60_000).toISOString(), row.qitemId);
+    expect(repo.getById(row.qitemId)!.pickup?.state).toBe("stalled-after-claim");
+    await runSweep();
+    expect(repo.getById(prior.qitemId)!.summary).toMatch(new RegExp(`^Stuck sweep: stalled-after-claim on ${row.qitemId} \\(\\d+ min\\)$`));
+    expect(reasonNotes()).toHaveLength(2);
+    expect(reasonNotes()[1]).toContain("the sweep reported overdue-claim before");
+    expect(await findingsFor(row.qitemId)).toHaveLength(1);
+  });
+
+  // The sweep's reason note is its own bookkeeping, not progress on the recovery: a claimed recovery whose parked-owner
+  // wake failed and whose wake ladder is exhausted stays exhausted when its reason changes.
+  it("a reason change on an exhausted claimed recovery doesn't restart its wake ladder", async () => {
+    const row = await mkRow();
+    repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
+    ageClaim(row.qitemId, 60);
+    await runSweep();
+    const recoveryId = (await findingsFor(row.qitemId))[0]!.qitemId;
+    repo.claim({ qitemId: recoveryId, destinationSession: "worker@r" });
+    const mod = await sweepMod();
+    repo.transitionLog.append({ qitemId: recoveryId, state: "in-progress", actorSession: "watchdog@system", transitionNote: "parked-owner wake delivery failed: fixture" });
+    repo.recordNudgeAttempt(recoveryId, "failed:tmux session not found");
+    repo.transitionLog.append({ qitemId: recoveryId, state: "in-progress", actorSession: "wake-ladder@system", transitionNote: `${mod.LADDER_EXHAUSTED_PREFIX} fixture` });
+    db.prepare("UPDATE queue_items SET last_nudge_attempt = ? WHERE qitem_id = ?").run(new Date(Date.now() - 10 * 60_000).toISOString(), recoveryId);
+    const ladder = await import("../src/domain/queue-wake-ladder.js");
+    const wakes: string[] = [];
+    const tick = () => ladder.runWakeLadderTick({
+      db, queueRepo: repo, attemptWake: async (q: string) => { wakes.push(q); return "failed:tmux session not found"; },
+      resolveOrchestrator: () => null, retryIntervalSeconds: 300, retryCap: 3, unconfirmedWindowMinutes: 30, swapGraceSeconds: 180, log: () => {},
+    } as never);
+    await tick();
+    expect(wakes).not.toContain(recoveryId);
+
+    makeOverdue(row.qitemId);
+    await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", transitionNote: "resumed work" });
+    await runSweep();
+    expect(repo.transitionLog.listForQitem(recoveryId).some((t) => t.transitionNote?.startsWith("stuck-sweep reason: overdue-claim"))).toBe(true);
+    await tick();
+    expect(wakes).not.toContain(recoveryId);
+  });
+
   // A dead owner with no delegates_to edge: the orchestrator resolves to nobody and the owner's session is gone. The
   // ladder covers handed-off batons only, so a finding addressed to the dead seat would reach nobody.
   function seedMissingSession(session: string): void {