diff --git a/docs/SURFACE.md b/docs/SURFACE.md index be2c74790..68aa8539e 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -426,6 +426,20 @@ example selects `stdout_tail`. JSON-emitting agent/LLM outputs use their declare value shape directly. Use matching SDK and kernel builds for input bindings; older kernels refuse the new field. +### Command timeouts + +`f.run(command, { timeout?: string | number })` gives each command its own +lease. The default is **30 seconds**. Use `await f.run(command, { timeout: '5m' })` +or `{ timeout: 300000 }` for longer work. Strings accept `ms`, `s`, and `m`; +the resolved value must be a positive whole number of milliseconds. + +The hard ceiling is **15 minutes** (900000 ms), inclusive. A larger timeout +is refused during step compilation, before dispatch, with `lease_exceeded`; +malformed durations are refused with `timeout_invalid`. On reaching its timeout, +the kernel kills the command's process group and journals `completionReason: timeout`; +`f.run` refuses with code `lease_exceeded`. The override applies only to that +invocation; calls without options retain the default. + ### The authored operation lifecycle An authored TypeScript body reaches `done()` only if every step it created was diff --git a/kernel/relayflowd-core/src/machine.rs b/kernel/relayflowd-core/src/machine.rs index 0d41a2561..e7579e714 100644 --- a/kernel/relayflowd-core/src/machine.rs +++ b/kernel/relayflowd-core/src/machine.rs @@ -280,7 +280,13 @@ fn start_actions(state: &RunState, step: &StepSpec, attempt: u32, now_ms: i64) - "unassigned" }; let lease_id = deterministic_ulid(&state.run_id, &step.id, attempt, now_ms, "lease"); - let lease_deadline_ms = now_ms.saturating_add(LEASE_DURATION_MS); + let lease_duration_ms = match &step.kind { + StepKind::Deterministic { + lease_ms: Some(ms), .. + } => i64::try_from(*ms).unwrap_or(i64::MAX), + _ => LEASE_DURATION_MS, + }; + let lease_deadline_ms = now_ms.saturating_add(lease_duration_ms); let started = JournalEntry::new( EntryType::StepAttemptStarted, state.run_id.clone(), diff --git a/kernel/relayflowd-core/src/machine/tests.rs b/kernel/relayflowd-core/src/machine/tests.rs index 2115ccf84..3907712e4 100644 --- a/kernel/relayflowd-core/src/machine/tests.rs +++ b/kernel/relayflowd-core/src/machine/tests.rs @@ -596,3 +596,43 @@ fn every_reason_label_matches_its_serialized_form() { ); } } + +#[test] +fn deterministic_lease_override_and_default_are_journaled() { + for (lease, duration) in [(None, 30_000), (Some(300_000), 300_000), (Some(10), 10)] { + let mut value = + json!({"steps": [{"id": "cmd", "type": "deterministic", "command": "true"}]}); + if let Some(ms) = lease { + value["steps"][0]["lease_ms"] = json!(ms); + } + let spec = crate::RunSpec::parse(&value).unwrap(); + spec.validate().unwrap(); + let state = RunState::fold("run", spec, &[]).unwrap(); + let clock = SimClock::new(1000); + let Action::Append(started) = &next_actions(&state, clock.now_ms())[0] else { + panic!("expected journaled lease"); + }; + let payload: AttemptStartedPayload = + serde_json::from_value(started.payload.clone()).unwrap(); + assert_eq!(payload.lease_deadline_ms, clock.now_ms() + duration); + } +} + +#[test] +fn deterministic_lease_rejects_invalid_and_foreign_fields() { + for ms in [0, u64::MAX] { + let spec = crate::RunSpec::parse(&json!({"steps": [{ + "id": "cmd", "type": "deterministic", "command": "true", "lease_ms": ms + }]})) + .unwrap(); + assert!(spec.validate().is_err()); + } + for kind in ["llm", "agent"] { + assert!( + crate::RunSpec::parse(&json!({"steps": [{ + "id": "worker", "type": kind, "lease_ms": 1000 + }]})) + .is_err() + ); + } +} diff --git a/kernel/relayflowd-core/src/spec.rs b/kernel/relayflowd-core/src/spec.rs index c4e549e83..d50f58ed2 100644 --- a/kernel/relayflowd-core/src/spec.rs +++ b/kernel/relayflowd-core/src/spec.rs @@ -146,6 +146,16 @@ impl RunSpec { if step.max_iterations == 0 { return Err(SpecError::ZeroIterations(step.id.clone())); } + if let StepKind::Deterministic { + lease_ms: Some(ms), .. + } = &step.kind + && (*ms == 0 || *ms > i64::MAX as u64) + { + return Err(SpecError::Malformed(format!( + "step {}: lease_ms must be positive and fit in i64", + step.id + ))); + } let cli = match &step.kind { StepKind::Llm { cli, .. } | StepKind::Agent { cli, .. } => cli, StepKind::Deterministic { .. } => &None, @@ -297,7 +307,7 @@ const STEP_COMMON_FIELDS: &[&str] = &[ "memory", "requirements", ]; -const STEP_DETERMINISTIC_FIELDS: &[&str] = &["command", "timeout_ms"]; +const STEP_DETERMINISTIC_FIELDS: &[&str] = &["command", "timeout_ms", "lease_ms"]; const STEP_LLM_FIELDS: &[&str] = &["prompt", "model", "cli"]; const STEP_AGENT_FIELDS: &[&str] = &[ "instruction", @@ -393,6 +403,8 @@ pub enum StepKind { command: CommandSpec, #[serde(default, skip_serializing_if = "Option::is_none")] timeout_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + lease_ms: Option, }, Llm { prompt: String, diff --git a/kernel/relayflowd/src/exec_det.rs b/kernel/relayflowd/src/exec_det.rs index 9a10a83b8..8445c793e 100644 --- a/kernel/relayflowd/src/exec_det.rs +++ b/kernel/relayflowd/src/exec_det.rs @@ -42,6 +42,7 @@ pub(crate) fn execute_placed_with_input( let StepKind::Deterministic { command, timeout_ms, + lease_ms, } = &step.kind else { return worker_error("deterministic executor received a non-deterministic step"); @@ -93,7 +94,11 @@ pub(crate) fn execute_placed_with_input( let stderr = child.stderr.take().expect("piped stderr"); let stdout_reader = thread::spawn(move || read_all(stdout)); let stderr_reader = thread::spawn(move || read_all(stderr)); - let timeout = Duration::from_millis(timeout_ms.unwrap_or(30_000)); + let timeout = Duration::from_millis(match (lease_ms, timeout_ms) { + (Some(lease), Some(command)) => (*lease).min(*command), + (Some(lease), None) => *lease, + (None, command) => command.unwrap_or(30_000), + }); let (status, timed_out) = match child.wait_timeout(timeout) { Ok(Some(status)) => (Some(status), false), Ok(None) => { @@ -196,6 +201,23 @@ mod tests { assert_eq!(result.failure_reason, Some(CompletionReason::Timeout)); } + #[test] + fn lease_override_bounds_execution_and_preserves_command_timeout() { + for (lease, command_timeout) in [(5, None), (1000, Some(5))] { + let mut value = json!({ + "id": "slow", "type": "deterministic", "command": "sleep 1", "lease_ms": lease + }); + if let Some(ms) = command_timeout { + value["timeout_ms"] = json!(ms); + } + let step: StepSpec = serde_json::from_value(value).unwrap(); + let started = std::time::Instant::now(); + let result = execute(&step); + assert_eq!(result.failure_reason, Some(CompletionReason::Timeout)); + assert!(started.elapsed() < Duration::from_millis(900)); + } + } + #[test] fn timeout_kills_the_whole_process_group() { // The backgrounded sleep inherits the stdout/stderr pipes. If a diff --git a/packages/sdk/src/authored-flow-error.ts b/packages/sdk/src/authored-flow-error.ts index 0a72c50c0..664f0d071 100644 --- a/packages/sdk/src/authored-flow-error.ts +++ b/packages/sdk/src/authored-flow-error.ts @@ -19,6 +19,7 @@ export type AuthoredFlowExecutionErrorCode = | 'operation_after_completion' | 'operation_callback_failed' | 'step_failed' + | 'lease_exceeded' | 'unsupported_completion' | 'unsupported_gate' | 'unsupported_header' diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index 4910dd229..5825ba7dd 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -10,8 +10,8 @@ import { checkMcpHeader, McpPreflightError } from './cli/check-typescript.js'; import { buildMcpProxy, runMcpEffect } from './authored-mcp.js'; import { AuthoredBudget } from './authored-budget.js'; import { assertMemoryReachable, authoredMemory, scriptMemoryScope } from './authored-memory.js'; -import { authoredWorkerRunner } from './authored-worker-step.js'; -import { readSuccessfulOutput, isSurfaceRunCompletionReason } from './authored-step-output.js'; +import { authoredDeterministicRunner, authoredWorkerRunner } from './authored-worker-step.js'; +import { isSurfaceRunCompletionReason } from './authored-step-output.js'; import { type LlmOptions, type CloudHelper, @@ -23,7 +23,7 @@ import { import type { FlowHandle } from '@relayflows/surface/runtime'; import { join } from 'node:path'; import { observeStep, type ProgressEvent } from './progress.js'; -import { compileSpec, toKernelSpec } from './compile.js'; +import { parseStepTimeout } from './compile.js'; import { getAuthoredFlowDefinition } from './authored-flow.js'; import type { GetFlowDefinition } from './authored-flow-loader.js'; import type { RunLifecycleOptions } from './cli/run.js'; @@ -42,7 +42,6 @@ import type { CompletionReason as ProtocolCompletionReason, RunCompletionReason as ProtocolRunCompletionReason, } from './protocol.js'; -import { SPEC_SCHEMA_VERSION } from './spec.js'; type Assert = T; type Equal = [A] extends [B] @@ -169,19 +168,7 @@ export async function executeAuthoredFlow( let nextStep = 1; let requestedCompletion: SurfaceRunCompletionReason | undefined; - const lowerDeterministic = async ( - id: string, - command: string, - terminal = false, - ): Promise => { - const spec = toKernelSpec(compileSpec({ - version: SPEC_SCHEMA_VERSION, - name: `${definition.name}/${id}`, - steps: [{ id, type: 'deterministic', command }], - })); - if (terminal) return readSuccessfulOutput(journal, await journal.runStart(spec), id, journalSteps); - return budget.execute(journal, spec, outcome => readSuccessfulOutput(journal, outcome, id, journalSteps)); - }; + const lowerDeterministic = authoredDeterministicRunner(definition.name, journal, journalSteps, budget); const worker = authoredWorkerRunner(definition, journal, flowPath, journalSteps, waitOptions, localAgentStream, budget, definition.header.budget); @@ -248,14 +235,15 @@ export async function executeAuthoredFlow( () => assertOperationAllowed('memory', definition.name, requestedCompletion), definition.header.memory?.script !== false, ), - run(command) { + run(command, runOptions) { assertOperationAllowed('run', definition.name, requestedCompletion); + const leaseMs = runOptions?.timeout === undefined ? undefined : parseStepTimeout(runOptions.timeout); const id = `run-${nextStep++}`; return trackStep(authoredSteps, new AuthoredFlowOperation( id, 'run', () => assertOperationAllowed('run', definition.name, requestedCompletion), - () => observeStep(id, 'deterministic', () => lowerDeterministic(id, command), options.onProgress), + () => observeStep(id, 'deterministic', () => lowerDeterministic(id, command, false, leaseMs), options.onProgress), lifecycle, )); }, diff --git a/packages/sdk/src/authored-step-output.ts b/packages/sdk/src/authored-step-output.ts index 8bc5ebcb5..3174d40bc 100644 --- a/packages/sdk/src/authored-step-output.ts +++ b/packages/sdk/src/authored-step-output.ts @@ -50,7 +50,14 @@ export async function readSuccessfulOutput( stepId: string, journalSteps: AuthoredFlowJournalStep[], ): Promise { - const output = await readCompletedStepOutput(journal, outcome.run_id, stepId, journalSteps); + const output = await readCompletedStepOutput(journal, outcome.run_id, stepId, journalSteps).catch(error => { + if (error instanceof AuthoredFlowExecutionError + && (error.completionReason === 'timeout' || error.completionReason === 'lease_expired')) { + throw new AuthoredFlowExecutionError('lease_exceeded', + `f.run step "${stepId}" exceeded its command timeout.`, error.completionReason, error.runId); + } + throw error; + }); if (!isRecord(output) || typeof output['stdout_tail'] !== 'string') { throw protocolViolation(outcome.run_id, `step "${stepId}" has no string stdout_tail`); } diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts index 01c965b44..85923446c 100644 --- a/packages/sdk/src/authored-worker-step.ts +++ b/packages/sdk/src/authored-worker-step.ts @@ -1,14 +1,14 @@ import type { AuthoredBudget } from './authored-budget.js'; import { parseBudget } from './budget.js'; import type { AgentOptions, AgentResult, LlmOptions } from '@relayflows/surface'; -import { toKernelSpec } from './compile.js'; +import { compileSpec, toKernelSpec } from './compile.js'; import { checkAuthoredFlow } from './cli/check.js'; import { classifyOutcome, type RunLifecycleOptions } from './cli/run.js'; import type { PreflightDiagnostic } from './preflight.js'; import { AuthoredFlowExecutionError } from './authored-flow-error.js'; import type { JournalClient } from './journal-client.js'; import { SPEC_SCHEMA_VERSION, type FlowSpec, type StepSpec } from './spec.js'; -import { isSurfaceCompletionReason, readCompletedStepOutput } from './authored-step-output.js'; +import { isSurfaceCompletionReason, readCompletedStepOutput, readSuccessfulOutput } from './authored-step-output.js'; import type { AuthoredFlowJournalStep } from './authored-flow-executor.js'; import { snapshotJsonValue } from './json-value.js'; @@ -142,3 +142,18 @@ export function authoredWorkerRunner( }, }; } + +/** Deterministic commands execute inline under their per-invocation lease. */ +export function authoredDeterministicRunner( + name: string, journal: JournalClient, journalSteps: AuthoredFlowJournalStep[], budget: AuthoredBudget, +) { + return async (id: string, command: string, terminal = false, leaseMs?: number): Promise => { + const spec = toKernelSpec(compileSpec({ + version: SPEC_SCHEMA_VERSION, + name: `${name}/${id}`, + steps: [{ id, type: 'deterministic', command, ...(leaseMs === undefined ? {} : { lease_ms: leaseMs }) }], + })); + if (terminal) return readSuccessfulOutput(journal, await journal.runStart(spec), id, journalSteps); + return budget.execute(journal, spec, outcome => readSuccessfulOutput(journal, outcome, id, journalSteps)); + }; +} diff --git a/packages/sdk/src/compile.ts b/packages/sdk/src/compile.ts index 502f34890..8f0710169 100644 --- a/packages/sdk/src/compile.ts +++ b/packages/sdk/src/compile.ts @@ -53,6 +53,48 @@ export class CompileError extends Error { } } +/** + * Parse the f.run timeout before submitting any command to the kernel. + * + * Accepts a positive whole-millisecond `number`, or a duration string of the + * form `` where unit is `ms` / `s` / `m` — matched by an + * anchored regex that captures the coefficient first (group 1) and the unit + * second (group 2), so a future edit cannot silently swap them. Fractional + * coefficients that resolve to integer milliseconds are accepted; the + * multiplication is rounded to the nearest integer so `1.1s → 1100` + * survives float-precision (`1.1 * 1000 = 1100.0000000000002`) rather than + * being rejected by `isSafeInteger`. + */ +export function parseStepTimeout(timeout: unknown): number { + const unitToMs: Record<'ms' | 's' | 'm', number> = { ms: 1, s: 1000, m: 60_000 }; + let milliseconds: number; + if (typeof timeout === 'number') { + milliseconds = timeout; + } else if (typeof timeout === 'string') { + const match = /^(\d+(?:\.\d+)?)(ms|s|m)$/.exec(timeout); + if (match === null) { + milliseconds = NaN; + } else { + const coefficient = Number(match[1]); + const unit = match[2] as 'ms' | 's' | 'm'; + // Round to defeat float precision (1.1 * 1000 = 1100.0000000000002). + // Fractional milliseconds themselves (e.g. `1.5ms`) still round to `2`, + // which the isSafeInteger check below accepts. Sub-ms precision is not + // a supported unit — authors expressing "1.5ms" get the closest int. + milliseconds = Math.round(coefficient * unitToMs[unit]); + } + } else { + milliseconds = NaN; + } + if (!Number.isSafeInteger(milliseconds) || milliseconds <= 0) { + throw new CompileError(['f.run timeout must be a positive whole number of milliseconds or a duration such as "10s" or "5m".'], 'timeout_invalid'); + } + if (milliseconds > 15 * 60_000) { + throw new CompileError(['f.run timeout exceeds the maximum of 15 minutes declared in SURFACE.md.'], 'lease_exceeded'); + } + return milliseconds; +} + /** * Compile a YAML string into a validated authoring `FlowSpec`. * Throws `CompileError` on a YAML parse error or any validation failure. @@ -156,6 +198,7 @@ function compileStep(step: StepSpec): StepSpec { // #138: `timeoutMs` is deterministic-only — worker-backed verbs own // their dispatch timeout. It must be spread HERE and nowhere in `base`. ...(s.timeoutMs !== undefined ? { timeoutMs: s.timeoutMs } : {}), + ...(s.lease_ms !== undefined ? { lease_ms: parseStepTimeout(s.lease_ms) } : {}), }; } case 'llm': { @@ -372,14 +415,14 @@ function kernelTriggerToAuthoring(value: unknown, at: string): unknown { function kernelStepToAuthoring(value: unknown, at: string): unknown { const unionKeys = [ 'id', 'type', 'depends_on', 'max_iterations', 'retry', 'verification', 'memory', 'requirements', 'input', - 'command', 'timeout_ms', 'prompt', 'model', 'cli', 'instruction', + 'command', 'timeout_ms', 'lease_ms', 'prompt', 'model', 'cli', 'instruction', 'recovery_mode', 'surfaces', 'permissions', ] as const; const step = requireKernelObject(value, unionKeys, at); const type = step['type']; const commonKeys = ['id', 'type', 'depends_on', 'max_iterations', 'retry', 'verification', 'memory', 'requirements', 'input'] as const; const typeKeys = type === 'deterministic' - ? ['command', 'timeout_ms'] as const + ? ['command', 'timeout_ms', 'lease_ms'] as const : type === 'llm' ? ['prompt', 'model', 'cli'] as const : type === 'agent' @@ -405,6 +448,7 @@ function kernelStepToAuthoring(value: unknown, at: string): unknown { ...common, command: step['command'], ...(step['timeout_ms'] !== undefined ? { timeoutMs: step['timeout_ms'] } : {}), + ...(step['lease_ms'] !== undefined ? { lease_ms: step['lease_ms'] } : {}), }; } if (type === 'llm') { @@ -554,6 +598,7 @@ function toKernelStep(step: StepSpec): KernelStepSpec { type: 'deterministic', command: step.command, ...(step.timeoutMs !== undefined ? { timeout_ms: step.timeoutMs } : {}), + ...(step.lease_ms !== undefined ? { lease_ms: parseStepTimeout(step.lease_ms) } : {}), }; case 'llm': { return { diff --git a/packages/sdk/src/spec.ts b/packages/sdk/src/spec.ts index 949785199..e372126ca 100644 --- a/packages/sdk/src/spec.ts +++ b/packages/sdk/src/spec.ts @@ -171,6 +171,8 @@ export interface DeterministicStepSpec extends BaseStepSpec { command: string; /** Wall-clock command timeout; worker-backed verbs own their dispatch timeout. */ timeoutMs?: number; + /** Per-invocation deterministic lease in milliseconds (maximum 15 minutes). */ + lease_ms?: number; /** Omit to get the implicit `exit_code` gate. */ verification?: VerificationSpec; } @@ -342,6 +344,7 @@ export interface KernelDeterministicStep extends KernelStepCommon { type: 'deterministic'; command: string; timeout_ms?: number; + lease_ms?: number; } export interface KernelLlmStep extends KernelStepCommon { diff --git a/packages/sdk/src/step-fields.ts b/packages/sdk/src/step-fields.ts index 4e567a963..becc010e1 100644 --- a/packages/sdk/src/step-fields.ts +++ b/packages/sdk/src/step-fields.ts @@ -33,7 +33,7 @@ export const STEP_COMMON_FIELDS = [ * per-verb boundary through a second, drifting allowlist. */ export const STEP_FIELDS_BY_TYPE = { - deterministic: ['command', 'timeoutMs'], + deterministic: ['command', 'timeoutMs', 'lease_ms'], llm: ['prompt', 'model', 'cli', 'output'], agent: ['instruction', 'agent', 'cli', 'model', 'surfaces', 'recoveryMode', 'permissions', 'output'], } as const satisfies Record; diff --git a/packages/sdk/src/validate.ts b/packages/sdk/src/validate.ts index bce5be7a3..34e52904f 100644 --- a/packages/sdk/src/validate.ts +++ b/packages/sdk/src/validate.ts @@ -423,6 +423,9 @@ class Validator { if (!isNonEmptyString(st.command)) { this.fail(`${at}.command: expected a non-empty string`); } + if (st.lease_ms !== undefined && !isPosInt(st.lease_ms)) { + this.fail(`${at}.lease_ms: expected a positive integer`); + } if (st.timeoutMs !== undefined && !isPosInt(st.timeoutMs)) { this.fail(`${at}.timeoutMs: expected a positive integer`); } diff --git a/packages/sdk/tests/step-lease.test.ts b/packages/sdk/tests/step-lease.test.ts new file mode 100644 index 000000000..2b8074050 --- /dev/null +++ b/packages/sdk/tests/step-lease.test.ts @@ -0,0 +1,136 @@ +import { flow } from '@relayflows/surface'; +import { describe, expect, it, vi } from 'vitest'; +import { executeAuthoredFlow } from '../src/authored-flow-executor.js'; +import { compileSpec, kernelToAuthoring, parseStepTimeout, toKernelSpec } from '../src/compile.js'; +import { JournalClient } from '../src/journal-client.js'; +import { chainFixture } from './flow-chain-fixture.js'; + +const commandSpec = (lease_ms?: number) => ({ + version: '0.1.0', steps: [{ id: 'cmd', type: 'deterministic', command: 'true', + ...(lease_ms === undefined ? {} : { lease_ms }) }], +}); + +describe('deterministic step lease compilation', () => { + it.each([ + ['5m', 300_000], [300_000, 300_000], ['10s', 10_000], ['250ms', 250], + ['1.5s', 1500], ['15m', 900_000], [900_000, 900_000], + // Bugbot #350 regression: fractional coefficients that hit float + // precision (1.1 * 1000 = 1100.0000000000002) must round to a clean + // integer, not be rejected as non-safe-integer. + ['1.1s', 1100], ['2.2s', 2200], ['1.1m', 66_000], + ['0.5s', 500], ['14.999m', 899_940], + // Explicit coefficient-vs-unit ordering — a regression that swapped + // match[1]/match[2] would produce 1000 (units["s"]) * 5 (coefficient + // read as unit) = NaN. The row below fails hard on that specific bug. + ['5s', 5000], + ])('parses %s as %i milliseconds', (timeout, expected) => { + expect(parseStepTimeout(timeout)).toBe(expected); + const kernel = toKernelSpec(compileSpec(commandSpec(expected))); + expect(kernel.steps[0]).toHaveProperty('lease_ms', expected); + expect(toKernelSpec(compileSpec(kernelToAuthoring(kernel)))).toEqual(kernel); + }); + + it.each(['15.001m', '16m', 900_001])('refuses %s before contacting the journal', async timeout => { + await expect(executeAuthoredFlow(flow('too-long', async f => { + await f.run('true', { timeout }); + f.done('success'); + }), new JournalClient('/must-not-connect'))).rejects.toMatchObject({ kind: 'lease_exceeded' }); + }); + + it.each([0, -1, NaN, Infinity, 0.5, '', '5', 'forever', '-1s', '0.0001s', null, {}])( + 'refuses invalid timeout %s', timeout => { + expect(() => parseStepTimeout(timeout)).toThrow(/positive whole number/); + }, + ); + + it('enforces the ceiling for direct specs and omits an unspecified lease', () => { + expect(() => compileSpec(commandSpec(900_001))).toThrow(/maximum of 15 minutes/); + expect(toKernelSpec(compileSpec(commandSpec())).steps[0]).not.toHaveProperty('lease_ms'); + }); +}); + +describe('deterministic step lease lowering', () => { + function journalStub(reason = 'success') { + const journal = new JournalClient('/unused'); + const starts = vi.spyOn(journal, 'runStart').mockImplementation(async () => ({ + run_id: `run-${starts.mock.calls.length}`, status: reason === 'success' ? 'completed' : 'failed', + completion_reason: reason === 'success' ? 'success' : 'step_failed', completed_steps: 1, + })); + vi.spyOn(journal, 'journalRead').mockImplementation(async () => ({ entries: [{ + entry_type: 'step.completed', step_id: starts.mock.calls.at(-1)![0].steps[0]!.id, + payload: { completionReason: reason, output: reason === 'success' ? { stdout_tail: 'ok' } : null }, + }] })); + return { journal, starts }; + } + + it('snapshots each invocation and preserves defaults for later commands', async () => { + const { journal, starts } = journalStub(); + await executeAuthoredFlow(flow('per-invocation', async f => { + const options = { timeout: '5m' }; + const step = f.run('long command', options); + options.timeout = '1ms'; + expect(await step).toBe('ok'); + await f.run('default command'); + await f.run('empty options', {}); + await f.run('numeric timeout', { timeout: 10_000 }); + f.done('success'); + }), journal); + expect(starts.mock.calls.map(([spec]) => { + const step = spec.steps[0]!; + return 'lease_ms' in step ? step.lease_ms : undefined; + })).toEqual([300_000, undefined, undefined, 10_000, undefined]); + }); + + it.each(['timeout', 'lease_expired'])('exposes %s as lease_exceeded with the journal reason', async reason => { + const { journal } = journalStub(reason); + await expect(executeAuthoredFlow(flow('expired', async f => { + await f.run('slow command', { timeout: '10s' }); + f.done('success'); + }), journal)).rejects.toMatchObject({ code: 'lease_exceeded', completionReason: reason, runId: 'run-1' }); + }); +}); + +describe('f.run leases against the live kernel', () => { + // These wall-clock cases deliberately cross the former 30s limit; a smaller + // mocked lease would not detect the executor's independent default timeout. + it.each([ + { command: 'sleep 5; printf ok', timeout: '10s', lease: 10_000, succeeds: true }, + { command: 'sleep 31; printf ok', timeout: '40s', lease: 40_000, succeeds: true }, + { command: 'sleep 31; printf ok', timeout: undefined, lease: 30_000, succeeds: false }, + { command: 'sleep 5', timeout: 100, lease: 100, succeeds: false }, + ])('enforces $lease ms for $command', async ({ command, timeout, lease, succeeds }) => { + const fixture = chainFixture(); + try { + const journal = await fixture.connect(); + let output: string | undefined; + const handle = flow('step-lease', async f => { + output = await (timeout === undefined ? f.run(command) : f.run(command, { timeout })); + f.done('success'); + }); + const started = Date.now(); + const execution = executeAuthoredFlow(handle, journal); + let runId: string; + if (succeeds) { + const result = await execution; + expect(output).toBe('ok'); + runId = result.journalSteps[0]!.runId; + } else { + const error = await execution.then(() => { throw new Error('expected lease refusal'); }, error => error); + expect(error).toMatchObject({ code: 'lease_exceeded', completionReason: 'timeout' }); + expect(Date.now() - started).toBeGreaterThanOrEqual(lease); + expect(Date.now() - started).toBeLessThan(lease + 5000); + runId = error.runId; + } + const entries = (await journal.journalRead(runId, 1)).entries as Array<{ + entry_type: string; at_ms: number; payload: Record; + }>; + const attempt = entries.find(entry => entry.entry_type === 'step.attempt.started')!; + expect(attempt).toBeDefined(); + expect(attempt.payload.lease_deadline_ms).toBe(attempt.at_ms + lease); + const completed = entries.find(entry => entry.entry_type === 'step.completed')!; + expect(completed.payload.completionReason).toBe(succeeds ? 'success' : 'timeout'); + } finally { + await fixture.close(); + } + }, 45_000); +}); diff --git a/packages/sdk/tests/verb-field-lint.test.ts b/packages/sdk/tests/verb-field-lint.test.ts index 0bbccf8d2..37eab6d28 100644 --- a/packages/sdk/tests/verb-field-lint.test.ts +++ b/packages/sdk/tests/verb-field-lint.test.ts @@ -71,6 +71,7 @@ const VERB_FIELD_VALUES: Record = { agent: 'reviewer', command: 'printf foreign', timeoutMs: 1_000, + lease_ms: 1_000, prompt: 'foreign prompt', model: 'foreign-model', cli: 'foreign-cli', @@ -190,12 +191,13 @@ describe('closed per-verb step fields', () => { 'requirements', ]); expect(STEP_FIELDS_BY_TYPE).toEqual({ - deterministic: ['command', 'timeoutMs'], + deterministic: ['command', 'timeoutMs', 'lease_ms'], llm: ['prompt', 'model', 'cli', 'output'], agent: ['instruction', 'agent', 'cli', 'model', 'surfaces', 'recoveryMode', 'permissions', 'output'], }); expect(CROSS_VERB_STEP_FIELDS.map(({ label }) => label).sort()).toEqual([ 'agent foreign command', + 'agent foreign lease_ms', 'agent foreign prompt', 'agent foreign timeoutMs', 'deterministic foreign agent', @@ -210,6 +212,7 @@ describe('closed per-verb step fields', () => { 'llm foreign agent', 'llm foreign command', 'llm foreign instruction', + 'llm foreign lease_ms', 'llm foreign permissions', 'llm foreign recoveryMode', 'llm foreign surfaces', diff --git a/packages/surface/src/context.ts b/packages/surface/src/context.ts index 0880ef3b4..222e5b89a 100644 --- a/packages/surface/src/context.ts +++ b/packages/surface/src/context.ts @@ -31,7 +31,8 @@ export interface LlmOptions { */ export interface Ctx extends Helpers { readonly mcp: Readonly Step>>>>; - run(command: string): Step; + /** Command lease: milliseconds or a duration such as "5m"; default 30s, maximum 15m. */ + run(command: string, options?: { timeout?: string | number }): Step; llm(strings: TemplateStringsArray, ...values: unknown[]): Step; /** JSON Schema validates the value at runtime; narrow unknown in author code. */ llm(prompt: string, options: LlmOptions): Step;