diff --git a/docs/reference/keeping-a-watch.md b/docs/reference/keeping-a-watch.md index 1626e6489..e4aa34c10 100644 --- a/docs/reference/keeping-a-watch.md +++ b/docs/reference/keeping-a-watch.md @@ -107,6 +107,28 @@ rig queue create --destination @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 --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 diff --git a/packages/daemon/src/domain/draft-aware-delivery.ts b/packages/daemon/src/domain/draft-aware-delivery.ts index f0d0b7d60..528277db3 100644 --- a/packages/daemon/src/domain/draft-aware-delivery.ts +++ b/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; /** Applicability needs the intended queue row and recipient, not a saved delivery receipt. */ export type QueueWakeFacts = Pick; @@ -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 }; } diff --git a/packages/daemon/src/domain/queue-repository.ts b/packages/daemon/src/domain/queue-repository.ts index bb905f116..f57b01995 100644 --- a/packages/daemon/src/domain/queue-repository.ts +++ b/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. diff --git a/packages/daemon/src/domain/session-transport.ts b/packages/daemon/src/domain/session-transport.ts index e3cfdca82..f5d495a84 100644 --- a/packages/daemon/src/domain/session-transport.ts +++ b/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} "" --dangerously-interact --reason "".${clearAttentionHint(sessionName, readiness.reason)} No text was sent.`, }; diff --git a/packages/daemon/src/domain/watchdog-delivery-receipts.ts b/packages/daemon/src/domain/watchdog-delivery-receipts.ts new file mode 100644 index 000000000..d38f65df6 --- /dev/null +++ b/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 = {}; + 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; +} diff --git a/packages/daemon/src/domain/watchdog-delivery.ts b/packages/daemon/src/domain/watchdog-delivery.ts index ee7fb3e66..f272b9b43 100644 --- a/packages/daemon/src/domain/watchdog-delivery.ts +++ b/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; + outbox?: OutboxHandler; + now?: () => Date; + retryIntervalSeconds?: number; + retryCap?: number; hasPendingPostCompactStage?: (sessionName: string) => boolean; isTerminalSeat?: (sessionName: string) => boolean; createContinuityAction?: (action: NonNullable) => Promise; }): 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; } diff --git a/packages/daemon/src/domain/watchdog-policy-engine.ts b/packages/daemon/src/domain/watchdog-policy-engine.ts index ccc86aa6c..8befc1b0e 100644 --- a/packages/daemon/src/domain/watchdog-policy-engine.ts +++ b/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; } 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; +) => Promise) & { + /** 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; @@ -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), diff --git a/packages/daemon/src/startup.ts b/packages/daemon/src/startup.ts index 1bad0e9d9..c7752f2a3 100644 --- a/packages/daemon/src/startup.ts +++ b/packages/daemon/src/startup.ts @@ -479,6 +479,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise // 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 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 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), diff --git a/packages/daemon/test/watchdog-delivery-recovery.test.ts b/packages/daemon/test/watchdog-delivery-recovery.test.ts new file mode 100644 index 000000000..db37ee0e8 --- /dev/null +++ b/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, 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) { + 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, 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); + }); +}); diff --git a/packages/daemon/test/watchdog-delivery-round1.test.ts b/packages/daemon/test/watchdog-delivery-round1.test.ts new file mode 100644 index 000000000..b51581c6d --- /dev/null +++ b/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, 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(); +}); diff --git a/packages/daemon/test/watchdog-park-reminder-digest.test.ts b/packages/daemon/test/watchdog-park-reminder-digest.test.ts index c9ea2596c..feda65567 100644 --- a/packages/daemon/test/watchdog-park-reminder-digest.test.ts +++ b/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 " diff --git a/packages/daemon/test/watchdog-target-gen-wiring.test.ts b/packages/daemon/test/watchdog-target-gen-wiring.test.ts index bca86c578..6afa9a2bf 100644 --- a/packages/daemon/test/watchdog-target-gen-wiring.test.ts +++ b/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/); + }); }); diff --git a/packages/test-system/ci/prompt-state.mjs b/packages/test-system/ci/prompt-state.mjs index 41c8ab963..fdcd89484 100644 --- a/packages/test-system/ci/prompt-state.mjs +++ b/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; diff --git a/scripts/pr-scenarios.test.mjs b/scripts/pr-scenarios.test.mjs index fa434961a..e4124ec32 100644 --- a/scripts/pr-scenarios.test.mjs +++ b/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'; },