diff --git a/packages/daemon/src/domain/queue-stuck-sweep.ts b/packages/daemon/src/domain/queue-stuck-sweep.ts index 248252ffa..676d51310 100644 --- a/packages/daemon/src/domain/queue-stuck-sweep.ts +++ b/packages/daemon/src/domain/queue-stuck-sweep.ts @@ -637,6 +637,16 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise { + // Real topology, so the repository's pre-topology transport shortcut doesn't apply. + db.prepare("INSERT INTO rigs(id,name) VALUES ('fixture-rig','r')").run(); + db.prepare("INSERT INTO nodes(id,rig_id,logical_id,runtime,profile) VALUES ('late-node','fixture-rig','late','codex','none')").run(); + const sent: string[] = []; + repo = new QueueRepository(db, new EventBus(db), { + validateRig: () => true, + loadHumanRegistry: () => ({ ok: true, entities: [] }), + transport: { send: async (destination) => { sent.push(destination); return { ok: true, verified: true }; } }, + }); + const row = await repo.create({ sourceSession: "sender@r", destinationSession: "late@r", body: "work" }); + expect(row.lastNudgeResult).toMatch(/^unroutable:/); + expect(sent).toEqual([]); + // The seat's pane comes up: its tmux binding is written. + db.prepare("INSERT INTO bindings(id,node_id,tmux_session) VALUES ('late-binding','late-node','late@r')").run(); + await runSweep(); + expect(sent).toEqual(["late@r"]); + expect(repo.getById(row.qitemId)!.lastNudgeResult).not.toMatch(/^unroutable:/); + expect(await findingsFor(row.qitemId)).toHaveLength(0); + }); + + // With the durable wake intents in place (the production path), the re-send goes through the row's own intent: it's + // claimed and settled, so a later startup drain doesn't wake the seat a second time. + async function lateSeatWithOutbox() { + db.prepare("INSERT INTO rigs(id,name) VALUES ('fixture-rig','r')").run(); + db.prepare("INSERT INTO nodes(id,rig_id,logical_id,runtime,profile) VALUES ('late-node','fixture-rig','late','codex','none')").run(); + const sent: string[] = []; + repo = new QueueRepository(db, new EventBus(db), { + validateRig: () => true, + loadHumanRegistry: () => ({ ok: true, entities: [] }), + transport: { send: async (destination) => { sent.push(destination); return { ok: true, verified: true }; } }, + }); + const outbox = new OutboxHandler(db); + repo.attachOutbox(outbox); + const created = await repo.create({ sourceSession: "sender@r", destinationSession: "late@r", body: "work" }); + // With an outbox, the create's wake runs after its commit: read the row back once that wake has settled. + await vi.waitFor(() => expect(outbox.getById(`wake-intent-${created.qitemId}`)?.deliveryState).not.toMatch(/^(pending|sending)$/)); + const row = repo.getById(created.qitemId)!; + expect(row.lastNudgeResult).toMatch(/^unroutable:/); + expect(outbox.getById(`wake-intent-${row.qitemId}`)?.deliveryState).toBe("failed"); + expect(sent).toEqual([]); + db.prepare("INSERT INTO bindings(id,node_id,tmux_session) VALUES ('late-binding','late-node','late@r')").run(); + return { sent, outbox, row }; + } + + it("re-sends through the row's wake intent: the intent reads delivered, and a later drain sends nothing", async () => { + const { sent, outbox, row } = await lateSeatWithOutbox(); + await runSweep(); + expect(sent).toEqual(["late@r"]); + expect(outbox.getById(`wake-intent-${row.qitemId}`)?.deliveryState).toBe("delivered"); + expect(repo.getById(row.qitemId)!.lastNudgeResult).not.toMatch(/^unroutable:/); + expect(await findingsFor(row.qitemId)).toHaveLength(0); + await repo.drainPendingWakeIntents(); + expect(sent).toEqual(["late@r"]); + }); + + // The row still raises the sweep's finding here; it goes to the orchestrator, so only the row's own wake could reach the seat. + it.each(["sending", "delivered"])("doesn't send again when the row's wake intent is already %s", async (state) => { + const { sent, outbox, row } = await lateSeatWithOutbox(); + db.prepare("UPDATE outbox_entries SET delivery_state = ? WHERE outbox_id = ?").run(state, `wake-intent-${row.qitemId}`); + await runSweep({ resolveOrchestrator: () => "orch@r" }); + expect(sent).not.toContain("late@r"); + expect(outbox.getById(`wake-intent-${row.qitemId}`)?.deliveryState).toBe(state); + expect(repo.getById(row.qitemId)!.lastNudgeAttempt).toBe(row.lastNudgeAttempt); + }); + // With no live recovery seat, the finding is still created (it's owed work) to the missing session, and says so. it("a dead seat's own row with no operator agent or seat stays with it, and its why says no live recovery seat is available", async () => { const row = await repo.create({ sourceSession: "worker@r", destinationSession: "worker@r", body: "work" });