fix(daemon): stop replaying failed wakes to seats with no session at start (#2620)

drainPendingWakeIntents skips a failed or indeterminate wake intent while its seat has no receiver: the destination is unroutable, or the seat's current session, resolved through physicalSessionName, is exited or detached. The check runs at every claim, including grouped and correlated claims, so a seat that comes back during the pass gets its later wakes, and only notifications nothing in the pass attempted are counted as waiting. A never-tried intent always gets its first attempt; running, unknown and non-pane-bound seats take the old path, and an error in the check fails open. The skipped rows stay pending and owed, with no write, and the start-up log line adds waiting=N and names the seats. The wake ladder and stuck sweep are unchanged. No new refusal, error code or setting.

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

@@ -10,6 +10,7 @@ import { resolveExternal } from "./gateway/external-admission.js";
 import type { AgentActivity, PersistedEvent } from "./types.js";
 import { QueueTransitionLog, type OwnerNotificationLevel, type QueueTransition, type RecentQueueTransitionScope } from "./queue-transition-log.js";
 import { WAKE_INTENT_PREFIX, type OutboxHandler, type OutboxEntry } from "./outbox-handler.js";
+import { physicalSessionName } from "./session-registry.js";
 import type { QueueWakeFacts } from "./draft-aware-delivery.js";
 import { derivePickup, type PickupReceipt } from "./queue-pickup.js";
 import { withRecoveryIndex } from "./queue-recovery.js";
@@ -1599,10 +1600,25 @@ export class QueueRepository {
 
   /** Recover current notifications once per pass. A snapshot horizon and seek cursor
    * prevent new arrivals or persistent failures from making this pass unbounded. `stopRequested` is checked before
-   * each delivery: a stopped pass leaves the notifications it has not attempted as they are, for the next start. */
-  async drainPendingWakeIntents(opts?: { stopRequested?: () => boolean }): Promise<{ delivered: number; indeterminate: number; failed: number; retained: number }> {
-    const tally = { delivered: 0, indeterminate: 0, failed: 0, retained: 0 };
+   * each delivery: a stopped pass leaves the notifications it has not attempted as they are, for the next start.
+   * A notification already tried for a seat that still has no session or transport would only fail again, so it isn't
+   * claimed, alone or in a group: it stays as it is, its row stays owed, and the pass reports it as waiting. The
+   * receiver is read at each claim, so a seat that starts during the pass gets its later notifications. */
+  async drainPendingWakeIntents(opts?: { stopRequested?: () => boolean }): Promise<{ delivered: number; indeterminate: number; failed: number; retained: number;
+    /** Present only when a notification waits for a seat with no session or transport. */
+    waiting?: { notifications: number; seats: string[] } }> {
+    const tally: { delivered: number; indeterminate: number; failed: number; retained: number;
+      waiting?: { notifications: number; seats: string[] } } = { delivered: 0, indeterminate: 0, failed: 0, retained: 0 };
     if (!this.outbox || !this.transport) return tally;
+    const attempted = new Set<string>();
+    // Notifications left alone for want of a receiver; one a later claim in this pass attempted isn't waiting.
+    const skipped = new Map<string, string>();
+    const result = () => {
+      const waitingBySeat = new Map<string, number>();
+      for (const [outboxId, seat] of skipped) if (!attempted.has(outboxId)) waitingBySeat.set(seat, (waitingBySeat.get(seat) ?? 0) + 1);
+      if (waitingBySeat.size) tally.waiting = { notifications: [...waitingBySeat.values()].reduce((a, b) => a + b, 0), seats: [...waitingBySeat.keys()] };
+      return tally;
+    };
     const BATCH = 200;
     const horizon = (this.db.prepare("SELECT COALESCE(MAX(rowid),0) AS id FROM outbox_entries").get() as { id: number }).id;
     // Retire never-attempted obsolete notifications without materializing historical intents.
@@ -1612,9 +1628,15 @@ export class QueueRepository {
         AND NOT EXISTS (SELECT 1 FROM queue_items q WHERE q.qitem_id = o.audit_pointer
           AND q.state = 'pending' AND q.destination_session = o.destination_session)`)
       .run(horizon, WAKE_INTENT_PREFIX.length, WAKE_INTENT_PREFIX).changes;
-    const attempted = new Set<string>();
     // A stop also refuses every later claim inside a delivery: coalesced peers and members it releases for their own send.
-    const eligible = () => !opts?.stopRequested?.();
+    // So does a missing receiver, for a notification already tried. One never tried still gets its attempt, so its row
+    // records why it didn't land.
+    const eligible = (entry: OutboxEntry): boolean => {
+      if (opts?.stopRequested?.()) return false;
+      if (entry.deliveryState === "pending" || !this.wakeHasNoReceiver(entry.destinationSession)) return true;
+      skipped.set(entry.outboxId, entry.destinationSession);
+      return false;
+    };
     let cursor = 0;
     for (;;) {
       const pending = this.db.prepare(`SELECT o.rowid AS id, o.outbox_id FROM outbox_entries o
@@ -1625,14 +1647,30 @@ export class QueueRepository {
         .all(cursor, horizon, WAKE_INTENT_PREFIX.length, WAKE_INTENT_PREFIX, BATCH) as Array<{ id: number; outbox_id: string }>;
       if (pending.length === 0) break;
       for (const intent of pending) {
-        if (opts?.stopRequested?.()) return tally;
+        if (opts?.stopRequested?.()) return result();
         cursor = intent.id;
         const outcome = await this.deliverWakeIntent(intent.outbox_id, attempted, eligible);
         if (outcome !== "skipped") tally[outcome]++;
       }
       if (pending.length < BATCH) break; // served short batch ⇒ drained
     }
-    return tally;
+    return result();
+  }
+
+  /** Positive evidence only that a wake can't land now: no transport can hold the address, or the newest session
+   * under the name the transport sends to has stopped (exited, or detached because its tmux session is gone). A
+   * canonical address resolves to its current physical session first. Anything unknown may have a receiver. */
+  private wakeHasNoReceiver(destinationSession: string): boolean {
+    try {
+      const destClass = this.classifyDestinationOf(destinationSession).class;
+      if (destClass === "unroutable") return true;
+      if (destClass !== "pane-bound") return false;
+      const newest = this.db.prepare("SELECT status FROM sessions WHERE session_name = ? ORDER BY id DESC LIMIT 1")
+        .get(physicalSessionName(this.db, destinationSession)) as { status: string } | undefined;
+      return newest?.status === "exited" || newest?.status === "detached";
+    } catch {
+      return false;
+    }
   }
 
   /**

packages/daemon/src/domain/wake-recovery.ts

@@ -8,6 +8,13 @@ export interface WakeRecovery {
   stop(): Promise<void>;
 }
 
+/** Notifications the pass didn't replay because their seats have no session or transport, named once. */
+function waitingNote(waiting?: { notifications: number; seats: string[] }): string {
+  if (!waiting) return "";
+  const named = waiting.seats.slice(0, 10).join(", ") + (waiting.seats.length > 10 ? `, and ${waiting.seats.length - 10} more` : "");
+  return ` waiting=${waiting.notifications} (not replayed: no session or transport for ${named})`;
+}
+
 /**
  * Each recovered notification is a real, verified seat delivery, so a large current backlog takes as long as its
  * deliveries. The daemon's readiness must not wait for that: index.ts starts this pass once the daemon listens and
@@ -24,8 +31,8 @@ export function createWakeRecovery(repo: Pick<QueueRepository, "drainPendingWake
     start() {
       if (running || stopping) return;
       running = repo.drainPendingWakeIntents({ stopRequested }).then((drained) => {
-        if (drained.delivered || drained.indeterminate || drained.failed || reconciled) {
-          log.log(`wake-intent recovery sweep: delivered=${drained.delivered} indeterminate=${drained.indeterminate} failed=${drained.failed} reconciled=${reconciled}${cutShort ? " (stopped before the rest)" : ""}`);
+        if (drained.delivered || drained.indeterminate || drained.failed || reconciled || drained.waiting) {
+          log.log(`wake-intent recovery sweep: delivered=${drained.delivered} indeterminate=${drained.indeterminate} failed=${drained.failed} reconciled=${reconciled}${waitingNote(drained.waiting)}${cutShort ? " (stopped before the rest)" : ""}`);
         }
       }, (err: unknown) => {
         // Non-fatal, as before: a failed pass leaves its notifications visible and retried on the next start.

packages/daemon/test/wake-recovery-no-receiver.test.ts

@@ -0,0 +1,183 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import Database from "better-sqlite3";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { migrate } from "../src/db/migrate.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { OutboxHandler, WAKE_INTENT_PREFIX } from "../src/domain/outbox-handler.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import { physicalSessionName } from "../src/domain/session-registry.js";
+import { createWakeRecovery } from "../src/domain/wake-recovery.js";
+
+const databases: Database.Database[] = [];
+afterEach(() => { for (const db of databases.splice(0)) db.close(); });
+
+const MISSING = (session: string) =>
+  `Session '${session}' not found: tmux reports no session with this name. No text was sent. Check available sessions with: rig ps --nodes`;
+
+/** One running seat, one stopped on purpose (exited), one whose tmux session was gone at start (detached), and an
+ *  address no transport holds. The transport behaves like tmux: only the running seat has a session. */
+function team() {
+  const db = new Database(":memory:");
+  databases.push(db);
+  migrate(db, ALL_MIGRATIONS);
+  db.prepare("INSERT INTO rigs (id, name) VALUES ('rig-1', 'rig1')").run();
+  for (const [seat, status] of [["a", "running"], ["b", "exited"], ["c", "detached"]] as const) {
+    db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime) VALUES (?, 'rig-1', ?, 'claude-code')").run(`node-${seat}`, `dev.${seat}`);
+    db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES (?, ?, ?, ?)").run(`sess-${seat}`, `node-${seat}`, `dev-${seat}@rig1`, status);
+    db.prepare("INSERT INTO bindings (id, node_id, attachment_type, tmux_session) VALUES (?, ?, 'tmux', ?)").run(`binding-${seat}`, `node-${seat}`, `dev-${seat}@rig1`);
+  }
+  const outbox = new OutboxHandler(db);
+  const repo = new QueueRepository(db, new EventBus(db), { validateRig: () => true, loadHumanRegistry: () => ({ ok: true as const, entities: [] }) });
+  repo.attachOutbox(outbox);
+  const live = new Set(["dev-a@rig1"]);
+  const send = vi.fn(async (session: string, _text?: string) => live.has(session)
+    ? { ok: true, verified: true }
+    : { ok: false, sent: false, reason: "session_missing", inputPhase: "not_reached" as const, error: MISSING(session) });
+  const state = (qitemId: string) => outbox.getById(`${WAKE_INTENT_PREFIX}${qitemId}`)!.deliveryState;
+  const transitions = (qitemId: string) => (db.prepare("SELECT COUNT(*) AS n FROM queue_transitions WHERE qitem_id = ?").get(qitemId) as { n: number }).n;
+  /** The seat starts again: a newer running session. */
+  const start = (seat: "b" | "c", id: string) => {
+    db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES (?, ?, ?, 'running')").run(id, `node-${seat}`, `dev-${seat}@rig1`);
+    live.add(`dev-${seat}@rig1`);
+  };
+  return { db, outbox, repo, send, live, state, transitions, start };
+}
+
+/** Rows owed by each address, each with its notification already tried once and failed (the daemon was up). */
+async function owe(t: ReturnType<typeof team>, destinations: string[]) {
+  t.repo.attachTransport({ send: t.send });
+  const wasLive = [...t.live];
+  t.live.clear();
+  const ids: string[] = [];
+  for (const [i, destinationSession] of destinations.entries()) {
+    ids.push((await t.repo.create({ sourceSession: "lead@rig1", destinationSession, body: `work ${i}` })).qitemId);
+  }
+  await vi.waitFor(() => expect(ids.map(t.state).every((s) => s === "failed")).toBe(true));
+  for (const seat of wasLive) t.live.add(seat);
+  t.send.mockClear();
+  return ids;
+}
+
+describe("start-up wake recovery and seats with nothing to receive a wake", () => {
+  it("doesn't replay failed wakes to stopped seats or unroutable addresses; a running seat's failed wake is still replayed", async () => {
+    const t = team();
+    const ids = await owe(t, ["dev-a@rig1", "dev-b@rig1", "dev-b@rig1", "dev-c@rig1", "gone@rig1"]);
+    const [a, ...waiting] = ids as [string, ...string[]];
+    const before = waiting.map((id) => ({ result: t.repo.getById(id)!.lastNudgeResult, transitions: t.transitions(id) }));
+    expect(before.map((b) => b.result)).toEqual([
+      `failed:${MISSING("dev-b@rig1")}`, `failed:${MISSING("dev-b@rig1")}`, `failed:${MISSING("dev-c@rig1")}`,
+      expect.stringMatching(/^unroutable: 'gone@rig1' has no terminal transport/),
+    ]);
+
+    const tally = await t.repo.drainPendingWakeIntents();
+
+    // One wake, to the running seat. Main sent three more into tmux (b twice, c once) and re-failed all four.
+    expect(t.send.mock.calls.map(([session]) => session)).toEqual(["dev-a@rig1"]);
+    expect(t.state(a)).toBe("delivered");
+    expect(tally).toEqual({ delivered: 1, indeterminate: 0, failed: 0, retained: 0,
+      waiting: { notifications: 4, seats: ["dev-b@rig1", "dev-c@rig1", "gone@rig1"] } });
+    // The waiting rows stay owed and untouched: same notification state, same last result, no new transitions.
+    expect(waiting.map(t.state)).toEqual(["failed", "failed", "failed", "failed"]);
+    expect(waiting.map((id) => ({ result: t.repo.getById(id)!.lastNudgeResult, transitions: t.transitions(id) }))).toEqual(before);
+    expect(waiting.map((id) => t.repo.getById(id)!.state)).toEqual(["pending", "pending", "pending", "pending"]);
+  });
+
+  it("replays the waiting wakes once the seat is running again", async () => {
+    const t = team();
+    const [b] = await owe(t, ["dev-b@rig1"]);
+    expect((await t.repo.drainPendingWakeIntents()).waiting).toEqual({ notifications: 1, seats: ["dev-b@rig1"] });
+    expect(t.send).not.toHaveBeenCalled();
+    // The seat starts again: a newer running session.
+    t.db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES ('sess-b2', 'node-b', 'dev-b@rig1', 'running')").run();
+    t.send.mockImplementation(async () => ({ ok: true, verified: true }));
+    const tally = await t.repo.drainPendingWakeIntents();
+    expect(tally).toEqual({ delivered: 1, indeterminate: 0, failed: 0, retained: 0 });
+    expect(t.state(b!)).toBe("delivered");
+  });
+
+  it("still gives a never-tried wake to a stopped seat its one attempt, so its row records why it didn't land", async () => {
+    const t = team();
+    const [b] = [(await t.repo.create({ sourceSession: "lead@rig1", destinationSession: "dev-b@rig1", body: "work" })).qitemId];
+    expect(t.state(b!)).toBe("pending");
+    t.repo.attachTransport({ send: t.send });
+    const tally = await t.repo.drainPendingWakeIntents();
+    expect(t.send.mock.calls.map(([session]) => session)).toEqual(["dev-b@rig1"]);
+    expect(tally).toEqual({ delivered: 0, indeterminate: 0, failed: 1, retained: 0 });
+    expect(t.repo.getById(b!)!.lastNudgeResult).toBe(`failed:${MISSING("dev-b@rig1")}`);
+  });
+
+  it.each([
+    ["separate", false, ["failed", "delivered", "delivered"], { waiting: { notifications: 1, seats: ["dev-b@rig1"] } }],
+    // The later wake's group takes the earlier one along once the seat is back, so nothing is left waiting.
+    ["sharing a return tag", true, ["delivered", "delivered", "delivered"], {}],
+  ] as const)("a seat that starts while the pass is running gets its later failed wakes in that pass (%s)", async (_, shared, states, waiting) => {
+    const t = team();
+    const [b1, a1, b2] = await owe(t, ["dev-b@rig1", "dev-a@rig1", "dev-b@rig1"]);
+    if (shared) {
+      for (const id of [b1, b2]) {
+        t.db.prepare("UPDATE outbox_entries SET tags = ? WHERE outbox_id = ?").run(JSON.stringify(["queue:return:qitem-x"]), `${WAKE_INTENT_PREFIX}${id}`);
+      }
+    }
+    // dev-b starts while the pass waits on dev-a's delivery.
+    t.send.mockImplementation(async (session: string) => {
+      if (session === "dev-a@rig1") t.start("b", "sess-b2");
+      return t.live.has(session) ? { ok: true, verified: true }
+        : { ok: false, sent: false, reason: "session_missing", inputPhase: "not_reached" as const, error: MISSING(session) };
+    });
+    const tally = await t.repo.drainPendingWakeIntents();
+    expect(t.send.mock.calls.map(([session]) => session)).toEqual(["dev-a@rig1", "dev-b@rig1"]);
+    expect([b1, a1, b2].map((id) => t.state(id!))).toEqual(states);
+    expect(tally).toEqual({ delivered: 2, indeterminate: 0, failed: 0, retained: 0, ...waiting });
+  });
+
+  it("a never-tried wake goes alone: its group leaves a failed peer whose seat has no receiver, and only untouched wakes count as waiting", async () => {
+    const t = team();
+    const peer = (await t.repo.create({ sourceSession: "lead@rig1", destinationSession: "dev-b@rig1", body: "earlier work" })).qitemId;
+    const fresh = (await t.repo.create({ sourceSession: "lead@rig1", destinationSession: "dev-b@rig1", body: "later work" })).qitemId;
+    // No transport yet, so both are still pending. They share one return tag, and the earlier one was tried and failed.
+    for (const id of [peer, fresh]) {
+      t.db.prepare("UPDATE outbox_entries SET tags = ? WHERE outbox_id = ?").run(JSON.stringify(["queue:return:qitem-x"]), `${WAKE_INTENT_PREFIX}${id}`);
+    }
+    expect(t.outbox.claimForDelivery(`${WAKE_INTENT_PREFIX}${peer}`)).toBe(true);
+    t.outbox.finalizeDelivery(`${WAKE_INTENT_PREFIX}${peer}`, "failed");
+    const untouched = () => ({ result: t.repo.getById(peer)!.lastNudgeResult, attempt: t.repo.getById(peer)!.lastNudgeAttempt, transitions: t.transitions(peer) });
+    const before = untouched();
+    t.repo.attachTransport({ send: t.send });
+    const tally = await t.repo.drainPendingWakeIntents();
+    expect(t.send).toHaveBeenCalledTimes(1);
+    expect(t.send.mock.calls[0]![1]).toContain(fresh);
+    expect(t.send.mock.calls[0]![1]).not.toContain(peer);
+    expect(t.state(peer)).toBe("failed");
+    expect(untouched()).toEqual(before);
+    expect(tally).toEqual({ delivered: 0, indeterminate: 0, failed: 1, retained: 0, waiting: { notifications: 1, seats: ["dev-b@rig1"] } });
+  });
+
+  it("a canonical address whose old session exited still gets its wake when its current physical session runs", async () => {
+    const t = team();
+    // dev-e@rig1 ran once under its own name; the same node then reconnected under the physical name e-phys.
+    t.db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime) VALUES ('node-e', 'rig-1', 'dev.e', 'claude-code')").run();
+    t.db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES ('sess-e1', 'node-e', 'dev-e@rig1', 'exited')").run();
+    t.db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES ('sess-e2', 'node-e', 'e-phys', 'running')").run();
+    t.db.prepare("INSERT INTO bindings (id, node_id, attachment_type, tmux_session) VALUES ('binding-e', 'node-e', 'tmux', 'e-phys')").run();
+    // The transport sends to the current physical session.
+    expect(physicalSessionName(t.db, "dev-e@rig1")).toBe("e-phys");
+    const [e] = await owe(t, ["dev-e@rig1"]);
+    t.live.add("dev-e@rig1");
+    const tally = await t.repo.drainPendingWakeIntents();
+    expect(t.send.mock.calls.map(([session]) => session)).toEqual(["dev-e@rig1"]);
+    expect(t.state(e!)).toBe("delivered");
+    expect(tally).toEqual({ delivered: 1, indeterminate: 0, failed: 0, retained: 0 });
+  });
+
+  it("names the waiting seats once in the start-up log line", async () => {
+    const t = team();
+    await owe(t, ["dev-b@rig1", "dev-c@rig1", "dev-c@rig1"]);
+    const log = { log: vi.fn(), warn: vi.fn() };
+    const recovery = createWakeRecovery(t.repo, 0, log);
+    recovery.start();
+    await vi.waitFor(() => expect(log.log).toHaveBeenCalled());
+    await recovery.stop();
+    expect(log.log).toHaveBeenCalledWith(
+      "wake-intent recovery sweep: delivered=0 indeterminate=0 failed=0 reconciled=0 waiting=3 (not replayed: no session or transport for dev-b@rig1, dev-c@rig1)");
+  });
+});