From ccabadadf00e3189695a5a41b0c341e6a874eda2 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Sun, 11 Oct 2026 21:39:13 -0700 Subject: [PATCH 1/2] fix(kernel,sdk): continue a CLI deadline in the dirty workspace An agent timeout used to settle the attempt, so the journal sealed and the edits died with the sandbox. A deadline now spends the transport budget: the worker journals the dirty end pin and the trajectory tail, and the next attempt inspects that tree even when the step declared reset. The default budget continues once. A second deadline seals. Pass transportRetries: 0 when the deadline itself is the result. --- docs/SURFACE.md | 20 +- kernel/DESIGN.md | 18 +- kernel/relayflowd-core/src/machine.rs | 18 +- .../src/machine/deadline_tests.rs | 240 ++++++++++++++++++ .../relayflowd-core/src/machine/recovery.rs | 17 ++ kernel/relayflowd-core/src/state.rs | 9 +- kernel/relayflowd-core/src/state/pins.rs | 25 +- packages/sdk/src/worker-input.ts | 16 +- packages/sdk/src/worker.ts | 23 +- .../tests/agent-deadline-inspect-live.test.ts | 107 ++++++++ packages/sdk/tests/agent-timeout-live.test.ts | 4 +- 11 files changed, 467 insertions(+), 30 deletions(-) create mode 100644 kernel/relayflowd-core/src/machine/deadline_tests.rs create mode 100644 packages/sdk/tests/agent-deadline-inspect-live.test.ts diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 1aed2ea5e..cb4f9770a 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -724,7 +724,11 @@ compiler/SDK validation limit; the kernel validates positive milliseconds fitting in `i64` and carries the declaration to the worker. ```ts -const repair = await f.agent('repair', { task: 'Repair failing checks.', timeout: '45m' }); +const repair = await f.agent('repair', { + task: 'Repair failing checks.', + timeout: '45m', + transportRetries: 0, // the deadline is the result; do not continue +}); if (repair.completionReason === 'timeout') { await f.run('git push'); // publish what the agent already committed } @@ -751,11 +755,15 @@ native artifact paths remain under `trajectory_tail.transcript.artifacts.paths` in the journal (a bounded list). Wrappers discard partial stdout on timeout and provide the deadline message; they have no transcript artifact list. A wrapper result envelope emitted before the deadline still supplies its usage. -No workspace reset occurs: committed work and uncommitted edits remain, and -the following step sees that potentially dirty tree even with -`recoveryMode: 'reset'`. Unlike crash/lease recovery in RFC Appendix A, timeout -settles the step with no successor attempt. Resume replays the recorded timeout -and does not execute that agent again. Incurred spend remains charged: +No workspace reset occurs on the settling attempt: committed work and +uncommitted edits remain, and the following step sees that potentially dirty +tree. The default transport budget is one. A deadline then journals `timeout` +with disposition `retry`, the dirty end pins, and the trajectory tail, and the +next attempt recovers as `inspect` even when `recoveryMode` is `reset`. The +continuation sees the edits and the journaled tail. A second deadline seals +the step. Set `transportRetries: 0` when the deadline itself is the result you +branch on: that attempt settles with no successor, and resume replays the +recorded timeout without executing the agent again. Incurred spend remains charged: reported usage is priced as usual, and usage never reported before the deadline is journaled as dollar-unmetered rather than as a measured $0. diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 6cf159274..609f7a47a 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -296,9 +296,16 @@ retry; `inspect` resumes inside the dirty workspace with the failed attempt's tail injected; `manual` parks as `wait.human` with `diff_ref`. Infrastructure retry is a separate, explicit budget (`retry.max_transport_retries`, default one so a single process loss remains resumable; set zero to disable). -Only `crashed` and `lease_expired` consume it. `worker_error`, timeout, budget, -cancellation, and an ordinary nonzero CLI exit are terminal regardless of the -budget; semantic verification retry remains bounded only by `max_iterations`. +`crashed`, `lease_expired`, and an agent CLI `timeout` consume it. A deadline +is not a judged result: it does not increment `semantic_executions`, and the +completion keeps the worker's end pins and trajectory tail. The next attempt +recovers as `inspect` even when the step declared `reset`, so the dirty pin +and the tail are what the continuation sees. The attempt-started record still +carries the declared mode. `manual` parks. A second deadline seals when the +transport budget is the default one. A deterministic command timeout, +`worker_error`, budget, cancellation, and an ordinary nonzero CLI exit stay +terminal regardless of the budget; semantic verification retry remains +bounded only by `max_iterations`. One kernel-side `worker_error` is not terminal: a dispatch the kernel refuses because the attached worker does not hold the attempt's journaled pins (Appendix A rule 2, `DispatchOutcome::PinMismatch`) ran nothing, so the step is @@ -308,7 +315,10 @@ re-elected against those same pins under `max_iterations` The kernel learns of a dead attempt two ways, and the recovery mode applies to both: the kernel notices an abandoned lease (`abandonment_actions`), or the worker reports its own loss through `step.complete` with `crashed` / -`lease_expired` (`completion_actions`). Under `manual` either path journals +`lease_expired` (`completion_actions`). A CLI deadline is a third report, +`step.complete` with `timeout`: unlike abandonment it keeps the end pins and +the tail, because the worker still holds the dirty tree. Under `manual` a +crash, an expired lease, or a CLI deadline journals `step.completed` with `disposition: park` and then the `wait.human` a human answers, each its own append. A process death between the two leaves the step folded to the placeholder wait id `park--` with nothing to diff --git a/kernel/relayflowd-core/src/machine.rs b/kernel/relayflowd-core/src/machine.rs index 36bf2f722..c21e64ccd 100644 --- a/kernel/relayflowd-core/src/machine.rs +++ b/kernel/relayflowd-core/src/machine.rs @@ -255,7 +255,9 @@ fn start_actions(state: &RunState, step: &StepSpec, attempt: u32, now_ms: i64) - surfaces, .. } => { - let retry_pins = match recovery_mode { + let effective_mode = + recovery::recovery_mode_for_attempt(*recovery_mode, runtime.last_completion_reason); + let retry_pins = match effective_mode { RecoveryMode::Inspect => runtime .last_end_pins .as_ref() @@ -267,9 +269,9 @@ fn start_actions(state: &RunState, step: &StepSpec, attempt: u32, now_ms: i64) - None => carried_pins(state.current_pins.as_ref(), surfaces), }; let recovery = RecoveryInstruction { - mode: *recovery_mode, + mode: effective_mode, restore_pins: (runtime.last_completion_reason.is_some() - && *recovery_mode == RecoveryMode::Reset) + && effective_mode == RecoveryMode::Reset) .then(|| pins.clone()), previous_completion_reason: runtime.last_completion_reason, trajectory_tail: runtime.trajectory_tail.clone(), @@ -445,10 +447,15 @@ fn settle_attempt( result.failure_reason, None | Some(CompletionReason::VerificationFailed) ); + // A CLI deadline stopped the process; it is not a result a gate judged. + // A deterministic command timeout stays that command's result and does + // not join this budget. + let agent_deadline = matches!(step.kind, StepKind::Agent { .. }) + && matches!(result.failure_reason, Some(CompletionReason::Timeout)); let transport_failure = matches!( result.failure_reason, Some(CompletionReason::Crashed | CompletionReason::LeaseExpired) - ); + ) || agent_deadline; // Appendix A rule 4: a dead attempt under `manual` parks as `needs_human` // with a diff, whichever way the kernel learned of the death. The // abandoned-lease path (`abandonment_actions`) always honoured that; a @@ -622,10 +629,13 @@ fn deterministic_ulid( } mod recovery; +pub(crate) use recovery::recovery_mode_for_attempt; pub use recovery::{abandonment_actions, recovery_actions, recovery_actions_filtered}; mod parallel; +#[cfg(test)] +mod deadline_tests; #[cfg(test)] mod parallel_tests; #[cfg(test)] diff --git a/kernel/relayflowd-core/src/machine/deadline_tests.rs b/kernel/relayflowd-core/src/machine/deadline_tests.rs new file mode 100644 index 000000000..7b7330cfe --- /dev/null +++ b/kernel/relayflowd-core/src/machine/deadline_tests.rs @@ -0,0 +1,240 @@ +//! An agent CLI deadline is a transport failure. It keeps the dirty pins and +//! the trajectory tail, and the next attempt inspects that workspace even +//! when the step declared `reset`. + +use serde_json::{Value, json}; + +use super::*; +use crate::{entry::AttemptStartedPayload, state::RunState}; + +fn deadline_spec(mode: &str) -> crate::RunSpec { + crate::RunSpec::parse(&json!({ + "steps": [{ + "id": "agent", + "type": "agent", + "instruction": "edit the workspace", + "recovery_mode": mode, + "max_iterations": 1, + "retry": { + "initial_backoff_ms": 0, + "max_backoff_ms": 0, + "multiplier": 1, + "jitter_percent": 0, + "max_transport_retries": 1 + }, + "surfaces": {"workspace": [{"surface": "repo"}]} + }] + })) + .unwrap() +} + +fn pins(revision_id: &str) -> Pins { + Pins { + workspace: vec![crate::WorkspacePin { + surface: "repo".to_owned(), + revision_id: revision_id.to_owned(), + }], + streams: vec![], + } +} + +fn timeout_result(end: &Pins) -> AttemptResult { + AttemptResult { + human_intervention: false, + output: Value::Null, + budget: Budget::default(), + completed_by: "worker".to_owned(), + end_pins: Some(end.clone()), + effects: vec![], + trajectory_tail: Some(json!({"artifacts": {"count": 1, "paths": ["kept.txt"]}})), + failure_reason: Some(CompletionReason::Timeout), + failure_detail: Some("CLI invocation timed out after 800ms.".to_owned()), + reported_cost: None, + } +} + +fn started(spec: &crate::RunSpec, revision: &str) -> JournalEntry { + let state = RunState::fold("run", spec.clone(), &[]).unwrap(); + let Action::Append(mut entry) = next_actions(&state, 10).remove(0) else { + panic!("agent start must be journaled") + }; + let mut payload: AttemptStartedPayload = serde_json::from_value(entry.payload.clone()).unwrap(); + payload.pins = pins(revision); + entry.payload = serde_json::to_value(payload).unwrap(); + entry +} + +fn appends(actions: Vec) -> Vec { + actions + .into_iter() + .filter_map(|action| match action { + Action::Append(entry) => Some(entry), + _ => None, + }) + .collect() +} + +#[test] +fn agent_deadline_retries_once_into_the_dirty_workspace() { + let spec = deadline_spec("reset"); + let clean = pins("rev-clean"); + let dirty = pins("rev-clean+deadbeef"); + let first = started(&spec, "rev-clean"); + let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&first)).unwrap(); + let declared: AttemptStartedPayload = serde_json::from_value(first.payload.clone()).unwrap(); + assert_eq!(declared.recovery_mode, Some(RecoveryMode::Reset)); + + let first_actions = completion_actions( + "run", + &spec.steps[0], + 1, + running.steps["agent"].semantic_executions, + Some(&clean), + timeout_result(&dirty), + 20, + ); + let Action::Append(completed) = &first_actions[0] else { + panic!("deadline must journal a completion") + }; + let payload: StepCompletedPayload = serde_json::from_value(completed.payload.clone()).unwrap(); + assert_eq!(payload.completion_reason, CompletionReason::Timeout); + assert_eq!(payload.disposition, Disposition::Retry); + assert_eq!(payload.end_pins.as_ref(), Some(&dirty)); + assert_eq!( + payload.trajectory_tail, + Some(json!({"artifacts": {"count": 1, "paths": ["kept.txt"]}})) + ); + + let mut entries = vec![first]; + entries.extend(appends(first_actions)); + let backoff = RunState::fold("run", spec.clone(), &entries).unwrap(); + assert_eq!(backoff.steps["agent"].semantic_executions, 0); + assert_eq!(backoff.steps["agent"].last_end_pins.as_ref(), Some(&dirty)); + let wake_at = payload + .next_attempt_at_ms + .expect("transport retry arms a timer"); + let Action::Append(woken) = next_actions(&backoff, wake_at).remove(0) else { + panic!("retry timer must fire") + }; + entries.push(woken); + let ready = RunState::fold("run", spec.clone(), &entries).unwrap(); + let retry = next_actions(&ready, wake_at); + let Action::Append(second_started) = &retry[0] else { + panic!("continuation must journal its start") + }; + let second_started: AttemptStartedPayload = + serde_json::from_value(second_started.payload.clone()).unwrap(); + let Action::Dispatch { + pins: dispatched, + recovery, + .. + } = &retry[1] + else { + panic!("continuation must dispatch") + }; + let recovery = recovery.as_ref().expect("agent dispatch carries recovery"); + assert_eq!(second_started.recovery_mode, Some(RecoveryMode::Reset)); + assert_eq!(recovery.mode, RecoveryMode::Inspect); + assert!(recovery.restore_pins.is_none()); + assert_eq!(dispatched, &dirty); + assert_eq!(second_started.pins, dirty); + assert_eq!( + recovery.previous_completion_reason, + Some(CompletionReason::Timeout) + ); + assert_eq!( + recovery.trajectory_tail, + Some(json!({"artifacts": {"count": 1, "paths": ["kept.txt"]}})) + ); + entries.extend(appends(retry)); + RunState::fold("run", spec.clone(), &entries) + .expect("the dirty start pin is the deadline's pin chain"); + + let second = completion_actions( + "run", + &spec.steps[0], + 2, + ready.steps["agent"].semantic_executions, + Some(&dirty), + timeout_result(&dirty), + 40, + ); + let Action::Append(sealed) = &second[0] else { + panic!("a second deadline must journal a completion") + }; + let sealed: StepCompletedPayload = serde_json::from_value(sealed.payload.clone()).unwrap(); + assert_eq!(sealed.completion_reason, CompletionReason::Timeout); + assert_eq!(sealed.disposition, Disposition::StepDone); + assert_eq!(sealed.next_attempt_at_ms, None); +} + +#[test] +fn agent_deadline_with_no_transport_budget_settles() { + let mut spec = deadline_spec("reset"); + spec.steps[0].retry.max_transport_retries = 0; + let actions = completion_actions( + "run", + &spec.steps[0], + 1, + 0, + None, + timeout_result(&pins("rev-clean+deadbeef")), + 10, + ); + let Action::Append(completed) = &actions[0] else { + panic!("a deadline with no transport budget must still journal") + }; + let payload: StepCompletedPayload = serde_json::from_value(completed.payload.clone()).unwrap(); + assert_eq!(payload.completion_reason, CompletionReason::Timeout); + assert_eq!(payload.disposition, Disposition::StepDone); +} + +#[test] +fn manual_agent_deadline_parks() { + let spec = deadline_spec("manual"); + let actions = completion_actions( + "run", + &spec.steps[0], + 1, + 0, + Some(&pins("rev-clean")), + timeout_result(&pins("rev-clean+deadbeef")), + 10, + ); + let Action::Append(completed) = &actions[0] else { + panic!("manual deadline must journal a completion") + }; + let payload: StepCompletedPayload = serde_json::from_value(completed.payload.clone()).unwrap(); + assert_eq!(payload.completion_reason, CompletionReason::Timeout); + assert_eq!(payload.disposition, Disposition::Park); +} + +#[test] +fn deterministic_command_timeout_stays_terminal() { + let spec = crate::RunSpec::parse(&json!({ + "steps": [{ + "id": "hello", + "type": "deterministic", + "command": "printf hello", + "max_iterations": 5, + "retry": { + "initial_backoff_ms": 0, + "max_backoff_ms": 0, + "multiplier": 1, + "jitter_percent": 0, + "max_transport_retries": 1 + } + }] + })) + .unwrap(); + let mut result = AttemptResult::successful(json!({"exit_code": -1}), "kernel"); + result.failure_reason = Some(CompletionReason::Timeout); + result.failure_detail = Some("command timed out".to_owned()); + let actions = completion_actions("run", &spec.steps[0], 1, 0, None, result, 10); + let Action::Append(completed) = &actions[0] else { + panic!("a command timeout must journal a completion") + }; + let payload: StepCompletedPayload = serde_json::from_value(completed.payload.clone()).unwrap(); + assert_eq!(payload.completion_reason, CompletionReason::Timeout); + assert_eq!(payload.disposition, Disposition::StepDone); +} diff --git a/kernel/relayflowd-core/src/machine/recovery.rs b/kernel/relayflowd-core/src/machine/recovery.rs index a4312cb23..cd21b5f88 100644 --- a/kernel/relayflowd-core/src/machine/recovery.rs +++ b/kernel/relayflowd-core/src/machine/recovery.rs @@ -14,6 +14,23 @@ use crate::{ state::{RunState, StepState, park_placeholder_wait_id}, }; +/// The mode the next attempt actually runs under. +/// +/// A CLI deadline journals `timeout` and keeps the dirty end pins. Declared +/// `reset` would restore the start pin and drop those edits, so the +/// continuation runs as `inspect`. The attempt-started record still carries +/// the declared mode. `manual` is unchanged: settlement parks before a new +/// attempt is started. +pub(crate) fn recovery_mode_for_attempt( + declared: RecoveryMode, + last_completion_reason: Option, +) -> RecoveryMode { + match (declared, last_completion_reason) { + (RecoveryMode::Reset, Some(CompletionReason::Timeout)) => RecoveryMode::Inspect, + (declared, _) => declared, + } +} + pub fn recovery_actions(state: &RunState, now_ms: i64) -> Vec { recovery_actions_filtered(state, now_ms, &|_, _| false) } diff --git a/kernel/relayflowd-core/src/state.rs b/kernel/relayflowd-core/src/state.rs index 64c010fa8..70e5c2e2c 100644 --- a/kernel/relayflowd-core/src/state.rs +++ b/kernel/relayflowd-core/src/state.rs @@ -328,12 +328,15 @@ impl RunState { step.last_completion_reason = Some(payload.completion_reason); step.trajectory_tail = payload.trajectory_tail.clone(); } + let agent_deadline = is_agent && payload.completion_reason == CompletionReason::Timeout; if !matches!( payload.completion_reason, CompletionReason::Crashed | CompletionReason::LeaseExpired - ) { - // The attempt ran to completion and produced a result the gate - // could judge; only these consume `max_iterations` allowance. + ) && !agent_deadline + { + // The attempt ran to a result a gate could judge. A crash, an + // expired lease, and an agent CLI deadline did not: those consume + // the transport budget, so `max_iterations: 1` can still continue. step.semantic_executions = step.semantic_executions.saturating_add(1); } step.state = match payload.disposition { diff --git a/kernel/relayflowd-core/src/state/pins.rs b/kernel/relayflowd-core/src/state/pins.rs index 9dd8a9d40..471a8c50e 100644 --- a/kernel/relayflowd-core/src/state/pins.rs +++ b/kernel/relayflowd-core/src/state/pins.rs @@ -36,15 +36,22 @@ impl RunState { // must carry that exact revision forward, surfaces the chain has never // produced are worker-sourced and unconstrained here. if let Some(previous) = runtime.last_completion_reason { - let expected = match spec.kind { - StepKind::Agent { - recovery_mode: RecoveryMode::Inspect, - .. - } => runtime - .last_end_pins - .as_ref() - .or(runtime.last_start_pins.as_ref()), - StepKind::Agent { .. } => runtime.last_start_pins.as_ref(), + // Same choice `start_actions` journals. A CLI deadline continues + // as inspect, so the retry's start pin is the dirty end pin even + // when the step declared reset. + let expected = match &spec.kind { + StepKind::Agent { recovery_mode, .. } => { + match crate::machine::recovery_mode_for_attempt(*recovery_mode, Some(previous)) + { + RecoveryMode::Inspect => runtime + .last_end_pins + .as_ref() + .or(runtime.last_start_pins.as_ref()), + RecoveryMode::Reset | RecoveryMode::Manual => { + runtime.last_start_pins.as_ref() + } + } + } _ => None, }; if let Some(expected) = expected diff --git a/packages/sdk/src/worker-input.ts b/packages/sdk/src/worker-input.ts index 6d33abd24..6cc68f5ff 100644 --- a/packages/sdk/src/worker-input.ts +++ b/packages/sdk/src/worker-input.ts @@ -4,12 +4,24 @@ import type { StepDispatchEvent } from './protocol.js'; export function workerInstruction(instruction: string, dispatch: StepDispatchEvent): string { const withMemory = dispatch.memory === undefined ? instruction : `${instruction}\n\nMemory context (journaled):\n${JSON.stringify(dispatch.memory.pack)}`; + const noted = `${withMemory}${recoveryNote(dispatch)}`; if (dispatch.input === undefined) { if (typeof dispatch.spec === 'object' && dispatch.spec !== null && 'input' in dispatch.spec) { throw new Error('Step declares input bindings but the kernel did not resolve them; upgrade relayflowd.'); } - return withMemory; + return noted; } - return `${withMemory}\n\ninput:\n${JSON.stringify(dispatch.input)}`; + return `${noted}\n\ninput:\n${JSON.stringify(dispatch.input)}`; +} + +/** Absent on the first attempt. Null and undefined are both "the kernel had nothing". */ +function recoveryNote(dispatch: StepDispatchEvent): string { + const reason = dispatch.recovery?.previous_completion_reason; + const tail = dispatch.recovery?.trajectory_tail; + if ((reason === undefined || reason === null) && (tail === undefined || tail === null)) return ''; + return `\n\nPrevious attempt (journaled):\n${JSON.stringify({ + completion_reason: reason ?? null, + trajectory_tail: tail ?? null, + })}`; } diff --git a/packages/sdk/src/worker.ts b/packages/sdk/src/worker.ts index 348317f62..97ff04381 100644 --- a/packages/sdk/src/worker.ts +++ b/packages/sdk/src/worker.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto'; import { communicationInstruction } from './communication/spec.js'; import { workerSpend } from './worker-spend.js'; import { reportedCost } from './reported-cost.js'; @@ -222,12 +223,32 @@ export class AgentWorker extends EventEmitter { ...(usage !== undefined ? { usage } : {}), ...(cost === undefined ? {} : { reported_cost: cost }), started_pins: dispatch.pins, - end_pins: dispatch.pins, + end_pins: completionReason === 'timeout' + ? deadlineEndPins(dispatch.pins, result.artifacts) + : dispatch.pins, }, ); } } +/** + * A deadline's end pin names the dirty tree the next attempt must inspect. + * The stamp is the artifact path list, not the file bytes: the tail is capped + * and the bytes stay on this worker's disk. An empty diff keeps the start pin. + */ +function deadlineEndPins(pins: Pins, artifacts: readonly string[] | undefined): Pins { + const workspace = pins.workspace ?? []; + if (artifacts === undefined || artifacts.length === 0 || workspace.length === 0) return pins; + if (workspace.every(pin => pin.revision_id === '')) return pins; + const stamp = createHash('sha256').update([...artifacts].sort().join('\n')).digest('hex').slice(0, 16); + return { + ...pins, + workspace: workspace.map(pin => pin.revision_id === '' + ? pin + : { ...pin, revision_id: `${pin.revision_id}+${stamp}` }), + }; +} + /** * The digest as journaled: the attempt and exit code (dropped from `output` on * the JSON path by design) and the artifact paths folded in, then bounded to diff --git a/packages/sdk/tests/agent-deadline-inspect-live.test.ts b/packages/sdk/tests/agent-deadline-inspect-live.test.ts new file mode 100644 index 000000000..c236fc837 --- /dev/null +++ b/packages/sdk/tests/agent-deadline-inspect-live.test.ts @@ -0,0 +1,107 @@ +import { createHash } from 'node:crypto'; +import { chmodSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { afterEach, expect, it } from 'vitest'; +import { AgentWorker } from '../src/worker.js'; +import { compileSpec, toKernelSpec } from '../src/compile.js'; +import type { KernelRunSpec } from '../src/spec.js'; +import { SPEC_SCHEMA_VERSION } from '../src/spec.js'; +import { chainFixture } from './flow-chain-fixture.js'; + +const cleanups: Array<() => Promise> = []; +afterEach(async () => { for (const close of cleanups.splice(0).reverse()) await close(); }); + +it('a CLI deadline journals the dirty pin and the next attempt inspects the file', async () => { + const f = chainFixture(); + cleanups.push(() => f.close()); + const workspace = join(f.root, 'workspace'); + const kept = join(workspace, 'kept.txt'); + const second = join(workspace, 'second.txt'); + const seen = join(f.root, 'instructions.log'); + mkdirSync(workspace); + const cli = join(f.root, 'claude'); + writeFileSync(cli, `#!/usr/bin/env node +import { appendFileSync, readFileSync, writeFileSync } from 'node:fs'; +const dash = process.argv.indexOf('--'); +const instruction = dash === -1 ? '' : process.argv[dash + 1] ?? ''; +appendFileSync(${JSON.stringify(seen)}, instruction + '\\n---\\n'); +if (instruction.includes('Previous attempt')) { + const body = readFileSync(${JSON.stringify(kept)}, 'utf8'); + writeFileSync(${JSON.stringify(second)}, body); + process.stdout.write(body); + process.exit(0); +} +writeFileSync(${JSON.stringify(kept)}, 'still-here\\n'); +setTimeout(() => {}, 30_000); +`); + chmodSync(cli, 0o755); + const spec = toKernelSpec(compileSpec({ + version: SPEC_SCHEMA_VERSION, + steps: [{ + id: 'agent', type: 'agent', instruction: 'edit the workspace', cli, + model: 'unpriced-test-model', recoveryMode: 'reset', maxIterations: 1, + timeoutMs: 800, surfaces: { workspace: [{ surface: 'repo' }] }, + }], + })) as KernelRunSpec; + const retry = spec.steps[0]?.retry; + if (retry === undefined) throw new Error('agent spec has no retry'); + retry.initial_backoff_ms = 0; + retry.max_backoff_ms = 0; + retry.multiplier = 1; + retry.jitter_percent = 0; + retry.max_transport_retries = 1; + + const client = await f.connect(); + const worker = new AgentWorker(client, { + workerId: 'deadline-inspect', + pins: { workspace: [{ surface: 'repo', revision_id: 'rev-clean' }], streams: [] }, + dataDir: f.data, + runRoot: workspace, + }); + const errors: unknown[] = []; + worker.on('error', error => errors.push(error)); + await worker.attach(); + const started = await client.runStart(spec); + + const deadline = Date.now() + 25_000; + let timedOut: { payload?: Record } | undefined; + while (timedOut === undefined) { + if (Date.now() > deadline) { + throw new Error(`deadline did not journal a retry: ${JSON.stringify(errors)} ${readFileSync(seen, 'utf8')}`); + } + const page = await client.journalRead(started.run_id, 1, 1000); + timedOut = (page.entries as Array<{ entry_type: string; payload?: Record }>) + .find(entry => entry.entry_type === 'step.completed' + && entry.payload?.['completionReason'] === 'timeout' + && entry.payload?.['disposition'] === 'retry'); + if (timedOut === undefined) await new Promise(resolve => setTimeout(resolve, 25)); + } + expect(errors, JSON.stringify(errors)).toEqual([]); + const tail = JSON.stringify(timedOut.payload?.['trajectory_tail']); + expect(tail).toContain('kept.txt'); + const endPins = timedOut.payload?.['end_pins'] as { workspace?: Array<{ revision_id?: string }> }; + const revision = endPins.workspace?.[0]?.revision_id ?? ''; + const stamp = createHash('sha256').update('kept.txt').digest('hex').slice(0, 16); + expect(revision).toBe(`rev-clean+${stamp}`); + const completions = (await client.journalRead(started.run_id, 1, 1000)).entries as Array<{ + entry_type: string; attempt?: number; payload?: { disposition?: string }; + }>; + expect(completions.filter(entry => entry.entry_type === 'step.completed' && entry.attempt === 1) + .map(entry => entry.payload?.disposition)).toEqual(['retry']); + + await client.runResume(started.run_id); + const until = Date.now() + 20_000; + while (true) { + try { + if (readFileSync(second, 'utf8') === 'still-here\n') break; + } catch { /* not written yet */ } + if (Date.now() > until) { + throw new Error(`continuation did not write second.txt: ${readFileSync(seen, 'utf8')}`); + } + await new Promise(resolve => setTimeout(resolve, 25)); + } + expect(readFileSync(second, 'utf8')).toBe('still-here\n'); + expect(readFileSync(seen, 'utf8')).toContain('Previous attempt (journaled):'); + expect(readFileSync(seen, 'utf8')).toContain('kept.txt'); + await worker.close(); +}, 45_000); diff --git a/packages/sdk/tests/agent-timeout-live.test.ts b/packages/sdk/tests/agent-timeout-live.test.ts index 83045a40f..7dacdc363 100644 --- a/packages/sdk/tests/agent-timeout-live.test.ts +++ b/packages/sdk/tests/agent-timeout-live.test.ts @@ -30,7 +30,9 @@ if (request?.instruction.includes('TIMEOUT_TEST')) { writeFileSync(join(f.root, 'flows.json'), JSON.stringify({ cli: f.wrapper, models: ['claude-opus-5'] })); writeFileSync(f.flowPath, `import { flow } from '@relayflows/surface'; export default flow('timeout-test', { budget: { wallclock: '2m' } }, async f => { - const result = await f.agent('repair', { task: 'TIMEOUT_TEST', timeout: '300ms', model: 'claude-opus-5' }) + // transportRetries 0: this deadline is the result the gate judges. The default + // budget would continue once in the dirty workspace before the gate runs. + const result = await f.agent('repair', { task: 'TIMEOUT_TEST', timeout: '300ms', model: 'claude-opus-5', transportRetries: 0 }) .gate(r => r.completionReason === 'timeout' && ${gate}); if (result.completionReason === 'timeout') { await f.run(${JSON.stringify(pause ? 'test -f paused || { touch paused; sleep 120; }; cat work.txt > published' : 'cat work.txt > published')}); From 07bed71dae2cf79bbde4944db5b9fb539c5f844f Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Sun, 11 Oct 2026 23:01:50 -0700 Subject: [PATCH 2/2] fix(kernel,sdk): keep a deadline's dirty tree distinct and dispatchable A path-only stamp treated a deletion as a clean tree and gave two different edits of the same paths one revision. The stamp is now the content signatures plus the deleted paths. A manual deadline still parks. After the human answers, the attempt inspects the dirty pins: the worker was already advanced to them, and asking for the clean start pin would refuse the dispatch. A crash under manual stays manual. A local agent with a deadline declares local-workspace, which the local worker holds at revision local until the deadline stamps it. Selection prefers a worker already standing at the attempt's revisions, so a fresh worker at local cannot take the continuation. A settled timeout still ends the run. It does not publish the pin chain to a later step. --- docs/SURFACE.md | 5 +- kernel/DESIGN.md | 19 +++-- .../src/machine/deadline_tests.rs | 69 +++++++++++++++++++ .../relayflowd-core/src/machine/recovery.rs | 14 ++-- .../relayflowd/src/server/session/matching.rs | 26 ++++++- packages/sdk/src/agent-artifacts.ts | 35 +++++++++- packages/sdk/src/agent-transcript.ts | 25 +++++-- packages/sdk/src/authored-worker-step.ts | 11 ++- packages/sdk/src/local-agent-pins.ts | 7 ++ packages/sdk/src/local-agent.ts | 12 ++-- packages/sdk/src/worker-cli.ts | 38 ++++++++-- packages/sdk/src/worker.ts | 25 ++++--- packages/sdk/tests/agent-artifacts.test.ts | 41 ++++++++++- .../tests/agent-deadline-inspect-live.test.ts | 7 +- 14 files changed, 295 insertions(+), 39 deletions(-) create mode 100644 packages/sdk/src/local-agent-pins.ts diff --git a/docs/SURFACE.md b/docs/SURFACE.md index cb4f9770a..c25d0451e 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -760,7 +760,10 @@ uncommitted edits remain, and the following step sees that potentially dirty tree. The default transport budget is one. A deadline then journals `timeout` with disposition `retry`, the dirty end pins, and the trajectory tail, and the next attempt recovers as `inspect` even when `recoveryMode` is `reset`. The -continuation sees the edits and the journaled tail. A second deadline seals +continuation sees the edits and the journaled tail. The dirty revision's +stamp includes file contents and deletions, so a deletion-only edit and two +different edits of the same paths are distinct pins. After a human answers a +`manual` park, that attempt inspects the dirty pins. A second deadline seals the step. Set `transportRetries: 0` when the deadline itself is the result you branch on: that attempt settles with no successor, and resume replays the recorded timeout without executing the agent again. Incurred spend remains charged: diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 609f7a47a..35868acc0 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -298,11 +298,20 @@ is a separate, explicit budget (`retry.max_transport_retries`, default one so a single process loss remains resumable; set zero to disable). `crashed`, `lease_expired`, and an agent CLI `timeout` consume it. A deadline is not a judged result: it does not increment `semantic_executions`, and the -completion keeps the worker's end pins and trajectory tail. The next attempt -recovers as `inspect` even when the step declared `reset`, so the dirty pin -and the tail are what the continuation sees. The attempt-started record still -carries the declared mode. `manual` parks. A second deadline seals when the -transport budget is the default one. A deterministic command timeout, +completion keeps the worker's end pins and trajectory tail. The end revision +is the start revision plus a stamp of the workspace delta: each added or +changed file by its content signature, each deletion by its path. An empty +diff keeps the start pin. The next attempt recovers as `inspect` even when +the step declared `reset`, so the dirty pin and the tail are what the +continuation sees. The attempt-started record still carries the declared +mode. `manual` still parks for a human. After that answer the attempt +inspects the dirty pins: the attached worker was advanced to them, and a +retry that asked for the clean start pin would be refused. A second deadline +seals when the transport budget is the default one. A local agent with a +deadline declares the surface `local-workspace`, which that worker holds at +revision `local` until a deadline stamps it. Selection prefers a worker +already standing at the attempt's revisions, so a fresh worker still at +`local` does not take the continuation. A deterministic command timeout, `worker_error`, budget, cancellation, and an ordinary nonzero CLI exit stay terminal regardless of the budget; semantic verification retry remains bounded only by `max_iterations`. diff --git a/kernel/relayflowd-core/src/machine/deadline_tests.rs b/kernel/relayflowd-core/src/machine/deadline_tests.rs index 7b7330cfe..2cf61ce8a 100644 --- a/kernel/relayflowd-core/src/machine/deadline_tests.rs +++ b/kernel/relayflowd-core/src/machine/deadline_tests.rs @@ -209,6 +209,75 @@ fn manual_agent_deadline_parks() { assert_eq!(payload.disposition, Disposition::Park); } +#[test] +fn manual_deadline_answer_inspects_the_dirty_pins() { + let spec = deadline_spec("manual"); + let clean = pins("rev-clean"); + let dirty = pins("rev-clean+deadbeef"); + let first = started(&spec, "rev-clean"); + let running = RunState::fold("run", spec.clone(), std::slice::from_ref(&first)).unwrap(); + let parked = completion_actions( + "run", + &spec.steps[0], + 1, + running.steps["agent"].semantic_executions, + Some(&clean), + timeout_result(&dirty), + 20, + ); + let Action::Append(completed) = &parked[0] else { + panic!("manual deadline must journal a completion") + }; + let payload: StepCompletedPayload = serde_json::from_value(completed.payload.clone()).unwrap(); + assert_eq!(payload.disposition, Disposition::Park); + assert_eq!(payload.end_pins.as_ref(), Some(&dirty)); + + let mut entries = vec![first]; + entries.extend(appends(parked)); + let waiting = RunState::fold("run", spec.clone(), &entries).unwrap(); + assert!(matches!( + waiting.steps["agent"].state, + StepState::NeedsHuman { .. } + )); + + entries.push(JournalEntry::new( + EntryType::WaitCompleted, + "run", + Some("agent".to_owned()), + Some(1), + 30, + WaitCompletedPayload { + wait_id: "answered".to_owned(), + completion_reason: WaitCompletionReason::HumanResponded, + result: json!("retry"), + }, + )); + let ready = RunState::fold("run", spec.clone(), &entries).unwrap(); + let retry = next_actions(&ready, 30); + let Action::Append(second_started) = &retry[0] else { + panic!("the answered park must journal a start") + }; + let second_started: AttemptStartedPayload = + serde_json::from_value(second_started.payload.clone()).unwrap(); + let Action::Dispatch { + pins: dispatched, + recovery, + .. + } = &retry[1] + else { + panic!("the answered park must dispatch") + }; + let recovery = recovery.as_ref().expect("agent dispatch carries recovery"); + assert_eq!(second_started.recovery_mode, Some(RecoveryMode::Manual)); + assert_eq!(recovery.mode, RecoveryMode::Inspect); + assert!(recovery.restore_pins.is_none()); + assert_eq!(dispatched, &dirty); + assert_eq!(second_started.pins, dirty); + entries.extend(appends(retry)); + RunState::fold("run", spec.clone(), &entries) + .expect("the dirty start pin is the pin chain after a manual answer"); +} + #[test] fn deterministic_command_timeout_stays_terminal() { let spec = crate::RunSpec::parse(&json!({ diff --git a/kernel/relayflowd-core/src/machine/recovery.rs b/kernel/relayflowd-core/src/machine/recovery.rs index cd21b5f88..2290c6cd0 100644 --- a/kernel/relayflowd-core/src/machine/recovery.rs +++ b/kernel/relayflowd-core/src/machine/recovery.rs @@ -17,16 +17,20 @@ use crate::{ /// The mode the next attempt actually runs under. /// /// A CLI deadline journals `timeout` and keeps the dirty end pins. Declared -/// `reset` would restore the start pin and drop those edits, so the -/// continuation runs as `inspect`. The attempt-started record still carries -/// the declared mode. `manual` is unchanged: settlement parks before a new -/// attempt is started. +/// `reset` would restore the start pin and drop those edits. Declared +/// `manual` still parks first; after the human answers, asking for the clean +/// start pin would refuse the worker, which the completion already advanced +/// to the dirty end pins. Both continuations run as `inspect`. The +/// attempt-started record still carries the declared mode. A crash or an +/// expired lease under `manual` stays `manual`. pub(crate) fn recovery_mode_for_attempt( declared: RecoveryMode, last_completion_reason: Option, ) -> RecoveryMode { match (declared, last_completion_reason) { - (RecoveryMode::Reset, Some(CompletionReason::Timeout)) => RecoveryMode::Inspect, + (RecoveryMode::Reset | RecoveryMode::Manual, Some(CompletionReason::Timeout)) => { + RecoveryMode::Inspect + } (declared, _) => declared, } } diff --git a/kernel/relayflowd/src/server/session/matching.rs b/kernel/relayflowd/src/server/session/matching.rs index c5cc632db..8289b1979 100644 --- a/kernel/relayflowd/src/server/session/matching.rs +++ b/kernel/relayflowd/src/server/session/matching.rs @@ -42,11 +42,35 @@ pub(super) fn select_worker_for_step<'a>( (load < worker.capacity).then_some((worker, load)) }) .min_by(|(left, left_load), (right, right_load)| { - normalized_load_order(*left_load, left.capacity, *right_load, right.capacity) + let left_match = revisions_match(left, required_pins); + let right_match = revisions_match(right, required_pins); + // A worker already at the attempt's revisions can take an inspect + // continuation. A restoring reset still needs a worker that only + // holds the surfaces, so a miss stays eligible and loses the tie. + right_match.cmp(&left_match).then_with(|| { + normalized_load_order(*left_load, left.capacity, *right_load, right.capacity) + }) }) .map(|(worker, _)| worker) } +/// Does this worker already stand at every revision and stream offset the +/// attempt requires? Surface presence is `worker_can_pin`; this is the value. +fn revisions_match(worker: &Worker, required: &Pins) -> bool { + required.workspace.iter().all(|pin| { + worker.pins.workspace.iter().any(|held| { + workspace_surfaces_equal(&held.surface, &pin.surface) + && held.revision_id == pin.revision_id + }) + }) && required.streams.iter().all(|pin| { + worker + .pins + .streams + .iter() + .any(|held| held.stream == pin.stream && held.read_offset == pin.read_offset) + }) +} + fn worker_load(sessions: &Sessions, connection_id: u64) -> usize { sessions .assignments diff --git a/packages/sdk/src/agent-artifacts.ts b/packages/sdk/src/agent-artifacts.ts index ba98f9156..9fc83297d 100644 --- a/packages/sdk/src/agent-artifacts.ts +++ b/packages/sdk/src/agent-artifacts.ts @@ -70,10 +70,43 @@ async function walk(root: string, current: string, out: Map): Pr if (isEnoent(error)) continue; throw error; } - out.set(relative(root, path).split(sep).join('/'), `${info.size}:${createHash('sha256').update(bytes).digest('hex')}`); + out.set(relative(root, path).split(sep).join('/'), workspaceFileSignature(bytes)); } } +/** Byte length and sha256. The deadline pin uses this, not the path alone. */ +export function workspaceFileSignature(bytes: Uint8Array): string { + return `${bytes.length}:${createHash('sha256').update(bytes).digest('hex')}`; +} + +/** + * Identity of a dirty tree. Each added or changed file contributes its content + * signature, and each deletion contributes its path. Undefined when nothing + * changed. Two trees that edit the same paths differently, or that only delete + * a file, receive different stamps. + */ +export function deadlineWorkspaceStamp( + before: ReadonlyMap, + after: ReadonlyMap, +): string | undefined { + const lines: string[] = []; + for (const path of [...new Set([...before.keys(), ...after.keys()])].sort()) { + const prior = before.get(path); + const next = after.get(path); + if (prior === next) continue; + lines.push(next === undefined ? `D ${path}` : `C ${next} ${path}`); + } + if (lines.length === 0) return undefined; + return createHash('sha256').update(lines.join('\n')).digest('hex').slice(0, 16); +} + +/** Paths present in `before` and absent in `after`, sorted. */ +export function deletedWorkspaceFiles(before: ReadonlyMap, after: ReadonlyMap): string[] { + const deleted: string[] = []; + for (const path of before.keys()) if (!after.has(path)) deleted.push(path); + return deleted.sort(); +} + /** `dir`-relative paths present in `after` that are new or changed since `before`, sorted. */ export function diffWorkspaceFiles(before: Map, after: Map): string[] { const changed: string[] = []; diff --git a/packages/sdk/src/agent-transcript.ts b/packages/sdk/src/agent-transcript.ts index 89c6d5f7c..6c9365d54 100644 --- a/packages/sdk/src/agent-transcript.ts +++ b/packages/sdk/src/agent-transcript.ts @@ -102,7 +102,7 @@ export interface TranscriptDigest { final_text?: string; final_text_truncated?: boolean; failure?: { kind: 'result' | 'tool_result' | 'stderr'; excerpt: string; truncated?: boolean }; - artifacts?: { count: number; paths: string[] }; + artifacts?: { count: number; paths: string[]; deleted?: string[] }; /** * Set when even the ordered reductions left the digest over its cap and it * was cut back to fixed-size fields — every provider-supplied string but a @@ -636,7 +636,13 @@ function coreDigest(d: TranscriptDigest): TranscriptDigest { counts: [], last_calls: [], total_calls: d.tools.total_calls, shown_calls: 0, complete: d.tools.complete, }; } - if (d.artifacts !== undefined) core.artifacts = { count: d.artifacts.count, paths: [] }; + if (d.artifacts !== undefined) { + core.artifacts = { + count: d.artifacts.count, + paths: [], + ...(d.artifacts.deleted === undefined ? {} : { deleted: [] }), + }; + } if (d.final_text !== undefined || d.final_text_truncated === true) core.final_text_truncated = true; if (d.failure !== undefined) core.failure = { kind: d.failure.kind, excerpt: '', truncated: true }; return core; @@ -649,8 +655,19 @@ function keepLastCalls(d: TranscriptDigest, n: number): TranscriptDigest { } function keepArtifactPaths(d: TranscriptDigest, n: number): TranscriptDigest { - if (d.artifacts === undefined || d.artifacts.paths.length <= n) return d; - return { ...d, artifacts: { count: d.artifacts.count, paths: d.artifacts.paths.slice(0, n) } }; + const artifacts = d.artifacts; + if (artifacts === undefined) return d; + const deleted = artifacts.deleted; + if (artifacts.paths.length <= n && (deleted === undefined || deleted.length <= n)) return d; + const kept = deleted === undefined ? undefined : deleted.slice(0, n); + return { + ...d, + artifacts: { + count: artifacts.count, + paths: artifacts.paths.slice(0, n), + ...(kept === undefined ? {} : { deleted: kept }), + }, + }; } function cutFinalText(d: TranscriptDigest, maxBytes: number): TranscriptDigest { diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts index 9b44f7eb4..878281292 100644 --- a/packages/sdk/src/authored-worker-step.ts +++ b/packages/sdk/src/authored-worker-step.ts @@ -18,6 +18,7 @@ import { snapshotJsonValue } from './json-value.js'; import { authoredChildAdmissionKey } from './authored-admission.js'; import { alsoRecord, recordAuthoredChild } from './authored-step-index.js'; import type { StepFailedDetails } from './failure-kinds.js'; +import { LOCAL_AGENT_WORKSPACE } from './local-agent-pins.js'; import { authoredWorkerSlots, type AuthoredWorkerSlots } from './worker-slots.js'; const AGENT_TIMEOUT = Symbol('declared agent timeout'); @@ -278,7 +279,15 @@ export function authoredWorkerRunner( id, type: 'agent', instruction: options.task, ...(timeoutMs === undefined ? {} : { timeoutMs }), ...(permissionsSnapshot === undefined ? {} : { permissions: permissionsSnapshot }), - ...(localAgentStream === undefined ? {} : { surfaces: { streams: [{ stream: localAgentStream }] } }), + ...(localAgentStream === undefined ? {} : { surfaces: { + streams: [{ stream: localAgentStream }], + // A deadline continues in this process's tree. Declare the surface + // the local worker holds so the dirty stamp is a pin the kernel can + // match, and a fresh worker at the clean revision cannot take it. + ...(timeoutMs !== undefined && options.workspace === undefined + ? { workspace: [{ surface: LOCAL_AGENT_WORKSPACE }] } + : {}), + } }), ...(options.workspace === undefined ? {} : { surfaces: { workspace: [{ surface: options.workspace }] } }), ...(options.cli === undefined ? {} : { cli: options.cli }), ...(options.model === undefined ? {} : { model: options.model }), diff --git a/packages/sdk/src/local-agent-pins.ts b/packages/sdk/src/local-agent-pins.ts new file mode 100644 index 000000000..70a2d369b --- /dev/null +++ b/packages/sdk/src/local-agent-pins.ts @@ -0,0 +1,7 @@ +/** Surface a timed-out local agent pins so a fresh worker at the clean tree + * cannot take the continuation. The local worker always holds it. A step + * declares it only when a deadline can continue. */ +export const LOCAL_AGENT_WORKSPACE = 'local-workspace'; + +/** Revision the local worker reports before a deadline stamps the dirty tree. */ +export const LOCAL_AGENT_REVISION = 'local'; diff --git a/packages/sdk/src/local-agent.ts b/packages/sdk/src/local-agent.ts index 9007ab930..97378c693 100644 --- a/packages/sdk/src/local-agent.ts +++ b/packages/sdk/src/local-agent.ts @@ -4,9 +4,10 @@ import type { JournalClient } from './journal-client.js'; import { AgentWorker } from './worker.js'; import type { KernelRunSpec } from './spec.js'; import { communicationInstruction } from './communication/spec.js'; +import { LOCAL_AGENT_REVISION, LOCAL_AGENT_WORKSPACE } from './local-agent-pins.js'; import { DEFAULT_LOCAL_AGENT_CAPACITY } from './worker-slots.js'; -/** A local worker for stream-only steps; no workspace recovery is claimed. */ +/** A local worker. It holds `local-workspace` at `local` so a deadline can stamp the dirty tree. */ export async function attachLocalAgent( client: JournalClient, dataDir?: string, @@ -24,7 +25,8 @@ export async function attachLocalAgent( // This worker has consumed no messages: its read offsets are zero, including // named declarative streams. The kernel retains ownership of dispatch pins // and rejects a resume whose required offsets this worker does not hold. - // No worktree revision is invented. + // `local-workspace` starts at `local`. A deadline stamps that revision from + // the dirty tree, so a replacement worker still at `local` cannot take it. const stream = requestedStream ?? `local-agent-${randomUUID()}`; const worker = new AgentWorker(client, { workerId: stream, @@ -33,8 +35,10 @@ export async function attachLocalAgent( // this same number (worker-slots.ts), so they never ask for more. capacity, dataDir, onPtyReady, runRoot, environment, - pins: { workspace: [], streams: [...new Set([stream, ...declaredStreams])] - .map(stream => ({ stream, read_offset: 0 })) }, + pins: { + workspace: [{ surface: LOCAL_AGENT_WORKSPACE, revision_id: LOCAL_AGENT_REVISION }], + streams: [...new Set([stream, ...declaredStreams])].map(stream => ({ stream, read_offset: 0 })), + }, }); let failure: unknown; // The cause travels with the close, so the flow's next request names it. diff --git a/packages/sdk/src/worker-cli.ts b/packages/sdk/src/worker-cli.ts index d3e811063..edc4bc23e 100644 --- a/packages/sdk/src/worker-cli.ts +++ b/packages/sdk/src/worker-cli.ts @@ -1,6 +1,6 @@ import { realpathSync } from 'node:fs'; import { basename, dirname, join, resolve, sep } from 'node:path'; -import { diffWorkspaceFiles, snapshotWorkspaceFiles } from './agent-artifacts.js'; +import { deadlineWorkspaceStamp, deletedWorkspaceFiles, diffWorkspaceFiles, snapshotWorkspaceFiles } from './agent-artifacts.js'; import { claudeResultOutcome, decodeProviderResult, decodeWrapperResult, requirePricedUsage } from './worker-usage.js'; import { openSidechannel, type SidechannelContext } from './pty-sidechannel.js'; import { @@ -87,6 +87,18 @@ export interface WorkerCliResult { * agent runs on another host. */ artifacts?: string[]; + /** + * Content-and-deletion identity of the dirty tree, 16 hex characters. + * Present when the workspace changed. A deadline appends it to the end-pin + * revision so two different edits of the same paths, and a deletion with no + * new file, do not look like the clean start. + */ + workspace_stamp?: string; + /** + * Paths removed during the attempt, cwd-relative and sorted. Journaled + * beside `artifacts`. Absent when nothing was deleted. + */ + deleted?: string[]; /** * The attempt's transcript digest (`agent-transcript.ts`): provider result * metadata, tool calls summarized, final text, failure excerpt, and — when @@ -167,9 +179,27 @@ export async function runAgentCli( const before = await snapshotWorkspaceFiles(artifactRoot); const result = await execute(); const kernelData = sidechannel === undefined ? undefined : realPath(resolve(sidechannel.dataDir)); - const artifacts = diffWorkspaceFiles(before, await snapshotWorkspaceFiles(artifactRoot)) - .filter(path => !under(realPath(resolve(artifactRoot, path)), kernelData)); - return { ...result, artifacts }; + const after = await snapshotWorkspaceFiles(artifactRoot); + const visible = (files: Map) => { + const out = new Map(); + for (const [path, signature] of files) { + // `canonicalTree` keeps a deleted path under the data dir excluded: + // the file is gone, so a plain realpath falls back to the spelling + // and can miss a symlinked data dir. + if (!under(canonicalTree(resolve(artifactRoot, path)), kernelData)) out.set(path, signature); + } + return out; + }; + const stampBefore = visible(before); + const stampAfter = visible(after); + const workspace_stamp = deadlineWorkspaceStamp(stampBefore, stampAfter); + const deleted = deletedWorkspaceFiles(stampBefore, stampAfter); + return { + ...result, + artifacts: diffWorkspaceFiles(stampBefore, stampAfter), + ...(workspace_stamp === undefined ? {} : { workspace_stamp }), + ...(deleted.length === 0 ? {} : { deleted }), + }; }); async function execute(): Promise { diff --git a/packages/sdk/src/worker.ts b/packages/sdk/src/worker.ts index 97ff04381..a06cf34f4 100644 --- a/packages/sdk/src/worker.ts +++ b/packages/sdk/src/worker.ts @@ -1,4 +1,3 @@ -import { createHash } from 'node:crypto'; import { communicationInstruction } from './communication/spec.js'; import { workerSpend } from './worker-spend.js'; import { reportedCost } from './reported-cost.js'; @@ -224,7 +223,7 @@ export class AgentWorker extends EventEmitter { ...(cost === undefined ? {} : { reported_cost: cost }), started_pins: dispatch.pins, end_pins: completionReason === 'timeout' - ? deadlineEndPins(dispatch.pins, result.artifacts) + ? deadlineEndPins(dispatch.pins, result.workspace_stamp) : dispatch.pins, }, ); @@ -233,14 +232,14 @@ export class AgentWorker extends EventEmitter { /** * A deadline's end pin names the dirty tree the next attempt must inspect. - * The stamp is the artifact path list, not the file bytes: the tail is capped - * and the bytes stay on this worker's disk. An empty diff keeps the start pin. + * The stamp covers file contents and deletions, not the file bytes: the tail + * is capped and the bytes stay on this worker's disk. An empty diff keeps + * the start pin. */ -function deadlineEndPins(pins: Pins, artifacts: readonly string[] | undefined): Pins { +function deadlineEndPins(pins: Pins, stamp: string | undefined): Pins { const workspace = pins.workspace ?? []; - if (artifacts === undefined || artifacts.length === 0 || workspace.length === 0) return pins; + if (stamp === undefined || workspace.length === 0) return pins; if (workspace.every(pin => pin.revision_id === '')) return pins; - const stamp = createHash('sha256').update([...artifacts].sort().join('\n')).digest('hex').slice(0, 16); return { ...pins, workspace: workspace.map(pin => pin.revision_id === '' @@ -259,9 +258,15 @@ function transcriptDigest(transcript: TranscriptDigest, attempt: number, result: attempt, exit_code: result.exit_code, ...transcript, - ...(result.artifacts === undefined ? {} : { artifacts: { - count: result.artifacts.length, paths: result.artifacts.slice(0, ARTIFACT_PATHS_MAX), - } }), + ...(result.artifacts === undefined && (result.deleted === undefined || result.deleted.length === 0) ? {} : { + artifacts: { + count: result.artifacts?.length ?? 0, + paths: (result.artifacts ?? []).slice(0, ARTIFACT_PATHS_MAX), + ...(result.deleted === undefined || result.deleted.length === 0 ? {} : { + deleted: result.deleted.slice(0, ARTIFACT_PATHS_MAX), + }), + }, + }), }); } diff --git a/packages/sdk/tests/agent-artifacts.test.ts b/packages/sdk/tests/agent-artifacts.test.ts index 432a7cccd..8123c7f01 100644 --- a/packages/sdk/tests/agent-artifacts.test.ts +++ b/packages/sdk/tests/agent-artifacts.test.ts @@ -1,9 +1,16 @@ +import { createHash } from 'node:crypto'; import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs'; import { chmod, open } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; -import { diffWorkspaceFiles, snapshotWorkspaceFiles } from '../src/agent-artifacts.js'; +import { + deadlineWorkspaceStamp, + deletedWorkspaceFiles, + diffWorkspaceFiles, + snapshotWorkspaceFiles, + workspaceFileSignature, +} from '../src/agent-artifacts.js'; const dirs: string[] = []; afterEach(() => { for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); }); @@ -14,6 +21,38 @@ function tempDir(): string { return dir; } +describe('deadlineWorkspaceStamp', () => { + it('is undefined when nothing changed', () => { + const files = new Map([['a.txt', '1:abc']]); + expect(deadlineWorkspaceStamp(files, new Map(files))).toBeUndefined(); + }); + + it('stamps a new file from its content signature', () => { + const signature = workspaceFileSignature(Buffer.from('still-here\n')); + const stamp = deadlineWorkspaceStamp(new Map(), new Map([['kept.txt', signature]])); + const line = `C ${signature} kept.txt`; + expect(stamp).toBe(createHash('sha256').update(line).digest('hex').slice(0, 16)); + }); + + it('differs when the same path has different contents', () => { + const one = deadlineWorkspaceStamp(new Map(), new Map([['fix.ts', '4:aaa']])); + const two = deadlineWorkspaceStamp(new Map(), new Map([['fix.ts', '4:bbb']])); + expect(one).toBeDefined(); + expect(two).toBeDefined(); + expect(one).not.toBe(two); + }); + + it('stamps a deletion that the path diff does not list', () => { + const before = new Map([['gone.txt', '1:abc']]); + const after = new Map(); + expect(deadlineWorkspaceStamp(before, after)).toBe( + createHash('sha256').update('D gone.txt').digest('hex').slice(0, 16), + ); + expect(deletedWorkspaceFiles(before, after)).toEqual(['gone.txt']); + expect(diffWorkspaceFiles(before, after)).toEqual([]); + }); +}); + describe('snapshotWorkspaceFiles / diffWorkspaceFiles', () => { it('reports a newly written nested file as an artifact', async () => { const dir = tempDir(); diff --git a/packages/sdk/tests/agent-deadline-inspect-live.test.ts b/packages/sdk/tests/agent-deadline-inspect-live.test.ts index c236fc837..75fa291bc 100644 --- a/packages/sdk/tests/agent-deadline-inspect-live.test.ts +++ b/packages/sdk/tests/agent-deadline-inspect-live.test.ts @@ -1,7 +1,7 @@ -import { createHash } from 'node:crypto'; import { chmodSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { afterEach, expect, it } from 'vitest'; +import { deadlineWorkspaceStamp, workspaceFileSignature } from '../src/agent-artifacts.js'; import { AgentWorker } from '../src/worker.js'; import { compileSpec, toKernelSpec } from '../src/compile.js'; import type { KernelRunSpec } from '../src/spec.js'; @@ -81,7 +81,10 @@ setTimeout(() => {}, 30_000); expect(tail).toContain('kept.txt'); const endPins = timedOut.payload?.['end_pins'] as { workspace?: Array<{ revision_id?: string }> }; const revision = endPins.workspace?.[0]?.revision_id ?? ''; - const stamp = createHash('sha256').update('kept.txt').digest('hex').slice(0, 16); + const stamp = deadlineWorkspaceStamp( + new Map(), + new Map([['kept.txt', workspaceFileSignature(Buffer.from('still-here\n'))]]), + ); expect(revision).toBe(`rev-clean+${stamp}`); const completions = (await client.journalRead(started.run_id, 1, 1000)).entries as Array<{ entry_type: string; attempt?: number; payload?: { disposition?: string };