feat(surface,sdk,kernel): deterministic-step timeout option (f.run with timeout) (#343)

Author: miyaontherelay

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

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(),

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()
+        );
+    }
+}

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<u64>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        lease_ms: Option<u64>,
     },
     Llm {
         prompt: String,

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

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'

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 extends true> = T;
 type Equal<A, B> = [A] extends [B]
@@ -169,19 +168,7 @@ export async function executeAuthoredFlow<Input = undefined>(
   let nextStep = 1;
   let requestedCompletion: SurfaceRunCompletionReason | undefined;
 
-  const lowerDeterministic = async (
-    id: string,
-    command: string,
-    terminal = false,
-  ): Promise<string> => {
-    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<Input = undefined>(
       () => 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,
       ));
     },

packages/sdk/src/authored-step-output.ts

@@ -50,7 +50,14 @@ export async function readSuccessfulOutput(
   stepId: string,
   journalSteps: AuthoredFlowJournalStep[],
 ): Promise<string> {
-  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`);
   }

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<string> => {
+    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));
+  };
+}

packages/sdk/src/compile.ts

@@ -53,6 +53,21 @@ export class CompileError extends Error {
   }
 }
 
+/** Parse the f.run timeout before submitting any command to the kernel. */
+export function parseStepTimeout(timeout: unknown): number {
+  const units: Record<string, number> = { ms: 1, s: 1000, m: 60_000 };
+  const match = typeof timeout === 'string' ? /^(\d+(?:\.\d+)?)(ms|s|m)$/.exec(timeout) : null;
+  const milliseconds = typeof timeout === 'number' ? timeout
+    : match === null ? NaN : Number(match[1]) * units[match[2]!]!;
+  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 +171,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 +388,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 +421,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 +571,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 {