fix(watchdog): retain undelivered periodic occurrences through bounded recovery (#2478)

A scheduled reminder that couldn't reach its seat, for example after a reboot where the daemon starts before tmux, now stays owed and recovers instead of silently moving on to the next cycle. Each periodic occurrence, including each parked-row digest member, gets its own receipt in the existing outbox and retries through the existing scheduler and wake retry settings (default three retries, five minutes apart). A confirmed delivery is never sent again after interrupted accounting or a restart, and an uncertain one is never replayed. Exhaustion or an unconfirmed result ends only that occurrence, with its cause and next action recorded, and a recurring reminder keeps its normal cadence under a new occurrence; only a permanent transport failure ends the job. A one-shot park timer records its failed wake on its row before it ends, and a pending retry leaves its binding armed with a retry note. The fire stamp and the queue wake receipt and one-shot retirement commit in one transaction, with input outside it. Refusals are read with the transport's own rule (sent false or input not reached), and the external-CLI and interactive-prompt refusals are labelled, which the queue wake classifier, draft-aware delivery, compaction preparation, claim hints and the transport receipt also read. Abandoned occurrence claims are marked indeterminate at startup, before anything sends. Limits: scope is periodic reminders and parked-row digests, while condition watches, the parked-owner policy and explicit sends are unchanged; no schema, API, scheduler or replay of earlier failures is added, and previously recorded failures can't be reconstructed; while a retry waits, the existing one-second loop still evaluates it; a later recurring cycle is a new occurrence, not confirmation of an uncertain one.

docs/reference/keeping-a-watch.md

@@ -107,6 +107,28 @@ rig queue create --destination <entityId>@external --human-intent update \
 - **`posted` means the connector posted it, not that they read it.** Don't say they saw it.
 - **If delivery fails or is indeterminate,** inspect that task before anything else. Never resend blindly.
 
+## If the timer cannot reach its recipient
+
+A `periodic-reminder` occurrence that definitely fails before input remains owed. The watchdog
+retries through its existing loop, using `queue.wake_retry_interval_seconds` and
+`queue.wake_retry_cap` (by default, three retries five minutes apart). A failed attempt does not
+advance the reminder to its next regular cycle. A confirmed send advances the normal interval;
+recovery does not replay every elapsed cycle.
+
+`rig watchdog show <jobId> --json` includes the failed attempt, its retained outbox IDs, retry
+time and next action. If input may have reached the recipient, that occurrence ends unconfirmed
+and is not automatically sent again. Inspect the recipient and exact outbox record before any
+manual replay. A recurring reminder keeps its next normal cycle under a new occurrence identity. A confirmed receipt survives a daemon restart, so
+finishing its accounting does not type the same occurrence twice.
+
+Exhausted retries end only the failed occurrence, with its cause and next action; a recurring
+reminder keeps its next normal cycle. A permanent transport failure leaves a terminal job:
+correct the destination and register a replacement reminder. A one-shot park timer records its
+failed or unconfirmed wake on the parked row before it ends. A retryable failure is recorded
+there while the timer stays armed. Deliberately held
+input keeps its existing retained-message custody. Condition watches and parked-owner policy
+behavior are unchanged. This does not replay failed occurrences recorded by older versions.
+
 ## "What are you watching for me?"
 
 List **every active state**, not just set-aside ones. A watch being checked can be in progress, and a new one can be

packages/daemon/src/domain/draft-aware-delivery.ts

@@ -10,7 +10,8 @@ export interface GuardedDeliverySummary {
 }
 
 /** Historical attempts contain outcomes, never executable options or callbacks. */
-interface Receipt { ids: string[]; result?: SendResult; expectedIdentity?: import("./expected-conversation.js").ExpectedConversation }
+interface Receipt { ids: string[]; result?: SendResult; expectedIdentity?: import("./expected-conversation.js").ExpectedConversation;
+  watchdog?: { nextAction?: string } }
 type Deliver = (session: string, text: string, opts: SendOpts) => Promise<SendResult>;
 /** Applicability needs the intended queue row and recipient, not a saved delivery receipt. */
 export type QueueWakeFacts = Pick<OutboxEntry, "outboxId" | "auditPointer" | "destinationSession">;
@@ -34,7 +35,7 @@ export class DraftAwareDelivery {
     const entry = this.outbox.getById(id)!;
     const state = entry.deliveryState === "retained" || entry.deliveryState === "retired" ? "held"
       : entry.deliveryState === "delivered" ? "complete" : entry.deliveryState === "sending" ? "sending" : entry.deliveryState === "failed" ? "failed" : "indeterminate";
-    const reason = entry.deliveryState === "retired" ? "message_retired" : receipt.result?.reason
+    const reason = entry.deliveryState === "retired" ? "message_retired" : receipt.result?.reason ?? receipt.watchdog?.nextAction
       ?? (state === "sending" ? "delivery_in_progress" : state === "complete" ? "delivered" : "interrupted_write");
     return { id: receipt.ids[0]!, state, reason };
   }

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

@@ -3756,7 +3756,7 @@ export class QueueRepository {
   /** Called by the watchdog engine after the delivery attempt is durably
    *  audited. The queue transition records that attempt independently of
    *  whether the HELD row's owner consumed it. */
-  recordWatchdogWakeAttempt(jobId: string, deliveryStatus: string): void {
+  recordWatchdogWakeAttempt(jobId: string, deliveryStatus: string, retryAt?: string): void {
     const bindings = this.wakeRepo.findBlockedQitemsByWatchdog(jobId);
     if (bindings.length === 0) return;
     // Both receipt ownership and timer lifecycle follow the current park.
@@ -3766,11 +3766,11 @@ export class QueueRepository {
         qitemId,
         state: "blocked",
         actorSession: "watchdog@system",
-        transitionNote: deliveryStatus === "retained"
+        transitionNote: retryAt ? `park wake failed: watchdog ${jobId}; delivery=${deliveryStatus}; occurrence remains owed; retry at ${retryAt}` : deliveryStatus === "retained"
           ? `park wake retained: watchdog ${jobId}; not delivered; blocked work unchanged`
           : `park wake fired: watchdog ${jobId}; delivery=${deliveryStatus}; awaiting owner consumption`,
       });
-      this.wakeRepo.record({
+      if (!retryAt) this.wakeRepo.record({
         transitionId: transition.transitionId,
         qitemId,
         phase: "fired",
@@ -3803,7 +3803,7 @@ export class QueueRepository {
     const parkGeneratedTimer = this.wakeRepo.findCurrentQitemsByGeneratedTimer(jobId).length > 0;
     const events = this.db.transaction(() => {
       const firedEvents = targets.map(recordFired);
-      if (deliveryStatus === "retained") return firedEvents;
+      if (retryAt || deliveryStatus === "retained") return firedEvents;
       if (usageLimitBlockers.length === 0) {
         const jobsRepo = this.watchdogJobsRepo ?? new WatchdogJobsRepository(this.db);
         if (parkGeneratedTimer && !backOffQueueWait(jobsRepo, jobId, deliveryStatus)) {
@@ -3839,7 +3839,9 @@ export class QueueRepository {
       );
       return [...firedEvents, ...resolutionEvents];
     })();
-    for (const event of events) this.eventBus.notifySubscribers(event);
+    for (const event of events) {
+      if (!this.eventBus.registerPersistedWithinActiveEnvelope(event)) this.eventBus.notifySubscribers(event);
+    }
   }
 
   /** Close a direct human request and retain its typed answer in the same transaction.

packages/daemon/src/domain/session-transport.ts

@@ -2693,6 +2693,7 @@ export class SessionTransport {
         sessionName,
         reason: "transport_unavailable",
         error: `Session '${sessionName}' is attached as an external CLI node. Inbound tmux transport is unavailable for this target.`,
+        sent: false, inputPhase: "not_reached",
       } };
     }
     const runtime = meta.runtime;
@@ -2737,6 +2738,7 @@ export class SessionTransport {
           sessionName,
           reason: "session_missing",
           error: `Session '${sessionName}' not found: tmux reports no session with this name. No text was sent. Check available sessions with: rig ps --nodes`,
+          sent: false, inputPhase: "not_reached",
         } };
       }
       if (probe.state === "transport_unavailable") {
@@ -2745,6 +2747,7 @@ export class SessionTransport {
           sessionName,
           reason: "tmux_unavailable",
           error: `The tmux server could not be reached (${probe.cause}). Whether session '${sessionName}' exists was not determined. No text was sent.`,
+          sent: false, inputPhase: "not_reached",
         } };
       }
     } catch (err) {
@@ -2753,6 +2756,7 @@ export class SessionTransport {
         sessionName,
         reason: "tmux_unavailable",
         error: `The tmux session probe failed unexpectedly (${err instanceof Error ? err.message : String(err)}). Whether session '${sessionName}' exists was not determined. No text was sent.`,
+        sent: false, inputPhase: "not_reached",
       } };
     }
     const deadlineAt = Date.now() + timeoutMs;
@@ -2870,6 +2874,7 @@ export class SessionTransport {
         sessionName,
         reason: "transport_unavailable",
         error: `Session '${sessionName}' is attached as an external CLI node. Inbound tmux transport is unavailable for this target.`,
+        sent: false, inputPhase: "not_reached",
       };
     }
 
@@ -2923,6 +2928,7 @@ export class SessionTransport {
           sessionName,
           reason: "session_missing",
           error: `Session '${sessionName}' not found: tmux reports no session with this name. No text was sent. Check available sessions with: rig ps --nodes`,
+          sent: false, inputPhase: "not_reached",
         };
       }
       if (probe.state === "transport_unavailable") {
@@ -2931,6 +2937,7 @@ export class SessionTransport {
           sessionName,
           reason: "tmux_unavailable",
           error: `The tmux server could not be reached (${probe.cause}). Whether session '${sessionName}' exists was not determined. No text was sent.`,
+          sent: false, inputPhase: "not_reached",
         };
       }
     } catch (err) {
@@ -2939,6 +2946,7 @@ export class SessionTransport {
         sessionName,
         reason: "tmux_unavailable",
         error: `The tmux session probe failed unexpectedly (${err instanceof Error ? err.message : String(err)}). Whether session '${sessionName}' exists was not determined. No text was sent.`,
+        sent: false, inputPhase: "not_reached",
       };
     }
 
@@ -3227,6 +3235,7 @@ export class SessionTransport {
             ok: false,
             sessionName,
             reason: "target_needs_input",
+            sent: false, inputPhase: "not_reached",
             activity: readiness,
             error: `Refused: '${sessionName}' is at an interactive prompt (${readiness.reason}). A message must not select or approve it. To deliberately drive the prompt: rig send ${sessionName} "<text>" --dangerously-interact --reason "<why>".${clearAttentionHint(sessionName, readiness.reason)} No text was sent.`,
           };

packages/daemon/src/domain/watchdog-delivery-receipts.ts

@@ -0,0 +1,128 @@
+import { OutboxHandler } from "./outbox-handler.js";
+import { resolveWakeRetryCap, resolveWakeRetryIntervalSeconds } from "./queue-wake-ladder.js";
+import type { DeliveryFn, DeliveryOutcome, WatchdogDeliverySource } from "./watchdog-policy-engine.js";
+
+const PREFIX = "watchdog-occurrence-";
+type Receipt = { attempts: number; retryAt?: string; message: string; nextAction?: string; terminalReason?: string; ended?: boolean; error?: string; heldIds?: string[] };
+
+/** Occurrence custody uses the existing outbox. The watchdog's own loop is its only retry driver.
+ * Condition policies keep their existing receipts; this is for scheduled periodic reminders, including digests. */
+export function withWatchdogReceipts(send: DeliveryFn, deps: {
+  outbox: OutboxHandler; now?: () => Date; retryIntervalSeconds?: number; retryCap?: number;
+}): DeliveryFn {
+  const { outbox } = deps;
+  const now = deps.now ?? (() => new Date());
+  const interval = deps.retryIntervalSeconds ?? resolveWakeRetryIntervalSeconds();
+  const cap = deps.retryCap ?? resolveWakeRetryCap();
+  const members = (source: WatchdogDeliverySource) => source.policy !== "periodic-reminder" || !source.occurrenceId ? []
+    : (source.members ?? [{ jobId: source.jobId, occurrenceId: source.occurrenceId }])
+      .map(member => ({ ...member, id: PREFIX + member.occurrenceId }));
+  const read = (id: string): Receipt | undefined => {
+    const row = outbox.db.prepare("SELECT guard_delivery FROM outbox_entries WHERE outbox_id=?").get(id) as { guard_delivery: string | null };
+    return row.guard_delivery ? JSON.parse(row.guard_delivery).watchdog : undefined;
+  };
+  const save = (id: string, receipt: Receipt) => outbox.db.prepare(
+    "UPDATE outbox_entries SET guard_delivery=json_set(COALESCE(guard_delivery,'{}'),'$.watchdog',json(?),'$.ids',json(?)) WHERE outbox_id=?",
+  ).run(JSON.stringify(receipt), JSON.stringify([id]), id);
+  const reconcile = (source: WatchdogDeliverySource): DeliveryOutcome | null => {
+    const selected = members(source);
+    if (!selected.length) return null;
+    const entries = selected.map(m => outbox.getById(m.id));
+    if (entries.every(e => e?.deliveryState === "delivered")) {
+      return { status: "ok", occurrenceTracked: true, reconciled: true, outboxIds: selected.map(m => m.id), sentMessage: read(selected[0]!.id)?.message };
+    }
+    if (entries.every(e => e?.deliveryState === "retained" || e?.deliveryState === "retired")) {
+      return { status: "retained", occurrenceTracked: true, reconciled: true,
+        outboxIds: [...new Set(selected.flatMap(m => [m.id, ...(read(m.id)?.heldIds ?? [])]))] };
+    }
+    for (let i = 0; i < entries.length; i++) {
+      const entry = entries[i];
+      if (!entry) continue;
+      const receipt = read(selected[i]!.id);
+      if (receipt?.ended || receipt?.terminalReason) return selected.length > 1 ? { status: "failed", occurrenceTracked: true, deferred: true }
+        : { status: "failed", occurrenceTracked: true, outboxIds: [entry.outboxId],
+          error: receipt.error ?? receipt.terminalReason, occurrenceComplete: true, terminalReason: receipt.terminalReason, nextAction: receipt.nextAction };
+      if (entry.deliveryState === "sending" || entry.deliveryState === "indeterminate") {
+        if (selected.length > 1) return { status: "failed", occurrenceTracked: true, deferred: true };
+        return { status: "failed", occurrenceTracked: true, occurrenceComplete: true,
+          inputPhase: "attempted", error: "Prior scheduled input is unconfirmed; no replay was attempted.",
+          outboxIds: selected.map(m => m.id),
+          nextAction: `Inspect outbox ${entry.outboxId} and the recipient before any manual replay; input is unconfirmed. This occurrence ends; the next regular cycle is a new occurrence.` };
+      }
+      if (entry.deliveryState === "delivered" || entry.deliveryState === "retained" || entry.deliveryState === "retired"
+        || (receipt?.retryAt && Date.parse(receipt.retryAt) > now().getTime())) {
+        // A pass can have collected new jobs before another member's receipt landed. Re-group next pass.
+        return { status: "failed", occurrenceTracked: true, deferred: true, retryAt: receipt?.retryAt };
+      }
+    }
+    return null;
+  };
+
+  const deliver: DeliveryFn = async (request, source) => {
+    const selected = members(source);
+    if (!selected.length) return send(request, source);
+    const prior = reconcile(source);
+    if (prior) return prior;
+    const claimed = outbox.db.transaction(() => {
+      for (const member of selected) {
+        if (!outbox.getById(member.id)) {
+          outbox.record({ outboxId: member.id, senderSession: "watchdog@system", destinationSession: request.targetSession,
+            body: request.message, auditPointer: member.jobId });
+          save(member.id, { attempts: 0, message: request.message });
+        }
+      }
+      // Synchronous transaction: no other drainer can claim a subset between the check and these writes.
+      if (selected.some(m => !["pending", "failed"].includes(outbox.getById(m.id)!.deliveryState))) return false;
+      for (const member of selected) {
+        outbox.claimForDelivery(member.id);
+        const receipt = read(member.id)!;
+        save(member.id, { attempts: (receipt?.attempts ?? 0) + 1, message: request.message });
+      }
+      return true;
+    })();
+    if (!claimed) return { status: "failed", occurrenceTracked: true, deferred: true };
+    let result: DeliveryOutcome;
+    try { result = await send(request, source); }
+    catch (error) { result = { status: "failed", error: (error as Error).message }; }
+    const noInput = result.sent === false || result.inputPhase === "not_reached";
+    const permanent = result.transportReason === "transport_unavailable" || result.code === "terminal_seat"
+      || result.code === "retained_quota_full";
+    const memberOutcomes: Record<string, DeliveryOutcome> = {};
+    outbox.db.transaction(() => {
+      for (const member of selected) {
+        const receipt = read(member.id)!;
+        let retryAt: string | undefined, nextAction: string | undefined, terminalReason: string | undefined;
+        const exhausted = receipt.attempts > cap;
+        const state = result.status === "ok" ? "delivered" : result.status === "retained" ? "retained"
+          : noInput ? "failed" : "indeterminate";
+        if (result.status === "retained") {
+          nextAction = `Retained by existing input protection; inspect held messages ${(result.outboxIds ?? []).join(", ")}. Nothing was delivered.`;
+        }
+        if (result.status === "failed") {
+          if (noInput && permanent) {
+            terminalReason = "watchdog_delivery_unavailable";
+            nextAction = `Inspect outbox ${member.id}: ${result.error ?? result.transportReason ?? "delivery failed"}. Fix the destination and register a replacement reminder.`;
+          } else if (noInput && exhausted) {
+            nextAction = `Inspect outbox ${member.id}: retry budget exhausted. This occurrence ends; the next regular cycle is a new occurrence. A one-shot timer records its failed wake and ends.`;
+          } else if (noInput) {
+            retryAt = new Date(now().getTime() + interval * 1000).toISOString();
+            nextAction = `Occurrence remains owed; retry at ${retryAt}. Inspect outbox ${member.id}.`;
+          } else {
+            nextAction = `Inspect outbox ${member.id} and the recipient before any manual replay; input is unconfirmed. This occurrence ends; the next regular cycle is a new occurrence.`;
+          }
+        }
+        const ended = result.status === "failed" && (!noInput || permanent || exhausted);
+        outbox.finalizeDelivery(member.id, state);
+        save(member.id, { ...receipt, ...(ended ? { ended: true, error: result.error } : {}), ...(result.status === "retained" ? { heldIds: result.outboxIds ?? [] } : {}),
+          ...(retryAt ? { retryAt } : {}), ...(nextAction ? { nextAction } : {}),
+          ...(terminalReason ? { terminalReason } : {}) });
+        memberOutcomes[member.jobId] = { ...result, occurrenceTracked: true, ...(ended ? { occurrenceComplete: true } : {}),
+          outboxIds: [...new Set([member.id, ...(result.outboxIds ?? [])])], ...(retryAt ? { retryAt } : {}),
+          ...(nextAction ? { nextAction } : {}), ...(terminalReason ? { terminalReason } : {}) };
+      }
+    })();
+    return { ...memberOutcomes[selected[0]!.jobId]!, memberOutcomes };
+  };
+  deliver.reconcile = reconcile;
+  return deliver;
+}

packages/daemon/src/domain/watchdog-delivery.ts

@@ -1,4 +1,5 @@
-import { OutboxHandlerError } from "./outbox-handler.js";
+import { OutboxHandlerError, type OutboxHandler } from "./outbox-handler.js";
+import { withWatchdogReceipts } from "./watchdog-delivery-receipts.js";
 import { DeliveryGuardError } from "./seat-delivery-guard.js";
 import { PARKED_OWNER_COMPACTION_PENDING, PARKED_OWNER_POLICY_NAME } from "./policies/parked-owner-consumer.js";
 import type { SessionTransport } from "./session-transport.js";
@@ -8,14 +9,18 @@ import {
   type DeliveryRequest,
 } from "./watchdog-policy-engine.js";
 
-/** Current reminders get an input attempt. Job accounting may be interrupted after input, so a later attempt can repeat. */
+/** The input adapter; scheduled occurrence custody is reconciled outside the actual transport call. */
 export function makeWatchdogDeliver(deps: {
   sessionTransport: Pick<SessionTransport, "send">;
+  outbox?: OutboxHandler;
+  now?: () => Date;
+  retryIntervalSeconds?: number;
+  retryCap?: number;
   hasPendingPostCompactStage?: (sessionName: string) => boolean;
   isTerminalSeat?: (sessionName: string) => boolean;
   createContinuityAction?: (action: NonNullable<DeliveryRequest["continuityAction"]>) => Promise<void>;
 }): DeliveryFn {
-  return async ({ targetSession, message, continuityAction }, source) => {
+  const deliver: DeliveryFn = async ({ targetSession, message, continuityAction }, source) => {
     let continuityActionCompleted = false;
     try {
       const checkTerminal = () => {
@@ -27,10 +32,10 @@ export function makeWatchdogDeliver(deps: {
         await deps.createContinuityAction?.(continuityAction);
         continuityActionCompleted = true;
       }
-      const payload = formatWatchdogDeliveryMessage(source,
-        `Delivery may repeat after an unconfirmed attempt.\n${message}`);
-      // SessionTransport still owns recipient, approval and draft protection. A historical delivery ID must not
-      // substitute an earlier result for this current request, including when several reminders share one message.
+      const warning = deps.outbox && source.policy === "periodic-reminder" ? "" : "Delivery may repeat after an unconfirmed attempt.\n";
+      const payload = formatWatchdogDeliveryMessage(source, warning + message);
+      // SessionTransport still owns recipient, approval and draft protection. Only an occurrence admitted by the
+      // outer receipt check reaches this current input request, including when several reminders share it.
       const beforeWrite = source.policy === PARKED_OWNER_POLICY_NAME && (deps.hasPendingPostCompactStage || deps.isTerminalSeat) ? () => {
         checkTerminal();
         if (deps.hasPendingPostCompactStage?.(targetSession)) throw new DeliveryGuardError("managed_compaction_pending", PARKED_OWNER_COMPACTION_PENDING);
@@ -40,7 +45,8 @@ export function makeWatchdogDeliver(deps: {
       if (result.outcome === "retained") return { status: "retained", outboxIds: result.outboxIds, continuityActionCompleted };
       return result.ok
         ? { status: "ok", continuityActionCompleted }
-        : { status: "failed", error: result.error, ...(result.inputPhase ? { inputPhase: result.inputPhase } : {}),
+        : { status: "failed", error: result.error, sent: result.sent, ...(result.reason ? { transportReason: result.reason } : {}),
+          ...(result.inputPhase ? { inputPhase: result.inputPhase } : {}),
           ...(result.sent === false && result.inputPhase === "not_reached" && result.noInputCode ? { code: result.noInputCode } : {}), continuityActionCompleted };
     } catch (err) {
       // Only these typed refusals establish the input boundary. Matching error text never does.
@@ -54,4 +60,5 @@ export function makeWatchdogDeliver(deps: {
       };
     }
   };
+  return deps.outbox ? withWatchdogReceipts(deliver, { ...deps, outbox: deps.outbox }) : deliver;
 }

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

@@ -65,6 +65,19 @@ export interface DeliveryOutcome {
   continuityActionCompleted?: boolean;
   /** The text reported by a delivery implementation, without its scheduler header. */
   sentMessage?: string;
+  /** Durable custody of this scheduled occurrence, separate from its recurrence. */
+  occurrenceTracked?: boolean;
+  sent?: boolean;
+  /** Failed occurrence is finished; a recurring job can start its next normal cycle. */
+  occurrenceComplete?: boolean;
+  /** A retained receipt or retry time was checked; no new input attempt happened. */
+  deferred?: boolean;
+  reconciled?: boolean;
+  transportReason?: string;
+  retryAt?: string;
+  nextAction?: string;
+  terminalReason?: string;
+  memberOutcomes?: Record<string, DeliveryOutcome>;
 }
 
 export interface WatchdogDeliverySource {
@@ -75,10 +88,13 @@ export interface WatchdogDeliverySource {
   members?: Array<{ jobId: string; occurrenceId: string }>;
 }
 
-export type DeliveryFn = (
+export type DeliveryFn = ((
   req: DeliveryRequest,
   source: WatchdogDeliverySource,
-) => Promise<DeliveryOutcome>;
+) => Promise<DeliveryOutcome>) & {
+  /** Resolve existing occurrence custody before grouping it with new reminders. */
+  reconcile?: (source: WatchdogDeliverySource) => DeliveryOutcome | null;
+};
 
 export function formatWatchdogDeliveryMessage(
   source: WatchdogDeliverySource,
@@ -148,7 +164,7 @@ interface WatchdogPolicyEngineDeps {
   /** Queue-side observer for a wake attempt. It appends resume evidence to
    *  every HELD row armed to this job; delivery outcome is preserved. */
   resolveQueueWait?: (input: { jobId: string }) => PolicyEvaluation | null | undefined;
-  onWakeAttempt?: (attempt: { jobId: string; deliveryStatus: string }) => void;
+  onWakeAttempt?: (attempt: { jobId: string; deliveryStatus: string; retryAt?: string }) => void;
   /** The parked row a due reminder is about, when it may share one message with other reminders due for the same
    *  seat in the same scheduler pass. Null, or absent, keeps the reminder's own send. */
   resolveDigestEntry?: (input: { jobId: string; message: string }) => ParkReminderEntry | null;
@@ -262,7 +278,7 @@ export class WatchdogPolicyEngine {
   private readonly resolveContextUsage?: WatchdogPolicyEngineDeps["resolveContextUsage"];
   private readonly resolvePreDeliveryTerminalReason?: (input: { jobId: string }) => string | null;
   private readonly resolveQueueWait?: WatchdogPolicyEngineDeps["resolveQueueWait"];
-  private readonly onWakeAttempt?: (attempt: { jobId: string; deliveryStatus: string }) => void;
+  private readonly onWakeAttempt?: (attempt: { jobId: string; deliveryStatus: string; retryAt?: string }) => void;
   private readonly resolveDigestEntry?: WatchdogPolicyEngineDeps["resolveDigestEntry"];
 
   constructor(deps: WatchdogPolicyEngineDeps) {
@@ -576,11 +592,15 @@ export class WatchdogPolicyEngine {
 
     const occurrence = [job.jobId, occupantGeneration, outcome.conditionReceipt ?? (job.policy === "periodic-reminder" ? job.lastFireAt : outcome.message)];
     const recurring = outcome.conditionReceipt === undefined && job.policy !== "periodic-reminder" && job.policy !== "context-usage-threshold";
-    // Identify cadences for digest grouping and history, not transport deduplication.
+    // Identify cadence for grouping/history and durable scheduled-occurrence custody, separate from explicit sends.
     if (recurring && job.lastFireAt !== null) {
       occurrence.push(job.lastFireAt);
     }
     const occurrenceId = createHash("sha256").update(JSON.stringify(occurrence)).digest("hex");
+    const recovered = this.deliver.reconcile?.({ jobId: job.jobId, policy: job.policy, occurrenceId });
+    if (recovered) return this.recordAttempt({ job, outcome, evaluatedAt,
+      deliverySchedule: this.jobsRepo.deliverySchedule(job.jobId), delivery: recovered,
+      isContextUsageThreshold: false, notice: outcome.notice === true });
     // A parked-row reminder due in a scheduler pass waits for the pass's end, when every reminder held for its seat
     // goes in one message. Nothing has changed for it yet, so a pass that stops early leaves it due.
     const digestEntry = job.policy === "periodic-reminder" && !continuityAction
@@ -718,7 +738,7 @@ export class WatchdogPolicyEngine {
       { jobId: digestId, policy: "periodic-reminder", occurrenceId: digestId, members },
     );
     digest.forEach((held, i) => this.recordAttempt({
-      ...held, deliverySchedule: schedules[i]!, delivery, isContextUsageThreshold: false,
+      ...held, deliverySchedule: schedules[i]!, delivery: delivery.memberOutcomes?.[held.job.jobId] ?? delivery, isContextUsageThreshold: false,
       digest: { deliveryId: digestId, size: digest.length, line: lines[i]! },
     }));
   }
@@ -735,7 +755,7 @@ export class WatchdogPolicyEngine {
     this.recordAttempt({ ...held, deliverySchedule: schedule, delivery, isContextUsageThreshold: false, sentMessage: message });
   }
 
-  /** Audit one job's delivery attempt and advance its schedule, wake receipt and actionable state from it. */
+  /** Account for one input attempt or a recovered receipt, then advance only accounted occurrences. */
   private recordAttempt(input: {
     job: WatchdogJob;
     outcome: Extract<PolicyEvaluation, { action: "send" }>;
@@ -751,18 +771,26 @@ export class WatchdogPolicyEngine {
     sentMessage?: string;
   }): EvaluationResult {
     const { job, outcome, evaluatedAt, deliverySchedule, delivery, isContextUsageThreshold, digest, sentMessage } = input;
+    if (delivery.deferred) {
+      this.jobsRepo.deliveryAccounted(job.jobId);
+      return { job: this.jobsRepo.getByIdOrThrow(job.jobId), outcome, history: null, delivery, meaningful: false };
+    }
     const notes = digest
       ? { ...(outcome.notes ?? {}), digestDeliveryId: digest.deliveryId, digestSize: digest.size }
       : outcome.notes;
     const deliveryNotes = delivery.inputPhase === undefined && delivery.code === undefined ? notes
       : { ...(notes ?? {}), ...(delivery.inputPhase === undefined ? {} : { deliveryInputPhase: delivery.inputPhase }),
         ...(delivery.code === undefined ? {} : { deliveryCode: delivery.code }) };
+    const receiptNotes = delivery.retryAt || delivery.nextAction || delivery.occurrenceTracked
+      ? { ...(deliveryNotes ?? {}), ...(delivery.retryAt ? { retryAt: delivery.retryAt } : {}),
+        ...(delivery.nextAction ? { nextAction: delivery.nextAction } : {}),
+        ...(delivery.occurrenceTracked ? { outboxIds: delivery.outboxIds ?? [] } : {}) } : deliveryNotes;
     const history = this.historyLog.record({
       jobId: job.jobId,
       evaluatedAt,
-      outcome: delivery.status === "retained" ? "skipped" : delivery.status === "ok" ? "sent"
+      outcome: delivery.reconciled ? "skipped" : delivery.status === "retained" ? "skipped" : delivery.status === "ok" ? "sent"
         : isRefusedInteractive(delivery.error) ? "refused" : "failed",
-      skipReason: delivery.status === "retained" ? "typing_guard_retained" : null,
+      skipReason: delivery.reconciled ? "delivery_receipt_reconciled" : delivery.status === "retained" ? "typing_guard_retained" : null,
       deliveryTargetSession: outcome.target.session,
       deliveryStatus: delivery.status,
       deliveryMessage: digest?.line ?? delivery.sentMessage ?? sentMessage ?? outcome.message,
@@ -771,16 +799,31 @@ export class WatchdogPolicyEngine {
       // refusal from a generic failure (status alone discards that identity).
       evaluationNotes:
         delivery.error !== undefined
-          ? { ...(deliveryNotes ?? {}), deliveryReason: delivery.error }
-          : delivery.status === "retained" ? { ...(deliveryNotes ?? {}), outboxIds: delivery.outboxIds ?? [], retainedNotDelivered: true } : deliveryNotes ?? null,
+          ? { ...receiptNotes, deliveryReason: delivery.error }
+          : delivery.status === "retained" ? { ...receiptNotes, outboxIds: delivery.outboxIds ?? [], retainedNotDelivered: true } : receiptNotes ?? null,
     });
-    // A notice stamps no fire time and leaves the job not actionable, so the active-wake throttle can't hold
-    // back the watched condition's own send after it.
-    if (input.notice) {
-      this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, false);
-    } else if (!isContextUsageThreshold) {
-      this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, true, deliverySchedule);
-    }
+    const retryPending = delivery.occurrenceTracked && delivery.status === "failed" && !delivery.occurrenceComplete;
+    const accountWake = () => {
+      const current = this.jobsRepo.getByIdOrThrow(job.jobId);
+      if (current.state === "active" && this.jobsRepo.scheduleUnchanged(current, deliverySchedule)
+        && (isContextUsageThreshold || current.lastEvaluationAt === evaluatedAt
+          || (retryPending && current.lastEvaluationAt === deliverySchedule.lastEvaluationAt))) {
+        this.onWakeAttempt?.({ jobId: job.jobId, deliveryStatus: delivery.status,
+          ...(retryPending && delivery.retryAt ? { retryAt: delivery.retryAt } : {}) });
+      }
+      if (delivery.terminalReason) this.jobsRepo.markTerminal(job.jobId, delivery.terminalReason);
+    };
+    const account = () => {
+      // The fire stamp and queue wake receipt/one-shot retirement commit together. A crash here leaves
+      // the old occurrence identity, whose durable positive receipt can finish accounting without input.
+      if (input.notice) this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, false);
+      else if (!isContextUsageThreshold && !retryPending) {
+        this.jobsRepo.recordEvaluation(job.jobId, evaluatedAt, true, deliverySchedule);
+      }
+      if (delivery.occurrenceTracked) accountWake();
+    };
+    if (delivery.occurrenceTracked) this.eventBus.withNotifyEnvelope(account);
+    else account();
     // OPR.0.5.8.1 S2 — a policy's condition receipt is banked ONLY on positive
     // delivery. `status` is "ok" | "failed"; anything that is not a definite ok
     // leaves the receipt untouched so the next scan retries. A policy that does
@@ -798,13 +841,7 @@ export class WatchdogPolicyEngine {
       targetSession: outcome.target.session,
       deliveryStatus: delivery.status,
     });
-    // The history above audits the actual attempt. A newer park/blocker change
-    // must not receive its receipt or be backed off as though we sent for it.
-    const current = this.jobsRepo.getByIdOrThrow(job.jobId);
-    if (current.state === "active" && this.jobsRepo.scheduleUnchanged(current, deliverySchedule)
-      && (isContextUsageThreshold || current.lastEvaluationAt === evaluatedAt)) {
-      this.onWakeAttempt?.({ jobId: job.jobId, deliveryStatus: delivery.status });
-    }
+    if (!delivery.occurrenceTracked) accountWake();
     this.jobsRepo.deliveryAccounted(job.jobId);
     return {
       job: this.jobsRepo.getByIdOrThrow(job.jobId),

packages/daemon/src/startup.ts

@@ -479,6 +479,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
   // db.transaction commits atomically with the close + successor create.
   const outboxHandlerInstance = new OutboxHandler(db);
   queueRepoInstance.attachOutbox(outboxHandlerInstance);
+  outboxHandlerInstance.reconcileAbandonedSending("watchdog-occurrence-");
   // Reconcile only the prior process's claims, before transport/producers/timers can start new sends.
   let reconciledWakeIntents = 0;
   try {
@@ -2292,6 +2293,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
       eventBus,
       deliver: makeWatchdogDeliver({
         sessionTransport,
+        outbox: outboxHandlerInstance,
         isTerminalSeat: (sessionName) => isBoundTerminalSeat(db, sessionName),
         hasPendingPostCompactStage: (sessionName) => deps.compactionEnforcer?.hasPendingPostCompactStage(sessionName) ?? false,
         createContinuityAction: (action) => createContinuityCutoverBaton(action, queueRepoInstance).then(() => {}),
@@ -2308,8 +2310,8 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
       resolvePreDeliveryTerminalReason: ({ jobId }) =>
         queueRepoInstance.resolveWatchdogPreDeliveryTerminalReason(jobId),
       resolveQueueWait: (input) => queueRepoInstance.evaluateWaitReminder(input),
-      onWakeAttempt: ({ jobId, deliveryStatus }) => {
-        queueRepoInstance.recordWatchdogWakeAttempt(jobId, deliveryStatus);
+      onWakeAttempt: ({ jobId, deliveryStatus, retryAt }) => {
+        queueRepoInstance.recordWatchdogWakeAttempt(jobId, deliveryStatus, retryAt);
       },
       // Due parked-row reminders for one seat in one scheduler pass go as one message.
       resolveDigestEntry: ({ jobId, message }) => queueRepoInstance.parkReminderDigestEntry(jobId, message),

packages/daemon/test/watchdog-delivery-recovery.test.ts

@@ -0,0 +1,197 @@
+// The real watchdog loop and receipt store; terminal input is inert.
+import Database from "better-sqlite3";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { migrate } from "../src/db/migrate.js";
+import { TmuxAdapter } from "../src/adapters/tmux.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { OutboxHandler } from "../src/domain/outbox-handler.js";
+import { RigRepository } from "../src/domain/rig-repository.js";
+import { SessionRegistry } from "../src/domain/session-registry.js";
+import { SessionTransport, type SendResult } from "../src/domain/session-transport.js";
+import { makeWatchdogDeliver } from "../src/domain/watchdog-delivery.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogPolicyEngine } from "../src/domain/watchdog-policy-engine.js";
+import { WatchdogScheduler } from "../src/domain/watchdog-scheduler.js";
+
+const seat = "worker@test";
+const unavailable: SendResult = { ok: false, sessionName: seat, sent: false, inputPhase: "not_reached",
+  reason: "tmux_unavailable", error: "The terminal service is not listening yet." };
+const delivered: SendResult = { ok: true, sessionName: seat, sent: true };
+const stores: Database.Database[] = [];
+beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-01-01T00:00:00Z")); });
+afterEach(() => { for (const db of stores.splice(0)) if (db.open) db.close(); vi.useRealTimers(); });
+const later = (seconds: number) => vi.setSystemTime(Date.now() + seconds * 1000);
+function database(saved?: Buffer) {
+  const db = new Database(saved ?? ":memory:"); stores.push(db);
+  if (!saved) migrate(db, ALL_MIGRATIONS);
+  return db;
+}
+function boot(db: Database.Database, send: (session: string, body: string) => Promise<SendResult>, digest = false) {
+  const jobs = new WatchdogJobsRepository(db), history = new WatchdogHistoryLog(db), outbox = new OutboxHandler(db);
+  const deps = { sessionTransport: { send }, outbox, now: () => new Date(), retryIntervalSeconds: 300, retryCap: 3 };
+  const engine = new WatchdogPolicyEngine({ jobsRepo: jobs, historyLog: history, eventBus: new EventBus(db),
+    deliver: makeWatchdogDeliver(deps), now: () => new Date(),
+    resolveDigestEntry: digest ? ({ jobId, message }) => ({ qitemId: jobId, summary: message }) : undefined });
+  const errors: unknown[] = [];
+  const scheduler = new WatchdogScheduler({ jobsRepo: jobs, policyEngine: engine, now: () => new Date(), onTickError: e => errors.push(e) });
+  return { db, jobs, history, outbox, scheduler, errors };
+}
+function arm(f: ReturnType<typeof boot>) {
+  const job = f.jobs.register({ policy: "periodic-reminder", specYaml: `target: { session: ${seat} }\nmessage: Check the work\n`,
+    targetSession: seat, intervalSeconds: 3600, registeredBySession: "owner@test" });
+  later(3600);
+  return job;
+}
+function occurrence(f: ReturnType<typeof boot>, jobId: string) {
+  return f.outbox.listForSender("watchdog@system").find(e => e.auditPointer === jobId)!;
+}
+
+describe("an undelivered scheduled occurrence remains owed", () => {
+  it("recovers after the recipient starts, with bounded repeated passes and durable identity across restart", async () => {
+    let result = unavailable;
+    const send = vi.fn(async () => result);
+    const first = boot(database(), send), job = arm(first);
+    await first.scheduler.runTickNow();
+    expect(first.history.listForJob(job.jobId)[0]?.deliveryStatus).toBe("failed");
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(first.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBeNull();
+    const id = occurrence(first, job.jobId).outboxId;
+    await first.scheduler.runTickNow(); later(299); await first.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(1);
+    const saved = first.db.serialize(); first.db.close();
+    const resumed = boot(database(saved), send);
+    expect(occurrence(resumed, job.jobId).outboxId).toBe(id);
+    result = delivered; later(1); await resumed.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(2);
+    expect(resumed.outbox.getById(id)?.deliveryState).toBe("delivered");
+    expect(resumed.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBe(new Date().toISOString());
+    await resumed.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(2);
+  });
+
+  it("reconciles a confirmed delivery after accounting was interrupted, without typing it twice", async () => {
+    const send = vi.fn(async () => delivered), first = boot(database(), send), job = arm(first);
+    vi.spyOn(first.jobs, "recordEvaluation").mockImplementationOnce(() => { throw new Error("accounting interrupted"); });
+    await first.scheduler.runTickNow();
+    expect(first.errors).toHaveLength(1);
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(first.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBeNull();
+    const saved = first.db.serialize(); first.db.close();
+    const resumed = boot(database(saved), send);
+    later(1); // Distinct evaluation times: random ULIDs do not order two records at the same fake instant.
+    await resumed.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(resumed.jobs.getByIdOrThrow(job.jobId).lastFireAt).not.toBeNull();
+    expect(resumed.history.listForJob(job.jobId)[0]?.skipReason).toBe("delivery_receipt_reconciled");
+  });
+
+  it("ends an uncertain occurrence across restart without treating it as unsent", async () => {
+    const send = vi.fn(async () => ({ ok: false, sessionName: seat, inputPhase: "attempted" as const,
+      reason: "delivery_indeterminate", error: "Input acknowledgement was lost." }));
+    const first = boot(database(), send), job = arm(first);
+    await first.scheduler.runTickNow();
+    expect(first.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBe("2026-01-01T01:00:00.000Z");
+    const id = occurrence(first, job.jobId).outboxId;
+    expect(first.outbox.getById(id)?.deliveryState).toBe("indeterminate");
+    const saved = first.db.serialize(); first.db.close();
+    const resumed = boot(database(saved), send);
+    later(600); await resumed.scheduler.runTickNow(); await resumed.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(resumed.outbox.getById(id)?.deliveryState).toBe("indeterminate");
+    expect(resumed.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBe("2026-01-01T01:00:00.000Z");
+    expect(resumed.history.listForJob(job.jobId)[0]?.evaluationNotes?.nextAction).toContain("unconfirmed");
+  });
+
+  it("keeps a non-retryable failure visible with its cause and next action", async () => {
+    const send = vi.fn(async () => ({ ...unavailable, reason: "transport_unavailable", error: "This external CLI target has no inbound terminal transport." }));
+    const f = boot(database(), send), job = arm(f);
+    await f.scheduler.runTickNow();
+    expect(f.history.listForJob(job.jobId)[0]?.evaluationNotes?.deliveryReason).toContain("external CLI");
+    expect(f.jobs.getByIdOrThrow(job.jobId).state).toBe("terminal");
+    expect(f.jobs.getByIdOrThrow(job.jobId).lastFireAt).not.toBeNull();
+    expect(f.history.listForJob(job.jobId)[0]?.evaluationNotes?.nextAction).toEqual(expect.any(String));
+    later(86_400); await f.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(1);
+  });
+
+  it("exhausts definite no-input retries instead of churning forever", async () => {
+    const send = vi.fn(async () => unavailable), f = boot(database(), send), job = arm(f);
+    await f.scheduler.runTickNow();
+    for (let i = 0; i < 6; i++) { later(300); await f.scheduler.runTickNow(); }
+    expect(send).toHaveBeenCalledTimes(4); // Original attempt and the existing three-retry cap.
+    expect(f.jobs.getByIdOrThrow(job.jobId).state).toBe("active");
+    expect(f.history.listForJob(job.jobId)[0]?.evaluationNotes?.nextAction).toContain("exhausted");
+    expect(f.jobs.getByIdOrThrow(job.jobId).lastFireAt).not.toBeNull();
+  });
+
+  it("keeps a successful reminder's selected recurrence unchanged", async () => {
+    const send = vi.fn(async () => delivered), f = boot(database(), send), job = arm(f);
+    await f.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(f.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBe(new Date().toISOString());
+    later(3599); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(1);
+    later(1); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(2);
+  });
+
+  it("records a real terminal-service preflight failure as definitely before input", async () => {
+    const db = database();
+    db.exec(`INSERT INTO rigs(id,name) VALUES ('r','test');
+      INSERT INTO nodes(id,rig_id,logical_id,runtime) VALUES ('n','r','worker','claude-code');
+      INSERT INTO sessions(id,node_id,session_name,status) VALUES ('s','n','worker@test','running');`);
+    const exec = vi.fn(async () => ""), tmux = new TmuxAdapter(exec);
+    vi.spyOn(tmux, "probeSession").mockResolvedValue({ state: "transport_unavailable", cause: "terminal service has not started" });
+    const transport = new SessionTransport({ db, tmuxAdapter: tmux, rigRepo: new RigRepository(db), sessionRegistry: new SessionRegistry(db) });
+    const result = await transport.send(seat, "The reminder is owed", { actorSession: "watchdog@system", openrigNotice: true });
+    expect(result.reason).toBe("tmux_unavailable");
+    expect(exec).not.toHaveBeenCalled();
+    expect(result).toMatchObject({ ok: false, sent: false, inputPhase: "not_reached" });
+  });
+
+  it("accounts for every confirmed digest member after interrupted accounting without repeating the digest", async () => {
+    const send = vi.fn(async () => delivered), first = boot(database(), send, true);
+    const register = () => first.jobs.register({ policy: "periodic-reminder", specYaml: `message: Check the work\n`,
+      targetSession: seat, intervalSeconds: 3600, registeredBySession: "owner@test" });
+    const a = register(), b = register(); later(3600);
+    vi.spyOn(first.jobs, "recordEvaluation").mockImplementationOnce(() => { throw new Error("digest accounting interrupted"); });
+    await first.scheduler.runTickNow();
+    expect(first.errors).toHaveLength(1); expect(send).toHaveBeenCalledTimes(1);
+    expect(first.jobs.getByIdOrThrow(a.jobId).lastFireAt).toBeNull();
+    expect(first.jobs.getByIdOrThrow(b.jobId).lastFireAt).toBeNull();
+    const saved = first.db.serialize(); first.db.close();
+    const resumed = boot(database(saved), send, true); await resumed.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(resumed.jobs.getByIdOrThrow(a.jobId).lastFireAt).not.toBeNull();
+    expect(resumed.jobs.getByIdOrThrow(b.jobId).lastFireAt).not.toBeNull();
+  });
+
+  it("keeps each digest member's own retry budget when an older failure joins a new reminder", async () => {
+    let result = unavailable;
+    const send = vi.fn(async () => result), f = boot(database(), send, true), old = arm(f);
+    await f.scheduler.runTickNow();
+    later(300); await f.scheduler.runTickNow(); later(300); await f.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(3);
+    later(299);
+    const fresh = f.jobs.register({ policy: "periodic-reminder", specYaml: `message: Fresh work\n`,
+      targetSession: seat, intervalSeconds: 1, registeredBySession: "owner@test" });
+    later(1); await f.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(4);
+    expect(f.jobs.getByIdOrThrow(old.jobId).state).toBe("active");
+    expect(f.jobs.getByIdOrThrow(fresh.jobId).state).toBe("active");
+    expect(f.jobs.getByIdOrThrow(fresh.jobId).lastFireAt).toBeNull();
+    result = delivered; later(300); await f.scheduler.runTickNow();
+    expect(send).toHaveBeenCalledTimes(5);
+    expect(f.jobs.getByIdOrThrow(fresh.jobId).lastFireAt).not.toBeNull();
+  });
+
+  it("keeps deliberate input retention in its existing custody", async () => {
+    const send = vi.fn(async () => ({ ok: true, sessionName: seat, outcome: "retained" as const,
+      sent: false, outboxIds: ["held-original"] }));
+    const f = boot(database(), send), job = arm(f); await f.scheduler.runTickNow();
+    expect(f.jobs.getByIdOrThrow(job.jobId).lastFireAt).not.toBeNull();
+    expect(occurrence(f, job.jobId).deliveryState).toBe("retained");
+    expect(f.history.listForJob(job.jobId)[0]?.evaluationNotes?.outboxIds).toContain("held-original");
+    later(300); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(1);
+  });
+});

packages/daemon/test/watchdog-delivery-round1.test.ts

@@ -0,0 +1,155 @@
+// Real migrated queue/watchdog accounting, with inert terminal input and a deterministic clock.
+import Database from "better-sqlite3";
+import { afterEach, beforeEach, expect, it, vi } from "vitest";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { migrate } from "../src/db/migrate.js";
+import { TmuxAdapter } from "../src/adapters/tmux.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { OutboxHandler } from "../src/domain/outbox-handler.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import { RigRepository } from "../src/domain/rig-repository.js";
+import { SessionRegistry } from "../src/domain/session-registry.js";
+import { SessionTransport, type SendResult } from "../src/domain/session-transport.js";
+import { makeWatchdogDeliver } from "../src/domain/watchdog-delivery.js";
+import { WatchdogHistoryLog } from "../src/domain/watchdog-history-log.js";
+import { WatchdogJobsRepository } from "../src/domain/watchdog-jobs-repository.js";
+import { WatchdogPolicyEngine } from "../src/domain/watchdog-policy-engine.js";
+import { WatchdogScheduler } from "../src/domain/watchdog-scheduler.js";
+
+const seat = "worker@test", stores: Database.Database[] = [];
+const absent: SendResult = { ok: false, sessionName: seat, sent: false, inputPhase: "not_reached", reason: "tmux_unavailable", error: "Terminal service not yet listening" };
+const delivered: SendResult = { ok: true, sessionName: seat, sent: true };
+beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-01-01T00:00:00Z")); });
+afterEach(() => { for (const db of stores.splice(0)) if (db.open) db.close(); vi.useRealTimers(); });
+const later = (s: number) => vi.setSystemTime(Date.now() + s * 1000);
+function fixture(send: (session: string, body: string) => Promise<SendResult>, saved?: Buffer, digest = false) {
+  const db = new Database(saved ?? ":memory:"); stores.push(db); if (!saved) migrate(db, ALL_MIGRATIONS);
+  const bus = new EventBus(db), outbox = new OutboxHandler(db), jobs = new WatchdogJobsRepository(db), history = new WatchdogHistoryLog(db);
+  const repo = new QueueRepository(db, bus);
+  repo.attachWatchdogJobsRepository(jobs);
+  // The optional argument is ignored by the old callback; the test still exercises its actual old behavior.
+  const account = repo.recordWatchdogWakeAttempt.bind(repo) as (id: string, status: string, retryAt?: string) => void;
+  const engine = new WatchdogPolicyEngine({ jobsRepo: jobs, historyLog: history, eventBus: bus,
+    deliver: makeWatchdogDeliver({ sessionTransport: { send }, outbox, now: () => new Date(), retryIntervalSeconds: 300, retryCap: 3 }),
+    resolvePreDeliveryTerminalReason: ({ jobId }) => repo.resolveWatchdogPreDeliveryTerminalReason(jobId),
+    onWakeAttempt: ({ jobId, deliveryStatus, retryAt }: { jobId: string; deliveryStatus: string; retryAt?: string }) => account(jobId, deliveryStatus, retryAt),
+    resolveDigestEntry: digest ? ({ jobId, message }) => ({ qitemId: jobId, summary: message }) : undefined });
+  const errors: unknown[] = [], scheduler = new WatchdogScheduler({ jobsRepo: jobs, policyEngine: engine, onTickError: e => errors.push(e) });
+  const park = async () => {
+    const row = await repo.create({ sourceSession: "owner@test", destinationSession: seat, body: "Continue work", nudge: false });
+    repo.update({ qitemId: row.qitemId, actorSession: seat, state: "blocked", blockedOn: "external:wait", transitionNote: "continuation: work", wakeAfterSeconds: 3600 } as never);
+    return { qitemId: row.qitemId, jobId: repo.getParkWakeStatus(row.qitemId)!.ref };
+  };
+  const reminder = () => jobs.register({ policy: "periodic-reminder", specYaml: `message: Check work\n`, targetSession: seat, intervalSeconds: 3600, registeredBySession: "owner@test" });
+  return { db, bus, jobs, history, outbox, repo, errors, scheduler, park, reminder };
+}
+
+it("exhausts one occurrence but delivers the next normal recurring cycle under a new identity", async () => {
+  let result = absent; const send = vi.fn(async () => result), f = fixture(send), job = f.reminder();
+  later(3600); await f.scheduler.runTickNow();
+  for (let i = 0; i < 3; i++) { later(300); await f.scheduler.runTickNow(); }
+  expect(send).toHaveBeenCalledTimes(4);
+  const old = f.outbox.listForSender("watchdog@system")[0]!;
+  expect(old.deliveryState).toBe("failed");
+  expect(f.jobs.getByIdOrThrow(job.jobId).state).toBe("active");
+  expect(f.history.listForJob(job.jobId)[0]?.evaluationNotes?.nextAction).toContain("next regular cycle");
+  result = delivered; later(3599); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(4);
+  later(1); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(5);
+  const entries = f.outbox.listForSender("watchdog@system");
+  expect(entries).toHaveLength(2); expect(f.outbox.getById(old.outboxId)?.deliveryState).toBe("failed");
+  expect(entries.filter(e => e.deliveryState === "delivered")).toHaveLength(1);
+});
+
+it("does not replay an uncertain occurrence, while keeping the next recurring cycle", async () => {
+  let result: SendResult = { ok: false, sessionName: seat, inputPhase: "attempted", error: "Acknowledgement lost" };
+  const send = vi.fn(async () => result), f = fixture(send), job = f.reminder();
+  later(3600); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(1);
+  const old = f.outbox.listForSender("watchdog@system")[0]!; expect(old.deliveryState).toBe("indeterminate");
+  expect(f.jobs.getByIdOrThrow(job.jobId).lastFireAt).toBe(new Date().toISOString());
+  expect(f.history.listForJob(job.jobId)[0]?.evaluationNotes?.nextAction).toContain("unconfirmed");
+  const saved = f.db.serialize(); f.db.close(); const resumed = fixture(send, saved);
+  later(3599); await resumed.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(1);
+  result = delivered; later(1); await resumed.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(2);
+  expect(resumed.outbox.getById(old.outboxId)?.deliveryState).toBe("indeterminate");
+  expect(resumed.outbox.listForSender("watchdog@system")).toHaveLength(2);
+});
+
+it("records an uncertain one-shot park failure on its row before retiring the timer", async () => {
+  const send = vi.fn(async () => ({ ok: false, sessionName: seat, inputPhase: "attempted" as const, error: "Acknowledgement lost" }));
+  const f = fixture(send), row = await f.park(); later(3600); await f.scheduler.runTickNow();
+  expect(send).toHaveBeenCalledTimes(1); expect(f.repo.getById(row.qitemId)?.state).toBe("blocked");
+  expect(f.history.listForJob(row.jobId)[0]?.deliveryStatus).toBe("failed");
+  expect(f.repo.getParkWakeStatus(row.qitemId)).toMatchObject({ phase: "fired", deliveryStatus: "failed", live: false, unconsumed: true });
+  expect(f.jobs.getByIdOrThrow(row.jobId).state).toBe("terminal");
+  later(7200); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(1);
+});
+
+it("records a retryable refusal on the parked row without consuming its one-shot binding", async () => {
+  let result = absent; const send = vi.fn(async () => result), f = fixture(send), row = await f.park();
+  later(3600); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(1);
+  expect(f.repo.getParkWakeStatus(row.qitemId)).toMatchObject({ phase: "armed", live: true });
+  const notes = f.db.prepare("SELECT transition_note FROM queue_transitions WHERE qitem_id=? ORDER BY transition_id DESC LIMIT 1").get(row.qitemId) as { transition_note: string };
+  expect(notes.transition_note).toContain("retry at");
+  result = delivered; later(300); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(2);
+  expect(f.repo.getParkWakeStatus(row.qitemId)).toMatchObject({ phase: "fired", deliveryStatus: "ok", live: false });
+});
+
+it("records exhausted one-shot retries as a failed wake before ending the timer", async () => {
+  const send = vi.fn(async () => absent), f = fixture(send), row = await f.park();
+  later(3600); await f.scheduler.runTickNow();
+  for (let i = 0; i < 3; i++) { later(300); await f.scheduler.runTickNow(); }
+  expect(f.errors).toHaveLength(0); expect(send).toHaveBeenCalledTimes(4);
+  expect(f.repo.getById(row.qitemId)?.state).toBe("blocked");
+  expect(f.repo.getParkWakeStatus(row.qitemId)).toMatchObject({ phase: "fired", deliveryStatus: "failed", live: false, unconsumed: true });
+  expect(f.jobs.getByIdOrThrow(row.jobId).terminalReason).toBe("park_timer_fired_once");
+  expect(f.history.listForJob(row.jobId)[0]?.evaluationNotes?.nextAction).toContain("exhausted");
+  later(7200); await f.scheduler.runTickNow(); expect(send).toHaveBeenCalledTimes(4);
+});
+
+it.each([false, true])("repairs a post-fire accounting crash without duplicate input (digest=%s)", async digest => {
+  const send = vi.fn(async () => delivered), f = fixture(send, undefined, digest), rows = [await f.park()];
+  if (digest) rows.push(await f.park());
+  const cut = rows.at(-1)!, original = f.jobs.recordEvaluation.bind(f.jobs); let stampSeen = false;
+  vi.spyOn(f.jobs, "recordEvaluation").mockImplementation((id, at, fired, schedule) => {
+    original(id, at, fired, schedule);
+    if (id === cut.jobId && fired && !stampSeen) {
+      stampSeen = true; expect(f.jobs.getByIdOrThrow(id).lastFireAt).toBe(at);
+      throw new Error("crash after fire stamp and before queue accounting");
+    }
+  });
+  later(3600); await f.scheduler.runTickNow();
+  expect(stampSeen).toBe(true); expect(f.errors).toHaveLength(1); expect(send).toHaveBeenCalledTimes(1);
+  expect(f.repo.getById(cut.qitemId)?.state).toBe("blocked");
+  expect(f.outbox.listForSender("watchdog@system").filter(e => e.deliveryState === "delivered")).toHaveLength(rows.length);
+  const saved = f.db.serialize(); f.db.close(); const resumed = fixture(send, saved, digest);
+  later(3600); await resumed.scheduler.runTickNow();
+  expect(send).toHaveBeenCalledTimes(1);
+  for (const row of rows) {
+    expect(resumed.repo.getParkWakeStatus(row.qitemId)).toMatchObject({ phase: "fired", deliveryStatus: "ok", live: false });
+    expect(resumed.jobs.getByIdOrThrow(row.jobId).state).toBe("terminal");
+  }
+});
+
+it.each(["mode", "external", "prompt"] as const)("classifies the actual %s transport refusal before input", async kind => {
+  const f = fixture(async () => delivered);
+  f.db.exec(`INSERT INTO rigs(id,name) VALUES ('r','test'); INSERT INTO nodes(id,rig_id,logical_id,runtime) VALUES ('n','r','worker','terminal');
+    INSERT INTO sessions(id,node_id,session_name,status) VALUES ('s','n','worker@test','running');
+    INSERT INTO bindings(id,node_id,tmux_session,tmux_pane,attachment_type) VALUES ('b','n','worker@test','%1','${kind === "external" ? "external_cli" : "tmux"}');`);
+  const exec = vi.fn(async () => ""), tmux = new TmuxAdapter(exec);
+  vi.spyOn(tmux, "probeSession").mockResolvedValue({ state: "present" });
+  vi.spyOn(tmux, "isPaneInMode").mockResolvedValue(kind === "mode");
+  vi.spyOn(tmux, "cancelPaneMode").mockResolvedValue(undefined);
+  const transport = new SessionTransport({ db: f.db, tmuxAdapter: tmux, rigRepo: new RigRepository(f.db), sessionRegistry: new SessionRegistry(f.db) });
+  vi.spyOn(transport as unknown as { classifySendReadiness: () => Promise<{ state: string; reason: string }> }, "classifySendReadiness")
+    .mockResolvedValue({ state: kind === "prompt" ? "needs_input" : "idle", reason: kind === "prompt" ? "permission_prompt" : "fixture idle" });
+  const probe = await transport.send(seat, "Preflight", { openrigNotice: true, actorSession: "watchdog@system" });
+  expect(probe.reason).toBe(kind === "mode" ? "pane_in_mode" : kind === "prompt" ? "target_needs_input" : "transport_unavailable"); expect(exec).not.toHaveBeenCalled();
+  const driven = fixture(transport.send.bind(transport)), job = driven.reminder(); later(3600); await driven.scheduler.runTickNow();
+  expect(driven.history.listForJob(job.jobId)[0]?.deliveryStatus).toBe("failed");
+  if (kind === "external") expect(driven.jobs.getByIdOrThrow(job.jobId).state).toBe("terminal");
+  else {
+    expect(driven.outbox.listForSender("watchdog@system")[0]?.deliveryState).toBe("failed");
+    expect(driven.history.listForJob(job.jobId)[0]?.evaluationNotes?.retryAt).toEqual(expect.any(String));
+  }
+  expect(exec).not.toHaveBeenCalled();
+});

packages/daemon/test/watchdog-park-reminder-digest.test.ts

@@ -137,16 +137,16 @@ describe("due parked-row reminders for one seat", () => {
     due(later!.jobId);
     await deps.watchdogScheduler!.runTickNow();
     expect(sends).toHaveLength(2);
-    expect(sends[1]!.text).toBe(`[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${later!.jobId}]\nDelivery may repeat after an unconfirmed attempt.\n`
+    expect(sends[1]!.text).toBe(`[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${later!.jobId}]\n`
       + `Wake timer fired for parked qitem ${later!.qitemId}. Resume the recorded continuation and update the row.`);
   });
 
-  it("sends a lone reminder with repeat-delivery context", async () => {
+  it("sends a lone tracked reminder without a repeat-delivery warning", async () => {
     const owner = seat();
     const only = await park(owner, "alone");
     due(only.jobId);
     await deps.watchdogScheduler!.runTickNow();
-    expect(sends).toEqual([{ session: owner, text: `[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${only.jobId}]\nDelivery may repeat after an unconfirmed attempt.\n`
+    expect(sends).toEqual([{ session: owner, text: `[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${only.jobId}]\n`
       + `Wake timer fired for parked qitem ${only.qitemId}. Resume the recorded continuation and update the row.` }]);
     expect(disposition(only.jobId, only.qitemId)).toMatchObject({ state: "terminal", terminalReason: "park_timer_fired_once", deliveryStatus: "ok" });
   });
@@ -204,7 +204,7 @@ describe("due parked-row reminders for one seat", () => {
     expect(sends.filter((s) => s.session === owner)).toHaveLength(1);
   });
 
-  it("retries current reminders after interrupted digest accounting and includes a newly due one", async () => {
+  it("ends unconfirmed digest occurrences without replay and sends only a newly due reminder", async () => {
     const owner = seat();
     const [a, b, c] = [await park(owner, "a"), await park(owner, "b"), await park(owner, "c")];
     due(a!.jobId);
@@ -228,12 +228,19 @@ describe("due parked-row reminders for one seat", () => {
     await again.deps.watchdogScheduler!.runTickNow();
     expect(later).toHaveLength(1);
     expect(later[0]!.session).toBe(owner);
-    expect(later[0]!.text).toContain("Delivery may repeat after an unconfirmed attempt.");
-    for (const r of [a!, b!, c!]) {
-      expect(later[0]!.text).toContain(r.qitemId);
-      expect(again.deps.watchdogJobsRepo!.getByIdOrThrow(r.jobId)).toMatchObject({ state: "terminal", terminalReason: "park_timer_fired_once" });
-      expect(again.deps.watchdogHistoryLog!.listForJob(r.jobId, 1)[0]).toMatchObject({ deliveryStatus: "ok" });
+    expect(later[0]!.text).not.toContain("Delivery may repeat after an unconfirmed attempt.");
+    expect(later[0]!.text).toContain(c!.qitemId);
+    for (const r of [a!, b!]) {
+      expect(later[0]!.text).not.toContain(r.qitemId);
+      expect(again.deps.watchdogHistoryLog!.listForJob(r.jobId, 1)[0]).toMatchObject({ deliveryStatus: "failed" });
+      expect(again.deps.watchdogHistoryLog!.listForJob(r.jobId, 1)[0]?.evaluationNotes?.nextAction).toContain("unconfirmed");
+      expect(again.db.prepare("SELECT delivery_state FROM outbox_entries WHERE audit_pointer=?").get(r.jobId)).toEqual({ delivery_state: "indeterminate" });
+      expect(again.deps.queueRepo.getParkWakeStatus(r.qitemId)).toMatchObject({ phase: "fired", deliveryStatus: "failed", live: false });
     }
+    for (const r of [a!, b!, c!]) {
+      expect(again.deps.watchdogJobsRepo!.getByIdOrThrow(r.jobId)).toMatchObject({ state: "terminal", terminalReason: "park_timer_fired_once" });
+    }
+    expect(again.deps.watchdogHistoryLog!.listForJob(c!.jobId, 1)[0]).toMatchObject({ deliveryStatus: "ok" });
     again.deps.watchdogScheduler?.stop?.();
     again.db.close();
     // Let the first process's abandoned pass end, so it holds nothing in this file's later tests.
@@ -241,7 +248,7 @@ describe("due parked-row reminders for one seat", () => {
     await lost;
   });
 
-  it("retries the current member whose accounting failed after a digest send", async () => {
+  it("reconciles confirmed members whose accounting failed after a digest send without input replay", async () => {
     const owner = seat();
     const [a, b] = [await park(owner, "a"), await park(owner, "b")];
     due(a.jobId);
@@ -251,12 +258,13 @@ describe("due parked-row reminders for one seat", () => {
     await deps.watchdogScheduler!.runTickNow();
     spy.mockImplementation(original);
     expect(sends.filter((s) => s.session === owner)).toHaveLength(1);
-    // B is still due; its old digest receipt cannot substitute for a new transport attempt.
+    // B is still due; its confirmed occurrence repairs accounting without repeating terminal input.
     expect(deps.watchdogJobsRepo!.getByIdOrThrow(b.jobId).state).toBe("active");
     due(b.jobId);
     await deps.watchdogScheduler!.runTickNow();
-    expect(sends.filter((s) => s.session === owner)).toHaveLength(2);
-    expect(sends.filter((s) => s.session === owner).at(-1)!.text).toContain(b.qitemId);
+    expect(sends.filter((s) => s.session === owner)).toHaveLength(1);
+    expect(sends.filter((s) => s.session === owner)[0]!.text).toContain(b.qitemId);
+    expect(disposition(a.jobId, a.qitemId)).toMatchObject({ state: "terminal", terminalReason: "park_timer_fired_once", deliveryStatus: "ok", wakePhase: "fired" });
     expect(disposition(b.jobId, b.qitemId)).toMatchObject({ state: "terminal", terminalReason: "park_timer_fired_once", deliveryStatus: "ok", wakePhase: "fired" });
   });
 
@@ -282,7 +290,7 @@ describe("due parked-row reminders for one seat", () => {
     expect(digest).toContain(b.qitemId);
     expect(digest).toContain(c.qitemId);
     expect(digest).not.toContain(a.qitemId);
-    expect(texts).toContain(`[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${a.jobId}]\nDelivery may repeat after an unconfirmed attempt.\n`
+    expect(texts).toContain(`[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${a.jobId}]\n`
       + `Wake timer fired for parked qitem ${a.qitemId}. Resume the recorded continuation and update the row.`);
     expect(deps.queueRepo.getParkWakeStatus(attached)).toMatchObject({ phase: "fired" });
   });
@@ -335,7 +343,7 @@ describe("due parked-row reminders for one seat", () => {
     };
     await deps.watchdogScheduler!.runTickNow();
     const texts = sends.filter((s) => s.session === owner).map((s) => s.text);
-    const lone = (r: { jobId: string; qitemId: string }) => `[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${r.jobId}]\nDelivery may repeat after an unconfirmed attempt.\n`
+    const lone = (r: { jobId: string; qitemId: string }) => `[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${r.jobId}]\n`
       + `Wake timer fired for parked qitem ${r.qitemId}. Resume the recorded continuation and update the row.`;
     expect(texts).toEqual([lone(b), lone(a), lone(c)]);
   });
@@ -372,7 +380,7 @@ describe("due parked-row reminders for one seat", () => {
 });
 
 describe("a park wake shows the park's recorded continuation", () => {
-  const header = (jobId: string) => `[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${jobId}]\nDelivery may repeat after an unconfirmed attempt.\n`;
+  const header = (jobId: string) => `[OpenRig watchdog scheduler · policy: periodic-reminder · job: ${jobId}]\n`;
   // Earlier tests leave other seats' reminders due on purpose, so each test reads only its own seat's sends.
   const to = (owner: string) => sends.filter((s) => s.session === owner);
   const guidance = "Check the row's current state and blocker first, then resume only that recorded work and record the "

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

@@ -22,4 +22,8 @@ describe("(i-c) wiring pin — startup injects the real target-generation resolv
       /resolvePreDeliveryTerminalReason:\s*\(\{ jobId \}\)\s*=>\s*queueRepoInstance\.resolveWatchdogPreDeliveryTerminalReason\(jobId\)/,
     );
   });
+
+  it("gives scheduled occurrence delivery the daemon's shared durable outbox", () => {
+    expect(src).toMatch(/deliver:\s*makeWatchdogDeliver\(\{\s*sessionTransport,\s*outbox:\s*outboxHandlerInstance/);
+  });
 });

packages/test-system/ci/prompt-state.mjs

@@ -50,7 +50,10 @@ export function verifyPromptState(proof) {
   assert.notEqual(waiting.send.code, 0, 'real prompt admitted send');
   assert.match(waiting.send.stdout + waiting.send.stderr, /target_needs_input/);
   assert.ok(!waiting.pane.includes(waiting.marker), 'send typed into real prompt');
-  assert.ok(waiting.wake.fired, 'refused wake never attempted');
+  assert.equal(waiting.wake.fired, false, 'retryable refusal consumed the one-shot timer');
+  assert.equal(waiting.wake.job?.state, 'active');
+  assert.equal(waiting.wake.job?.lastFireAt, null);
+  assert.match(waiting.wake.retryNote ?? '', /occurrence remains owed; retry at/);
   assert.match(waiting.wake.delivery, /needs_input|refused|retained|held|failed/i);
   const attempt = waiting.wake.attempt;
   assert.ok(attempt?.historyId && waiting.wake.jobId, 'missing watchdog refusal attempt');
@@ -59,6 +62,8 @@ export function verifyPromptState(proof) {
   assert.equal(attempt.outcome, 'refused');
   assert.equal(attempt.deliveryStatus, 'failed');
   assert.ok(attempt.deliveryMessage.includes(waiting.wake.id), 'attempt belongs to another queue row');
+  assert.ok(Number.isFinite(Date.parse(attempt.evaluationNotes?.retryAt)), 'refusal has no bounded retry time');
+  assert.match(attempt.evaluationNotes?.nextAction ?? '', /Occurrence remains owed; retry at/);
   const reason = attempt.evaluationNotes?.deliveryReason;
   assert.ok(typeof reason === 'string' && reason.startsWith(`Refused: '${proof.claude}' is at an interactive prompt (`));
   assert.match(reason, /interactive prompt \((?:selection_prompt|permission_prompt)\)/);
@@ -178,14 +183,18 @@ export async function runPromptState(rigBin, frames, proof = {}) {
       const read = async () => {
         const transitions = await jsonCli('queue', 'transitions', id, '--full');
         const fired = entries(transitions).find(row => row.wake?.phase === 'fired');
-        const jobId = fired?.wake?.ref;
-        const history = !expectDelivery && jobId ? (await get(`/api/watchdog/${encodeURIComponent(jobId)}/status`)).recentHistory : [];
+        const armed = entries(transitions).find(row => row.wake?.phase === 'armed');
+        const jobId = fired?.wake?.ref ?? armed?.wake?.ref;
+        const status = !expectDelivery && jobId ? await get(`/api/watchdog/${encodeURIComponent(jobId)}/status`) : null;
+        const history = status?.recentHistory ?? [];
         // A fresh generated one-shot timer must have exactly one matching attempt, not an unrelated delivery error.
         const matches = history.filter(row => row.jobId === jobId && row.deliveryTargetSession === claude && row.deliveryMessage?.includes(id));
         return { id, fired: Boolean(fired), jobId, attempt: matches.length === 1 ? matches[0] : null,
-          delivery: fired?.wake?.deliveryStatus ?? '', pane: await capture(claude) };
+          delivery: fired?.wake?.deliveryStatus ?? matches[0]?.deliveryStatus ?? '', job: status?.job,
+          retryNote: entries(transitions).find(row => row.transitionNote?.includes('occurrence remains owed; retry at'))?.transitionNote,
+          pane: await capture(claude) };
       };
-      const observed = await until(read, row => expectDelivery ? row.pane.includes(id) : row.fired && row.delivery && row.attempt,
+      const observed = await until(read, row => expectDelivery ? row.pane.includes(id) : row.delivery && row.attempt && row.retryNote,
         'timer wake not observed', 35_000);
       await cli(['queue', 'update', id, '--state', 'done', '--closure-reason', 'no-follow-on'], claude);
       return observed;

scripts/pr-scenarios.test.mjs

@@ -30,10 +30,13 @@ function promptStateProof() {
     working: ['quoted-selector', 'quoted-permission'].map(kind => ({ kind, marker: kind, before: ['Frame', '❯ ', '⏵⏵ accept edits on (shift+tab to cycle) · ← for agents', ...Array(42).fill('')].join('\n') + '\n',
       send: { code: 0 }, pane: kind, wake: { id: 'wake-' + kind, pane: 'wake-' + kind } })),
     waiting: { marker: 'must-not-type', send: { code: 1, stdout: 'target_needs_input', stderr: '' },
-      pane: 'real prompt', wake: { id: 'must-not-wake', fired: true, delivery: 'failed', pane: 'real prompt', jobId: 'watchdog-waiting',
+      pane: 'real prompt', wake: { id: 'must-not-wake', fired: false, delivery: 'failed', pane: 'real prompt', jobId: 'watchdog-waiting',
+        job: { state: 'active', lastFireAt: null },
+        retryNote: 'park wake failed: occurrence remains owed; retry at 2026-01-01T00:05:00.000Z',
         attempt: { historyId: 'attempt-waiting', jobId: 'watchdog-waiting', deliveryTargetSession: 'dev-claude@prompt-state',
           outcome: 'refused', deliveryStatus: 'failed', deliveryMessage: 'Wake must-not-wake',
-          evaluationNotes: { deliveryReason: "Refused: 'dev-claude@prompt-state' is at an interactive prompt (selection_prompt). No text was sent." } } } },
+          evaluationNotes: { deliveryReason: "Refused: 'dev-claude@prompt-state' is at an interactive prompt (selection_prompt). No text was sent.",
+            retryAt: '2026-01-01T00:05:00.000Z', nextAction: 'Occurrence remains owed; retry at 2026-01-01T00:05:00.000Z.' } } } },
   };
 }
 
@@ -373,7 +376,19 @@ test('prompt-state result rejects every escaped regression and missing control',
     p => { p.working[0].wake.pane = ''; },
     p => { p.waiting.send.code = 0; },
     p => { p.waiting.pane += p.waiting.marker; },
-    p => { p.waiting.wake.fired = false; },
+    p => { p.waiting.wake.fired = true; },
+    p => { delete p.waiting.wake.fired; },
+    p => { delete p.waiting.wake.job; },
+    p => { delete p.waiting.wake.job.state; },
+    p => { p.waiting.wake.job.state = 'terminal'; },
+    p => { delete p.waiting.wake.job.lastFireAt; },
+    p => { p.waiting.wake.job.lastFireAt = '2026-01-01T00:00:00.000Z'; },
+    p => { delete p.waiting.wake.retryNote; },
+    p => { p.waiting.wake.retryNote = 'No retry scheduled'; },
+    p => { delete p.waiting.wake.attempt.evaluationNotes.retryAt; },
+    p => { p.waiting.wake.attempt.evaluationNotes.retryAt = 'later'; },
+    p => { delete p.waiting.wake.attempt.evaluationNotes.nextAction; },
+    p => { p.waiting.wake.attempt.evaluationNotes.nextAction = 'No retry scheduled'; },
     p => { p.waiting.wake.delivery = 'ok'; },
     p => { p.waiting.wake.attempt = null; },
     p => { p.waiting.wake.attempt.outcome = 'failed'; p.waiting.wake.attempt.evaluationNotes.deliveryReason = 'transport unavailable'; },