fix(watchdog): refuse legacy terminal-row timers

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

@@ -2635,6 +2635,17 @@ export class QueueRepository {
     return this.wakeRepo.getStatus(qitemId);
   }
 
+  /** Refuse a legacy park-generated timer only when every row bound to it is
+   *  terminal. Current exits retire these timers transactionally; this is the
+   *  delivery-seam backstop for residue persisted by an older daemon. A timer
+   *  still bound to any actionable row, and every operator-attached watchdog,
+   *  remains deliverable. */
+  resolveWatchdogPreDeliveryTerminalReason(jobId: string): string | null {
+    const targets = this.wakeRepo.findQitemsByGeneratedTimer(jobId);
+    if (targets.length === 0 || targets.some(({ state }) => !isTerminalState(state))) return null;
+    return "park_timer_target_terminal";
+  }
+
   listTransitions(qitemId: string): Array<ReturnType<QueueTransitionLog["listForQitem"]>[number] & { wake?: ReturnType<QueueWakeRepository["getForTransition"]> }> {
     return this.transitionLog.listForQitem(qitemId).map((transition) => {
       const wake = this.wakeRepo.getForTransition(transition.transitionId);

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

@@ -133,6 +133,23 @@ export class QueueWakeRepository {
     });
   }
 
+  /** Queue rows bound to a park-generated timer, regardless of whether that
+   *  timer has fired before. Legacy daemons could leave a repeating timer live
+   *  after its row closed, so delivery-time defense must consult the original
+   *  armed binding rather than only the current blocked/fired projection. */
+  findQitemsByGeneratedTimer(jobId: string): Array<{ qitemId: string; state: string }> {
+    if (!this.available) return [];
+    return this.db.prepare(
+      `SELECT DISTINCT w.qitem_id, q.state
+         FROM queue_transition_wakes w
+         JOIN queue_items q ON q.qitem_id = w.qitem_id
+        WHERE w.phase = 'armed' AND w.wake_ref = ? AND w.wake_kind = 'timer'`,
+    ).all(jobId).map((row) => {
+      const r = row as { qitem_id: string; state: string };
+      return { qitemId: r.qitem_id, state: r.state };
+    });
+  }
+
   private isLive(kind: ParkWakeKind, ref: string): boolean {
     if (kind === "blocker") {
       const row = this.db.prepare("SELECT state FROM queue_items WHERE qitem_id = ?").get(ref) as

packages/daemon/src/domain/watchdog-policy-engine.ts

@@ -93,6 +93,10 @@ interface WatchdogPolicyEngineDeps {
    * SKIPPING, so an unknown target-generation must never skip a legitimate wake (note-2 inversion).
    */
   resolveTargetGeneration?: (sessionName: string) => string | null;
+  /** Last defense before transport for persisted jobs whose owning domain can
+   *  prove they are no longer actionable. A reason terminals the job without
+   *  delivering; null preserves the existing send path. */
+  resolvePreDeliveryTerminalReason?: (input: { jobId: string }) => string | null;
   /** Queue-side observer for a wake attempt. It appends resume evidence to
    *  every HELD row armed to this job; delivery outcome is preserved. */
   onWakeAttempt?: (attempt: { jobId: string; deliveryStatus: string }) => void;
@@ -171,6 +175,7 @@ export class WatchdogPolicyEngine {
   private readonly now: () => Date;
   private readonly policies: Map<string, Policy>;
   private readonly resolveTargetGeneration?: (sessionName: string) => string | null;
+  private readonly resolvePreDeliveryTerminalReason?: (input: { jobId: string }) => string | null;
   private readonly onWakeAttempt?: (attempt: { jobId: string; deliveryStatus: string }) => void;
 
   constructor(deps: WatchdogPolicyEngineDeps) {
@@ -179,6 +184,7 @@ export class WatchdogPolicyEngine {
     this.eventBus = deps.eventBus;
     this.deliver = deps.deliver;
     this.resolveTargetGeneration = deps.resolveTargetGeneration;
+    this.resolvePreDeliveryTerminalReason = deps.resolvePreDeliveryTerminalReason;
     this.onWakeAttempt = deps.onWakeAttempt;
     this.parseSpec = deps.parseSpec ?? parseWatchdogSpec;
     this.now = deps.now ?? (() => new Date());
@@ -432,6 +438,35 @@ export class WatchdogPolicyEngine {
       }
     }
 
+    // OPR.0.5.8.1 S1c — current queue transitions retire park timers at every
+    // exit. This narrow pre-transport guard handles only legacy persisted
+    // residue, where the row is already terminal and no transition remains to
+    // intercept. It runs after normal policy/throttle/generation decisions but
+    // before any custody or message delivery side effect.
+    const preDeliveryTerminalReason = this.resolvePreDeliveryTerminalReason?.({ jobId: job.jobId });
+    if (preDeliveryTerminalReason) {
+      this.jobsRepo.markTerminal(job.jobId, preDeliveryTerminalReason);
+      const history = this.historyLog.record({
+        jobId: job.jobId,
+        evaluatedAt,
+        outcome: "terminal",
+        skipReason: preDeliveryTerminalReason,
+      });
+      this.eventBus.emit({
+        type: "watchdog.evaluation_terminal",
+        jobId: job.jobId,
+        policy: job.policy,
+        terminalReason: preDeliveryTerminalReason,
+      });
+      return {
+        job: this.jobsRepo.getByIdOrThrow(job.jobId),
+        outcome: { action: "terminal", reason: preDeliveryTerminalReason },
+        history,
+        delivery: null,
+        meaningful: true,
+      };
+    }
+
     const continuityAction = parseContinuityAction(
       parsed.context["continuity_action"],
       job,

packages/daemon/src/startup.ts

@@ -1757,6 +1757,8 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
       // UNKNOWN (null) fails open → deliver. A drop of this line disables the gate silently (gen-bound
       // wakes would fire at the successor) — pinned in watchdog-target-gen-wiring.test.ts.
       resolveTargetGeneration: (s) => sessionRegistry.currentOccupantGenerationForSession(s),
+      resolvePreDeliveryTerminalReason: ({ jobId }) =>
+        queueRepoInstance.resolveWatchdogPreDeliveryTerminalReason(jobId),
       onWakeAttempt: ({ jobId, deliveryStatus }) => {
         queueRepoInstance.recordWatchdogWakeAttempt(jobId, deliveryStatus);
       },

packages/daemon/test/queue-park-wake.test.ts

@@ -8,6 +8,7 @@ import { queueItemsSchema } from "../src/db/migrations/024_queue_items.js";
 import { queueTransitionsSchema } from "../src/db/migrations/025_queue_transitions.js";
 import { outboxEntriesSchema } from "../src/db/migrations/027_outbox_entries.js";
 import { watchdogJobsSchema } from "../src/db/migrations/031_watchdog_jobs.js";
+import { watchdogHistorySchema } from "../src/db/migrations/032_watchdog_history.js";
 import { queueItemSummarySchema } from "../src/db/migrations/044_queue_item_summary.js";
 import { queueItemEvidenceRefSchema } from "../src/db/migrations/048_queue_item_evidence_ref.js";
 import { EventBus } from "../src/domain/event-bus.js";
@@ -17,6 +18,8 @@ import { USAGE_LIMIT_BLOCKER_TAG } from "../src/domain/queue-wake-repository.js"
 import { isDue } from "../src/domain/watchdog-scheduler.js";
 import type { PersistedEvent } from "../src/domain/types.js";
 import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+import { WatchdogPolicyEngine } from "../src/domain/watchdog-policy-engine.js";
 
 // S03 R25 RED-FIRST fixture: this is the contract shape migration 073 will ship.
 // Keeping the table local in the RED commit lets every behavior fail on its own
@@ -55,6 +58,7 @@ describe("S03 R25 — a park records its wake on the append-only transition", ()
       queueTransitionsSchema,
       outboxEntriesSchema,
       watchdogJobsSchema,
+      watchdogHistorySchema,
       queueItemSummarySchema,
       queueItemEvidenceRefSchema,
     ]);
@@ -221,6 +225,103 @@ describe("S03 R25 — a park records its wake on the append-only transition", ()
     expect(wakes(row.qitemId).some((w) => (w as { phase: string }).phase === "fired")).toBe(false);
   });
 
+  it("OPR.0.5.8.1 S1c — a legacy park timer bound to a terminal row is refused before transport", async () => {
+    const row = await item("worker@rig");
+    repo.update({
+      qitemId: row.qitemId, actorSession: "worker@rig", state: "blocked",
+      blockedOn: "external:cooldown", transitionNote: "continuation: legacy timer", wakeAfterSeconds: 90,
+    } as never);
+    const timerId = (wakes(row.qitemId)[0] as { wake_ref: string }).wake_ref;
+
+    // Exact legacy residue: an older daemon closed the row without retiring its
+    // generated timer. The current delivery seam must defend this persisted
+    // preimage even though there is no new row transition left to intercept.
+    db.prepare("UPDATE queue_items SET state = 'done', blocked_on = NULL WHERE qitem_id = ?").run(row.qitemId);
+    expect(jobs.getById(timerId)?.state).toBe("active");
+
+    const guardCalls: string[] = [];
+    const deliveries: Array<{ targetSession: string; message: string }> = [];
+    const engine = new WatchdogPolicyEngine({
+      jobsRepo: jobs,
+      historyLog: new WatchdogHistoryLog(db),
+      eventBus: bus,
+      deliver: async (request) => {
+        deliveries.push(request);
+        return { status: "ok" };
+      },
+      resolvePreDeliveryTerminalReason: ({ jobId }: { jobId: string }) => {
+        guardCalls.push(jobId);
+        return repo.resolveWatchdogPreDeliveryTerminalReason(jobId);
+      },
+      onWakeAttempt: ({ jobId, deliveryStatus }) => repo.recordWatchdogWakeAttempt(jobId, deliveryStatus),
+    });
+
+    const result = await engine.evaluate(jobs.getByIdOrThrow(timerId));
+
+    expect(guardCalls).toEqual([timerId]);
+    expect(deliveries).toEqual([]);
+    expect(result.outcome).toEqual({ action: "terminal", reason: "park_timer_target_terminal" });
+    expect(jobs.getById(timerId)).toMatchObject({
+      state: "terminal",
+      terminalReason: "park_timer_target_terminal",
+    });
+    expect(wakes(row.qitemId).some((w) => (w as { phase: string }).phase === "fired")).toBe(false);
+  });
+
+  it("OPR.0.5.8.1 S1c — the delivery guard preserves actionable timers and attached watchdogs", async () => {
+    const deliveries: Array<{ targetSession: string; message: string }> = [];
+    const engine = new WatchdogPolicyEngine({
+      jobsRepo: jobs,
+      historyLog: new WatchdogHistoryLog(db),
+      eventBus: bus,
+      deliver: async (request) => {
+        deliveries.push(request);
+        return { status: "ok" };
+      },
+      resolvePreDeliveryTerminalReason: ({ jobId }) =>
+        repo.resolveWatchdogPreDeliveryTerminalReason(jobId),
+      onWakeAttempt: ({ jobId, deliveryStatus }) => repo.recordWatchdogWakeAttempt(jobId, deliveryStatus),
+    });
+
+    const timerRow = await item("timer-owner@rig");
+    repo.update({
+      qitemId: timerRow.qitemId, actorSession: "timer-owner@rig", state: "blocked",
+      blockedOn: "external:cooldown", transitionNote: "continuation: actionable", wakeAfterSeconds: 90,
+    } as never);
+    const timerId = (wakes(timerRow.qitemId)[0] as { wake_ref: string }).wake_ref;
+    const timerResult = await engine.evaluate(jobs.getByIdOrThrow(timerId));
+
+    expect(timerResult.outcome.action).toBe("send");
+    expect(deliveries).toEqual([
+      expect.objectContaining({ targetSession: "timer-owner@rig" }),
+    ]);
+    expect(wakes(timerRow.qitemId)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ phase: "fired", wake_kind: "timer", wake_ref: timerId }),
+    ]));
+
+    const operatorJob = jobs.register({
+      policy: "periodic-reminder",
+      specYaml: "policy: periodic-reminder\ntarget:\n  session: attached-owner@rig\nmessage: operator-owned\n",
+      targetSession: "attached-owner@rig",
+      intervalSeconds: 90,
+      registeredBySession: "operator@rig",
+    });
+    const attachedRow = await item("attached-owner@rig");
+    repo.update({
+      qitemId: attachedRow.qitemId, actorSession: "attached-owner@rig", state: "blocked",
+      blockedOn: "external:cooldown", transitionNote: "continuation: attached watchdog",
+      wakeWatchdogId: operatorJob.jobId,
+    } as never);
+    const attachedResult = await engine.evaluate(jobs.getByIdOrThrow(operatorJob.jobId));
+
+    expect(attachedResult.outcome.action).toBe("send");
+    expect(deliveries).toEqual([
+      expect.objectContaining({ targetSession: "timer-owner@rig" }),
+      expect.objectContaining({ targetSession: "attached-owner@rig" }),
+    ]);
+    expect(jobs.getById(operatorJob.jobId)?.state).toBe("active");
+  });
+
   it("OPR.0.5.8.1 S1b — an unpark (blocked -> in-progress) also ends the timer", async () => {
     const row = await item("worker@rig");
     repo.update({
@@ -336,9 +437,9 @@ describe("S03 R25 — a park records its wake on the append-only transition", ()
     expect(terminalReason(ref)).toBe("park_ended:auto-unparked");
   });
 
-  it("OPR.0.5.8.1 S1b — every exit route leaves an operator's ATTACHED watchdog alone", async () => {
-    // The one way this class repair could do harm: retiring jobs it does not own,
-    // now from six call sites instead of one.
+  it("OPR.0.5.8.1 S1b — the handoff exit leaves an operator's ATTACHED watchdog alone", async () => {
+    // The shared ownership predicate is used by every exit caller; this test
+    // executes one real handoff route rather than claiming a six-route matrix.
     const job = jobs.register({
       policy: "periodic-reminder",
       specYaml: "policy: periodic-reminder\ntarget:\n  session: \"worker@rig\"\nmessage: \"operator's own\"\n",

packages/daemon/test/watchdog-target-gen-wiring.test.ts

@@ -16,4 +16,10 @@ describe("(i-c) wiring pin — startup injects the real target-generation resolv
   it("passes resolveTargetGeneration to the WatchdogPolicyEngine, wired to the live occupant-generation", () => {
     expect(src).toMatch(/resolveTargetGeneration:\s*\(s\)\s*=>\s*sessionRegistry\.currentOccupantGenerationForSession\(s\)/);
   });
+
+  it("passes the queue's terminal-timer resolver to the pre-delivery seam", () => {
+    expect(src).toMatch(
+      /resolvePreDeliveryTerminalReason:\s*\(\{ jobId \}\)\s*=>\s*queueRepoInstance\.resolveWatchdogPreDeliveryTerminalReason\(jobId\)/,
+    );
+  });
 });