From 6d3b25c50b88ab8036aef1661dafc530d67fde48 Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 30 Sep 2026 13:01:51 +0200 Subject: [PATCH 1/5] fix: follow lost worker leases and retried attempts instead of abandoning the run Customer rw_3a0fcb71 issue 1: an f.agent attempt lost its lease, the kernel retried it, and the CLI exited `worker wait canceled` / protocol_error with no run id while attempt 2 went on to succeed -- its output was discarded, and the failure reported `attempt=2/1`. Lease timing (kernel + SDK): - relayflowd sends `lease_ttl_ms` on step.dispatch and step.heartbeat (additive). - withWorkerLease times the lease on the worker's monotonic clock from that duration instead of comparing the daemon's absolute deadline with its own wall clock. A worker 45s ahead of the daemon refused every dispatch as already expired; 75s behind, it renewed after the daemon swept the lease. Old daemons keep the old comparison. Following retries: - A lost authored ROOT lease is not a failed flow: the root driver waits for the kernel's re-dispatch and re-drives the body (completed steps replay from receipts; the in-flight child is re-adopted). It no longer terminalizes the root as worker_error, which burned every root attempt and made the run unrecoverable. If no retry arrives it reports `root_lease_lost` (status running, root run id, exact `flows resume` command). - classifyOutcome follows a child returned `running` (re-admitted mid-attempt) and a step in retry `backoff`, instead of "status running without a classifiable completion" after 2s. - A worker wait whose lease looks expired only by this process's clock keeps following the step; it fails only when the deadline stops moving for a whole lease (never renewed, never swept). - Cancellation errors carry the abort reason ("worker wait canceled: Agent lease expired before renewal ...") and withWorkerLease reports a lost lease as the cause, not the body's generic wait error. Attempt accounting: - stepFailureDetails counts attempts the kernel declared dead (crashed/lease_expired, completed_by kernel) as `unchargedAttempts`, and the report prints `attempt=2 (iteration 1/1; 1 earlier attempt lost to a dead worker, not charged)` instead of `attempt=2/1`. Builds on #590 (terminal-completion read). Co-Authored-By: Claude Opus 5.5 (1M context) --- kernel/DESIGN.md | 2 +- kernel/relayflowd/src/server.rs | 7 +- .../src/server/session/assignments.rs | 19 ++- kernel/relayflowd/src/server/tests.rs | 25 +++ packages/sdk/src/authored-flow-error.ts | 6 + packages/sdk/src/authored-node-runner.ts | 3 +- packages/sdk/src/authored-root.ts | 133 +++++++++++++++- packages/sdk/src/authored-worker-step.ts | 1 + packages/sdk/src/cli/direct-run.ts | 5 +- packages/sdk/src/cli/run.ts | 97 +++++++++++- packages/sdk/src/cli/step-failure.ts | 18 +++ packages/sdk/src/failure-kinds.ts | 14 ++ packages/sdk/src/protocol.ts | 9 ++ packages/sdk/src/worker-lease.ts | 56 +++++-- packages/sdk/tests/classify-outcome.test.ts | 53 +++++++ packages/sdk/tests/cli.test.ts | 11 +- .../sdk/tests/worker-lease-drift-live.test.ts | 143 ++++++++++++++++++ packages/sdk/tests/worker-lease-sweep.test.ts | 28 +++- packages/sdk/tests/worker-lease.test.ts | 22 ++- 19 files changed, 611 insertions(+), 41 deletions(-) create mode 100644 packages/sdk/tests/worker-lease-drift-live.test.ts diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index d3cd99d64..ac35b3e6f 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -386,7 +386,7 @@ Minimal verb set for gate 1: | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility | | `run.watch` | `{run_id}` → stream of `{event: "entry", data: Entry}` | every appended entry, pushed | | `worker.attach` | `{worker_id, step_types: ["llm","agent"], pins}` → `{}` | connection becomes a worker; agent workers **must** supply opaque initial workspace revisions/stream offsets (refused otherwise) and receive `step.dispatch` events with pins plus recovery context. A step whose declared surfaces no attached worker holds parks — it is not dispatched | -| `step.heartbeat` | `{run_id, step_id, attempt, lease_id}` → `{lease_deadline_ms}` | renew the lease; the one lease primitive | +| `step.heartbeat` | `{run_id, step_id, attempt, lease_id}` → `{lease_deadline_ms, lease_ttl_ms}` | renew the lease; the one lease primitive. `lease_ttl_ms` (also on `step.dispatch`) is the same deadline as a duration on the daemon's clock: workers time the lease on their own monotonic clock from it, never by comparing `lease_deadline_ms` with their wall clock | | `effect.record` | `{run_id, step_id, attempt, idempotency_key, surface_path, revision_before, revision_after}` → `{deduped}` | phase one: agent records a writeback before performing it; the journal boundary elects one provider-call winner. `deduped:false` means this attempt owes the call **and** the `effect.confirm` that closes it | | `effect.confirm` | `{run_id, step_id, attempt, idempotency_key, surface_path}` → `{confirmed}` | phase two: the elected attempt made the provider call. Until it lands the election is reclaimable by a later attempt, so a worker that dies mid-writeback loses nothing; only the election holder may confirm | | `step.complete` | `{run_id, step_id, attempt, idempotency_key, completionReason, output, usage, started_pins, end_pins, effects, trajectory_tail}` → `{}` | completes a dispatched step — **also the out-of-band path**: the lease holder reports its actual start/end pins and journaled effect refs; kernel verifies them, runs the gate, and decides the edge | diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index a0960c53a..01691b760 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -381,12 +381,13 @@ fn handle_request( let lock = hub.run_lock(¶ms.run_id); let _guard = lock.lock().expect("run lock"); ensure_mutable(&engine, ¶ms.run_id)?; + let now = now_ms(); let (deadline, run_deadline) = hub .heartbeat( connection_id, &(params.run_id.clone(), params.step_id, params.attempt), ¶ms.lease_id, - now_ms(), + now, ) .map_err(protocol_conflict)?; // Durable renewal: persist the extended deadline so the lease the @@ -396,7 +397,9 @@ fn handle_request( engine .renew_lease(¶ms.run_id, run_deadline) .map_err(internal_error)?; - Ok(json!({"lease_deadline_ms": deadline})) + // `lease_ttl_ms`: the same deadline as a duration, so the worker can + // time it on its own clock (see the dispatch frame in assignments.rs). + Ok(json!({"lease_deadline_ms": deadline, "lease_ttl_ms": deadline.saturating_sub(now)})) } "step.complete" => { let params: StepCompleteParams = decode_params(request.params)?; diff --git a/kernel/relayflowd/src/server/session/assignments.rs b/kernel/relayflowd/src/server/session/assignments.rs index 70209c72f..94372523d 100644 --- a/kernel/relayflowd/src/server/session/assignments.rs +++ b/kernel/relayflowd/src/server/session/assignments.rs @@ -243,11 +243,20 @@ impl StepDispatcher for ProtocolHub { sessions.reservations.remove(&key); return Ok(DispatchOutcome::PinMismatch { detail }); } - if let Err(error) = write_frame( - &worker.writer, - &json!({"event": "step.dispatch", "data": dispatch}), - ) - .with_context(|| format!("dispatch step to worker {}", worker.worker_id)) + // `lease_ttl_ms` restates the deadline as a duration on this daemon's + // clock. The worker times its lease against its OWN clock, so an + // absolute deadline alone is only as good as the two clocks' agreement + // -- a skew of one lease length refused every dispatch locally as + // already expired, or renewed after the daemon had swept the lease. + let mut frame = json!({"event": "step.dispatch", "data": dispatch}); + frame["data"]["lease_ttl_ms"] = json!( + dispatch + .lease_deadline_ms + .saturating_sub(super::super::protocol::now_ms()) + .max(0) + ); + if let Err(error) = write_frame(&worker.writer, &frame) + .with_context(|| format!("dispatch step to worker {}", worker.worker_id)) { sessions.reservations.remove(&key); return Err(error); diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index 33f46594a..682a95797 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -572,6 +572,31 @@ fn run_resume_refuses_a_valid_journal_that_belongs_to_another_run() { ); } +/// The dispatch restates its lease as a duration on the daemon's clock, so a +/// worker can time the lease on its own clock instead of comparing the +/// absolute deadline with a wall clock that may disagree (customer +/// rw_3a0fcb71: a skewed worker refused or under-renewed every lease). +#[test] +fn dispatch_carries_the_lease_as_a_duration() { + let directory = tempdir().unwrap(); + let data_dir = directory.path(); + let hub = Arc::new(ProtocolHub::default()); + let worker_peer = attach_llm_worker(data_dir, &hub, 1); + let mut worker_reader = BufReader::new(worker_peer); + let before = now_ms(); + start_llm_run(data_dir, &hub); + let dispatch = read_frame(&mut worker_reader); + assert_eq!(dispatch["event"], "step.dispatch"); + let ttl = dispatch["data"]["lease_ttl_ms"] + .as_i64() + .expect("lease_ttl_ms"); + let deadline = dispatch["data"]["lease_deadline_ms"].as_i64().unwrap(); + assert!( + ttl > 0 && ttl <= deadline - before, + "ttl {ttl} deadline {deadline} before {before}" + ); +} + /// Finding 3: a hung worker that stops heartbeating past its lease deadline — /// socket still open, so no disconnect fires — must not leave the run in /// waiting_worker forever. The reconciler journals a `lease_expired` diff --git a/packages/sdk/src/authored-flow-error.ts b/packages/sdk/src/authored-flow-error.ts index cd8d3236d..d43aeeb96 100644 --- a/packages/sdk/src/authored-flow-error.ts +++ b/packages/sdk/src/authored-flow-error.ts @@ -30,6 +30,12 @@ export type AuthoredFlowExecutionErrorCode = | 'operation_callback_failed' | 'step_failed' | 'lease_exceeded' + /** + * The CLI lost the authored root's worker lease and the kernel did not + * re-dispatch the root to it. The run is not failed: completed steps are + * journaled and `flows resume ` continues it. + */ + | 'root_lease_lost' | 'unsupported_completion' | 'unsupported_gate' | 'gate_failed' diff --git a/packages/sdk/src/authored-node-runner.ts b/packages/sdk/src/authored-node-runner.ts index 9ad762cbf..d0cf146ea 100644 --- a/packages/sdk/src/authored-node-runner.ts +++ b/packages/sdk/src/authored-node-runner.ts @@ -240,7 +240,8 @@ export function stepFailedFrame(value: unknown): StepFailedDetails | undefined { for (const [key, parsed] of [ ['stepId', frameText(frame, 'stepId')], ['stepType', frameText(frame, 'stepType')], ['completionReason', frameText(frame, 'completionReason')], ['attempt', frameCount(frame, 'attempt')], - ['maxIterations', frameCount(frame, 'maxIterations')], ['exitCode', frameCount(frame, 'exitCode')], + ['maxIterations', frameCount(frame, 'maxIterations')], + ['unchargedAttempts', frameCount(frame, 'unchargedAttempts')], ['exitCode', frameCount(frame, 'exitCode')], ['stdoutTail', frameText(frame, 'stdoutTail')], ['stderrTail', frameText(frame, 'stderrTail')], ['detail', frameText(frame, 'detail')], ['transcriptPath', frameText(frame, 'transcriptPath')], ['attempts', attemptFrames(frame['attempts'])], diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 547016e7d..2288208bb 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -16,9 +16,9 @@ import { JournalClient } from './journal-client.js'; import type { RunOutcome, StepDispatchEvent } from './protocol.js'; import { SPEC_SCHEMA_VERSION } from './spec.js'; import type { RunLifecycleOptions } from './cli/run.js'; -import { withWorkerLease } from './worker-lease.js'; +import { isLeaseLost, withWorkerLease } from './worker-lease.js'; import { AuthoredFlowExecutionError, AuthoredHumanParked } from './authored-flow-error.js'; -import { readOpenHumanWaits } from './authored-human.js'; +import { readOpenHumanWaits, resumeCommand } from './authored-human.js'; import { isSurfaceCompletionReason } from './authored-step-output.js'; import { readSubscriptionPark } from './authored-subscription-park.js'; import { localAgentCredentialEnvironment } from './local-agent-environment.js'; @@ -134,7 +134,7 @@ export async function executeDurableAuthoredFlow( if (parked !== undefined) return { state: 'suspended', name: metadata.flowName, suspension: parked, journalSteps: [], rootRunId: outcome.run_id }; const dispatch = await dispatchWait.promise; - return await driveRoot(loaded, metadata, journal, peer, dispatch, options); + return await driveRootFollowingRetries(loaded, metadata, journal, peer, dispatch, options); } finally { cancelDispatch(); peer.close(); @@ -177,7 +177,7 @@ export async function resumeDurableAuthoredFlow( if (parked !== undefined) return { state: 'suspended', name: metadata.flowName, suspension: parked, journalSteps: [], rootRunId }; const dispatch = await dispatchWait.promise; - return await driveRoot(loaded, metadata, journal, peer, dispatch, options); + return await driveRootFollowingRetries(loaded, metadata, journal, peer, dispatch, options); } finally { cancelDispatch(); peer.close(); @@ -214,6 +214,91 @@ export async function readAuthoredRootMetadata( return Object.freeze({ ...value, extensions: value.extensions ?? [] }); } +/** + * How many times one CLI re-drives a root whose lease it lost. Each re-drive + * replays completed steps from their journaled receipts, so the bound guards a + * lease that keeps being lost, not the cost of the work. + */ +const MAX_ROOT_REDRIVES = 3; +/** + * How long to wait for the kernel to re-dispatch a root whose lease was lost: + * a full lease (30s, relayflowd LEASE_RENEWAL_MS) in case the daemon still + * held it, plus its sweep and grace. + */ +const ROOT_REDISPATCH_TIMEOUT_MS = 45_000; + +/** + * Drive the root, and FOLLOW the kernel when it retries the root attempt. + * + * A lost root lease is not a failed flow. The kernel journals the attempt + * `lease_expired` with `disposition: retry` without charging an iteration + * (relayflowd-core recovery.rs) and re-dispatches the root to the worker + * still attached on `peer` -- this process. Exiting there with + * "worker wait canceled" (customer rw_3a0fcb71) threw away a run the kernel + * was about to continue, including a child agent already on its second + * attempt: the re-driven body replays completed steps from their receipts and + * adopts that in-flight child (authored-step-output.ts). Only when no retry + * arrives does the error leave, as `root_lease_lost` naming the resume. + */ +async function driveRootFollowingRetries( + loaded: LoadedAuthoredFlow, + metadata: AuthoredRootMetadata, + journal: JournalClient, + peer: JournalClient, + firstDispatch: StepDispatchEvent, + options: Omit, +): Promise { + let dispatch = firstDispatch; + for (let redrives = 0; ; redrives += 1) { + // Listen before driving: the retry can be dispatched the moment the + // daemon sweeps the lease, before the aborted body has unwound. The + // timeout starts only once there is a lost lease to wait on. + const next = rootRedispatch(peer); + try { + const result = await driveRoot(loaded, metadata, journal, peer, dispatch, options); + next.cancel(); + return result; + } catch (error) { + if (!isLeaseLost(error) || options.lifecycle?.signal?.aborted === true) { + next.cancel(); + throw error; + } + const lost = sentence(error instanceof Error ? error.message : String(error)); + const resume = resumeCommand(dispatch.run_id, options.dataDir, options.localAgentStream !== undefined); + if (redrives >= MAX_ROOT_REDRIVES) { + next.cancel(); + throw rootLeaseLost(dispatch, `${lost} It was lost ${redrives + 1} times in this process; continue the run with: ${resume}`, error); + } + process.emitWarning( + `authored root run_id=${dispatch.run_id} attempt=${dispatch.attempt}: ${lost} Waiting for the kernel to retry it.`, + { code: 'FLOWS_ROOT_LEASE_LOST' }, + ); + try { + dispatch = await next.wait(ROOT_REDISPATCH_TIMEOUT_MS); + } catch { + throw rootLeaseLost(dispatch, `${lost} The kernel did not re-dispatch the root within ${ROOT_REDISPATCH_TIMEOUT_MS / 1000}s; continue the run with: ${resume}`, error); + } + if (dispatch.run_id !== firstDispatch.run_id) { + throw new Error('authored root retry was dispatched for a different run'); + } + } + } +} + +function sentence(text: string): string { + return /[.!?]$/.test(text) ? text : `${text}.`; +} + +function rootLeaseLost(dispatch: StepDispatchEvent, message: string, cause: unknown): AuthoredFlowExecutionError { + const error = new AuthoredFlowExecutionError('root_lease_lost', + `authored root run ${dispatch.run_id} lost its worker lease on attempt ${dispatch.attempt}. ${message}` + + ' Completed steps are journaled and are not re-run.', + 'lease_expired'); + error.rootRunId = dispatch.run_id; + (error as { cause?: unknown }).cause = cause; + return error; +} + async function driveRoot( loaded: LoadedAuthoredFlow, metadata: AuthoredRootMetadata, @@ -269,6 +354,9 @@ async function driveRoot( ); return Object.freeze({ ...result, rootRunId: dispatch.run_id }); } catch (error) { + // A lost lease is owned by the kernel's retry path, not by this attempt's + // terminalization or suspension handling. + if (isLeaseLost(error)) throw error; if (error instanceof AuthoredFlowExecutionError && error.code === 'subscription_suspended' && error.suspension !== undefined) { @@ -376,7 +464,40 @@ function hasRootStream(step: Record): boolean { try { rootStreamFromMetadata(step); return true; } catch { return false; } } -function nextRootDispatch(peer: JournalClient): { +/** + * Buffers the next root dispatch from the moment it is created; `wait` + * bounds only the time spent waiting for it. Unlike `nextRootDispatch`, no + * timer runs while the body is being driven -- a body longer than the + * timeout must not leave a rejected promise nobody awaits. + */ +function rootRedispatch(peer: JournalClient): { + wait(timeoutMs: number): Promise; + cancel(): void; +} { + let received: StepDispatchEvent | undefined; + let deliver: ((dispatch: StepDispatchEvent) => void) | undefined; + const onDispatch = (dispatch: StepDispatchEvent): void => { + if (dispatch.step_id !== 'authored-root') return; + peer.off('step.dispatch', onDispatch); + if (deliver === undefined) received = dispatch; + else deliver(dispatch); + }; + peer.on('step.dispatch', onDispatch); + const cancel = (): void => { peer.off('step.dispatch', onDispatch); }; + return { + cancel, + wait: timeoutMs => new Promise((resolve, reject) => { + if (received !== undefined) { resolve(received); return; } + const timer = setTimeout(() => { + cancel(); + reject(new Error('authored root was not re-dispatched')); + }, timeoutMs); + deliver = dispatch => { clearTimeout(timer); resolve(dispatch); }; + }), + }; +} + +function nextRootDispatch(peer: JournalClient, timeoutMs = 30_000): { promise: Promise; cancel(): void; } { @@ -385,7 +506,7 @@ function nextRootDispatch(peer: JournalClient): { const timer = setTimeout(() => { peer.off('step.dispatch', onDispatch); reject(new Error('authored root was not dispatched')); - }, 30_000); + }, timeoutMs); const onDispatch = (dispatch: StepDispatchEvent) => { if (dispatch.step_id !== 'authored-root') return; clearTimeout(timer); diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts index 2daed1e81..c90e41f60 100644 --- a/packages/sdk/src/authored-worker-step.ts +++ b/packages/sdk/src/authored-worker-step.ts @@ -343,6 +343,7 @@ function stepDetails( ...(found.completionReason === undefined ? {} : { completionReason: found.completionReason }), ...(found.attempt === undefined ? {} : { attempt: found.attempt }), ...(found.maxIterations === undefined ? {} : { maxIterations: found.maxIterations }), + ...(found.unchargedAttempts === undefined ? {} : { unchargedAttempts: found.unchargedAttempts }), ...(found.exitCode === undefined ? {} : { exitCode: found.exitCode }), ...(found.stdoutTail === undefined ? {} : { stdoutTail: found.stdoutTail }), ...(found.stderrTail === undefined ? {} : { stderrTail: found.stderrTail }), diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index d2d5a8e5d..904573700 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -21,7 +21,7 @@ import { authoredInput, authoredWorkerRemedy, localAgentRemedy } from './local-a import { authoredCompletion, authoredHumanParked, - authoredStepFailure, + authoredStepFailure, rootLeaseLostReport, connect, emptyReport, fromCheckReport, @@ -212,6 +212,9 @@ export async function runDirectFlow( if (error instanceof AuthoredFlowExecutionError && (error.code === 'step_failed' || error.code === 'gate_failed')) { return authoredStepFailure('run', base, socketPath, error); } + if (error instanceof AuthoredFlowExecutionError && error.code === 'root_lease_lost') { + return rootLeaseLostReport('run', base, socketPath, error); + } if (error instanceof AuthoredHumanParked) { return authoredHumanParked('run', base, socketPath, error, { dataDir, localAgent: options.localAgent === true }); } diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index 716504e1b..d74b69b11 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -408,6 +408,9 @@ export async function resumeFlow( if (error instanceof AuthoredFlowExecutionError && (error.code === 'step_failed' || error.code === 'gate_failed')) { return authoredStepFailure('resume', base, socketPath, error, runId); } + if (error instanceof AuthoredFlowExecutionError && error.code === 'root_lease_lost') { + return rootLeaseLostReport('resume', base, socketPath, error, runId); + } // An authored resume that parks for want of a worker is a park, reported // like the run path's — and with the same remedy, because an authored root // admits its worker at run start: resuming again cannot attach one, so the @@ -522,6 +525,38 @@ function authoredInputArgument(metadata: AuthoredRootMetadata | undefined): Auth * drove correctly, and it produces a report with no `status`, which is the * whole of what `RUN unknown` ever meant. */ +/** + * The authored root's lease was lost and not re-dispatched to this process + * (authored-root.ts driveRootFollowingRetries). Exit 1 because this CLI did + * not finish the run, but the report says what is true of it: still + * `running` in the journal, and resumable -- never `protocol_error` with the + * generic "worker wait canceled" and no run id, which is what it was. + */ +export function rootLeaseLostReport( + command: RunCommand, + base: CheckReport | RunReport, + socketPath: string, + error: AuthoredFlowExecutionError, + fallbackRunId?: string, +): RunExecution { + const rootRunId = error.rootRunId ?? fallbackRunId; + return { + exitCode: 1, + report: { + ...fromBase(command, base), + ok: false, + ...(rootRunId === undefined ? {} : { runId: rootRunId, rootRunId }), + socketPath, + status: 'running', + diagnostics: [...base.diagnostics, { + severity: 'failure', + kind: 'root_lease_lost', + message: error.message.replace(/^root_lease_lost: /, ''), + }], + }, + }; +} + export function authoredStepFailure( command: RunCommand, base: CheckReport | RunReport, @@ -798,7 +833,12 @@ export async function classifyOutcome( let parkedStep: ParkedStep | undefined; let needsHuman = false; let unclassifiedPolls = 0; - while (current.status === 'parked') { + // `running` too: an authored body re-driven after a lost root lease re-admits + // its child by admission key and gets the EXISTING run back, mid-attempt. + // Only `parked` was followed, so that child -- on the attempt 2 the kernel + // retried it to -- fell straight through to "status running without a + // classifiable completion" and failed the flow (customer rw_3a0fcb71). + while (current.status === 'parked' || current.status === 'running') { const inspection = await inspectOutOfBandStep(client, current.run_id); // A runnable sibling may only be waiting for capacity held by a live // attempt. Keep driving until that lease releases before declaring a park. @@ -812,6 +852,20 @@ export async function classifyOutcome( current = await client.runResume(current.run_id, command === 'run' || options.allowHumanInfluenced); continue; } + // A step in `backoff` is between attempts: the kernel has scheduled the + // next one (a dead worker's retry is due at once, but waits for the + // worker slot the dead attempt held; a failed attempt waits out the + // retry policy's delay, bounded by its max_backoff_ms). That is progress, + // not an unclassifiable state -- treating it as one gave up after 2s on a + // child whose attempt 2 then succeeded (customer rw_3a0fcb71). Once due, + // the step is dispatched (`running`, followed above) or, with no worker, + // left `runnable` and reported as a park. + if (inspection?.backoffStep !== undefined && inspection.runningStep === undefined) { + throwIfCanceled(options.signal, inspection.backoffStep.id); + await delay(EXPIRED_LEASE_POLL_MS, options.signal); + current = await client.runResume(current.run_id, command === 'run' || options.allowHumanInfluenced); + continue; + } if (inspection?.status === 'completed' || inspection?.status === 'failed') { current = await client.runResume(current.run_id, command === 'run' || options.allowHumanInfluenced); continue; @@ -954,6 +1008,8 @@ interface OutOfBandInspection { parkedStep?: ParkedStep; needsHuman: boolean; runningStep?: RunningStep; + /** A step waiting out a retry backoff: its next attempt is scheduled. */ + backoffStep?: ParkedStep; } interface RunningStep extends ParkedStep { @@ -981,6 +1037,9 @@ async function inspectOutOfBandStep( const runningEntry = entries.find(([, step]) => step.type !== 'deterministic' && step.state === 'running', ); + const backoffEntry = entries.find(([, step]) => + step.type !== 'deterministic' && step.state === 'backoff', + ); const parkedEntry = humanEntry ?? runnableEntry; const parkedStep = parkedEntry === undefined ? undefined : { id: parkedEntry[0], @@ -996,11 +1055,21 @@ async function inspectOutOfBandStep( needsHuman: humanEntry !== undefined, ...(parkedStep !== undefined ? { parkedStep } : {}), ...(runningStep !== undefined ? { runningStep } : {}), + ...(backoffEntry !== undefined ? { backoffStep: { + id: backoffEntry[0], + type: backoffEntry[1].type as Extract, + } } : {}), }; } // Match kernel/relayflowd/src/server/client.rs: allow the lease sweep to dispatch a retry. export const LEASE_SWEEP_GRACE_MS = 5_000; +// How often to re-ask the daemon about a step whose lease looks expired here. +const EXPIRED_LEASE_POLL_MS = 250; +// A deadline unchanged for a whole lease (relayflowd LEASE_RENEWAL_MS, 30s) +// plus the sweep grace was renewed by nobody: the attempt is dead and the +// daemon never swept it. Fail rather than hang. +const STALE_LEASE_MS = 30_000 + LEASE_SWEEP_GRACE_MS; async function waitForRunningStep( client: JournalClient, @@ -1018,20 +1087,29 @@ async function waitForRunningStep( stepType: runningStep.type, leaseDeadlineMs, }); + // When the lease deadline last changed, on this process's monotonic clock. + // A live worker renews every ~10s, moving the deadline; a dead lease the + // daemon has not swept keeps it. That -- not a comparison of the daemon's + // deadline with our wall clock -- is what separates the two under skew. + let deadlineSeenAt = performance.now(); while (true) { throwIfCanceled(options.signal, runningStep.id); const remainingMs = leaseDeadlineMs + LEASE_SWEEP_GRACE_MS - Date.now(); - if (remainingMs <= 0) { + if (remainingMs <= 0 && performance.now() - deadlineSeenAt > STALE_LEASE_MS) { throw new Error( `worker lease for step "${runningStep.id}" expired at ${leaseDeadlineMs} without completion`, ); } - await delay(Math.min(50, remainingMs), options.signal); + // Past the deadline by THIS clock alone is not an expiry: the daemon may + // still hold the lease (skew) or be sweeping it into a retry. Keep + // following the step, at a slower poll, until the daemon says otherwise. + await delay(remainingMs <= 0 ? EXPIRED_LEASE_POLL_MS : Math.min(50, remainingMs), options.signal); const snapshot = await client.runGet(runId); const step = snapshot.steps[runningStep.id]; if (step?.state !== 'running') return; if (step.lease_deadline_ms !== undefined && step.lease_deadline_ms !== leaseDeadlineMs) { leaseDeadlineMs = step.lease_deadline_ms; + deadlineSeenAt = performance.now(); options.onWait?.({ runId, stepId: runningStep.id, @@ -1098,7 +1176,14 @@ function errorMessage(error: unknown): string { } function throwIfCanceled(signal: AbortSignal | undefined, stepId: string): void { - if (signal?.aborted === true) throw new Error(`waiting for running step "${stepId}" was canceled`); + if (signal?.aborted === true) { + throw new Error(`waiting for running step "${stepId}" was canceled${cancelCause(signal)}`, { cause: signal.reason }); + } +} + +/** Why a wait was canceled, when whoever aborted it said (a lost lease does). */ +function cancelCause(signal: AbortSignal): string { + return signal.reason instanceof Error && signal.reason.name !== 'AbortError' ? `: ${signal.reason.message}` : ''; } export function delay(ms: number, signal?: AbortSignal): Promise { @@ -1112,7 +1197,9 @@ export function delay(ms: number, signal?: AbortSignal): Promise { const cancel = (): void => { clearTimeout(timer); signal.removeEventListener('abort', cancel); - rejectDelay(new Error('worker wait canceled')); + // Carries the abort reason: a bare "worker wait canceled" was all a + // lost root lease ever reported (customer rw_3a0fcb71). + rejectDelay(new Error(`worker wait canceled${cancelCause(signal)}`, { cause: signal.reason })); }; if (signal.aborted) cancel(); else signal.addEventListener('abort', cancel, { once: true }); diff --git a/packages/sdk/src/cli/step-failure.ts b/packages/sdk/src/cli/step-failure.ts index 35eae1cff..78c38e420 100644 --- a/packages/sdk/src/cli/step-failure.ts +++ b/packages/sdk/src/cli/step-failure.ts @@ -63,6 +63,8 @@ export async function stepFailureDetails( // Keyed by step, so interleaved steps never pool their attempts and a page // boundary never splits one step's history. const histories = new Map(); + // Attempts the kernel itself declared dead (see `unchargedAttempts`). + const uncharged = new Map(); while (true) { const { entries } = await client.journalRead(runId, fromSeq, 100); if (entries.length === 0) break; @@ -93,6 +95,7 @@ export async function stepFailureDetails( // holds every entry; only this diagnostic's candidate is cleared. if (payload === undefined || typeof completionReason !== 'string' || completionReason === 'success') { histories.delete(stepId); + uncharged.delete(stepId); continue; } const history = histories.get(stepId) ?? { records: [], causes: [] }; @@ -101,6 +104,10 @@ export async function stepFailureDetails( history.records.push(attemptFailure(entry['attempt'], completionReason, payload['disposition'], payload)); history.causes.push(failureCause(completionReason, payload)); histories.set(stepId, history); + if (payload['disposition'] !== 'step_done' && payload['completed_by'] === 'kernel' + && (completionReason === 'crashed' || completionReason === 'lease_expired')) { + uncharged.set(stepId, (uncharged.get(stepId) ?? 0) + 1); + } // A terminal completion that is not a success is the failure, whatever // its step type. The old predicate also demanded a non-zero `exit_code`, // which no agent completion carries and which a deterministic step that @@ -116,6 +123,7 @@ export async function stepFailureDetails( ...(stepType === undefined ? {} : { stepType }), ...(typeof attempt === 'number' && Number.isSafeInteger(attempt) && attempt > 0 ? { attempt } : {}), ...(maxIterations === undefined ? {} : { maxIterations }), + ...((uncharged.get(stepId) ?? 0) > 0 ? { unchargedAttempts: uncharged.get(stepId) } : {}), ...terminalEvidence(payload), // A single failed attempt is already fully described by the scalars // above; repeating it as a one-element history would add a clause to @@ -169,6 +177,16 @@ export function renderStepEvidence(details: StepFailedDetails): string { function renderAttempt(details: StepFailedDetails): string { if (details.attempt === undefined) return ''; + const uncharged = details.unchargedAttempts ?? 0; + // `attempt=N/M` compares like with like only when every attempt was charged. + // A worker that died (crashed / lease_expired) costs no iteration, so a + // one-iteration step can legitimately run attempt 2 -- printed as + // `attempt=2/1` it read as the budget being overrun. Say which is which. + if (uncharged > 0 && uncharged < details.attempt) { + return ` attempt=${details.attempt}` + + ` (iteration ${details.attempt - uncharged}${details.maxIterations === undefined ? '' : `/${details.maxIterations}`};` + + ` ${uncharged} earlier attempt${uncharged === 1 ? '' : 's'} lost to a dead worker, not charged)`; + } return ` attempt=${details.attempt}` + (details.maxIterations === undefined ? '' : `/${details.maxIterations}`); } diff --git a/packages/sdk/src/failure-kinds.ts b/packages/sdk/src/failure-kinds.ts index 9d8995c5e..cbbe9c84f 100644 --- a/packages/sdk/src/failure-kinds.ts +++ b/packages/sdk/src/failure-kinds.ts @@ -144,6 +144,12 @@ export const RUN_FAILURE_KINDS = [ * attaches and before the resume touches the journal. */ 'local_agent_unavailable', + /** + * The authored root's worker lease was lost and the kernel's retry did not + * reach this process. The run is NOT failed (`status: running`): its + * completed steps are journaled and `flows resume ` continues it. + */ + 'root_lease_lost', ] as const; /** @@ -271,6 +277,14 @@ export interface StepFailedDetails { * `retries_exhausted`. */ maxIterations?: number; + /** + * Attempts before this one that died without a result -- the kernel + * journaled them `crashed` or `lease_expired` itself -- and so, by design, + * consumed none of `maxIterations` (relayflowd-core recovery.rs). Present + * only when non-zero. Without it `attempt` can exceed `maxIterations` + * (`attempt=2/1`) and read as the kernel overrunning its own budget. + */ + unchargedAttempts?: number; exitCode?: number; /** * Terminal-safe UTF-8 excerpt of the captured stdout, at most diff --git a/packages/sdk/src/protocol.ts b/packages/sdk/src/protocol.ts index d4296647c..c418d796b 100644 --- a/packages/sdk/src/protocol.ts +++ b/packages/sdk/src/protocol.ts @@ -253,6 +253,8 @@ export interface StepDispatchEvent { lease_id: string; idempotency_key: string; pins: Pins; + /** `lease_deadline_ms` as a duration at dispatch; see StepHeartbeatResult. */ + lease_ttl_ms?: number; /** * The `wake_context` payload from the run's `subscription.matched` * journal entry — carries the triggering event and any epoch summary @@ -291,6 +293,13 @@ export interface StepHeartbeatParams { } export interface StepHeartbeatResult { lease_deadline_ms: number; + /** + * The renewed lease as a duration on the daemon's clock. Absent from older + * daemons; a worker that has it times the lease on its own monotonic clock + * instead of comparing the absolute deadline with a wall clock that may + * disagree with the daemon's. + */ + lease_ttl_ms?: number; } /** completionReason mirrors kernel DESIGN.md §1.3. */ diff --git a/packages/sdk/src/worker-lease.ts b/packages/sdk/src/worker-lease.ts index a1ef1c7db..2e65f292c 100644 --- a/packages/sdk/src/worker-lease.ts +++ b/packages/sdk/src/worker-lease.ts @@ -2,8 +2,12 @@ import { JournalProtocolError, type JournalClient } from './journal-client.js'; import type { StepDispatchEvent } from './protocol.js'; export class WorkerLeaseLostError extends Error { - constructor(readonly reason: 'already_expired' | 'renewal_expired' | 'completion_expired', message: string) { - super(message); + constructor( + readonly reason: 'already_expired' | 'renewal_expired' | 'completion_expired', + message: string, + options?: { cause?: unknown }, + ) { + super(message, options); this.name = 'WorkerLeaseLostError'; } } @@ -36,13 +40,19 @@ export async function withWorkerLease( ): Promise { const controller = new AbortController(); let stopped = false; - let latestDeadline = dispatch.lease_deadline_ms; + // Every deadline below is on THIS process's monotonic clock. The daemon + // issues `lease_deadline_ms` on its own wall clock, and comparing it with + // ours made the lease exactly as good as the two clocks' agreement: a + // worker 45s ahead refused every dispatch as already expired, one 75s + // behind scheduled its first renewal after the daemon had swept the lease + // (customer rw_3a0fcb71; tests/worker-lease-drift-live.test.ts). + let latestDeadline = Number.NaN; let renewalTimer: NodeJS.Timeout | undefined; let expiryTimer: NodeJS.Timeout | undefined; let pending: Promise = Promise.resolve(); const fail = (error: unknown): void => { controller.abort(error); }; const armExpiry = (deadline: number): number => { - const remaining = deadline - Date.now(); + const remaining = deadline - performance.now(); if (!Number.isFinite(remaining)) { throw new Error(`Agent lease is already expired for ${dispatch.run_id}/${dispatch.step_id}.`); } @@ -57,16 +67,19 @@ export async function withWorkerLease( return remaining; }; const renew = async (): Promise => { + // Anchored BEFORE the request: the daemon starts the renewed lease no + // earlier than it receives this, so the local deadline errs early. + const sentAt = performance.now(); const result = await untilAborted(client.stepHeartbeat( dispatch.run_id, dispatch.step_id, dispatch.attempt, dispatch.lease_id, ), controller.signal); controller.signal.throwIfAborted(); // A response handled after local expiry cannot revive ownership, even // if its future deadline was issued before this event loop stalled. - if (Date.now() >= latestDeadline) { + if (performance.now() >= latestDeadline) { throw new WorkerLeaseLostError('renewal_expired', `Agent lease expired before renewal for ${dispatch.run_id}/${dispatch.step_id}.`); } - const remaining = armExpiry(result.lease_deadline_ms); + const remaining = armExpiry(localDeadline(result.lease_deadline_ms, result.lease_ttl_ms, sentAt)); if (!stopped) { renewalTimer = setTimeout(() => { pending = renew().catch(fail); @@ -74,10 +87,23 @@ export async function withWorkerLease( } }; try { - armExpiry(dispatch.lease_deadline_ms); + armExpiry(localDeadline(dispatch.lease_deadline_ms, dispatch.lease_ttl_ms, performance.now())); // Establish ownership before starting an effectful process. await renew(); - const result = await execute(controller.signal); + let result: T; + try { + result = await execute(controller.signal); + } catch (error) { + // A body that rejected because the lease was lost under it reports + // whatever its wait happened to say ("worker wait canceled"). The lease + // is the cause, and callers branch on it (isLeaseLost): an attempt the + // kernel will retry must not be recorded as the body's own failure. + const lost = controller.signal.reason; + if (lost instanceof WorkerLeaseLostError && error !== lost) { + throw new WorkerLeaseLostError(lost.reason, lost.message, { cause: error }); + } + throw error; + } stopped = true; if (renewalTimer !== undefined) clearTimeout(renewalTimer); // Drain any renewal before the caller sends step.complete. A renewal @@ -86,7 +112,7 @@ export async function withWorkerLease( controller.signal.throwIfAborted(); // Timer callbacks can be delayed behind a resolved subprocess promise. // Check the clock itself before permitting step.complete. - if (Date.now() >= latestDeadline) { + if (performance.now() >= latestDeadline) { throw new WorkerLeaseLostError('completion_expired', `Agent lease expired before completion for ${dispatch.run_id}/${dispatch.step_id}.`); } return result; @@ -99,6 +125,18 @@ export async function withWorkerLease( } } +/** + * A daemon lease deadline on this process's monotonic clock. With the + * daemon's `lease_ttl_ms` the wall clocks never meet: the duration is added to + * `anchor`, a moment no later than the daemon started it. An older daemon + * sends only the absolute deadline, which keeps the old wall-clock comparison + * -- read NOW, not at `anchor`, or the round trip would be subtracted twice. + */ +function localDeadline(daemonDeadlineMs: number, ttlMs: number | undefined, anchor: number): number { + if (typeof ttlMs === 'number' && Number.isFinite(ttlMs)) return anchor + ttlMs; + return performance.now() + (daemonDeadlineMs - Date.now()); +} + function untilAborted(request: Promise, signal: AbortSignal): Promise { return new Promise((resolve, reject) => { const abort = (): void => { reject(signal.reason); }; diff --git a/packages/sdk/tests/classify-outcome.test.ts b/packages/sdk/tests/classify-outcome.test.ts index a79d9b354..f28129cc9 100644 --- a/packages/sdk/tests/classify-outcome.test.ts +++ b/packages/sdk/tests/classify-outcome.test.ts @@ -140,4 +140,57 @@ describe('the remedy on a worker park', () => { expect(message).toContain('no attached worker was eligible'); expect(message).not.toMatch(/flows (run|resume)/); }); + + /// Customer rw_3a0fcb71. A worker died, the kernel journaled `lease_expired` + /// with `disposition: retry`, and the step sat in `backoff` waiting for the + /// worker slot the dead attempt held. That lasted longer than the 2s + /// unclassified bound, so the CLI reported `status running without a + /// classifiable completion` over a child whose attempt 2 then succeeded. + it('follows a step through a retry backoff longer than the unclassified bound', async () => { + const backoff: RunGetResult = { + run_id: RUN_ID, + status: 'running', + steps: { answer: { type: 'agent', state: 'backoff' } as never }, + budget: { tokens_in: 0, tokens_out: 0, dollars: '0' }, + }; + const completed: RunGetResult = { ...backoff, status: 'completed', steps: { answer: { type: 'agent', state: 'done' } as never } }; + // 3s of backoff by the clock, past the 2s unclassified bound. + const until = Date.now() + 3_000; + let resumes = 0; + const client = { + runGet: async () => (Date.now() < until ? backoff : completed), + runResume: async (): Promise => { + resumes += 1; + return Date.now() < until + ? parked + : { run_id: RUN_ID, status: 'completed', completion_reason: 'success', completed_steps: 2 }; + }, + } as unknown as JournalClient; + const result = await classifyOutcome(client, 'run', parked, base, '/unused', {}); + expect(result.exitCode).toBe(0); + expect(result.report.completionReason).toBe('success'); + expect(resumes).toBeGreaterThan(1); + }, 20_000); + + /// The re-admitted child: `run.start` on an existing admission key returns + /// the run as it is -- `running`, its retried attempt leased to a worker. + it('follows a run.start outcome that is already running on a retried attempt', async () => { + const until = Date.now() + 500; + const running: RunGetResult = { + run_id: RUN_ID, + status: 'running', + steps: { answer: { type: 'agent', state: 'running', lease_deadline_ms: Date.now() + 30_000 } as never }, + budget: { tokens_in: 0, tokens_out: 0, dollars: '0' }, + }; + const client = { + runGet: async () => (Date.now() < until ? running + : { ...running, status: 'completed', steps: { answer: { type: 'agent', state: 'done' } } }), + runResume: async (): Promise => + ({ run_id: RUN_ID, status: 'completed', completion_reason: 'success', completed_steps: 1 }), + } as unknown as JournalClient; + const result = await classifyOutcome(client, 'run', + { run_id: RUN_ID, status: 'running', completion_reason: null, completed_steps: 0 }, base, '/unused', {}); + expect(result.exitCode).toBe(0); + expect(result.report.completionReason).toBe('success'); + }); }); diff --git a/packages/sdk/tests/cli.test.ts b/packages/sdk/tests/cli.test.ts index 569e7511e..c8b6138ad 100644 --- a/packages/sdk/tests/cli.test.ts +++ b/packages/sdk/tests/cli.test.ts @@ -1049,7 +1049,11 @@ describe('flows run/resume CLI over the journal protocol', () => { expect(output.stderr.join('\n')).not.toContain('protocol_error'); }); - it('bounds a worker wait by its lease plus sweep grace and reports what it is waiting for', async () => { + // A deadline already past by this process's clock is not the daemon's + // verdict (clock skew, a sweep in progress): the CLI keeps following the + // step until the daemon settles it, instead of exiting 1 over a run that + // then completed. The never-swept bound is in worker-lease-sweep.test.ts. + it('follows a worker wait past a locally expired lease until the daemon settles it', async () => { const dataDir = temporaryProject('flows-run-lease-'); const leaseDeadlineMs = Date.now() - 5_001; let snapshots = 0; @@ -1087,10 +1091,11 @@ describe('flows run/resume CLI over the journal protocol', () => { 'run', '--data-dir', dataDir, join(TESTDATA, 'hello-llm.flow.yaml'), ], output.io); - expect(code).toBe(1); + expect(code).toBe(0); expect(output.stderr.join('\n')).toContain('WAITING [worker_lease]'); expect(output.stderr.join('\n')).toContain(`until ${leaseDeadlineMs}`); - expect(output.stderr.join('\n')).toContain('worker lease for step "answer" expired'); + expect(output.stderr.join('\n')).not.toContain('worker lease for step "answer" expired'); + expect(snapshots).toBeGreaterThan(4); }); it('allows a caller to cancel a worker-lease wait', async () => { diff --git a/packages/sdk/tests/worker-lease-drift-live.test.ts b/packages/sdk/tests/worker-lease-drift-live.test.ts new file mode 100644 index 000000000..9aa737d01 --- /dev/null +++ b/packages/sdk/tests/worker-lease-drift-live.test.ts @@ -0,0 +1,143 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { compileSpec, toKernelSpec } from '../src/compile.js'; +import type { JournalClient } from '../src/journal-client.js'; +import type { StepDispatchEvent } from '../src/protocol.js'; +import { SPEC_SCHEMA_VERSION } from '../src/spec.js'; +import { withWorkerLease } from '../src/worker-lease.js'; +import { renderStepEvidence, stepFailureDetails } from '../src/cli/step-failure.js'; +import { chainFixture } from './flow-chain-fixture.js'; + +const closes: Array<() => Promise | void> = []; +afterEach(async () => { + vi.restoreAllMocks(); + for (const close of closes.splice(0).reverse()) await close(); +}); + +/** + * Customer rw_3a0fcb71 issue 1: an `f.agent` attempt lost its lease while the + * worker was healthy and heartbeating, the kernel journaled `lease_expired` + * and retried it. The daemon issues lease deadlines on ITS wall clock and the + * worker judged them against ITS OWN `Date.now()`, so any skew between the two + * clocks (a container VM that fell behind after sleep, an NTP step) moved the + * worker's idea of the deadline: + * + * - worker clock AHEAD of the daemon by more than the 30s renewal: every + * dispatch is refused locally as `already_expired`, and nothing renews; + * - worker clock BEHIND by more than ~60s: `remaining / 3` exceeds the lease, + * so the first renewal lands after the daemon has already swept it. + * + * These drive a real relayflowd and the real `withWorkerLease`; only the + * worker's wall clock is skewed. + */ +describe('worker lease under clock skew between worker and daemon', () => { + async function leasedAgentRun(): Promise<{ + journal: JournalClient; + worker: JournalClient; + runId: string; + dispatch: StepDispatchEvent; + }> { + const fixture = chainFixture(); + closes.push(() => fixture.close()); + const journal = await fixture.connect(); + const stream = 'drift-stream'; + const worker = journal.createPeer(); + await worker.connect(); + await worker.hello('drift-worker'); + closes.push(() => worker.close()); + const dispatched = new Promise(resolve => worker.once('step.dispatch', resolve)); + await worker.workerAttach('drift-worker', ['agent'], + { workspace: [], streams: [{ stream, read_offset: 0 }] }, 1); + const spec = toKernelSpec(compileSpec({ + version: SPEC_SCHEMA_VERSION, name: 'drift', + steps: [{ id: 'agent-1', type: 'agent', instruction: 'do it', maxIterations: 1, recoveryMode: 'reset', + surfaces: { streams: [{ stream }] } }], + })); + const outcome = await journal.runStart(spec); + return { journal, worker, runId: outcome.run_id, dispatch: await dispatched }; + } + + function skewWorkerClock(offsetMs: number): void { + const real = Date.now.bind(Date); + vi.spyOn(Date, 'now').mockImplementation(() => real() + offsetMs); + } + + async function completions(journal: JournalClient, runId: string): Promise> { + const entries = (await journal.journalRead(runId, 1, 1000)).entries as Array<{ + entry_type: string; payload?: { completionReason?: string; disposition?: string }; + }>; + return entries.filter(entry => entry.entry_type === 'step.completed') + .map(entry => [entry.payload?.completionReason, entry.payload?.disposition]); + } + + it('holds the lease when the worker clock runs 45s ahead of the daemon', async () => { + const { journal, worker, runId, dispatch } = await leasedAgentRun(); + skewWorkerClock(45_000); + const output = await withWorkerLease(worker, dispatch, async () => 'done'); + expect(output).toBe('done'); + await worker.stepComplete(runId, dispatch.step_id, dispatch.attempt, dispatch.idempotency_key, 'success', { + output: { stdout_tail: 'ok' }, started_pins: dispatch.pins, end_pins: dispatch.pins, + }); + expect(await completions(journal, runId)).toEqual([['success', 'step_done']]); + }, 30_000); + + it('renews before the daemon deadline when the worker clock runs 75s behind the daemon', async () => { + const { journal, worker, runId, dispatch } = await leasedAgentRun(); + skewWorkerClock(-75_000); + // Longer than one 30s lease: survives only if a renewal lands in time. + const output = await withWorkerLease(worker, dispatch, + () => new Promise(resolve => setTimeout(() => resolve('done'), 36_000))); + expect(output).toBe('done'); + await worker.stepComplete(runId, dispatch.step_id, dispatch.attempt, dispatch.idempotency_key, 'success', { + output: { stdout_tail: 'ok' }, started_pins: dispatch.pins, end_pins: dispatch.pins, + }); + expect(await completions(journal, runId)).toEqual([['success', 'step_done']]); + }, 60_000); +}); + +describe('attempt accounting after a lost worker', () => { + it('does not report attempt=2/1 when the first attempt died uncharged', async () => { + // max_iterations 1; the first worker disconnects mid-attempt (kernel: + // `crashed`, retry, no iteration charged) and the second attempt fails on + // its own merits. Attempt 2 of a 1-iteration budget is correct kernel + // behaviour, and the report has to say why rather than print `2/1`. + const fixture = chainFixture(); + closes.push(() => fixture.close()); + const journal = await fixture.connect(); + const stream = 'accounting-stream'; + const attach = async (onDispatch: (worker: JournalClient, event: StepDispatchEvent) => void) => { + const worker = journal.createPeer(); + await worker.connect(); + await worker.hello('accounting-worker'); + worker.on('step.dispatch', (event: StepDispatchEvent) => onDispatch(worker, event)); + await worker.workerAttach(`worker-${Math.random()}`, ['agent'], + { workspace: [], streams: [{ stream, read_offset: 0 }] }, 1); + return worker; + }; + let firstDispatched!: () => void; + const dispatchedToFirst = new Promise(resolve => { firstDispatched = resolve; }); + const first = await attach(() => firstDispatched()); + const outcome = await journal.runStart(toKernelSpec(compileSpec({ + version: SPEC_SCHEMA_VERSION, name: 'accounting', + steps: [{ id: 'agent-1', type: 'agent', instruction: 'do it', maxIterations: 1, recoveryMode: 'reset', + surfaces: { streams: [{ stream }] } }], + }))); + await dispatchedToFirst; + const second = await attach((worker, event) => { + void worker.stepComplete(event.run_id, event.step_id, event.attempt, event.idempotency_key, 'worker_error', { + output: { exit_code: 1, stdout_tail: '', stderr_tail: 'boom' }, started_pins: event.pins, end_pins: event.pins, + }); + }); + closes.push(() => second.close()); + first.close(); + const deadline = Date.now() + 30_000; + while ((await journal.runResume(outcome.run_id, true)).status !== 'failed') { + if (Date.now() > deadline) throw new Error('run did not fail'); + await new Promise(resolve => setTimeout(resolve, 25)); + } + const details = (await stepFailureDetails(journal, outcome.run_id))!; + expect(details).toMatchObject({ stepId: 'agent-1', attempt: 2, maxIterations: 1, unchargedAttempts: 1 }); + const rendered = renderStepEvidence(details); + expect(rendered).not.toContain('attempt=2/1'); + expect(rendered).toContain('attempt=2 (iteration 1/1; 1 earlier attempt lost to a dead worker, not charged)'); + }, 60_000); +}); diff --git a/packages/sdk/tests/worker-lease-sweep.test.ts b/packages/sdk/tests/worker-lease-sweep.test.ts index d169a3de3..b9160b029 100644 --- a/packages/sdk/tests/worker-lease-sweep.test.ts +++ b/packages/sdk/tests/worker-lease-sweep.test.ts @@ -5,7 +5,7 @@ import type { JournalClient } from '../src/journal-client.js'; afterEach(() => vi.useRealTimers()); function fixture() { - vi.useFakeTimers(); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }); const deadline = Date.now() - 1; const snapshot = (state: string, lease = deadline) => ({ run_id: 'run', status: 'parked', @@ -31,13 +31,35 @@ it('waits through an expired snapshot until the kernel retries and completes', a .mockResolvedValue({ run_id: 'run', status: 'completed', completion_reason: 'success', completed_steps: 1 }); const execution = run(); const assertion = expect(execution).resolves.toMatchObject({ exitCode: 0, report: { completionReason: 'success' } }); - await Promise.all([assertion, vi.advanceTimersByTimeAsync(100)]); + await Promise.all([assertion, vi.advanceTimersByTimeAsync(1_000)]); expect(client.runResume).toHaveBeenCalledTimes(2); }); it('still fails when the kernel never resolves an expired lease after sweep grace', async () => { const { run } = fixture(); const assertion = expect(run()).rejects.toThrow('without completion'); - await vi.advanceTimersByTimeAsync(5_001); + await vi.advanceTimersByTimeAsync(35_500); await assertion; }); + +it('follows a live step whose renewed deadline looks expired on a skewed CLI clock', async () => { + // The CLI's wall clock runs 60s ahead of the daemon: every deadline the + // worker renews reads as already past here. It used to throw + // `expired ... without completion` five seconds in, abandoning a healthy + // attempt; the moving deadline is what says it is alive. + const { client, snapshot, run } = fixture(); + let renewed = Date.now() - 60_000; + client.runGet.mockImplementation(async () => { + renewed += 250; + return snapshot('running', renewed); + }); + const execution = run(); + let settled = false; + void execution.then(() => { settled = true; }, () => { settled = true; }); + await vi.advanceTimersByTimeAsync(60_000); + expect(settled).toBe(false); + client.runGet.mockResolvedValue(snapshot('completed')); + client.runResume.mockResolvedValue({ run_id: 'run', status: 'completed', completion_reason: 'success', completed_steps: 1 }); + const assertion = expect(execution).resolves.toMatchObject({ exitCode: 0, report: { completionReason: 'success' } }); + await Promise.all([assertion, vi.advanceTimersByTimeAsync(1_000)]); +}); diff --git a/packages/sdk/tests/worker-lease.test.ts b/packages/sdk/tests/worker-lease.test.ts index ef4c2f93a..f35d73cfb 100644 --- a/packages/sdk/tests/worker-lease.test.ts +++ b/packages/sdk/tests/worker-lease.test.ts @@ -6,10 +6,22 @@ import { AgentWorker } from '../src/worker.js'; import { runAgentCli } from '../src/worker-cli.js'; vi.mock('../src/worker-cli.js', () => ({ runAgentCli: vi.fn() })); -afterEach(() => { vi.useRealTimers(); vi.resetAllMocks(); }); +afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); vi.resetAllMocks(); }); + +// A stalled event loop: both clocks jump while queued timer callbacks wait. +// The lease is timed on the monotonic clock (worker-lease.ts), so a stall +// that moved only the wall clock would not be one. +let stalledMs = 0; +function stall(ms: number): void { + vi.setSystemTime(Date.now() + ms); + stalledMs += ms; +} function setup() { - vi.useFakeTimers(); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'setInterval', 'clearInterval', 'Date', 'performance'] }); + stalledMs = 0; + const fakeNow = performance.now.bind(performance); + vi.spyOn(performance, 'now').mockImplementation(() => fakeNow() + stalledMs); const client = Object.assign(new EventEmitter(), { workerAttach: vi.fn(async () => ({})), stepHeartbeat: vi.fn(async () => ({ lease_deadline_ms: Date.now() + 30_000 })), @@ -107,7 +119,7 @@ describe('worker lease ownership', () => { it('refuses completion past the deadline even before the expiry timer runs', async () => { const { client, worker, errors, dispatch } = setup(); vi.mocked(runAgentCli).mockImplementation(async () => { - vi.setSystemTime(Date.now() + 30_001); // changes the clock WITHOUT running timers + stall(30_001); // changes the clock WITHOUT running timers return { exit_code: 0, stdout_tail: 'late', stderr_tail: '' }; }); await worker.attach(); @@ -121,7 +133,7 @@ it('refuses completion past the deadline even before the expiry timer runs', asy it('does not revive ownership when the initial heartbeat response is handled late', async () => { const { client, worker, errors, dispatch } = setup(); client.stepHeartbeat.mockImplementationOnce(async () => { - vi.setSystemTime(Date.now() + 30_001); // leave timer callbacks queued + stall(30_001); // leave timer callbacks queued return { lease_deadline_ms: Date.now() + 30_000 }; }); await worker.attach(); @@ -136,7 +148,7 @@ it('aborts the CLI when a later heartbeat response would revive an expired lease const { client, worker, errors, dispatch } = setup(); client.stepHeartbeat.mockResolvedValueOnce({ lease_deadline_ms: Date.now() + 30_000 }) .mockImplementationOnce(async () => { - vi.setSystemTime(Date.now() + 20_001); // renewal starts at t=10s + stall(20_001); // renewal starts at t=10s return { lease_deadline_ms: Date.now() + 30_000 }; }); const signals = runningCli(60_000); From ac87f25309eef1fcb2dabaf8f1a1a3e66e20634b Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 30 Sep 2026 13:16:21 +0200 Subject: [PATCH 2/5] fix(sdk): cancel the root re-dispatch wait on caller abort Bugbot on #593: after a lost root lease the driver waited up to 45s for the kernel's re-dispatch without observing the lifecycle signal, so a Ctrl-C in that window blocked until the timer fired. The wait now rejects at once on abort, and the driver rethrows it as the caller's cancel rather than as root_lease_lost. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/sdk/src/authored-root.ts | 30 +++++++++++++++++++----- packages/sdk/tests/authored-root.test.ts | 24 ++++++++++++++++++- 2 files changed, 47 insertions(+), 7 deletions(-) diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 2288208bb..80fb8fc7b 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -274,8 +274,10 @@ async function driveRootFollowingRetries( { code: 'FLOWS_ROOT_LEASE_LOST' }, ); try { - dispatch = await next.wait(ROOT_REDISPATCH_TIMEOUT_MS); - } catch { + dispatch = await next.wait(ROOT_REDISPATCH_TIMEOUT_MS, options.lifecycle?.signal); + } catch (waitError) { + // A caller cancel while waiting is the caller's, not a lost lease. + if (aborted(options.lifecycle?.signal)) throw waitError; throw rootLeaseLost(dispatch, `${lost} The kernel did not re-dispatch the root within ${ROOT_REDISPATCH_TIMEOUT_MS / 1000}s; continue the run with: ${resume}`, error); } if (dispatch.run_id !== firstDispatch.run_id) { @@ -285,6 +287,11 @@ async function driveRootFollowingRetries( } } +/** Read at call time: `signal.aborted` changes while this function awaits. */ +function aborted(signal: AbortSignal | undefined): boolean { + return signal?.aborted === true; +} + function sentence(text: string): string { return /[.!?]$/.test(text) ? text : `${text}.`; } @@ -471,7 +478,7 @@ function hasRootStream(step: Record): boolean { * timeout must not leave a rejected promise nobody awaits. */ function rootRedispatch(peer: JournalClient): { - wait(timeoutMs: number): Promise; + wait(timeoutMs: number, signal?: AbortSignal): Promise; cancel(): void; } { let received: StepDispatchEvent | undefined; @@ -486,13 +493,24 @@ function rootRedispatch(peer: JournalClient): { const cancel = (): void => { peer.off('step.dispatch', onDispatch); }; return { cancel, - wait: timeoutMs => new Promise((resolve, reject) => { + wait: (timeoutMs, signal) => new Promise((resolve, reject) => { if (received !== undefined) { resolve(received); return; } - const timer = setTimeout(() => { + const settle = (): void => { + clearTimeout(timer); + signal?.removeEventListener('abort', onAbort); cancel(); + }; + const onAbort = (): void => { + settle(); + reject(new Error('waiting for the authored root to be re-dispatched was canceled', { cause: signal?.reason })); + }; + const timer = setTimeout(() => { + settle(); reject(new Error('authored root was not re-dispatched')); }, timeoutMs); - deliver = dispatch => { clearTimeout(timer); resolve(dispatch); }; + deliver = dispatch => { settle(); resolve(dispatch); }; + if (signal?.aborted === true) onAbort(); + else signal?.addEventListener('abort', onAbort, { once: true }); }), }; } diff --git a/packages/sdk/tests/authored-root.test.ts b/packages/sdk/tests/authored-root.test.ts index c09dae9dc..9d27553fe 100644 --- a/packages/sdk/tests/authored-root.test.ts +++ b/packages/sdk/tests/authored-root.test.ts @@ -12,7 +12,7 @@ import { resumeDurableAuthoredFlow, type AuthoredRootMetadata, } from '../src/authored-root.js'; -import type { JournalClient } from '../src/journal-client.js'; +import { JournalProtocolError, type JournalClient } from '../src/journal-client.js'; import type { RunOutcome, StepDispatchEvent } from '../src/protocol.js'; import { AuthoredHumanParked } from '../src/authored-flow-error.js'; @@ -166,6 +166,28 @@ describe('durable authored root', () => { expect(journal.peer.completions).toEqual([{ attempt: 1, reason: 'success' }]); }); + it('stops waiting for a root re-dispatch when the caller cancels', async () => { + // The root lease is lost (the daemon refuses the renewal), so the driver + // waits up to 45s for the kernel to re-dispatch the root. A caller cancel + // during that wait must end it at once, as a cancel -- never a + // `root_lease_lost` report, and never after the timeout. + const loaded = await fixture(); + const journal = new RootJournal(); + journal.peer.stepHeartbeat = async () => { + throw new JournalProtocolError('lease_conflict', 'attempt has no active worker lease'); + }; + const controller = new AbortController(); + const started = Date.now(); + const execution = executeDurableAuthoredFlow( + loaded, journal as unknown as JournalClient, undefined, + { dataDir: '/unused', admissionKey: 'cancel-while-redispatch', lifecycle: { signal: controller.signal } }, + ); + setTimeout(() => controller.abort(), 200); + await expect(execution).rejects.toThrow('re-dispatched was canceled'); + expect(Date.now() - started).toBeLessThan(5_000); + expect(journal.peer.completions).toEqual([]); + }); + it('journals plugin digests from composed extensions', async () => { const loaded = await fixture(); const extension = { From affaae9a3dc71db3912b131f4607225f06c9aee1 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Thu, 1 Oct 2026 04:51:12 +0200 Subject: [PATCH 3/5] fix(sdk): follow retries before parking --- packages/sdk/src/authored-root.ts | 8 +++-- packages/sdk/src/cli/run.ts | 6 ++-- packages/sdk/tests/authored-root.test.ts | 30 +++++++++++++++--- packages/sdk/tests/classify-outcome.test.ts | 34 +++++++++++++++++++++ 4 files changed, 70 insertions(+), 8 deletions(-) diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 80fb8fc7b..5471faa84 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -12,7 +12,7 @@ import { type LoadedAuthoredFlow, type SurfaceModuleAuthority, } from './authored-flow-loader.js'; -import { JournalClient } from './journal-client.js'; +import { JournalClient, JournalProtocolError } from './journal-client.js'; import type { RunOutcome, StepDispatchEvent } from './protocol.js'; import { SPEC_SCHEMA_VERSION } from './spec.js'; import type { RunLifecycleOptions } from './cli/run.js'; @@ -259,7 +259,11 @@ async function driveRootFollowingRetries( next.cancel(); return result; } catch (error) { - if (!isLeaseLost(error) || options.lifecycle?.signal?.aborted === true) { + // `run_terminal` from a step verb means the run has already reached a + // terminal state. It is lease-shaped for worker cleanup, but the kernel + // cannot re-dispatch this root, so do not turn it into a 45s retry wait. + const terminal = error instanceof JournalProtocolError && error.code === 'run_terminal'; + if (!isLeaseLost(error) || terminal || options.lifecycle?.signal?.aborted === true) { next.cancel(); throw error; } diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index d74b69b11..d8d469749 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -841,8 +841,10 @@ export async function classifyOutcome( while (current.status === 'parked' || current.status === 'running') { const inspection = await inspectOutOfBandStep(client, current.run_id); // A runnable sibling may only be waiting for capacity held by a live - // attempt. Keep driving until that lease releases before declaring a park. - if (inspection?.parkedStep !== undefined && (inspection.needsHuman || inspection.runningStep === undefined)) { + // attempt or by a retry in backoff. Keep driving until that active work + // settles before declaring a park. + if (inspection?.parkedStep !== undefined && (inspection.needsHuman + || (inspection.runningStep === undefined && inspection.backoffStep === undefined))) { parkedStep = inspection.parkedStep; needsHuman = inspection.needsHuman; break; diff --git a/packages/sdk/tests/authored-root.test.ts b/packages/sdk/tests/authored-root.test.ts index 9d27553fe..88357d469 100644 --- a/packages/sdk/tests/authored-root.test.ts +++ b/packages/sdk/tests/authored-root.test.ts @@ -79,6 +79,7 @@ class RootJournal { startStatus: RunOutcome = outcome(this.runId, 'parked', null); resumeStatus: RunOutcome = outcome(this.runId, 'parked', null); dispatchOnStart = true; + dispatchLeaseMs = 30; resumeCalls = 0; entries: Array> = []; readonly childEntries = new Map>>(); @@ -98,14 +99,14 @@ class RootJournal { } if (this.dispatchOnStart && (this.startStatus.status === 'running' || this.startStatus.status === 'parked')) { - queueMicrotask(() => this.peer.emit('step.dispatch', dispatch(this.runId, 1))); + queueMicrotask(() => this.peer.emit('step.dispatch', dispatch(this.runId, 1, this.dispatchLeaseMs))); } return this.startStatus; } async runResume(runId: string): Promise { this.resumeCalls += 1; if (this.resumeStatus.status === 'running' || this.resumeStatus.status === 'parked') { - queueMicrotask(() => this.peer.emit('step.dispatch', dispatch(runId, 2))); + queueMicrotask(() => this.peer.emit('step.dispatch', dispatch(runId, 2, this.dispatchLeaseMs))); } return this.resumeStatus; } @@ -188,6 +189,27 @@ describe('durable authored root', () => { expect(journal.peer.completions).toEqual([]); }); + it('does not wait for a re-dispatch after the root run is terminal', async () => { + const loaded = await fixture(); + const journal = new RootJournal(); + journal.dispatchLeaseMs = 30_000; + const terminal = Object.assign(new JournalProtocolError('run_terminal', 'run is already terminal'), { + verb: 'step.complete', + }); + journal.peer.stepHeartbeat = async () => ({ lease_deadline_ms: Date.now() + 30_000 }); + journal.peer.stepComplete = async () => { throw terminal; }; + const controller = new AbortController(); + const abort = setTimeout(() => controller.abort(), 2_000); + try { + await expect(executeDurableAuthoredFlow( + loaded, journal as unknown as JournalClient, undefined, + { dataDir: '/unused', admissionKey: 'terminal-root', lifecycle: { signal: controller.signal } }, + )).rejects.toBe(terminal); + } finally { + clearTimeout(abort); + } + }); + it('journals plugin digests from composed extensions', async () => { const loaded = await fixture(); const extension = { @@ -507,11 +529,11 @@ async function fixture( }; } -function dispatch(runId: string, attempt: number): StepDispatchEvent { +function dispatch(runId: string, attempt: number, leaseMs = 30): StepDispatchEvent { return { run_id: runId, step_id: 'authored-root', attempt, step_type: 'agent', spec: {}, lease_id: `lease-${attempt}`, idempotency_key: `key-${attempt}`, - lease_deadline_ms: Date.now() + 30, + lease_deadline_ms: Date.now() + leaseMs, pins: { workspace: [], streams: [{ stream: 'authored-root-stream', read_offset: 0 }] }, }; } diff --git a/packages/sdk/tests/classify-outcome.test.ts b/packages/sdk/tests/classify-outcome.test.ts index f28129cc9..894e1e760 100644 --- a/packages/sdk/tests/classify-outcome.test.ts +++ b/packages/sdk/tests/classify-outcome.test.ts @@ -172,6 +172,40 @@ describe('the remedy on a worker park', () => { expect(resumes).toBeGreaterThan(1); }, 20_000); + it('follows a backoff retry instead of parking on a runnable sibling', async () => { + const retrying: RunGetResult = { + run_id: RUN_ID, + status: 'running', + steps: { + waitingForWorker: { type: 'agent', state: 'runnable' } as never, + retrying: { type: 'agent', state: 'backoff' } as never, + }, + budget: { tokens_in: 0, tokens_out: 0, dollars: '0' }, + }; + const completed: RunGetResult = { + ...retrying, + status: 'completed', + steps: { + waitingForWorker: { type: 'agent', state: 'done' } as never, + retrying: { type: 'agent', state: 'done' } as never, + }, + }; + let resumed = false; + const client = { + runGet: async () => (resumed ? completed : retrying), + runResume: async (): Promise => { + resumed = true; + return { run_id: RUN_ID, status: 'completed', completion_reason: 'success', completed_steps: 2 }; + }, + } as unknown as JournalClient; + + const result = await classifyOutcome(client, 'run', parked, base, '/unused', {}); + + expect(result.exitCode).toBe(0); + expect(result.report.completionReason).toBe('success'); + expect(result.report.parkCause).toBeUndefined(); + }); + /// The re-admitted child: `run.start` on an existing admission key returns /// the run as it is -- `running`, its retried attempt leased to a worker. it('follows a run.start outcome that is already running on a retried attempt', async () => { From 25f4764b89b820b01081f2a33cc054e4d30ec5ea Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Thu, 1 Oct 2026 05:53:44 +0200 Subject: [PATCH 4/5] fix: close lease recovery review gaps --- kernel/relayflowd/src/server.rs | 5 +- .../src/server/session/assignments.rs | 7 +++ kernel/relayflowd/src/server/tests.rs | 63 ++++++++++++++++++- packages/sdk/src/cli/run.ts | 3 +- packages/sdk/src/protocol.ts | 2 +- packages/sdk/src/worker-lease.ts | 7 ++- packages/sdk/tests/classify-outcome.test.ts | 12 ++++ packages/sdk/tests/worker-lease-lost.test.ts | 15 +++++ packages/sdk/tests/worker-lease-sweep.test.ts | 8 +++ packages/sdk/tests/worker-lease.test.ts | 2 +- 10 files changed, 115 insertions(+), 9 deletions(-) diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index 01691b760..59e977a6e 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -399,7 +399,10 @@ fn handle_request( .map_err(internal_error)?; // `lease_ttl_ms`: the same deadline as a duration, so the worker can // time it on its own clock (see the dispatch frame in assignments.rs). - Ok(json!({"lease_deadline_ms": deadline, "lease_ttl_ms": deadline.saturating_sub(now)})) + Ok(json!({ + "lease_deadline_ms": deadline, + "lease_ttl_ms": deadline.saturating_sub(now_ms()).max(0), + })) } "step.complete" => { let params: StepCompleteParams = decode_params(request.params)?; diff --git a/kernel/relayflowd/src/server/session/assignments.rs b/kernel/relayflowd/src/server/session/assignments.rs index 94372523d..ab0901e4c 100644 --- a/kernel/relayflowd/src/server/session/assignments.rs +++ b/kernel/relayflowd/src/server/session/assignments.rs @@ -28,6 +28,13 @@ impl ProtocolHub { if assignment.connection_id != connection_id || assignment.lease_id != lease_id { bail!("heartbeat does not match the active worker lease") } + // A late heartbeat cannot revive ownership after the daemon-side + // deadline. In particular, this closes the transport-delay gap + // between a dispatch TTL being written and the worker receiving + // it; workers heartbeat before beginning effectful execution. + if now_ms >= assignment.lease_deadline_ms { + bail!("attempt worker lease expired before heartbeat") + } assignment.lease_deadline_ms = now_ms.saturating_add(LEASE_RENEWAL_MS); assignment.lease_deadline_ms }; diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index 682a95797..f43b1770a 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -583,17 +583,74 @@ fn dispatch_carries_the_lease_as_a_duration() { let hub = Arc::new(ProtocolHub::default()); let worker_peer = attach_llm_worker(data_dir, &hub, 1); let mut worker_reader = BufReader::new(worker_peer); - let before = now_ms(); start_llm_run(data_dir, &hub); let dispatch = read_frame(&mut worker_reader); + let after = now_ms(); assert_eq!(dispatch["event"], "step.dispatch"); let ttl = dispatch["data"]["lease_ttl_ms"] .as_i64() .expect("lease_ttl_ms"); let deadline = dispatch["data"]["lease_deadline_ms"].as_i64().unwrap(); + let observed_remaining = deadline.saturating_sub(after).max(0); assert!( - ttl > 0 && ttl <= deadline - before, - "ttl {ttl} deadline {deadline} before {before}" + ttl > 0 && ttl >= observed_remaining && ttl - observed_remaining <= 1_000, + "ttl {ttl} deadline {deadline} after {after} observed {observed_remaining}" + ); +} + +#[test] +fn heartbeat_cannot_revive_an_expired_assignment() { + let directory = tempdir().unwrap(); + let data_dir = directory.path(); + let hub = Arc::new(ProtocolHub::default()); + let worker_peer = attach_llm_worker(data_dir, &hub, 1); + let mut worker_reader = BufReader::new(worker_peer); + let run_id = start_llm_run(data_dir, &hub); + let dispatch = read_frame(&mut worker_reader); + let step_id = dispatch["data"]["step_id"].as_str().unwrap().to_owned(); + let attempt = dispatch["data"]["attempt"].as_u64().unwrap() as u32; + let lease_id = dispatch["data"]["lease_id"].as_str().unwrap(); + let deadline = dispatch["data"]["lease_deadline_ms"].as_i64().unwrap(); + + let error = hub + .heartbeat(1, &(run_id, step_id, attempt), lease_id, deadline) + .unwrap_err(); + + assert!(error.to_string().contains("lease expired"), "{error:#}"); +} + +#[test] +fn heartbeat_carries_the_remaining_lease_as_a_duration() { + let directory = tempdir().unwrap(); + let data_dir = directory.path(); + let hub = Arc::new(ProtocolHub::default()); + let worker_peer = attach_llm_worker(data_dir, &hub, 1); + let mut worker_reader = BufReader::new(worker_peer); + let run_id = start_llm_run(data_dir, &hub); + let dispatch = read_frame(&mut worker_reader); + let (writer, _peer) = shared_writer(); + let response = request( + data_dir, + &hub, + 1, + &writer, + &json!({"id":"heartbeat","verb":"step.heartbeat","params":{ + "run_id":run_id, + "step_id":dispatch["data"]["step_id"], + "attempt":dispatch["data"]["attempt"], + "lease_id":dispatch["data"]["lease_id"], + }}) + .to_string(), + ); + let after = now_ms(); + assert!(response.ok, "heartbeat failed: {:?}", response.error); + let result = response.result.unwrap(); + let ttl = result["lease_ttl_ms"].as_i64().expect("lease_ttl_ms"); + let deadline = result["lease_deadline_ms"].as_i64().unwrap(); + let observed_remaining = deadline.saturating_sub(after).max(0); + assert!( + ttl > 0 && ttl >= observed_remaining && ttl - observed_remaining <= 1_000, + "ttl {ttl} deadline {deadline} after {after} observed {observed_remaining}" ); } diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index d8d469749..ee3e16a4d 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -847,6 +847,7 @@ export async function classifyOutcome( || (inspection.runningStep === undefined && inspection.backoffStep === undefined))) { parkedStep = inspection.parkedStep; needsHuman = inspection.needsHuman; + current = { ...current, status: inspection.status }; break; } if (inspection?.runningStep !== undefined) { @@ -1097,7 +1098,7 @@ async function waitForRunningStep( while (true) { throwIfCanceled(options.signal, runningStep.id); const remainingMs = leaseDeadlineMs + LEASE_SWEEP_GRACE_MS - Date.now(); - if (remainingMs <= 0 && performance.now() - deadlineSeenAt > STALE_LEASE_MS) { + if (performance.now() - deadlineSeenAt > STALE_LEASE_MS) { throw new Error( `worker lease for step "${runningStep.id}" expired at ${leaseDeadlineMs} without completion`, ); diff --git a/packages/sdk/src/protocol.ts b/packages/sdk/src/protocol.ts index c418d796b..454002d91 100644 --- a/packages/sdk/src/protocol.ts +++ b/packages/sdk/src/protocol.ts @@ -253,7 +253,7 @@ export interface StepDispatchEvent { lease_id: string; idempotency_key: string; pins: Pins; - /** `lease_deadline_ms` as a duration at dispatch; see StepHeartbeatResult. */ + /** Remaining lease duration at dispatch; `lease_deadline_ms` stays absolute. */ lease_ttl_ms?: number; /** * The `wake_context` payload from the run's `subscription.matched` diff --git a/packages/sdk/src/worker-lease.ts b/packages/sdk/src/worker-lease.ts index 2e65f292c..8396f165a 100644 --- a/packages/sdk/src/worker-lease.ts +++ b/packages/sdk/src/worker-lease.ts @@ -99,8 +99,11 @@ export async function withWorkerLease( // is the cause, and callers branch on it (isLeaseLost): an attempt the // kernel will retry must not be recorded as the body's own failure. const lost = controller.signal.reason; - if (lost instanceof WorkerLeaseLostError && error !== lost) { - throw new WorkerLeaseLostError(lost.reason, lost.message, { cause: error }); + if (isLeaseLost(lost) && error !== lost) { + if (lost instanceof WorkerLeaseLostError) { + throw new WorkerLeaseLostError(lost.reason, lost.message, { cause: error }); + } + throw lost; } throw error; } diff --git a/packages/sdk/tests/classify-outcome.test.ts b/packages/sdk/tests/classify-outcome.test.ts index 894e1e760..c92d4d2f9 100644 --- a/packages/sdk/tests/classify-outcome.test.ts +++ b/packages/sdk/tests/classify-outcome.test.ts @@ -206,6 +206,18 @@ describe('the remedy on a worker park', () => { expect(result.report.parkCause).toBeUndefined(); }); + it('uses the inspected parked status after a running outcome loses its worker', async () => { + const { client } = clientReturning([parkedOnAgent]); + + const result = await classifyOutcome(client, 'run', + { ...parked, status: 'running' }, base, '/unused', {}); + + expect(result.exitCode).toBe(3); + expect(result.report.status).toBe('parked'); + expect(result.report.parkCause).toBe('worker_unavailable'); + expect(result.report.parkedStep?.id).toBe('work'); + }); + /// The re-admitted child: `run.start` on an existing admission key returns /// the run as it is -- `running`, its retried attempt leased to a worker. it('follows a run.start outcome that is already running on a retried attempt', async () => { diff --git a/packages/sdk/tests/worker-lease-lost.test.ts b/packages/sdk/tests/worker-lease-lost.test.ts index e023c098f..1c86523d7 100644 --- a/packages/sdk/tests/worker-lease-lost.test.ts +++ b/packages/sdk/tests/worker-lease-lost.test.ts @@ -94,6 +94,21 @@ it('preserves the heartbeat rejection through abort and finally', async () => { await assertion; }); +it('preserves a protocol lease loss over the body cancellation error', async () => { + const { client, dispatch } = setup('llm'); + const lost = Object.assign(new JournalProtocolError('lease_conflict', 'lost'), { + verb: 'step.heartbeat', + }); + const canceled = new Error('worker wait canceled'); + client.stepHeartbeat.mockResolvedValueOnce({ lease_deadline_ms: Date.now() + 30_000 }) + .mockRejectedValueOnce(lost); + const result = withWorkerLease(client as unknown as JournalClient, dispatch, signal => + new Promise((_resolve, reject) => signal.addEventListener('abort', () => reject(canceled)))); + const assertion = expect(result).rejects.toBe(lost); + await vi.advanceTimersByTimeAsync(10_000); + await assertion; +}); + it.each(['step.heartbeat', 'step.complete', 'step.wait'])('tags %s refusals at the client wrapper', async verb => { const client = new JournalClient('/unused'); const error = new JournalProtocolError('run_terminal', 'finished'); diff --git a/packages/sdk/tests/worker-lease-sweep.test.ts b/packages/sdk/tests/worker-lease-sweep.test.ts index b9160b029..1cc66b931 100644 --- a/packages/sdk/tests/worker-lease-sweep.test.ts +++ b/packages/sdk/tests/worker-lease-sweep.test.ts @@ -42,6 +42,14 @@ it('still fails when the kernel never resolves an expired lease after sweep grac await assertion; }); +it('bounds an unchanged lease when the CLI clock is behind the daemon', async () => { + const { client, snapshot, run } = fixture(); + client.runGet.mockResolvedValue(snapshot('running', Date.now() + 75_000)); + const assertion = expect(run()).rejects.toThrow('without completion'); + await vi.advanceTimersByTimeAsync(35_500); + await assertion; +}); + it('follows a live step whose renewed deadline looks expired on a skewed CLI clock', async () => { // The CLI's wall clock runs 60s ahead of the daemon: every deadline the // worker renews reads as already past here. It used to throw diff --git a/packages/sdk/tests/worker-lease.test.ts b/packages/sdk/tests/worker-lease.test.ts index f35d73cfb..0cb168b81 100644 --- a/packages/sdk/tests/worker-lease.test.ts +++ b/packages/sdk/tests/worker-lease.test.ts @@ -6,7 +6,7 @@ import { AgentWorker } from '../src/worker.js'; import { runAgentCli } from '../src/worker-cli.js'; vi.mock('../src/worker-cli.js', () => ({ runAgentCli: vi.fn() })); -afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); vi.resetAllMocks(); }); +afterEach(() => { vi.restoreAllMocks(); vi.resetAllMocks(); vi.useRealTimers(); }); // A stalled event loop: both clocks jump while queued timer callbacks wait. // The lease is timed on the monotonic clock (worker-lease.ts), so a stall From 00a214735dd56fd590564aa898a987fdafa806d6 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Thu, 1 Oct 2026 06:29:40 +0200 Subject: [PATCH 5/5] fix(sdk): preserve worker park classification --- packages/sdk/src/cli/run.ts | 8 +++++++- packages/sdk/tests/classify-outcome.test.ts | 16 ++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index ee3e16a4d..5ec492868 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -847,7 +847,13 @@ export async function classifyOutcome( || (inspection.runningStep === undefined && inspection.backoffStep === undefined))) { parkedStep = inspection.parkedStep; needsHuman = inspection.needsHuman; - current = { ...current, status: inspection.status }; + // `run.get` reports the live snapshot as `running` for an ordinary + // runnable worker wait, even when `run.start`/`run.resume` already + // classified the outcome as parked. Conversely, a re-admitted run can + // arrive here with a stale `running` outcome after its worker vanished. + // The step-level inspection is authoritative that this is a park; do + // not copy its broader run status and lose that classification. + current = { ...current, status: 'parked' }; break; } if (inspection?.runningStep !== undefined) { diff --git a/packages/sdk/tests/classify-outcome.test.ts b/packages/sdk/tests/classify-outcome.test.ts index c92d4d2f9..47cb65562 100644 --- a/packages/sdk/tests/classify-outcome.test.ts +++ b/packages/sdk/tests/classify-outcome.test.ts @@ -94,6 +94,11 @@ const parkedOnAgent: RunGetResult = { budget: { tokens_in: 0, tokens_out: 0, dollars: '0' }, }; +const runningOnRunnableAgent: RunGetResult = { + ...parkedOnAgent, + status: 'running', +}; + async function classifyPark( command: 'run' | 'resume', report: Partial, @@ -218,6 +223,17 @@ describe('the remedy on a worker park', () => { expect(result.report.parkedStep?.id).toBe('work'); }); + it('preserves a parked outcome when its runnable snapshot still says running', async () => { + const { client } = clientReturning([runningOnRunnableAgent]); + + const result = await classifyOutcome(client, 'run', parked, base, '/unused', {}); + + expect(result.exitCode).toBe(3); + expect(result.report.status).toBe('parked'); + expect(result.report.parkCause).toBe('worker_unavailable'); + expect(result.report.parkedStep?.id).toBe('work'); + }); + /// The re-admitted child: `run.start` on an existing admission key returns /// the run as it is -- `running`, its retried attempt leased to a worker. it('follows a run.start outcome that is already running on a retried attempt', async () => {