fix(delivery): keep one held reminder per watchdog job, and let a handoff through a full held quota (#2438)

A watchdog job now holds at most one reminder per seat: holding a newer reminder retires that job's earlier held one, inside the retention transaction and before the capacity check, so it frees the slot even at a full quota. A queue handoff (`rig queue handoff`, and handoff-and-complete) is no longer blocked by the recipient's full held quota: it stages its wake in a savepoint, so a full quota records that the wake wasn't retained on the successor and the handoff commits, while any other staging failure still rolls it back. That deliberately changes "one act or none" for this one failure. Mission control's route and handoff verbs and workflow step projections still roll back on a full quota; a follow-up gives them the same savepoint. Reported in #2430 by @billbenetti.

packages/daemon/src/domain/outbox-handler.ts

@@ -14,6 +14,8 @@ import type Database from "better-sqlite3";
  * select it. Single source of truth for the drain query only.
  */
 export const WAKE_INTENT_PREFIX = "wake-intent-";
+/** The sender a watchdog's reminders carry; their audit pointer is the job. */
+const WATCHDOG_SENDER = "watchdog@system";
 
 // Outcomes retain unresolved uncertainty. Wake recovery can claim failed/indeterminate
 // work again; it never steals a live sending claim or interprets unknown as unsent.
@@ -217,6 +219,14 @@ export class OutboxHandler {
       } else if (precommitted) {
         throw new OutboxHandlerError("outbox_not_found", "Precommitted delivery is missing; no replacement custody is created.");
       }
+      // One held reminder per watchdog job: a newer reminder retires the one held before it, so a held seat's quota isn't
+      // filled by repeats of one job (#2430).
+      if (!existing && !precommitted && input.senderSession === WATCHDOG_SENDER && input.auditPointer) {
+        this.db.prepare(`UPDATE outbox_entries SET delivery_state='retired', retired_at=?, retired_by=?, retirement_reason=?
+          WHERE delivery_state='retained' AND sender_session=? AND audit_pointer=? AND json_extract(guard_binding,'$.nodeId')=?`)
+          .run(new Date().toISOString(), WATCHDOG_SENDER, "superseded by a newer reminder from the same watchdog job",
+            WATCHDOG_SENDER, input.auditPointer, binding.nodeId);
+      }
       if (!precommitted) this.assertRetentionCapacity(binding.nodeId, input.body);
       if (!existing) this.record(input);
       this.db.prepare(`UPDATE outbox_entries SET delivery_state='retained', guard_binding=?, delivered_at=NULL

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

@@ -47,6 +47,9 @@ function humanAnswerId(askId: string, episodeKey?: string): string {
   return `qitem-human-answer-${createHash("sha256").update(episodeKey ? `${askId}\n${episodeKey}` : askId).digest("hex").slice(0, 20)}`;
 }
 
+/** A row's nudge result when its wake couldn't be staged: the row stands, and its wake is owed. */
+const WAKE_NOT_RETAINED = "failed:wake not retained: ";
+
 export const QUEUE_STATES = [
   "pending",
   "in-progress",
@@ -1255,6 +1258,18 @@ export class QueueRepository {
     });
   }
 
+  /** A handoff's wake, staged as create stages one: a recipient whose held-message quota is full still gets the row,
+   *  which records its wake as not retained, so work keeps moving between other seats (#2430). Any other staging
+   *  failure rolls the act back. */
+  private stageHandoffWake(successorQitemId: string, fromSession: string, toSession: string, identityProvenance: string | null, nudge: boolean | undefined): void {
+    try {
+      this.db.transaction(() => this.stageWakeIntent(successorQitemId, fromSession, toSession, identityProvenance, nudge))();
+    } catch (err) {
+      if ((err as { code?: string }).code !== "retained_quota_full") throw err;
+      this.recordNudgeAttempt(successorQitemId, `${WAKE_NOT_RETAINED}${(err as Error).message}`);
+    }
+  }
+
   private recordWakeIntent(input: {
     outboxId: string;
     auditPointer: string;
@@ -1361,7 +1376,8 @@ export class QueueRepository {
     const src = this.getById(sourceQitemId);
     if (!src || !isTerminalState(src.state)) return; // not a terminal close
     const intent = this.outbox.getById(`${WAKE_INTENT_PREFIX}${successorQitemId}`);
-    if (!intent) {
+    // A successor whose wake a full held-message quota refused records that instead, and stands (#2430).
+    if (!intent && !this.getById(successorQitemId)?.lastNudgeResult?.startsWith(WAKE_NOT_RETAINED)) {
       throw new QueueRepositoryError(
         "terminal_close_without_wake_intent",
         `terminal closure of ${sourceQitemId} committed without staging its wake intent for ${successorQitemId} — one act or none is violated`,
@@ -1904,7 +1920,7 @@ export class QueueRepository {
           this.db.transaction(() => this.stageWakeIntent(created.qitemId, wakeFrom, input.destinationSession, sideEffectProvenance(input.identityProvenance), input.nudge, input.wakeTrigger))();
           wakeStaged = true;
         } catch (err) {
-          this.recordNudgeAttempt(created.qitemId, `failed:wake not retained: ${err instanceof Error ? err.message : String(err)}`);
+          this.recordNudgeAttempt(created.qitemId, `${WAKE_NOT_RETAINED}${err instanceof Error ? err.message : String(err)}`);
         }
       }
       return created;
@@ -2340,8 +2356,9 @@ export class QueueRepository {
 
       // W1-a: the durable wake intent joins the SAME transaction as the close +
       // successor create. If this (or anything above) throws, the whole act rolls
-      // back — one act or none.
-      this.stageWakeIntent(newId, input.fromSession, input.toSession, sideEffectProvenance(input.identityProvenance), input.nudge);
+      // back — one act or none. A recipient's full held-message quota is the one
+      // exception: the successor records its wake as not retained (#2430).
+      this.stageHandoffWake(newId, input.fromSession, input.toSession, sideEffectProvenance(input.identityProvenance), input.nudge);
 
       const handoffEvent = this.eventBus.persistWithinTransaction({
         type: "queue.handed_off",
@@ -2517,8 +2534,8 @@ export class QueueRepository {
       });
 
       // W1-a: the durable wake intent joins the SAME transaction as the close +
-      // successor create — one act or none (symmetric with handoff()).
-      this.stageWakeIntent(newId, input.fromSession, input.toSession, sideEffectProvenance(input.identityProvenance), input.nudge);
+      // successor create — one act or none (symmetric with handoff(), including its full-quota exception).
+      this.stageHandoffWake(newId, input.fromSession, input.toSession, sideEffectProvenance(input.identityProvenance), input.nudge);
 
       const handoffEvent = this.eventBus.persistWithinTransaction({
         type: "queue.handed_off",

packages/daemon/test/seat-delivery-guard.test.ts

@@ -91,6 +91,15 @@ describe("delivery pause and durable custody", () => {
     expect(outbox.heldForNode("a").total).toBe(0);
     expect(outbox.listPending("id")).toEqual([]); db.close();
   });
+  it("keeps one held reminder per watchdog job: a newer one retires the one held before it (#2430)", () => {
+    const { db, targets, outbox } = fixture();
+    const reminder = (id: string, job: string) => outbox.retain({ outboxId: id, senderSession: "watchdog@system", destinationSession: "a", body: `reminder ${id}`, auditPointer: job }, targets.a!);
+    reminder("r1", "job-1"); reminder("r2", "job-1"); reminder("other", "job-2");
+    outbox.retain({ outboxId: "agent", senderSession: "planner@test", destinationSession: "a", body: "real work", auditPointer: "job-1" }, targets.a!);
+    expect(outbox.heldForNode("a").items.map((entry) => entry.outboxId).sort()).toEqual(["agent", "other", "r2"]);
+    expect(outbox.getById("r1")).toMatchObject({ deliveryState: "retired", body: "reminder r1", retiredBy: "watchdog@system" });
+    db.close();
+  });
   it("rolls back rejected staging but retains committed overflow members with original bodies", () => {
     const { db, targets, outbox } = fixture();
     for(let i=0;i<100;i++) outbox.retain({ outboxId: `h${i}`, senderSession: "s", destinationSession:"a", body:`body${i}` },targets.a!);
@@ -270,17 +279,18 @@ describe("real queue transaction and retained delivery",()=>{
     const row=await createWithFullHeldQuota();
     expect(row.lastNudgeResult).toContain("wake not retained");
   });
-  it("precommit quota refusal rolls back closure, successor, transitions and events",async()=>{
+  // #2430 reverses the earlier contract here, which rolled the whole handoff back: a full held quota is about one
+  // seat's pane, and the row is durable, so the work moves and only its wake is recorded as not retained.
+  it("a handoff to a seat whose held quota is full still moves the work, its wake recorded as not retained",async()=>{
     const f=queueFixture();const source=await f.repo.create({sourceSession:"sender@test",destinationSession:"relay@test",body:"work",nudge:false});
     await f.guard.set("a",true,"person","draft");
     for(let i=0;i<100;i++) f.outbox.retain({outboxId:`full${i}`,senderSession:"sender@test",destinationSession:"worker@test",body:"held"},f.guard.target("a"));
-    const before=JSON.stringify(f.db.prepare("SELECT * FROM queue_transitions").all());
-    const events=JSON.stringify(f.db.prepare("SELECT * FROM events").all());
-    await expect(f.repo.handoff({qitemId:source.qitemId,fromSession:"relay@test",toSession:"worker@test"})).rejects.toThrow("quota");
-    expect(f.repo.getById(source.qitemId)!.state).toBe("pending");
-    expect(f.db.prepare("SELECT count(*) n FROM queue_items").get()).toEqual({n:1});
-    expect(JSON.stringify(f.db.prepare("SELECT * FROM queue_transitions").all())).toBe(before);
-    expect(JSON.stringify(f.db.prepare("SELECT * FROM events").all())).toBe(events);
+    const {closed,created}=await f.repo.handoff({qitemId:source.qitemId,fromSession:"relay@test",toSession:"worker@test"});
+    expect(f.repo.getById(closed.qitemId)!.state).toBe("handed-off");
+    expect(f.repo.getById(created.qitemId)).toMatchObject({state:"pending",destinationSession:"worker@test",body:"work",
+      lastNudgeResult:expect.stringMatching(/^failed:wake not retained: Held-message quota is full/)});
+    expect(f.outbox.getById(`wake-intent-${created.qitemId}`)).toBeNull();
+    expect(f.outbox.heldForNode("a").total).toBe(100);
     expect(f.writes).toEqual([]);f.db.close();
   });
   it("handoff retains one audited intent without duplicate transport record or delivered label",async()=>{