From eca10cc662ad957586edfcaa6de3f62d7d71d237 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Fri, 18 Sep 2026 05:29:13 -0700 Subject: [PATCH 1/4] feat(kernel,sdk,surface): f.human parks the run on wait.human; flows answer closes it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The authored body's f.human no longer throws unsupported_verb. It reads the root journal for a wait.completed{human_responded} under its ordinal (human-N); when none is recorded it throws AuthoredHumanParked, the durable root turns that into the new step.wait verb, and the kernel journals wait.human for the running attempt, releases the lease and parks the run (needs_human, exit 3). event.emit keyed by the wait id closes a wait.human as human_responded — the contract DESIGN.md already documented — and the root is dispatched again; the resumed body finds the answer and lowers it as a memoized human-N deterministic step so the boolean the author branches on is journaled evidence. flows answer yes|no [--note] records it. Unknown runs now surface run_not_found from every verb instead of a journal open failure. Co-Authored-By: Claude Opus 5 (1M context) --- kernel/relayflowd/src/engine.rs | 34 ++- kernel/relayflowd/src/engine/remote.rs | 110 +++++++++- kernel/relayflowd/src/lib.rs | 3 +- kernel/relayflowd/src/server.rs | 41 +++- kernel/relayflowd/src/server/protocol.rs | 6 + kernel/relayflowd/src/server/tests.rs | 121 +++++++++++ kernel/relayflowd/src/server/wire.rs | 19 ++ packages/sdk/src/authored-flow-error.ts | 29 +++ packages/sdk/src/authored-flow-executor.ts | 50 ++++- packages/sdk/src/authored-human.ts | 136 ++++++++++++ packages/sdk/src/authored-node-entry.ts | 5 +- packages/sdk/src/authored-node-runner.ts | 25 ++- packages/sdk/src/authored-root.ts | 29 ++- packages/sdk/src/cli.ts | 70 ++++++ packages/sdk/src/cli/answer.ts | 89 ++++++++ packages/sdk/src/cli/direct-run.ts | 5 + packages/sdk/src/cli/run.ts | 50 ++++- packages/sdk/src/failure-kinds.ts | 2 + packages/sdk/src/journal-client.ts | 20 ++ packages/sdk/src/progress.ts | 15 +- packages/sdk/src/protocol.ts | 23 ++ packages/sdk/tests/authored-flow.test.ts | 4 - packages/sdk/tests/authored-human.test.ts | 200 ++++++++++++++++++ .../sdk/tests/authored-node-runtime.test.ts | 35 +++ packages/sdk/tests/authored-root.test.ts | 106 +++++++++- packages/sdk/tests/human-live.test.ts | 166 +++++++++++++++ packages/sdk/tests/journal-client-loopback.ts | 1 + packages/surface/src/context.ts | 9 +- 28 files changed, 1362 insertions(+), 41 deletions(-) create mode 100644 packages/sdk/src/authored-human.ts create mode 100644 packages/sdk/src/cli/answer.ts create mode 100644 packages/sdk/tests/authored-human.test.ts create mode 100644 packages/sdk/tests/human-live.test.ts diff --git a/kernel/relayflowd/src/engine.rs b/kernel/relayflowd/src/engine.rs index 34e202e3d..708f5d824 100644 --- a/kernel/relayflowd/src/engine.rs +++ b/kernel/relayflowd/src/engine.rs @@ -32,6 +32,22 @@ pub struct RunTerminalError { pub run_id: String, } +/// No registry row and no journal file: the run never existed here. Distinct +/// from a journal that exists but cannot be opened, which stays a journal +/// failure so a corrupt run is never reported as an absent one. +#[derive(Debug)] +pub struct RunNotFoundError { + pub run_id: String, +} + +impl std::fmt::Display for RunNotFoundError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "run {} does not exist", self.run_id) + } +} + +impl std::error::Error for RunNotFoundError {} + impl std::fmt::Display for RunTerminalError { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( @@ -92,7 +108,7 @@ mod wake; pub use channels::ChannelCommandError; pub use model::{RunOutcome, RunSnapshot, RunStatus, StepSnapshot, StepStatus}; use model::{outcome_from_state, snapshot_from_state}; -pub use remote::OutOfBandCompletion; +pub use remote::{OutOfBandCompletion, OutOfBandHumanWait}; pub use wake::EventSubmitOutcome; #[derive(Debug, Clone, Default)] @@ -643,9 +659,19 @@ impl Engine { fn open_run(&self, run_id: &str) -> Result { let registered = self.registry()?.lookup(run_id)?; - let path = registered - .map(|record| record.file) - .unwrap_or_else(|| self.run_path(run_id)); + let path = match registered { + Some(record) => record.file, + None => { + let path = self.run_path(run_id); + if !path.exists() { + return Err(RunNotFoundError { + run_id: run_id.to_owned(), + } + .into()); + } + path + } + }; SqliteJournal::open(&path).with_context(|| format!("open run journal {}", path.display())) } diff --git a/kernel/relayflowd/src/engine/remote.rs b/kernel/relayflowd/src/engine/remote.rs index 19035996a..8d51816a0 100644 --- a/kernel/relayflowd/src/engine/remote.rs +++ b/kernel/relayflowd/src/engine/remote.rs @@ -2,7 +2,7 @@ use anyhow::{Context, Result, anyhow, bail}; use relayflowd_core::{ AttemptResult, Budget, Clock, CompletionReason, EffectRef, EntryType, JournalEntry, Pins, StepKind, StepState, StreamAppendedPayload, WaitCompletedPayload, WaitCompletionReason, - WaitEventPayload, abandonment_actions, completion_actions, + WaitEventPayload, WaitHumanPayload, abandonment_actions, completion_actions, }; use relayflowd_journal::SqliteJournal; use serde_json::Value; @@ -30,7 +30,88 @@ pub struct OutOfBandCompletion { pub trajectory_tail: Option, } +/// A human question a leased attempt parks on. The worker names the wait so +/// its own replay can find the answer; the kernel refuses a reused name. +#[derive(Debug, Clone)] +pub struct OutOfBandHumanWait { + pub attempt: u32, + pub idempotency_key: String, + pub wait_id: String, + pub prompt: String, + pub requested_of: String, + pub options: Option>, + pub timeout_at_ms: Option, +} + impl Engine { + /// Park a currently leased step on a durable `wait.human` (DESIGN.md §1.5). + /// + /// The attempt is not completed: no `step.completed` is appended and no + /// semantic iteration is charged, because the attempt produced nothing a + /// gate could judge — it asked a question. The step folds to `NeedsHuman`, + /// the run parks, and the answer arrives through `emit_event` keyed by + /// `wait_id`, which folds the step back to `Runnable` for a fresh attempt. + pub fn wait_human_out_of_band( + &self, + run_id: &str, + step_id: &str, + wait: OutOfBandHumanWait, + ) -> Result { + if wait.wait_id.is_empty() || wait.prompt.is_empty() || wait.requested_of.is_empty() { + bail!("step wait requires a non-empty wait_id, prompt and requested_of"); + } + let mut journal = self.open_run(run_id)?; + let spec = journal.run_spec().context("read run spec")?; + let state = self.load_state(&journal, spec.clone())?; + if state.cancel_requested.is_some() || state.completion.is_some() { + bail!("run {run_id} no longer accepts step waits") + } + if spec.step(step_id).is_none() { + bail!("run {run_id} has no step {step_id}") + } + let StepState::Running { + attempt, + ref idempotency_key, + .. + } = state.steps[step_id].state + else { + bail!("step {step_id} has no active lease") + }; + if attempt != wait.attempt || idempotency_key != &wait.idempotency_key { + bail!("step {step_id} wait does not match its active lease") + } + // A wait id names one question for the life of the run. A second + // `wait.human` under the same id would let one answer close both, so + // the reuse is refused rather than journaled. + for entry in journal.scan_all().context("read prior waits")? { + if entry.entry_type == EntryType::WaitHuman { + let prior: WaitHumanPayload = serde_json::from_value(entry.payload)?; + if prior.wait_id == wait.wait_id { + bail!("wait {} was already asked in run {run_id}", wait.wait_id) + } + } + } + self.append( + &mut journal, + &JournalEntry::new( + EntryType::WaitHuman, + run_id, + Some(step_id.to_owned()), + Some(attempt), + self.clock.now_ms(), + WaitHumanPayload { + wait_id: wait.wait_id, + prompt: wait.prompt, + requested_of: wait.requested_of, + options: wait.options, + timeout_at_ms: wait.timeout_at_ms, + diff_ref: None, + }, + ), + )?; + self.drive(journal, spec, DriveOptions::default()) + } + /// Complete a currently leased step through the same verification/retry /// state-machine path used by in-process deterministic execution. pub fn complete_out_of_band( @@ -298,19 +379,38 @@ impl Engine { let mut journal = self.open_run(run_id)?; let spec = journal.run_spec().context("read run spec")?; let entries = journal.scan_all().context("read event waits")?; + // `wait.event` matches on its `event_key`; `wait.human` has none, so a + // human response is addressed to the wait itself: `event_key` is the + // `wait_id` (DESIGN.md §5, `event.emit`). Each open wait remembers + // which reason closes it. let mut open = Vec::new(); for entry in &entries { if entry.entry_type == EntryType::WaitEvent { let wait: WaitEventPayload = serde_json::from_value(entry.payload.clone())?; if wait.event_key == event_key { - open.push((wait.wait_id, entry.step_id.clone(), entry.attempt)); + open.push(( + wait.wait_id, + entry.step_id.clone(), + entry.attempt, + WaitCompletionReason::EventReceived, + )); + } + } else if entry.entry_type == EntryType::WaitHuman { + let wait: WaitHumanPayload = serde_json::from_value(entry.payload.clone())?; + if wait.wait_id == event_key { + open.push(( + wait.wait_id, + entry.step_id.clone(), + entry.attempt, + WaitCompletionReason::HumanResponded, + )); } } else if entry.entry_type == EntryType::WaitCompleted { let done: WaitCompletedPayload = serde_json::from_value(entry.payload.clone())?; - open.retain(|(wait_id, _, _)| wait_id != &done.wait_id); + open.retain(|(wait_id, _, _, _)| wait_id != &done.wait_id); } } - for (wait_id, step_id, attempt) in &open { + for (wait_id, step_id, attempt, completion_reason) in &open { self.append( &mut journal, &JournalEntry::new( @@ -321,7 +421,7 @@ impl Engine { self.clock.now_ms(), WaitCompletedPayload { wait_id: wait_id.clone(), - completion_reason: WaitCompletionReason::EventReceived, + completion_reason: *completion_reason, result: payload.clone(), }, ), diff --git a/kernel/relayflowd/src/lib.rs b/kernel/relayflowd/src/lib.rs index 004cdcb63..8cf501261 100644 --- a/kernel/relayflowd/src/lib.rs +++ b/kernel/relayflowd/src/lib.rs @@ -8,7 +8,8 @@ pub mod trigger_watcher; pub mod worker; pub use engine::{ - CancelOptions, DriveOptions, Engine, OutOfBandCompletion, RunOutcome, RunSnapshot, RunStatus, + CancelOptions, DriveOptions, Engine, OutOfBandCompletion, OutOfBandHumanWait, RunOutcome, + RunSnapshot, RunStatus, StepSnapshot, StepStatus, }; diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index 7655d742c..881c960fd 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -4,7 +4,7 @@ use anyhow::{Context, Result}; use relayflowd_core::{CompletionReason, PROTOCOL_VERSION, RunSpec, StepType}; use serde_json::{Value, json}; -use crate::{Engine, OutOfBandCompletion}; +use crate::{Engine, OutOfBandCompletion, OutOfBandHumanWait}; #[cfg(unix)] mod cancel; @@ -407,6 +407,45 @@ fn handle_request( } to_value(outcome) } + "step.wait" => { + let params: StepWaitParams = decode_params(request.params)?; + let key = ( + params.run_id.clone(), + params.step_id.clone(), + params.attempt, + ); + let lock = hub.run_lock(¶ms.run_id); + let _guard = lock.lock().expect("run lock"); + ensure_mutable(&engine, ¶ms.run_id)?; + // Only the lease holder may park its attempt, exactly as only it + // may complete it. + hub.completion_worker(connection_id, &key) + .map_err(protocol_conflict)?; + let outcome = engine + .wait_human_out_of_band( + ¶ms.run_id, + ¶ms.step_id, + OutOfBandHumanWait { + attempt: params.attempt, + idempotency_key: params.idempotency_key, + wait_id: params.wait_id, + prompt: params.prompt, + requested_of: params.requested_of, + options: params.options, + timeout_at_ms: params.timeout_at_ms, + }, + ) + .map_err(internal_error)?; + // The wait is durable; the in-memory lease is released so the + // connection closing later is not read as a crashed attempt. + hub.finish(&key); + if let Some(deadline) = hub.earliest_lease_deadline(¶ms.run_id) { + engine + .renew_lease(¶ms.run_id, deadline) + .map_err(internal_error)?; + } + to_value(outcome) + } "effect.record" => { let params: EffectRecordParams = decode_params(request.params)?; let key = ( diff --git a/kernel/relayflowd/src/server/protocol.rs b/kernel/relayflowd/src/server/protocol.rs index 5d54e5059..3ce0bb141 100644 --- a/kernel/relayflowd/src/server/protocol.rs +++ b/kernel/relayflowd/src/server/protocol.rs @@ -83,6 +83,12 @@ pub(super) fn run_start_error(error: anyhow::Error) -> (&'static str, String) { } pub(super) fn internal_error(error: anyhow::Error) -> (&'static str, String) { + if error + .downcast_ref::() + .is_some() + { + return ("run_not_found", error.to_string()); + } let journal_failure = error.chain().any(|cause| { cause.is::() || cause.is::() diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index 2489e013a..feb791279 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -500,6 +500,127 @@ fn stopped_heartbeats_past_the_deadline_journal_lease_expired_and_release_the_st assert_eq!(redispatch["data"]["attempt"], 2); } +/// `step.wait` parks a leased attempt on a durable human question without +/// completing it; `event.emit` keyed by the wait id closes it as +/// `human_responded` and the step is dispatched again as a fresh attempt. +#[test] +fn step_wait_parks_the_attempt_and_a_human_answer_redispatches_it() { + let directory = tempdir().unwrap(); + let data_dir = directory.path(); + let hub = Arc::new(ProtocolHub::default()); + let (worker_writer, worker_peer) = shared_writer(); + let attach = request( + data_dir, + &hub, + 1, + &worker_writer, + r#"{"id":"attach","verb":"worker.attach","params":{"worker_id":"unit-stub","step_types":["llm"]}}"#, + ); + assert!(attach.ok, "worker.attach failed: {:?}", attach.error); + let mut worker_reader = BufReader::new(worker_peer); + let run_id = start_llm_run(data_dir, &hub); + + let dispatch = read_frame(&mut worker_reader); + assert_eq!(dispatch["event"], "step.dispatch"); + assert_eq!(dispatch["data"]["attempt"], 1); + let idempotency_key = dispatch["data"]["idempotency_key"].as_str().unwrap().to_owned(); + let wait = |connection: u64, wait_id: &str| { + request( + data_dir, + &hub, + connection, + &worker_writer, + &json!({"id": "wait", "verb": "step.wait", "params": { + "run_id": run_id, "step_id": "model", "attempt": 1, + "idempotency_key": idempotency_key, "wait_id": wait_id, + "prompt": "Ship this?", "requested_of": "khaliq", + }}) + .to_string(), + ) + }; + + // Only the lease holder may park the attempt. + let stranger = wait(9, "human-1"); + assert!(!stranger.ok); + assert_eq!(stranger.error.unwrap().code, "lease_conflict"); + + let parked = wait(1, "human-1"); + assert!(parked.ok, "step.wait failed: {:?}", parked.error); + assert_eq!(parked.result.unwrap()["status"], "parked"); + // Nothing completed: the attempt asked a question, it did not fail. + assert!(step_completions(data_dir, &run_id).is_empty()); + let (writer, _peer) = shared_writer(); + let snapshot = request( + data_dir, + &hub, + 2, + &writer, + &json!({"id": "get", "verb": "run.get", "params": {"run_id": run_id}}).to_string(), + ); + let snapshot = snapshot.result.unwrap(); + assert_eq!(snapshot["status"], "parked"); + assert_eq!(snapshot["steps"]["model"]["state"], "needs_human"); + // The lease is released, so the worker connection closing later is not a crash. + assert!(!hub.lease_active(&run_id, "model", 1, now_ms())); + + // The same question cannot be asked twice under one id, and a released + // attempt holds no lease to park again anyway. + let again = wait(1, "human-1"); + assert!(!again.ok); + + // An answer to a different wait matches nothing and changes nothing. + let stray = request( + data_dir, + &hub, + 2, + &writer, + &json!({"id": "emit", "verb": "event.emit", "params": { + "run_id": run_id, "event_key": "human-2", "payload": {"answer": true}, + }}) + .to_string(), + ); + assert_eq!(stray.result.unwrap()["matched"], 0); + + let answered = request( + data_dir, + &hub, + 2, + &writer, + &json!({"id": "emit", "verb": "event.emit", "params": { + "run_id": run_id, "event_key": "human-1", "payload": {"answer": true}, + }}) + .to_string(), + ); + assert!(answered.ok, "event.emit failed: {:?}", answered.error); + assert_eq!(answered.result.unwrap()["matched"], 1); + let redispatch = read_frame(&mut worker_reader); + assert_eq!(redispatch["event"], "step.dispatch"); + assert_eq!(redispatch["data"]["attempt"], 2); + + let entries = Engine::new(data_dir) + .journal_entries(&run_id, 1, usize::MAX) + .unwrap(); + let completed = entries + .iter() + .find(|entry| entry.entry_type == EntryType::WaitCompleted) + .expect("wait.completed journaled"); + assert_eq!(completed.payload["wait_id"], "human-1"); + assert_eq!(completed.payload["completionReason"], "human_responded"); + assert_eq!(completed.payload["result"]["answer"], true); + // Answering again matches nothing: the wait is closed. + let repeat = request( + data_dir, + &hub, + 2, + &writer, + &json!({"id": "emit", "verb": "event.emit", "params": { + "run_id": run_id, "event_key": "human-1", "payload": {"answer": false}, + }}) + .to_string(), + ); + assert_eq!(repeat.result.unwrap()["matched"], 0); +} + /// Finding 4: an entry committed before `run.watch` registration but whose /// hub notification is delayed is delivered exactly once. The run lock makes /// the journal commit and notification indivisible from watch registration. diff --git a/kernel/relayflowd/src/server/wire.rs b/kernel/relayflowd/src/server/wire.rs index 9768b6b4f..0f7fb9eac 100644 --- a/kernel/relayflowd/src/server/wire.rs +++ b/kernel/relayflowd/src/server/wire.rs @@ -128,6 +128,25 @@ pub(super) struct EffectConfirmParams { pub surface_path: String, } +/// A lease holder parks its running attempt on a durable human question +/// (`wait.human`, DESIGN.md §1.5). The lease is released once the wait is +/// journaled; the answer arrives through `event.emit` keyed by `wait_id`. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct StepWaitParams { + pub run_id: String, + pub step_id: String, + pub attempt: u32, + pub idempotency_key: String, + pub wait_id: String, + pub prompt: String, + pub requested_of: String, + #[serde(default)] + pub options: Option>, + #[serde(default)] + pub timeout_at_ms: Option, +} + #[derive(Deserialize)] #[serde(deny_unknown_fields)] pub(super) struct EventEmitParams { diff --git a/packages/sdk/src/authored-flow-error.ts b/packages/sdk/src/authored-flow-error.ts index 5bc794a42..6edbbaed5 100644 --- a/packages/sdk/src/authored-flow-error.ts +++ b/packages/sdk/src/authored-flow-error.ts @@ -24,6 +24,10 @@ export type AuthoredFlowExecutionErrorCode = | 'unsupported_completion' | 'unsupported_gate' | 'gate_failed' + /** The body reached an unanswered `f.human`; the root parked on it. */ + | 'human_parked' + /** A recorded answer to `f.human` was not `{ answer: boolean }`. */ + | 'human_answer_invalid' | 'unsupported_header' | 'unsettled_derived_work' | 'unsupported_promise_lifecycle' @@ -42,3 +46,28 @@ export class AuthoredFlowExecutionError extends Error { this.name = 'AuthoredFlowExecutionError'; } } + +/** The question an authored body parked on, as the kernel journals it. */ +export interface AuthoredHumanWait { + readonly waitId: string; + readonly question: string; + readonly to: string; +} + +/** + * Thrown by `f.human` when no answer is journaled yet. Not a failure: the + * durable root catches it, parks its attempt on the kernel's `wait.human`, and + * the CLI reports exit 3 naming the question and the command that answers it. + * A resumed body re-runs to the same call and finds the recorded answer. + */ +export class AuthoredHumanParked extends AuthoredFlowExecutionError { + constructor(readonly wait: AuthoredHumanWait, rootRunId: string) { + super( + 'human_parked', + `flow is waiting for ${wait.to} to answer ${JSON.stringify(wait.question)} (${wait.waitId})`, + undefined, + rootRunId, + ); + this.name = 'AuthoredHumanParked'; + } +} diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index 5f1ee722c..48c87dee0 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -32,8 +32,10 @@ import type { GetFlowDefinition } from './authored-flow-loader.js'; import type { RunLifecycleOptions } from './cli/run.js'; import { AuthoredFlowExecutionError, + AuthoredHumanParked, type AuthoredFlowExecutionErrorCode, } from './authored-flow-error.js'; +import { readHumanAnswer } from './authored-human.js'; import { AuthoredFlowOperation, stopAuthoredOperations, @@ -406,9 +408,53 @@ export async function executeAuthoredFlow( ); return trackStep(authoredSteps, agentOp); }, - human() { + /** + * `f.human` (docs/SURFACE.md §1, §7). The question is not a child run: it + * is the ROOT attempt parking on the kernel's `wait.human`. The body + * cannot park itself — it holds no lease — so when no answer is journaled + * it throws `AuthoredHumanParked`, the durable root turns that into + * `step.wait`, and the CLI reports the question with exit 3. + * + * The wait id is the operation's ordinal (`human-N`), so a resumed body + * re-executes to the same call and finds the recorded + * `wait.completed{human_responded}`. The ANSWER is then lowered as a + * `human-N` deterministic step carrying `{"answer":…}` on stdout: the + * boolean the author's code branches on is a journaled, memoized fact in + * the same shape as every other authored step, so the IPC verifier and a + * later replay hold it to the same evidence. + */ + human(question, humanOptions) { assertOperationAllowed('human', definition.name, requestedCompletion); - throw unsupportedVerb('human'); + if (typeof question !== 'string' || question.trim() === '') { + throw new AuthoredFlowExecutionError('human_answer_invalid', 'f.human requires a non-empty question'); + } + if (typeof humanOptions?.to !== 'string' || humanOptions.to.trim() === '') { + throw new AuthoredFlowExecutionError('human_answer_invalid', 'f.human requires { to } naming who answers'); + } + const id = `human-${nextStep++}`; + const to = humanOptions.to; + return trackStep(authoredSteps, new AuthoredFlowOperation( + id, 'human', + () => assertOperationAllowed('human', definition.name, requestedCompletion), + () => observeStep(id, 'deterministic', async () => { + const rootRunId = options.rootRunId; + if (rootRunId === undefined) { + throw new AuthoredFlowExecutionError( + 'unsupported_verb', + `f.human needs a durable root to park in; "${id}" has no run to wait on. Run the flow with flows run.`, + ); + } + const recorded = await readHumanAnswer(journal, rootRunId, id); + if (recorded === undefined) throw new AuthoredHumanParked({ waitId: id, question, to }, rootRunId); + const record = { human: id, to, answer: recorded.answer, + ...(recorded.note === undefined ? {} : { note: recorded.note }), + ...(recorded.answeredBy === undefined ? {} : { answeredBy: recorded.answeredBy }) }; + const literal = `'${JSON.stringify(record).replaceAll("'", "'\\''")}'`; + await lowerDeterministic(id, `printf '%s' ${literal}`, false); + return recorded.answer; + }, options.onProgress), + lifecycle, + )); }, dispatch() { assertOperationAllowed('dispatch', definition.name, requestedCompletion); diff --git a/packages/sdk/src/authored-human.ts b/packages/sdk/src/authored-human.ts new file mode 100644 index 000000000..510fb33d5 --- /dev/null +++ b/packages/sdk/src/authored-human.ts @@ -0,0 +1,136 @@ +import { AuthoredFlowExecutionError, type AuthoredHumanWait } from './authored-flow-error.js'; +import type { JournalClient } from './journal-client.js'; + +/** + * The answer contract for a parked `f.human`. This is the `payload` of the + * `event.emit` that closes the wait (kernel DESIGN.md §5: `event_key` is the + * `wait_id`) and therefore the `result` of the journaled + * `wait.completed{human_responded}` the resumed body reads back. `flows + * answer` and Cloud's answer route both produce exactly this shape. + */ +export interface HumanAnswer { + readonly answer: boolean; + readonly note?: string; + readonly answeredBy?: string; + /** ISO-8601 instant the answer was given. */ + readonly at?: string; +} + +/** An open `wait.human` on a run: asked, not yet answered. */ +export interface OpenHumanWait extends AuthoredHumanWait { + readonly stepId: string; + readonly attempt: number; +} + +export const HUMAN_WAIT_ID = /^human-[1-9][0-9]*$/; + +export function humanAnswerPayload(answer: boolean, extra: { note?: string; answeredBy?: string } = {}): HumanAnswer { + return { + answer, + ...(extra.note === undefined || extra.note === '' ? {} : { note: extra.note }), + ...(extra.answeredBy === undefined || extra.answeredBy === '' ? {} : { answeredBy: extra.answeredBy }), + at: new Date().toISOString(), + }; +} + +/** Narrow an untrusted journal `result` to the answer contract, or refuse it. */ +export function parseHumanAnswer(value: unknown, waitId: string): HumanAnswer { + const record = value as Partial | null; + if (typeof record !== 'object' || record === null || typeof record.answer !== 'boolean' + || (record.note !== undefined && typeof record.note !== 'string') + || (record.answeredBy !== undefined && typeof record.answeredBy !== 'string') + || (record.at !== undefined && typeof record.at !== 'string')) { + throw new AuthoredFlowExecutionError( + 'human_answer_invalid', + `the recorded answer to ${waitId} is not { answer: boolean }; answer it again with flows answer`, + ); + } + return { + answer: record.answer, + ...(record.note === undefined ? {} : { note: record.note }), + ...(record.answeredBy === undefined ? {} : { answeredBy: record.answeredBy }), + ...(record.at === undefined ? {} : { at: record.at }), + }; +} + +interface WaitEntry { + entry_type?: string; + step_id?: string | null; + attempt?: number | null; + payload?: { + wait_id?: unknown; prompt?: unknown; requested_of?: unknown; + completionReason?: unknown; result?: unknown; + }; +} + +async function readWaitEntries(journal: JournalClient, runId: string): Promise { + const waits: WaitEntry[] = []; + let fromSeq = 1; + for (;;) { + const page = (await journal.journalRead(runId, fromSeq, 1000)).entries as Array; + if (page.length === 0) break; + for (const entry of page) { + if (typeof entry.seq !== 'number' || entry.seq < fromSeq) { + throw new AuthoredFlowExecutionError('journal_protocol_violation', `journal.read for ${runId} returned out-of-order entries`); + } + fromSeq = entry.seq + 1; + if (entry.entry_type === 'wait.human' || entry.entry_type === 'wait.completed') waits.push(entry); + } + } + return waits; +} + +/** + * The recorded answer to `waitId`, or `undefined` while the question is open + * or not yet asked. The first `human_responded` completion wins: the kernel + * closes a wait once, so a second answer never reaches the journal. + */ +export async function readHumanAnswer( + journal: JournalClient, + runId: string, + waitId: string, +): Promise { + for (const entry of await readWaitEntries(journal, runId)) { + if (entry.entry_type === 'wait.completed' && entry.payload?.wait_id === waitId + && entry.payload.completionReason === 'human_responded') { + return parseHumanAnswer(entry.payload.result, waitId); + } + } + return undefined; +} + +/** Every `wait.human` on the run that no `wait.completed` has closed. */ +export async function readOpenHumanWaits(journal: JournalClient, runId: string): Promise { + const open = new Map(); + for (const entry of await readWaitEntries(journal, runId)) { + const waitId = entry.payload?.wait_id; + if (typeof waitId !== 'string') continue; + if (entry.entry_type === 'wait.human') { + open.set(waitId, { + waitId, + question: typeof entry.payload?.prompt === 'string' ? entry.payload.prompt : '', + to: typeof entry.payload?.requested_of === 'string' ? entry.payload.requested_of : '', + stepId: typeof entry.step_id === 'string' ? entry.step_id : '', + attempt: typeof entry.attempt === 'number' ? entry.attempt : 0, + }); + } else { + open.delete(waitId); + } + } + return [...open.values()]; +} + +/** How to answer a parked question from a shell, with the run's own data dir. */ +export function answerCommand(runId: string, waitId: string, dataDir?: string): string { + const dir = dataDir === undefined ? '' : ` --data-dir ${shellWord(dataDir)}`; + return `flows answer${dir} ${runId} ${waitId} yes|no`; +} + +export function resumeCommand(runId: string, dataDir?: string, localAgent = false): string { + const dir = dataDir === undefined ? '' : ` --data-dir ${shellWord(dataDir)}`; + return `flows resume${dir}${localAgent ? ' --local-agent' : ''} ${runId}`; +} + +function shellWord(value: string): string { + return /^[A-Za-z0-9_./=:@%+,-]+$/.test(value) ? value : `'${value.replaceAll("'", "'\\''")}'`; +} diff --git a/packages/sdk/src/authored-node-entry.ts b/packages/sdk/src/authored-node-entry.ts index 3259d697f..fe465bce8 100644 --- a/packages/sdk/src/authored-node-entry.ts +++ b/packages/sdk/src/authored-node-entry.ts @@ -5,7 +5,7 @@ import { JournalClient } from './journal-client.js'; import { executeAuthoredFlow } from './authored-flow-executor.js'; import { loadPinnedAuthoredSource } from './authored-source-authority.js'; import { assertAuthoredNodeVersion, parseAuthoredParentPid } from './authored-runtime-capability.js'; -import { AuthoredFlowExecutionError } from './authored-flow-error.js'; +import { AuthoredFlowExecutionError, AuthoredHumanParked } from './authored-flow-error.js'; import type { AuthoredRootMetadata } from './authored-root.js'; let channelKey: string | undefined, sequence = 0; @@ -82,7 +82,8 @@ try { const prefix = error instanceof AuthoredFlowExecutionError ? `${error.code}: ` : ''; send({ type: 'error', message: prefix && message.startsWith(prefix) ? message.slice(prefix.length) : message, ...(error instanceof AuthoredFlowExecutionError ? { code: error.code, - completionReason: error.completionReason, runId: error.runId } : {}) }); + completionReason: error.completionReason, runId: error.runId } : {}), + ...(error instanceof AuthoredHumanParked ? { wait: error.wait } : {}) }); process.exitCode = 1; } finally { finished = true; await watchdog.terminate(); client?.close(); process.stdin.destroy(); diff --git a/packages/sdk/src/authored-node-runner.ts b/packages/sdk/src/authored-node-runner.ts index d646635ce..8ca12ef86 100644 --- a/packages/sdk/src/authored-node-runner.ts +++ b/packages/sdk/src/authored-node-runner.ts @@ -9,7 +9,11 @@ import type { Readable } from 'node:stream'; import type { AuthoredRootMetadata } from './authored-root.js'; import type { AuthoredExecutionRuntime, AuthoredFlowExecutionResult, ExecuteAuthoredFlowOptions } from './authored-flow-executor.js'; import { completionMarker, isLoweredCompletion } from './authored-flow-executor.js'; -import { AuthoredFlowExecutionError, type AuthoredFlowExecutionErrorCode } from './authored-flow-error.js'; +import { + AuthoredFlowExecutionError, AuthoredHumanParked, + type AuthoredFlowExecutionErrorCode, type AuthoredHumanWait, +} from './authored-flow-error.js'; +import { HUMAN_WAIT_ID } from './authored-human.js'; import { assertAuthoredPromiseHooks } from './authored-runtime-capability.js'; let embeddedSource: string | undefined; @@ -131,10 +135,12 @@ export async function runAuthoredInNode( else if (message.type === 'wait') options.onWait?.(message.event); else if (message.type === 'result') result = { ...message.result, executionRuntime: runtime }; else if (message.type === 'error') { - failure = typeof message.code === 'string' - ? new AuthoredFlowExecutionError(message.code as AuthoredFlowExecutionErrorCode, - message.message, message.completionReason, message.runId) - : new Error(message.message); + failure = message.code === 'human_parked' && isHumanWaitFrame(message.wait) && message.runId === rootRunId + ? new AuthoredHumanParked(message.wait, rootRunId) + : typeof message.code === 'string' + ? new AuthoredFlowExecutionError(message.code as AuthoredFlowExecutionErrorCode, + message.message, message.completionReason, message.runId) + : new Error(message.message); } else throw new Error('unknown authored runtime message'); } catch (error) { stop(error instanceof Error ? error : new Error('invalid authored runtime message')); } } @@ -156,6 +162,15 @@ export async function runAuthoredInNode( } finally { await rm(directory, { recursive: true, force: true }); } } +/** A park signal from the child names the question the parent must journal. */ +function isHumanWaitFrame(value: unknown): value is AuthoredHumanWait { + const wait = value as Partial | null; + return typeof wait === 'object' && wait !== null + && typeof wait.waitId === 'string' && HUMAN_WAIT_ID.test(wait.waitId) + && typeof wait.question === 'string' && wait.question !== '' + && typeof wait.to === 'string' && wait.to !== ''; +} + /** The IPC frame is a claim, not a durable terminal fact or a sandbox boundary. */ export async function verifyAuthoredNodeResult( result: AuthoredFlowExecutionResult, metadata: AuthoredRootMetadata, diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 4fe9ff388..f6e80ceca 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -15,6 +15,8 @@ import type { RunOutcome, StepDispatchEvent } from './protocol.js'; import { SPEC_SCHEMA_VERSION } from './spec.js'; import type { RunLifecycleOptions } from './cli/run.js'; import { withWorkerLease } from './worker-lease.js'; +import { AuthoredHumanParked } from './authored-flow-error.js'; +import { readOpenHumanWaits } from './authored-human.js'; import { isSurfaceCompletionReason } from './authored-step-output.js'; const ROOT_KIND = 'relayflows.authored-root.v1'; @@ -94,7 +96,8 @@ export async function executeDurableAuthoredFlow( // active run but receives no second dispatch from start itself. Resume is // safe for the first caller too: the daemon preserves a live lease and // redelivers only when the former worker connection is gone. - await journal.runResume(outcome.run_id); + const resumed = await journal.runResume(outcome.run_id); + await assertNoOpenHumanWait(journal, resumed); const dispatch = await dispatchWait.promise; return await driveRoot(loaded, metadata, journal, peer, dispatch, options); } finally { @@ -133,6 +136,7 @@ export async function resumeDurableAuthoredFlow( return await completedRootResult(journal, rootRunId); } assertRootCanDispatch(outcome); + await assertNoOpenHumanWait(journal, outcome); const dispatch = await dispatchWait.promise; return await driveRoot(loaded, metadata, journal, peer, dispatch, options); } finally { @@ -141,6 +145,17 @@ export async function resumeDurableAuthoredFlow( } } +/** + * A root parked on an unanswered `f.human` will not be dispatched: the + * kernel holds it in `needs_human` until `event.emit` closes the wait. Report + * the open question instead of waiting for a dispatch that cannot arrive. + */ +async function assertNoOpenHumanWait(journal: JournalClient, outcome: RunOutcome): Promise { + if (outcome.status !== 'parked') return; + const [open] = await readOpenHumanWaits(journal, outcome.run_id); + if (open !== undefined) throw new AuthoredHumanParked(open, outcome.run_id); +} + export async function readAuthoredRootMetadata( journal: JournalClient, runId: string, @@ -206,6 +221,18 @@ async function driveRoot( ); return Object.freeze({ ...result, rootRunId: dispatch.run_id }); } catch (error) { + if (error instanceof AuthoredHumanParked) { + // Not a failure: the body reached a question nobody has answered. Park + // THIS attempt on the kernel's `wait.human` under the body's own wait + // id, so the answer (`event.emit` keyed by it) re-dispatches the root + // and the re-run body finds it. The lease is released by the verb; the + // signal propagates so the CLI reports the question with exit 3. + await peer.stepWait(dispatch.run_id, dispatch.step_id, dispatch.attempt, dispatch.idempotency_key, { + wait_id: error.wait.waitId, prompt: error.wait.question, requested_of: error.wait.to, + options: ['yes', 'no'], + }); + throw error; + } await terminalizeRootFailure(peer, dispatch, error); throw error; } diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts index 023e78e1d..d7f5a9977 100644 --- a/packages/sdk/src/cli.ts +++ b/packages/sdk/src/cli.ts @@ -18,6 +18,7 @@ import { type RunProgress, type RunReport, } from './cli/run.js'; +import { answerFlow } from './cli/answer.js'; import { checkAuthoredTriggers } from './cli/check-triggers.js'; import { parseWebhookArgs, runServeWebhook } from './cli/serve-webhook.js'; import { runDirectFlow } from './cli/direct-run.js'; @@ -65,6 +66,7 @@ type ParsedArgs = | { command: 'check'; json: boolean; watch: boolean; value: string } | { command: 'run'; bucket: string | undefined; reuseFromRunId: string | undefined; localAgent: boolean; dataDir: string; input: string | undefined; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } | { command: 'resume'; localAgent: boolean; dataDir: string; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } + | { command: 'answer'; dataDir: string; json: boolean; spawn: boolean; note: string | undefined; runId: string; waitId: string; answer: boolean } | { command: 'observer'; dataDir: string } | { command: 'hn-monitor'; sub: 'start'; dataDir: string; specPath: string; pollIntervalMs: number | undefined } | { command: 'tick'; sub: 'start'; dataDir: string; specPath: string; scheduleId: string; @@ -93,6 +95,7 @@ const USAGE = [ 'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent] --input ', 'flows tick start --schedule-id --interval-ms [--epoch-ms ] [--max-catch-up ] [--poll-interval-ms ] [--data-dir ] ', 'flows resume [--allow-human-influenced] [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent] ', + 'flows answer [--json] [--no-spawn] [--data-dir ] [--note ] ', 'flows replay [--allow-human-influenced] [--json] [--data-dir ] [--at ]', 'flows observer [--data-dir ]', 'flows hn-monitor start [--data-dir ] [--poll-interval-ms ] ', @@ -142,6 +145,14 @@ export async function runCli( if (parsed.command === 'schedules') return runCloudSchedulesCli(parsed, io); if (parsed.command === 'unschedule') return runCloudUnscheduleCli(parsed, io); if (parsed.command === 'replay') return replayJournal(parsed, io); + if (parsed.command === 'answer') { + const execution = await answerFlow(parsed.runId, parsed.waitId, parsed.answer, parsed.dataDir, { + ...(parsed.note === undefined ? {} : { note: parsed.note }), + daemon: { spawn: parsed.spawn && spawnAllowedByEnv() }, + }); + emitRunReport(execution, parsed.json, io); + return execution.exitCode; + } if (parsed.command === 'build') return runBuild(parsed, io); if (parsed.command === 'deploy') return runDeploy(parsed, io); @@ -479,6 +490,7 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { if (command === 'tick') return parseTickArgs(args.slice(1)); if (command === 'observer') return parseObserverArgs(args.slice(1)); if (command === 'sync') return parseSyncArgs(args.slice(1)); + if (command === 'answer') return parseAnswerArgs(args.slice(1)); if (command !== 'check' && command !== 'run' && command !== 'resume') return undefined; let json = false; @@ -596,6 +608,55 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { : { command, localAgent, dataDir, json, spawn, noObserverLink, allowHumanInfluenced, value: positionals[0]! }; } +/** + * `flows answer [--json] [--no-spawn] [--data-dir ] [--note ] `. + * The answer is a literal `yes`/`no` (also `true`/`false`) so a shell cannot + * hand the kernel an ambiguous word as a decision. + */ +function parseAnswerArgs(rest: readonly string[]): ParsedArgs | undefined { + let json = false; + let spawn = true; + let dataDir = DEFAULT_DATA_DIR; + let sawDataDir = false; + let note: string | undefined; + const positionals: string[] = []; + for (let index = 0; index < rest.length; index += 1) { + const argument = rest[index]!; + if (argument === '--json') { + if (json) return undefined; + json = true; + continue; + } + if (argument === '--no-spawn') { + if (!spawn) return undefined; + spawn = false; + continue; + } + if (argument === '--data-dir') { + const value = rest[index + 1]; + if (sawDataDir || value === undefined || value.startsWith('-')) return undefined; + dataDir = value; + sawDataDir = true; + index += 1; + continue; + } + if (argument === '--note') { + const value = rest[index + 1]; + if (note !== undefined || value === undefined || value.startsWith('--')) return undefined; + note = value; + index += 1; + continue; + } + if (argument.startsWith('-')) return undefined; + positionals.push(argument); + } + if (positionals.length !== 3) return undefined; + const [runId, waitId, word] = positionals as [string, string, string]; + const answer = word === 'yes' || word === 'true' ? true : word === 'no' || word === 'false' ? false : undefined; + if (answer === undefined) return undefined; + return { command: 'answer', dataDir, json, spawn, note, runId, waitId, answer }; +} + function parseHnMonitorArgs(rest: readonly string[]): ParsedArgs | undefined { const sub = rest[0]; if (sub !== 'start') return undefined; @@ -810,6 +871,15 @@ function emitRunReport( return; } if (report.runId === undefined) return; + if (report.answer !== undefined) { + io.stdout(`ANSWERED ${report.runId} ${report.answer.waitId} ${report.answer.answer ? 'yes' : 'no'}` + + (report.answer.note === undefined ? '' : ` (${report.answer.note})`)); + if (report.next !== undefined) io.stdout(`Continue with: ${report.next}`); + return; + } + // A refused answer changed nothing about the run, so there is no run + // outcome to summarize; the refusal above is the whole report. + if (report.command === 'answer') return; const completed = report.completedSteps === undefined ? '' : ` (${report.completedSteps} ${report.completedSteps === 1 ? 'step' : 'steps'})`; const reason = report.completionReason === undefined ? '' diff --git a/packages/sdk/src/cli/answer.ts b/packages/sdk/src/cli/answer.ts new file mode 100644 index 000000000..2206b36db --- /dev/null +++ b/packages/sdk/src/cli/answer.ts @@ -0,0 +1,89 @@ +import { userInfo } from 'node:os'; +import { readAuthoredRootMetadata } from '../authored-root.js'; +import { HUMAN_WAIT_ID, humanAnswerPayload, readHumanAnswer, readOpenHumanWaits, resumeCommand } from '../authored-human.js'; +import { JournalClient, JournalProtocolError } from '../journal-client.js'; +import type { EnsureDaemonOptions } from '../daemon-lifecycle.js'; +import { connect, emptyReport, protocolFailure, socketFor, type RunExecution } from './run.js'; + +export interface AnswerOptions { + readonly note?: string; + readonly daemon?: EnsureDaemonOptions; +} + +/** + * `flows answer yes|no`: record a person's answer to a + * parked `f.human`. The answer is an `event.emit` keyed by the wait id; the + * kernel closes the `wait.human` as `human_responded` and the root becomes + * runnable. Nothing runs here — this process attaches no worker — so the + * report names the `flows resume` that continues the body with the answer. + */ +export async function answerFlow( + runId: string, + waitId: string, + answer: boolean, + dataDir: string, + options: AnswerOptions = {}, +): Promise { + const socketPath = socketFor(dataDir); + const base = emptyReport('answer'); + if (!HUMAN_WAIT_ID.test(waitId)) { + return { exitCode: 2, report: { ...base, runId, socketPath, diagnostics: [{ + severity: 'refusal', kind: 'human_wait_unknown', + message: `"${waitId}" is not an f.human wait id; they are named human- in the order the body asked.`, + }] } }; + } + const client = new JournalClient(socketPath); + const connected = await connect(client, 'answer', dataDir, base, { daemon: options.daemon ?? {} }); + if (connected !== undefined) return connected; + try { + const open = await readOpenHumanWaits(client, runId); + const wait = open.find(candidate => candidate.waitId === waitId); + if (wait === undefined) { + const answered = await readHumanAnswer(client, runId, waitId).catch(() => undefined); + const others = open.map(candidate => `${candidate.waitId} (${candidate.to}: ${JSON.stringify(candidate.question)})`); + return { exitCode: 2, report: { ...base, runId, socketPath, diagnostics: [{ + severity: 'refusal', kind: 'human_wait_unknown', + message: answered !== undefined + ? `Run "${runId}" already has an answer to ${waitId} (${answered.answer ? 'yes' : 'no'}); the kernel closes a wait once.` + : `Run "${runId}" is not asking ${waitId}.` + + (others.length === 0 ? ' It has no open question.' : ` Open: ${others.join(', ')}.`), + }] } }; + } + const payload = humanAnswerPayload(answer, { + ...(options.note === undefined ? {} : { note: options.note }), + answeredBy: safeUsername(), + }); + const emitted = await client.eventEmit(runId, waitId, payload); + if (emitted.matched !== 1) { + return protocolFailure('answer', base, socketPath, new Error( + `event.emit matched ${emitted.matched} waits for ${waitId}; expected the one open question`, + ), runId); + } + const snapshot = await client.runGet(runId); + const root = await readAuthoredRootMetadata(client, runId).catch(() => undefined); + return { + exitCode: 0, + report: { + ...base, ok: true, runId, socketPath, status: snapshot.status, + answer: { waitId, answer, ...(options.note === undefined ? {} : { note: options.note }) }, + humanWait: { waitId, question: wait.question, to: wait.to }, + next: resumeCommand(runId, dataDir, root?.localAgentStream !== undefined), + diagnostics: [], + }, + }; + } catch (error) { + if (error instanceof JournalProtocolError && error.code === 'run_not_found') { + return { exitCode: 2, report: { ...base, runId, socketPath, diagnostics: [{ + severity: 'refusal', kind: 'run_unavailable', + message: `Run "${runId}" is not known to relayflowd at "${socketPath}".`, + }] } }; + } + return protocolFailure('answer', base, socketPath, error, runId); + } finally { + client.close(); + } +} + +function safeUsername(): string { + try { return userInfo().username; } catch { return ''; } +} diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index 2afa1a313..f79a34ee8 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -8,6 +8,7 @@ import { AuthoredFlowExecutionError, } from '../authored-flow-executor.js'; import { executeDurableAuthoredFlow } from '../authored-root.js'; +import { AuthoredHumanParked } from '../authored-flow-error.js'; import { AuthoredFlowLoadError } from '../authored-flow-loader.js'; import { DirectInputError, parseDirectInput } from '../direct-input.js'; import { JournalClient } from '../journal-client.js'; @@ -15,6 +16,7 @@ import { inputFailureReport } from './check.js'; import { checkAuthoredTriggers } from './check-triggers.js'; import { authoredCompletion, + authoredHumanParked, authoredStepFailure, connect, emptyReport, @@ -175,6 +177,9 @@ export async function runDirectFlow( if (error instanceof AuthoredFlowExecutionError && (error.code === 'step_failed' || error.code === 'gate_failed')) { return authoredStepFailure('run', base, socketPath, error); } + if (error instanceof AuthoredHumanParked) { + return authoredHumanParked('run', base, socketPath, error, { dataDir, localAgent: options.localAgent === true }); + } const runId = error instanceof AuthoredFlowExecutionError ? error.runId : undefined; return protocolFailure('run', base, socketPath, error, runId); } finally { diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index 2420987ff..254ac0e29 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -2,7 +2,8 @@ import { parseDigestReference } from '../bundle-transport.js'; import { prepareDigestRun } from './run-digest.js'; import { reuseSummary } from './reuse.js'; import { resumeHelperEffect } from '../authored-helper-effect.js'; -import { AuthoredFlowExecutionError } from '../authored-flow-error.js'; +import { AuthoredFlowExecutionError, AuthoredHumanParked, type AuthoredHumanWait } from '../authored-flow-error.js'; +import { answerCommand, resumeCommand } from '../authored-human.js'; import { join, resolve } from 'node:path'; import type { ProgressEvent } from '../progress.js'; import { toKernelSpec } from '../compile.js'; @@ -29,7 +30,7 @@ import { } from './check.js'; export type RunExitCode = 0 | 1 | 2 | 3; -export type RunCommand = 'run' | 'resume'; +export type RunCommand = 'run' | 'resume' | 'answer'; export interface ParkedStep { id: string; @@ -53,6 +54,12 @@ export interface RunReport { completedSteps?: number; reuse?: { fromRunId: string; reusedSteps: number; executedSteps: number }; parkedStep?: ParkedStep; + /** The open `f.human` question a parked authored run is waiting on. */ + humanWait?: AuthoredHumanWait; + /** `flows answer` only: the answer it recorded. */ + answer?: { waitId: string; answer: boolean; note?: string }; + /** The invocation that continues this run, when one is known. */ + next?: string; projectConfigPath?: string; resolutions: CheckReport['resolutions']; diagnostics: Array; @@ -226,6 +233,9 @@ export async function resumeFlow( if (error instanceof AuthoredFlowExecutionError && (error.code === 'step_failed' || error.code === 'gate_failed')) { return authoredStepFailure('resume', base, socketPath, error, runId); } + if (error instanceof AuthoredHumanParked) { + return authoredHumanParked('resume', base, socketPath, error, { dataDir, localAgent: options.localAgent === true }); + } if (!(error instanceof JournalProtocolError) || error.code !== 'run_not_found') { return protocolFailure('resume', base, socketPath, error, runId); } @@ -298,6 +308,42 @@ export function authoredStepFailure( }; } +/** + * An authored body parked on an unanswered `f.human`: exit 3, like every + * other park, but the diagnostic names the question, who it is for, and the + * exact `flows answer` invocation — the run is waiting on a person, not on a + * worker. Shared by `run` and `resume` so the two never drift. + */ +export function authoredHumanParked( + command: RunCommand, + base: CheckReport | RunReport, + socketPath: string, + error: AuthoredHumanParked, + where: { dataDir?: string; localAgent: boolean }, +): RunExecution { + const runId = error.runId!; + return { + exitCode: 3, + report: { + ...fromBase(command, base), + ok: false, + runId, + socketPath, + status: 'parked', + parkedStep: { id: 'authored-root', type: 'agent' }, + humanWait: error.wait, + diagnostics: [...base.diagnostics, { + severity: 'parked', + kind: 'run_parked', + message: `Run "${runId}" is waiting for ${error.wait.to} to answer ${error.wait.waitId}: ` + + `${JSON.stringify(error.wait.question)}\n` + + `Answer with: ${answerCommand(runId, error.wait.waitId, where.dataDir)}\n` + + `Then continue with: ${resumeCommand(runId, where.dataDir, where.localAgent)}`, + }], + }, + }; +} + /** * An authored body that returned its own terminal verdict, reported as that verdict. * diff --git a/packages/sdk/src/failure-kinds.ts b/packages/sdk/src/failure-kinds.ts index 5cb48c759..6674a7bd1 100644 --- a/packages/sdk/src/failure-kinds.ts +++ b/packages/sdk/src/failure-kinds.ts @@ -113,6 +113,8 @@ export const RUN_FAILURE_KINDS = [ 'run_unavailable', /** A predicate `.gate(fn)` judged false; the verdict is journaled as `.gate`. */ 'gate_failed', + /** `flows answer` named a wait the run is not asking: unknown, or already answered. */ + 'human_wait_unknown', ] as const; /** diff --git a/packages/sdk/src/journal-client.ts b/packages/sdk/src/journal-client.ts index 0c8e32868..f349fdbb3 100644 --- a/packages/sdk/src/journal-client.ts +++ b/packages/sdk/src/journal-client.ts @@ -386,6 +386,26 @@ export class JournalClient extends EventEmitter { } /** Satisfy `wait.event`; a human response arrives here too. */ + /** + * Park the dispatched attempt this connection holds on a human question. + * Releases the lease; the caller must not heartbeat or complete afterwards. + */ + stepWait( + runId: string, + stepId: string, + attempt: number, + idempotencyKey: string, + wait: { wait_id: string; prompt: string; requested_of: string; options?: string[]; timeout_at_ms?: number }, + ): Promise { + return this.request('step.wait', { + run_id: runId, + step_id: stepId, + attempt, + idempotency_key: idempotencyKey, + ...wait, + }); + } + eventEmit(runId: string, eventKey: string, payload: unknown): Promise { return this.request('event.emit', { run_id: runId, event_key: eventKey, payload }); } diff --git a/packages/sdk/src/progress.ts b/packages/sdk/src/progress.ts index 2839f21a7..209abf197 100644 --- a/packages/sdk/src/progress.ts +++ b/packages/sdk/src/progress.ts @@ -1,8 +1,10 @@ import type { CompletionReason } from './protocol.js'; import type { StepType } from './spec.js'; +import { AuthoredHumanParked } from './authored-flow-error.js'; export interface ProgressEvent { - type: 'step.started' | 'step.running' | 'step.completed' | 'step.failed'; + /** `step.parked`: an `f.human` reached without an answer; the run parks, nothing failed. */ + type: 'step.started' | 'step.running' | 'step.completed' | 'step.failed' | 'step.parked'; stepId: string; stepType: StepType; elapsedMs: number; @@ -12,13 +14,15 @@ export interface ProgressEvent { /** Pure terminal rendering: caller owns the event source, clock, and output. */ export function renderProgress(events: Iterable): string[] { return Array.from(events, event => { - const icon = { 'step.started': '○', 'step.running': '↻', 'step.completed': '✓', 'step.failed': '✗' }[event.type]; + const icon = { 'step.started': '○', 'step.running': '↻', 'step.completed': '✓', 'step.failed': '✗', 'step.parked': '⏸' }[event.type]; const state = event.type.slice('step.'.length); + // A parked `f.human` is a question, not a step type the kernel ran. + const kind = event.type === 'step.parked' ? 'human' : event.stepType; const agent = event.stepType === 'agent' ? ` [agent: ${state === 'started' ? 'preparing' : state}]` : ''; const reason = event.completionReason ? ` completionReason: ${event.completionReason}` : ''; // Agent-authored names cannot inject terminal control sequences. const name = event.stepId.replace(/[\x00-\x1f\x7f-\x9f]/g, '?'); - return `${icon} ${name} (${event.stepType})${agent} ${(Math.max(0, event.elapsedMs) / 1000).toFixed(2)}s${reason}`; + return `${icon} ${name} (${kind})${agent} ${(Math.max(0, event.elapsedMs) / 1000).toFixed(2)}s${reason}`; }); } @@ -45,7 +49,10 @@ export async function observeStep( publish({ type: 'step.completed', stepId, stepType, elapsedMs: performance.now() - started, completionReason: 'success' }); return result; } catch (error) { - publish({ type: 'step.failed', stepId, stepType, elapsedMs: performance.now() - started }); + publish({ + type: error instanceof AuthoredHumanParked ? 'step.parked' : 'step.failed', + stepId, stepType, elapsedMs: performance.now() - started, + }); throw error; } } diff --git a/packages/sdk/src/protocol.ts b/packages/sdk/src/protocol.ts index 89105709c..94d6e7202 100644 --- a/packages/sdk/src/protocol.ts +++ b/packages/sdk/src/protocol.ts @@ -52,6 +52,7 @@ export type Verb = | 'effect.record' | 'effect.confirm' | 'step.complete' + | 'step.wait' | 'event.emit' | 'event.submit' | 'stream.append' @@ -329,6 +330,27 @@ export interface StepCompleteParams { } export type StepCompleteResult = RunOutcome; +/** + * Park a leased attempt on a durable human question (`wait.human`, kernel + * DESIGN.md §1.5). The attempt is not completed and no iteration is charged; + * the lease is released once the wait is journaled. The answer arrives through + * `event.emit` with `event_key` = `wait_id`, which closes the wait as + * `human_responded` and makes the step runnable for a fresh attempt. + */ +export interface StepWaitParams { + run_id: string; + step_id: string; + attempt: number; + idempotency_key: string; + /** Names the question for the life of the run; the kernel refuses a reuse. */ + wait_id: string; + prompt: string; + requested_of: string; + options?: string[]; + timeout_at_ms?: number; +} +export type StepWaitResult = RunOutcome; + export interface EventEmitParams { run_id: string; event_key: string; @@ -401,6 +423,7 @@ export interface VerbContract { 'effect.record': { params: EffectRecordParams; result: EffectRecordResult }; 'effect.confirm': { params: EffectConfirmParams; result: EffectConfirmResult }; 'step.complete': { params: StepCompleteParams; result: StepCompleteResult }; + 'step.wait': { params: StepWaitParams; result: StepWaitResult }; 'event.emit': { params: EventEmitParams; result: EventEmitResult }; 'event.submit': { params: EventSubmitParams; result: EventSubmitResult }; 'stream.append': { params: StreamAppendParams; result: StreamAppendResult }; diff --git a/packages/sdk/tests/authored-flow.test.ts b/packages/sdk/tests/authored-flow.test.ts index c3a11dae7..ebe1fb449 100644 --- a/packages/sdk/tests/authored-flow.test.ts +++ b/packages/sdk/tests/authored-flow.test.ts @@ -264,10 +264,6 @@ describe('authored flow journal executor', () => { it('refuses unsupported promise verbs synchronously even when their results are ignored', async () => { const disconnectedJournal = new JournalClient('/journal-must-not-be-contacted'); const cases = [ - flow('unawaited-human', async (f) => { - f.human('approve?', { to: 'owner' }); - f.done('success'); - }), flow('unawaited-dispatch', async (f) => { f.dispatch('child', {}); f.done('success'); diff --git a/packages/sdk/tests/authored-human.test.ts b/packages/sdk/tests/authored-human.test.ts new file mode 100644 index 000000000..35423f46a --- /dev/null +++ b/packages/sdk/tests/authored-human.test.ts @@ -0,0 +1,200 @@ +import { rmSync } from 'node:fs'; +import type { Server } from 'node:net'; +import { flow } from '@relayflows/surface'; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { executeAuthoredFlow } from '../src/authored-flow-executor.js'; +import { AuthoredHumanParked } from '../src/authored-flow-error.js'; +import { humanAnswerPayload, parseHumanAnswer, readOpenHumanWaits } from '../src/authored-human.js'; +import { JournalClient } from '../src/journal-client.js'; +import { sendOk, sendResult, sockPath, startLoopback } from './journal-client-loopback.js'; + +/** + * `f.human` against a loopback kernel. The ROOT journal is scripted per test: + * what `journal.read` returns for the root run is the only thing that decides + * whether the body parks or continues, exactly as with the real daemon. + */ +describe('f.human lowering', () => { + const ROOT = 'root-run'; + let path: string; + let server: Server; + let rootEntries: Array> = []; + const started: Array<{ id: string; command: string; admissionKey?: string }> = []; + const outputs = new Map(); + const stepOf = new Map(); + let nextRun = 1; + + beforeAll(() => { + path = sockPath(); + server = startLoopback(path, { + hello: (ctx) => sendOk(ctx), + 'run.start': (ctx, params) => { + const spec = params.spec as { steps: Array<{ id: string; command: string }> }; + const step = spec.steps[0]!; + const runId = `child-${nextRun++}`; + started.push({ id: step.id, command: step.command, ...(typeof params.admission_key === 'string' ? { admissionKey: params.admission_key } : {}) }); + // The fake "executes" printf '%s' '' by unquoting the literal. + const printed = /^printf '%s' '((?:[^']|'\\'')*)'$/.exec(step.command)?.[1]?.replaceAll("'\\''", "'") ?? ''; + outputs.set(runId, printed); + stepOf.set(runId, step.id); + sendResult(ctx, { run_id: runId, status: 'completed', completion_reason: 'success', completed_steps: 1 }); + }, + 'journal.read': (ctx, params) => { + const runId = params.run_id as string; + if (runId === ROOT) { + const from = (params.from_seq as number) ?? 1; + sendResult(ctx, { entries: rootEntries.map((entry, index) => ({ seq: index + 1, ...entry })).filter(entry => entry.seq >= from) }); + return; + } + sendResult(ctx, { entries: [{ + entry_type: 'step.completed', step_id: stepOf.get(runId), + payload: { completionReason: 'success', disposition: 'step_done', + output: { exit_code: 0, stdout_tail: outputs.get(runId) ?? '', stderr_tail: '' } }, + }] }); + }, + 'stream.read': (ctx) => sendResult(ctx, { messages: [], next_offset: 0 }), + }); + }); + + afterAll(async () => { + await new Promise((resolve) => server.close(() => resolve())); + rmSync(path, { force: true }); + }); + + beforeEach(() => { rootEntries = []; started.length = 0; }); + + async function client(): Promise { + const journal = new JournalClient(path, { requestTimeoutMs: 2000 }); + await journal.connect(); + await journal.hello('authored-human-test'); + return journal; + } + + const asked = { entry_type: 'wait.human', step_id: 'authored-root', attempt: 1, + payload: { wait_id: 'human-1', prompt: 'Ship it?', requested_of: 'khaliq', options: ['yes', 'no'] } }; + const answered = (result: unknown) => ({ entry_type: 'wait.completed', step_id: 'authored-root', attempt: 1, + payload: { wait_id: 'human-1', completionReason: 'human_responded', result } }); + + it('parks the body on an unanswered question without starting a child run', async () => { + const journal = await client(); + try { + const handle = flow('ask', async f => { + const ok = await f.human('Ship it?', { to: 'khaliq' }); + f.done(ok ? 'success' : 'declined'); + }); + const failure = await executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT }).catch(error => error); + expect(failure).toBeInstanceOf(AuthoredHumanParked); + expect(failure).toMatchObject({ code: 'human_parked', runId: ROOT, + wait: { waitId: 'human-1', question: 'Ship it?', to: 'khaliq' } }); + expect(started).toEqual([]); + } finally { journal.close(); } + }); + + it('still parks while the question is asked but not yet answered', async () => { + rootEntries = [asked]; + const journal = await client(); + try { + const handle = flow('ask', async f => { await f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); + await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toBeInstanceOf(AuthoredHumanParked); + expect(started).toEqual([]); + expect(await readOpenHumanWaits(journal, ROOT)).toEqual([ + { waitId: 'human-1', question: 'Ship it?', to: 'khaliq', stepId: 'authored-root', attempt: 1 }, + ]); + } finally { journal.close(); } + }); + + it('continues with the recorded answer, lowered as a memoized human-N step', async () => { + rootEntries = [asked, answered({ answer: true, note: 'lgtm', answeredBy: 'khaliq', at: '2026-09-18T00:00:00Z' })]; + const journal = await client(); + try { + const handle = flow('ask', async f => { + const ok = await f.human('Ship it?', { to: 'khaliq' }); + f.done(ok ? 'success' : 'declined'); + }); + const result = await executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT }); + expect(result.completionReason).toBe('success'); + expect(result.journalSteps.map(step => step.id)).toEqual(['human-1', 'complete-2']); + expect(started[0]).toMatchObject({ id: 'human-1' }); + expect(started[0]!.admissionKey).toMatch(/^[a-z0-9:_-]+$/i); + expect(JSON.parse(outputs.get('child-1')!)).toEqual({ human: 'human-1', to: 'khaliq', answer: true, note: 'lgtm', answeredBy: 'khaliq' }); + expect(await readOpenHumanWaits(journal, ROOT)).toEqual([]); + } finally { journal.close(); } + }); + + it('a negative answer is a value, not a failure', async () => { + rootEntries = [asked, answered({ answer: false })]; + const journal = await client(); + try { + const handle = flow('ask', async f => { + const ok = await f.human('Ship it?', { to: 'khaliq' }); + f.done(ok ? 'success' : 'declined'); + }); + expect((await executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).completionReason).toBe('declined'); + } finally { journal.close(); } + }); + + it('asks the second question only after the first is answered, under its own id', async () => { + rootEntries = [asked, answered({ answer: true })]; + const journal = await client(); + try { + const handle = flow('ask-twice', async f => { + await f.human('Ship it?', { to: 'khaliq' }); + await f.human('Announce it?', { to: 'marketing' }); + f.done('success'); + }); + const failure = await executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT }).catch(error => error); + expect(failure).toMatchObject({ wait: { waitId: 'human-2', question: 'Announce it?', to: 'marketing' } }); + expect(started.map(step => step.id)).toEqual(['human-1']); + } finally { journal.close(); } + }); + + it('refuses a recorded answer that is not { answer: boolean }', async () => { + rootEntries = [asked, answered({ answer: 'yes' })]; + const journal = await client(); + try { + const handle = flow('ask', async f => { await f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); + await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toMatchObject({ code: 'human_answer_invalid' }); + } finally { journal.close(); } + }); + + it('refuses an empty question or an unnamed recipient before touching the journal', async () => { + const journal = new JournalClient('/journal-must-not-be-contacted'); + for (const handle of [ + flow('blank', async f => { await f.human(' ', { to: 'khaliq' }); f.done('success'); }), + flow('nobody', async f => { await f.human('Ship it?', { to: '' }); f.done('success'); }), + ]) { + await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toMatchObject({ code: 'human_answer_invalid' }); + } + }); + + it('needs a durable root to park in', async () => { + const journal = await client(); + try { + const handle = flow('rootless', async f => { await f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); + await expect(executeAuthoredFlow(handle, journal)).rejects.toMatchObject({ code: 'unsupported_verb' }); + } finally { journal.close(); } + }); + + it('an unawaited f.human is an unawaited step', async () => { + rootEntries = [asked, answered({ answer: true })]; + const journal = await client(); + try { + const handle = flow('fire-and-forget', async f => { f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); + await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toMatchObject({ code: 'unawaited_step' }); + } finally { journal.close(); } + }); +}); + +describe('the answer contract', () => { + it('is { answer, note?, answeredBy?, at }', () => { + const payload = humanAnswerPayload(true, { note: 'ok', answeredBy: 'khaliq' }); + expect(payload).toMatchObject({ answer: true, note: 'ok', answeredBy: 'khaliq' }); + expect(Date.parse(payload.at!)).not.toBeNaN(); + expect(humanAnswerPayload(false, { note: '' })).not.toHaveProperty('note'); + expect(parseHumanAnswer(payload, 'human-1')).toEqual(payload); + }); + it('refuses every other shape', () => { + for (const bad of [null, 'yes', { answer: 'yes' }, { answer: true, note: 3 }, {}]) { + expect(() => parseHumanAnswer(bad, 'human-1')).toThrow(/human_answer_invalid/); + } + }); +}); diff --git a/packages/sdk/tests/authored-node-runtime.test.ts b/packages/sdk/tests/authored-node-runtime.test.ts index 7efc543e8..3771d178d 100644 --- a/packages/sdk/tests/authored-node-runtime.test.ts +++ b/packages/sdk/tests/authored-node-runtime.test.ts @@ -123,6 +123,41 @@ f.done('success');`); const resumed = f.invoke(['resume', report.runId]); expect(resumed.status, resumed.stderr + resumed.stdout).toBe(0); }, 90_000); + it('parks an f.human across the IPC boundary, answers it, and resumes the Node body with the answer', async () => { + const f = fixture(`await f.agent('worker',{task:'local fixture'}); +const ok = await f.human('Ship it?', { to: 'khaliq' }); +appendFileSync('answers', String(ok)); +f.done(ok ? 'success' : 'declined');`); + const first = f.run(); expect(first.status, first.stderr + first.stdout).toBe(3); + const parked = JSON.parse(first.stdout); + expect(parked).toMatchObject({ ok: false, status: 'parked', humanWait: { waitId: 'human-2', question: 'Ship it?', to: 'khaliq' } }); + // The Bun parent, not the Node child, holds the lease: the park signal + // crossed the authenticated frame and became the kernel's wait.human. + const asked = await entries(f.directory, parked.runId); + expect(asked.filter(e => e.entry_type === 'wait.human').map(e => e.payload)).toEqual([ + expect.objectContaining({ wait_id: 'human-2', prompt: 'Ship it?', requested_of: 'khaliq' }), + ]); + expect(asked.filter(e => e.entry_type === 'step.completed')).toHaveLength(0); + expect(existsSync(join(f.directory, 'answers'))).toBe(false); + + // `answer` attaches no worker, so the worker flags do not apply to it. + const answered = spawnSync(cli, ['answer', parked.runId, 'human-2', 'yes', '--data-dir', join(f.directory, 'data'), '--json'], + { cwd: f.directory, env: f.env, encoding: 'utf8', timeout: 60_000 }); + expect(answered.status, answered.stderr + answered.stdout).toBe(0); + const resumed = f.invoke(['resume', parked.runId]); expect(resumed.status, resumed.stderr + resumed.stdout).toBe(0); + expect(JSON.parse(resumed.stdout)).toMatchObject({ ok: true, runId: parked.runId, completionReason: 'success' }); + // The agent ran once; the answer reached author code once; the human + // step is verified as a durable child like every other. + expect(readFileSync(join(f.directory, 'agent-effects'), 'utf8')).toBe('once\n'); + expect(readFileSync(join(f.directory, 'answers'), 'utf8')).toBe('true'); + const rootEntries = await entries(f.directory, parked.runId); + expect(rootEntries.filter(e => e.entry_type === 'step.attempt.started')).toHaveLength(2); + expect(rootEntries.find(e => e.entry_type === 'wait.completed')!.payload).toMatchObject({ wait_id: 'human-2', completionReason: 'human_responded', result: { answer: true } }); + const output = rootEntries.find(e => e.entry_type === 'step.completed')!.payload['output']; + expect(output.executionRuntime).toMatchObject({ kind: 'node' }); + expect(output.journalSteps.map((s: { id: string }) => s.id)).toEqual(['agent-1', 'human-2', 'complete-3']); + }, 120_000); + it.each([ ['SIGKILL', 'success'], ['SIGTERM', 'success'], ['blocked-SIGKILL', 'success'], ['SIGKILL', 'declined'], diff --git a/packages/sdk/tests/authored-root.test.ts b/packages/sdk/tests/authored-root.test.ts index d67e9e5f7..b790ca0a9 100644 --- a/packages/sdk/tests/authored-root.test.ts +++ b/packages/sdk/tests/authored-root.test.ts @@ -14,6 +14,7 @@ import { } from '../src/authored-root.js'; import type { JournalClient } from '../src/journal-client.js'; import type { RunOutcome, StepDispatchEvent } from '../src/protocol.js'; +import { AuthoredHumanParked } from '../src/authored-flow-error.js'; vi.mock('../src/authored-flow-loader.js', async importOriginal => ({ ...await importOriginal(), @@ -30,6 +31,7 @@ const surface = Object.freeze({ class RootPeer extends EventEmitter { readonly completions: Array<{ attempt: number; reason: string }> = []; + readonly waits: Array<{ attempt: number; wait: Record }> = []; heartbeats = 0; async connect() {} @@ -41,6 +43,17 @@ class RootPeer extends EventEmitter { } close() {} + async stepWait( + runId: string, + _stepId: string, + attempt: number, + _idempotencyKey: string, + wait: Record, + ): Promise { + this.waits.push({ attempt, wait }); + return outcome(runId, 'parked', null); + } + async stepComplete( runId: string, _stepId: string, @@ -92,8 +105,12 @@ class RootJournal { } return this.resumeStatus; } - async journalRead(runId: string): Promise<{ entries: Array> }> { - return { entries: this.childEntries.get(runId) ?? this.entries }; + async journalRead(runId: string, fromSeq = 1): Promise<{ entries: Array> }> { + // Numbered like the kernel's `journal.read`, so paginating readers + // (`readOpenHumanWaits`) terminate against the fake too. + const entries = (this.childEntries.get(runId) ?? this.entries) + .map((entry, index) => ({ seq: index + 1, ...entry })); + return { entries: entries.filter(entry => entry.seq >= fromSeq) }; } } @@ -232,6 +249,71 @@ describe('durable authored root', () => { expect(journal.peer.completions).toEqual([{ attempt: 2, reason: 'success' }]); }); + it('parks the root attempt on an unanswered f.human instead of completing it', async () => { + const loaded = await fixture(false, 0, async f => { await f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); + const journal = new RootJournal(); + + const failure = await executeDurableAuthoredFlow( + loaded, journal as unknown as JournalClient, undefined, + { dataDir: '/unused', admissionKey: 'ask' }, + ).catch(error => error); + + expect(failure).toBeInstanceOf(AuthoredHumanParked); + expect((failure as AuthoredHumanParked).wait).toEqual({ waitId: 'human-1', question: 'Ship it?', to: 'khaliq' }); + // The attempt asked a question: `step.wait` under the body's own wait id, + // never `step.complete`, and never the worker_error retry ladder. + expect(journal.peer.waits).toEqual([{ attempt: 1, wait: { + wait_id: 'human-1', prompt: 'Ship it?', requested_of: 'khaliq', options: ['yes', 'no'], + } }]); + expect(journal.peer.completions).toEqual([]); + }); + + it('reports the open question on resume instead of waiting for a dispatch that cannot come', async () => { + const loaded = await fixture(); + const journal = new RootJournal(); + journal.entries = [spawnedEntry(loaded), { + entry_type: 'wait.human', step_id: 'authored-root', attempt: 1, + payload: { wait_id: 'human-1', prompt: 'Ship it?', requested_of: 'khaliq' }, + }]; + journal.resumeStatus = outcome(journal.runId, 'parked', null); + // A parked root is not dispatched by the kernel; the fake must not either. + journal.runResume = async function (this: RootJournal) { this.resumeCalls += 1; return this.resumeStatus; }; + vi.mocked(loadAuthoredFlow).mockResolvedValue(loaded); + + const failure = await resumeDurableAuthoredFlow( + journal.runId, journal as unknown as JournalClient, { dataDir: '/unused' }, + ).catch(error => error); + + expect(failure).toBeInstanceOf(AuthoredHumanParked); + expect((failure as AuthoredHumanParked).wait).toMatchObject({ waitId: 'human-1', question: 'Ship it?', to: 'khaliq' }); + expect(journal.peer.completions).toEqual([]); + }); + + it('resumes past an answered question: the wait is closed, the body re-runs and completes', async () => { + const loaded = await fixture(false, 0, async f => { + const ok = await f.human('Ship it?', { to: 'khaliq' }); + f.done(ok ? 'success' : 'declined'); + }); + const journal = new RootJournal(); + journal.entries = [spawnedEntry(loaded), { + entry_type: 'wait.human', step_id: 'authored-root', attempt: 1, + payload: { wait_id: 'human-1', prompt: 'Ship it?', requested_of: 'khaliq' }, + }, { + entry_type: 'wait.completed', step_id: 'authored-root', attempt: 1, + payload: { wait_id: 'human-1', completionReason: 'human_responded', result: { answer: false } }, + }]; + vi.mocked(loadAuthoredFlow).mockResolvedValue(loaded); + + const result = await resumeDurableAuthoredFlow( + journal.runId, journal as unknown as JournalClient, { dataDir: '/unused' }, + ); + + expect(result).toMatchObject({ rootRunId: 'root-run', completionReason: 'declined' }); + expect(result!.journalSteps.map(step => step.id)).toEqual(['human-1', 'complete-2']); + expect(journal.peer.waits).toEqual([]); + expect(journal.peer.completions).toEqual([{ attempt: 2, reason: 'success' }]); + }); + it('fails closed on malformed metadata under the reserved authored-root stream', async () => { const journal = new RootJournal(); journal.entries = [{ @@ -246,17 +328,23 @@ describe('durable authored root', () => { }); }); -async function fixture(fails = false, delayMs = 0): Promise { +async function fixture( + fails = false, + delayMs = 0, + body?: Parameters[1], +): Promise { const directory = await mkdtemp(join(tmpdir(), 'authored-root-')); directories.push(directory); const sourcePath = join(directory, 'flagship.flow.ts'); await writeFile(sourcePath, 'export default "exact source";\n'); - const handle = fails - ? flow('flagship', async () => { throw new Error('child failed'); }) - : flow('flagship', async f => { - if (delayMs > 0) await new Promise(resolve => setTimeout(resolve, delayMs)); - f.done('success'); - }); + const handle = body !== undefined + ? flow('flagship', body) + : fails + ? flow('flagship', async () => { throw new Error('child failed'); }) + : flow('flagship', async f => { + if (delayMs > 0) await new Promise(resolve => setTimeout(resolve, delayMs)); + f.done('success'); + }); const getDefinition = getFlowDefinition as LoadedAuthoredFlow['getDefinition']; return { sourcePath, handle, getDefinition, surfaceAuthority: surface, diff --git a/packages/sdk/tests/human-live.test.ts b/packages/sdk/tests/human-live.test.ts new file mode 100644 index 000000000..8a3b11904 --- /dev/null +++ b/packages/sdk/tests/human-live.test.ts @@ -0,0 +1,166 @@ +import { execFileSync, spawnSync } from 'node:child_process'; +import { existsSync, mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { JournalClient } from '../src/journal-client.js'; +import { socketFor } from '../src/cli/run.js'; + +const roots: string[] = []; +const sdk = resolve('.'); +const cli = process.env['FLOWS_TEST_CLI'] ?? join(sdk, 'dist/cli.js'); + +function resolveDaemon(): string { + if (process.env['RELAYFLOWD_BIN']) return process.env['RELAYFLOWD_BIN']; + try { + return join(JSON.parse(execFileSync('sh', [ + resolve('../../ops/cargo.sh'), 'metadata', '--format-version=1', '--no-deps', '--locked', '--offline', + ], { cwd: resolve('../../kernel'), encoding: 'utf8', + env: { ...process.env, RELAYFLOWS_NO_TOOLCHAIN_INSTALL: '1' }, + })).target_directory, 'debug', 'relayflowd'); + } catch (cause) { + throw new Error('Live CLI tests require npm run test:prep or an explicit RELAYFLOWD_BIN.', { cause }); + } +} + +afterEach(() => { + for (const root of roots.splice(0)) { + const connection = join(root, 'data/connection.json'); + if (existsSync(connection)) { + const { pid } = JSON.parse(readFileSync(connection, 'utf8')); + if (typeof pid === 'number') { + try { process.kill(pid, 'SIGTERM'); } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ESRCH') throw error; + } + } + } + rmSync(root, { recursive: true, force: true }); + } +}); + +function fixture() { + const relayflowd = resolveDaemon(); + const root = mkdtempSync(join(tmpdir(), 'flows-human-')); + roots.push(root); + symlinkSync(join(sdk, 'node_modules'), join(root, 'node_modules')); + writeFileSync(join(root, 'package.json'), '{"type":"module"}'); + writeFileSync(join(root, 'ask.flow.ts'), `import { flow } from '@relayflows/surface'; +export default flow('ask', async (f, input: { plan: string }) => { + const summary = await f.run(\`printf '%s' "plan: \${input.plan}"\`); + const ok = await f.human(\`Ship this?\\n\${summary}\`, { to: 'khaliq' }); + if (!ok) return f.done('declined'); + await f.run("printf '%s' shipped >> effects"); + f.done('success'); +}); +`); + const dataDir = join(root, 'data'); + const invoke = (...args: string[]) => spawnSync(process.execPath, + [cli, ...args, '--data-dir', dataDir, '--json'], + { cwd: root, encoding: 'utf8', timeout: 90_000, env: { ...process.env, RELAYFLOWD_BIN: relayflowd } }); + return { + root, dataDir, + run: () => invoke('run', 'ask.flow.ts', '--input', '{"plan":"v2"}', '--no-observer-link'), + resume: (runId: string) => invoke('resume', runId, '--no-observer-link'), + answer: (runId: string, waitId: string, word: string, ...extra: string[]) => invoke('answer', runId, waitId, word, ...extra), + text: (...args: string[]) => spawnSync(process.execPath, [cli, ...args, '--data-dir', dataDir], + { cwd: root, encoding: 'utf8', timeout: 90_000, env: { ...process.env, RELAYFLOWD_BIN: relayflowd } }), + }; +} + +async function waitEntries(dataDir: string, runId: string) { + const client = new JournalClient(socketFor(dataDir)); + await client.connect(); await client.hello('human-live-test'); + try { + const entries = (await client.journalRead(runId, 1, 1000)).entries as Array<{ entry_type: string; step_id?: string; attempt?: number; payload: Record }>; + return entries.filter(entry => entry.entry_type.startsWith('wait.') || entry.entry_type === 'step.attempt.started' || entry.entry_type === 'step.completed'); + } finally { client.close(); } +} + +describe('f.human against a real daemon', () => { + it('parks with the question, refuses wrong answers, records one, and resumes to success', async () => { + const f = fixture(); + const first = f.run(); + expect(first.status, first.stderr + first.stdout).toBe(3); + const parked = JSON.parse(first.stdout); + expect(parked).toMatchObject({ ok: false, status: 'parked', + parkedStep: { id: 'authored-root', type: 'agent' }, + humanWait: { waitId: 'human-2', question: 'Ship this?\nplan: v2', to: 'khaliq' } }); + const runId: string = parked.runId; + expect(parked.diagnostics.at(-1)).toMatchObject({ severity: 'parked', kind: 'run_parked' }); + expect(parked.diagnostics.at(-1).message).toContain(`flows answer --data-dir ${f.dataDir} ${runId} human-2 yes|no`); + + // The kernel holds the question: a wait.human under the root attempt, no completion. + const asked = await waitEntries(f.dataDir, runId); + expect(asked.map(entry => entry.entry_type)).toEqual(['step.attempt.started', 'wait.human']); + expect(asked[1]!.payload).toMatchObject({ wait_id: 'human-2', prompt: 'Ship this?\nplan: v2', requested_of: 'khaliq', options: ['yes', 'no'] }); + + // Resuming before anyone answers reports the same open question. + const early = f.resume(runId); + expect(early.status, early.stderr + early.stdout).toBe(3); + expect(JSON.parse(early.stdout).humanWait).toMatchObject({ waitId: 'human-2' }); + + const wrong = f.answer(runId, 'human-1', 'yes'); + expect(wrong.status, wrong.stderr + wrong.stdout).toBe(2); + expect(JSON.parse(wrong.stdout).diagnostics[0]).toMatchObject({ kind: 'human_wait_unknown' }); + expect(wrong.stderr).toContain('Open: human-2'); + + const garbage = f.answer(runId, 'human-2', 'maybe'); + expect(garbage.status).toBe(2); + + const answered = f.answer(runId, 'human-2', 'yes', '--note', 'looks good'); + expect(answered.status, answered.stderr + answered.stdout).toBe(0); + expect(JSON.parse(answered.stdout)).toMatchObject({ ok: true, command: 'answer', runId, + answer: { waitId: 'human-2', answer: true, note: 'looks good' } }); + expect(JSON.parse(answered.stdout).next).toBe(`flows resume --data-dir ${f.dataDir} ${runId}`); + + const twice = f.answer(runId, 'human-2', 'no'); + expect(twice.status, twice.stderr + twice.stdout).toBe(2); + expect(twice.stderr).toContain('already has an answer to human-2 (yes)'); + + const resumed = f.resume(runId); + expect(resumed.status, resumed.stderr + resumed.stdout).toBe(0); + expect(JSON.parse(resumed.stdout)).toMatchObject({ ok: true, runId, status: 'completed', completionReason: 'success', completedSteps: 4 }); + expect(readFileSync(join(f.root, 'effects'), 'utf8')).toBe('shipped'); + + const closed = await waitEntries(f.dataDir, runId); + expect(closed.map(entry => `${entry.entry_type}@${entry.attempt}`)).toEqual([ + 'step.attempt.started@1', 'wait.human@1', 'wait.completed@1', 'step.attempt.started@2', 'step.completed@2', + ]); + expect(closed[2]!.payload).toMatchObject({ wait_id: 'human-2', completionReason: 'human_responded', + result: { answer: true, note: 'looks good' } }); + expect(closed[4]!.payload).toMatchObject({ completionReason: 'success' }); + // A resume after completion replays nothing and reports the durable result. + const again = f.resume(runId); + expect(again.status, again.stderr + again.stdout).toBe(0); + expect(readFileSync(join(f.root, 'effects'), 'utf8')).toBe('shipped'); + }, 120_000); + + it('a "no" is a value the body branches on: declined, exit 0, no effect', () => { + const f = fixture(); + const first = f.text('run', 'ask.flow.ts', '--input', '{"plan":"v2"}', '--no-observer-link'); + expect(first.status, first.stderr + first.stdout).toBe(3); + // The question is rendered as a park, never as a failed step. + expect(first.stderr).toContain('⏸ human-2 (human)'); + expect(first.stderr).not.toContain('✗'); + expect(first.stderr).toContain('PARKED [run_parked]'); + const runId = /^RUN (\S+) parked$/m.exec(first.stdout)![1]!; + const answered = f.text('answer', runId, 'human-2', 'no'); + expect(answered.status, answered.stderr + answered.stdout).toBe(0); + expect(answered.stdout).toContain(`ANSWERED ${runId} human-2 no`); + expect(answered.stdout).toContain('Continue with: flows resume'); + const resumed = f.resume(runId); + expect(resumed.status, resumed.stderr + resumed.stdout).toBe(0); + expect(JSON.parse(resumed.stdout)).toMatchObject({ ok: true, completionReason: 'success', + diagnostics: [{ kind: 'run_declined' }] }); + expect(existsSync(join(f.root, 'effects'))).toBe(false); + }, 120_000); + + it('refuses to answer a run the daemon does not know', () => { + const f = fixture(); + // Start the daemon with a real run first so the refusal is the run, not the socket. + expect(f.run().status).toBe(3); + const missing = f.answer('01UNKNOWNRUN00000000000000', 'human-1', 'yes'); + expect(missing.status, missing.stderr + missing.stdout).toBe(2); + expect(JSON.parse(missing.stdout).diagnostics[0]).toMatchObject({ kind: 'run_unavailable' }); + }, 120_000); +}); diff --git a/packages/sdk/tests/journal-client-loopback.ts b/packages/sdk/tests/journal-client-loopback.ts index 84df23336..d5c13d836 100644 --- a/packages/sdk/tests/journal-client-loopback.ts +++ b/packages/sdk/tests/journal-client-loopback.ts @@ -35,6 +35,7 @@ export interface LoopbackHandlers { 'effect.record'?: (ctx: FrameCtx, params: Record) => void; 'effect.confirm'?: (ctx: FrameCtx, params: Record) => void; 'step.complete'?: (ctx: FrameCtx, params: Record) => void; + 'step.wait'?: (ctx: FrameCtx, params: Record) => void; 'event.emit'?: (ctx: FrameCtx, params: Record) => void; 'journal.read'?: (ctx: FrameCtx, params: Record) => void; 'stream.append'?: (ctx: FrameCtx, params: Record) => void; diff --git a/packages/surface/src/context.ts b/packages/surface/src/context.ts index eb23bf019..38b48ce74 100644 --- a/packages/surface/src/context.ts +++ b/packages/surface/src/context.ts @@ -63,7 +63,14 @@ export interface Ctx extends Helpers { /** JSON Schema validates the value at runtime; narrow unknown in author code. */ llm(prompt: string, options: LlmOptions): Step; agent(name: string, options: AgentOptions): Step; - human(question: string, options: { to: string }): Promise; + /** + * Ask a person a yes/no question and park until they answer. The run + * parks durably (kernel `wait.human`); `flows answer yes|no` + * (or Cloud's answer route) records the answer and the resumed body + * continues from this line with it. `to` names who is asked and is + * recorded with the question; it is not a delivery address. + */ + human(question: string, options: { to: string }): Step; dispatch(flow: string, input: unknown): Promise; done(reason: FlowCompletionReason): void; cloud: CloudHelper; From c632c47c1dac1da316ba117de75a2e7ad7401c30 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Fri, 18 Sep 2026 05:32:05 -0700 Subject: [PATCH 2/4] docs(surface,kernel,examples): f.human parks on wait.human; flows answer Co-Authored-By: Claude Opus 5 (1M context) --- docs/EVENT-AWAIT.md | 19 ++++--- docs/SURFACE.md | 43 +++++++++++++- examples/social-post-pipeline/README.md | 31 +++++----- .../social-post-pipeline.flow.ts | 15 +++-- kernel/DESIGN.md | 6 +- packages/sdk/tests/cli-answer.test.ts | 56 +++++++++++++++++++ 6 files changed, 139 insertions(+), 31 deletions(-) create mode 100644 packages/sdk/tests/cli-answer.test.ts diff --git a/docs/EVENT-AWAIT.md b/docs/EVENT-AWAIT.md index b80976ae2..63721c291 100644 --- a/docs/EVENT-AWAIT.md +++ b/docs/EVENT-AWAIT.md @@ -39,15 +39,19 @@ Verified against `main` at `85e7e372`: - `relayflowd-core/src/state.rs` folds `wait.event` into `StepState::Waiting`, and `relayflowd/src/engine/remote.rs` `emit_event` closes every open wait whose `event_key` matches. -- **No step produces `wait.event`.** Only `wait.human` is appended (manual - recovery, `machine/recovery.rs`). +- **No step produces `wait.event`.** `wait.human` is appended by manual + recovery (`machine/recovery.rs`) and by the `step.wait` verb, which a lease + holder uses to park its attempt on an `f.human` question; `emit_event` + closes a `wait.human` whose `wait_id` equals the `event_key` with + `human_responded`. - **`timeout_at_ms` is never read.** A wait with a timeout would wait forever. - **An event with no open wait is dropped.** `emit_event` returns `matched: 0` and journals nothing, so an event that lands while the body is running a fix step is lost. - **The authored `Ctx` has no `on`.** `packages/surface/src/context.ts` - exposes `human`, `dispatch`, and `done`; the authored executor throws - `unsupported_verb` for `human` and `dispatch` (#400). + exposes `human`, `dispatch`, and `done`; `human` lowers to `wait.human` + (SURFACE.md §5 *Human gates*); the authored executor still throws + `unsupported_verb` for `dispatch`. ## 3. Surface @@ -296,6 +300,7 @@ implementation proves: - **Pattern language.** v0 `wait.event` is exact-match; triggers already carry a recursive-subset `pattern`. Leaning: reuse `pattern` for subscriptions and keep `event_key` for exact-match waits. -- **Relationship to `f.human`.** #400 needs a durable approval wait. Leaning: - `f.human(question, { to, timeout })` lowers to `wait.human` with the timers - from §5.5, and resolves to `false` on timeout. +- **Relationship to `f.human`.** `f.human(question, { to })` lowers to + `wait.human` today (#400) and is answered by `event.emit` keyed by the wait + id. Open: `timeout` should reuse the timers from §5.5 and resolve to `false` + on timeout. diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 2ce871460..18b6eefc9 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -630,6 +630,7 @@ flows check [--watch] [--json] flows run [--json] [--no-spawn] [--data-dir ] flows run [--json] [--no-spawn] [--data-dir ] --input flows resume [--json] [--no-spawn] [--data-dir ] +flows answer [--json] [--no-spawn] [--data-dir ] [--note ] flows observer [--data-dir ] ``` @@ -787,7 +788,7 @@ The exit codes are part of the surface contract: | `0` | The run completed with `completionReason: success`; deliberate declination also carries a `run_declined` diagnostic locally. | | `1` | The run failed with a declared `completionReason`, or a transport, runtime, or daemon protocol error left the outcome unknown. A `step_failed` run names the failing step and its per-step `completionReason`, plus the exit code and output tails the journal recorded for it. An authored `done("step_failed")` exits `1` as well, and says so without naming a step, because no step failed — the body declared the verdict. | | `2` | The command was refused before a journal write: invalid input, failed preflight, unreachable daemon, or a `run_not_found` resume target. | -| `3` | The run parked. `PARKED [run_parked]` names the step and its `llm` or `agent` type, and distinguishes an unavailable worker from a `needs_human` recovery wait. | +| `3` | The run parked. `PARKED [run_parked]` names the step and its `llm` or `agent` type, and distinguishes an unavailable worker from a `needs_human` recovery wait. An authored body parked on `f.human` reports the question, who it is for, and the `flows answer` invocation that records the decision (see *Human gates* below). | Without an attached worker, reaching an `llm` or `agent` step returns a durable parked outcome. For authored TypeScript, `--local-agent` attaches both local @@ -807,6 +808,46 @@ If the lease expires without a completion, the command fails closed instead of polling forever. A manual-recovery agent whose worker dies parks in `needs_human`; the same exit-3 report says it is waiting for human recovery. +### Human gates: `f.human` + +```ts +const ok = await f.human(`Ship this?\n${plan.summary}`, { to: "khaliq" }); +if (!ok) return f.done("declined"); +``` + +`f.human(question, { to })` is the declared approval gate of RFC covenant 3. +It is not a child run. When the body reaches it with no answer on record, the +ROOT attempt parks on the kernel's durable `wait.human` (DESIGN.md §1.5) under +the call's own ordinal (`human-N`, counted with every other authored +operation), the lease is released, and the run is `parked` — exit 3, with a +`humanWait` field in the JSON report and a diagnostic that reads: + +``` +PARKED [run_parked] Run "" is waiting for khaliq to answer human-2: "Ship this?\n…" +Answer with: flows answer human-2 yes|no +Then continue with: flows resume +``` + +No process waits. `flows answer yes|no [--note ]` +records the decision as the answer contract `{ answer: boolean, note?, +answeredBy?, at }` — an `event.emit` keyed by the wait id, which the kernel +journals as `wait.completed{human_responded}` and closes the wait once: a +second answer is refused (`human_wait_unknown`), as is a wait the run is not +asking. `flows resume` then re-runs the body; every step before the gate is +memoized under its admission key, so nothing upstream repeats, and `f.human` +resolves from the journaled answer. The boolean the author branches on is +lowered as a `human-N` deterministic step carrying the answer on stdout, so it +is journal evidence in the same shape as every other authored step. A `no` is +a value the body decides on — `done("declined")` exits 0 — never a failure. + +`to` names who is asked and is recorded with the question; the local kit does +not deliver it anywhere (that is the channel-delivery work in RFC covenant 3 +and `ops/BACKLOG.md`). Authority is the journal socket: whoever can reach the +daemon can answer, and `answeredBy` records the OS user who did. On Cloud the +same wait is answered through the run's answer route with the caller's +identity. `timeout` is not yet enforced (DESIGN.md §1.4). `f.dispatch` still +fails closed as `unsupported_verb`. + `flows resume` reports `run_unavailable` only when relayflowd returns the typed `run_not_found` refusal. A dropped connection, request failure, or `journal_write_failed` response exits 1 as `protocol_error`, because the diff --git a/examples/social-post-pipeline/README.md b/examples/social-post-pipeline/README.md index 04ff0fa96..f1e417804 100644 --- a/examples/social-post-pipeline/README.md +++ b/examples/social-post-pipeline/README.md @@ -32,21 +32,24 @@ part isn't the agents — it's what each `.gate()` checks: - The human gate (`f.human`) is the last word. Everything upstream can pass every gate and the flow still won't publish without a yes. -## Status: typechecks, does not run yet +## Running it ```sh -cd packages/surface && npm run typecheck:examples +flows run --local-agent social-post-pipeline.flow.ts \ + --input '{"brand":"acme","topic":"launch week","approver":"khaliq"}' +# … agents run; the flow reaches f.human and parks: +# PARKED [run_parked] Run "" is waiting for khaliq to answer human-5: "Ready to publish …" +# Answer with: flows answer human-5 yes|no +flows answer human-5 yes --note "copy approved" +flows resume --local-agent ``` -- `f.agent(...)` builds a real step, but parks without a worker attached — - same as every other example in this repo today. -- `f.human(...)` currently **throws** `unsupported_verb` in the SDK's - authored-flow executor (`packages/sdk/src/authored-flow-executor.ts`). The - human-approval gate is declared surface, not yet wired to any execution - path — it's sequenced behind gates 2-4 and the `f.human` channel-delivery - work tracked for gate 6 (`ops/BACKLOG.md`). - -This flow documents the intended shape so that when `f.human` and the agent -worker land, this example starts working with no rewrite — just like -`docs/SURFACE.md` §1 promises: *"a simple flow becomes a harness by -accretion, never a rewrite."* +`f.human` parks the run on the kernel's durable `wait.human` (exit 3, the same +park code as every other wait). The question, who it is for, the answer, who +gave it and when are all journal facts; the resumed body re-runs to the same +line, finds the recorded answer, and every agent step before it is memoized — +nothing upstream is repeated. A `no` is a flow decision (`done("declined")`, +exit 0), not a failure. On Cloud the same wait is answered through the run's +answer route instead of the CLI. + +`f.agent(...)` needs a worker: `--local-agent` locally, or run it on Cloud. diff --git a/examples/social-post-pipeline/social-post-pipeline.flow.ts b/examples/social-post-pipeline/social-post-pipeline.flow.ts index c0fb3532f..0bf723e86 100644 --- a/examples/social-post-pipeline/social-post-pipeline.flow.ts +++ b/examples/social-post-pipeline/social-post-pipeline.flow.ts @@ -8,13 +8,11 @@ // substring of what it said — the same lesson examples/research/README.md // documents: "substring gates on model output are fail-open." // -// STATUS: typechecks against the real `@relayflows/surface` package (see -// ../tsconfig.json / `npm --prefix packages/surface run typecheck:examples`) -// but does not run yet — `f.agent` parks without an attached worker, and -// `f.human` throws `unsupported_verb` in the current authored-flow executor -// (packages/sdk/src/authored-flow-executor.ts). This flow documents the -// intended shape; see README.md for exactly what's real today vs. what gate -// work this is waiting on. +// STATUS: runs. `f.agent` needs a worker (`flows run --local-agent`, or +// Cloud), and `f.human` parks the run durably on the kernel's `wait.human`: +// `flows run` exits 3 naming the question, `flows answer human-N yes|no` +// records the decision, and `flows resume` continues from the same line with +// it. See README.md. import { flow } from "@relayflows/surface"; @@ -89,7 +87,8 @@ export default flow( `Fact-check: ${factCheck.summary}\nGraphic: drafts/graphic.png\n\nPublish?`, { to: input.approver }, ); - if (!approved) return f.done("canceled"); + // A "no" is a decision the flow made, not a kernel cancellation. + if (!approved) return f.done("declined"); f.done("success"); }, diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 497459a3f..35ae33f94 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -88,6 +88,9 @@ subscriptions (`subscription.opened` / `subscription.closed` over a durable stream) and stream-backed waits with `settle_ms`, `idle_at_ms`, and `deadline_at_ms`. As of v2.0.14 no step produces `wait.event`, and `timeout_at_ms` is not yet enforced for either `wait.event` or `wait.human`. +`wait.human` is produced by manual recovery (Appendix A) and by `step.wait` +(§5), the verb an authored root uses to park its attempt on an `f.human` +question. ### 1.5 `wait.human` Durable human await. Payload: `wait_id`, `prompt` (what is being asked), @@ -387,7 +390,8 @@ Minimal verb set for gate 1: | `effect.record` | `{run_id, step_id, attempt, idempotency_key, surface_path, revision_before, revision_after}` → `{deduped}` | phase one: agent records a writeback before performing it; the journal boundary elects one provider-call winner. `deduped:false` means this attempt owes the call **and** the `effect.confirm` that closes it | | `effect.confirm` | `{run_id, step_id, attempt, idempotency_key, surface_path}` → `{confirmed}` | phase two: the elected attempt made the provider call. Until it lands the election is reclaimable by a later attempt, so a worker that dies mid-writeback loses nothing; only the election holder may confirm | | `step.complete` | `{run_id, step_id, attempt, idempotency_key, completionReason, output, usage, started_pins, end_pins, effects, trajectory_tail}` → `{}` | completes a dispatched step — **also the out-of-band path**: the lease holder reports its actual start/end pins and journaled effect refs; kernel verifies them, runs the gate, and decides the edge | -| `event.emit` | `{run_id, event_key, payload}` → `{matched: n}` | satisfies `wait.event`; a human response arrives here too, closing `wait.human` with `completionReason: human_responded` | +| `step.wait` | `{run_id, step_id, attempt, idempotency_key, wait_id, prompt, requested_of, options?, timeout_at_ms?}` → run outcome | the lease holder parks its attempt on a `wait.human` (§1.5) instead of completing it: no `step.completed`, no iteration charged, lease released, run `parked`. `wait_id` names the question for the life of the run; a reuse is refused. Only the lease holder may call it | +| `event.emit` | `{run_id, event_key, payload}` → `{matched: n}` | satisfies `wait.event` by `event_key`; a human response arrives here too, with `event_key` = the `wait_id`, closing `wait.human` with `completionReason: human_responded` and `result` = the payload. The step folds to `Runnable` and is dispatched as a fresh attempt | | `stream.append` | `{run_id, stream, message}` → `{offset}` | durable channel write; journals `stream.appended` | | `stream.read` | `{run_id, stream, from_offset, limit}` → `{messages, next_offset}` | at-least-once replayable read; committing the consumer offset happens via the reader's step pins, not a verb | | `journal.read` | `{run_id, from_seq, limit}` → `{entries}` | raw journal access — replay, audit, the report step | diff --git a/packages/sdk/tests/cli-answer.test.ts b/packages/sdk/tests/cli-answer.test.ts new file mode 100644 index 000000000..ac9af7380 --- /dev/null +++ b/packages/sdk/tests/cli-answer.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; +import { runCli, type CliIo } from '../src/cli.js'; + +function capture(): { io: CliIo; stdout: string[]; stderr: string[] } { + const stdout: string[] = []; + const stderr: string[] = []; + return { io: { stdout: (line) => stdout.push(line), stderr: (line) => stderr.push(line) }, stdout, stderr }; +} + +/** + * `flows answer` argument shape. Every refusal here is a parse refusal: no + * daemon is contacted, so the only exit is 2 with the usage text. + */ +describe('flows answer invocation', () => { + it.each([ + ['no arguments', ['answer']], + ['missing the answer word', ['answer', 'run-1', 'human-1']], + ['an ambiguous word', ['answer', 'run-1', 'human-1', 'maybe']], + ['a fourth positional', ['answer', 'run-1', 'human-1', 'yes', 'extra']], + ['--note without text', ['answer', 'run-1', 'human-1', 'yes', '--note']], + ['--note followed by a flag', ['answer', 'run-1', 'human-1', 'yes', '--note', '--json']], + ['--data-dir without a path', ['answer', 'run-1', 'human-1', 'yes', '--data-dir']], + ['a repeated --json', ['answer', '--json', '--json', 'run-1', 'human-1', 'yes']], + ['worker flags that mean nothing here', ['answer', 'run-1', 'human-1', 'yes', '--local-agent']], + ['--allow-human-influenced', ['answer', 'run-1', 'human-1', 'yes', '--allow-human-influenced']], + ])('refuses %s with the usage text', async (_name, args) => { + const output = capture(); + expect(await runCli(args, output.io)).toBe(2); + expect(output.stderr.join('\n')).toContain('[invalid_invocation]'); + expect(output.stderr.join('\n')).toContain('flows answer [--json] [--no-spawn] [--data-dir ] [--note ] '); + }); + + it('refuses a wait id that is not human- before contacting any daemon', async () => { + const output = capture(); + expect(await runCli(['answer', '--no-spawn', '--data-dir', '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/nonexistent/data', 'run-1', 'approval', 'yes'], output.io)).toBe(2); + expect(output.stderr.join('\n')).toContain('[human_wait_unknown] "approval" is not an f.human wait id'); + expect(output.stdout).toEqual([]); + }); + + it('accepts true/false as yes/no and reaches the daemon seam', async () => { + // With `--no-spawn` and no daemon, the parse succeeded and the attach + // refused: that is the first thing past argument validation. + for (const word of ['true', 'false', 'yes', 'no']) { + const output = capture(); + expect(await runCli(['answer', '--no-spawn', '--data-dir', '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/nonexistent/data', 'run-1', 'human-1', word], output.io)).toBe(2); + expect(output.stderr.join('\n')).not.toContain('[invalid_invocation]'); + expect(output.stderr.join('\n')).toMatch(/\[(daemon_unreachable|relayflowd_not_found)]/); + } + }); + + it('lists the command in --help', async () => { + const output = capture(); + expect(await runCli(['--help'], output.io)).toBe(0); + expect(output.stdout.join('\n')).toContain('flows answer'); + }); +}); From f7b632a300bee256b81641c97532b3788b7d41d7 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Fri, 18 Sep 2026 05:51:52 -0700 Subject: [PATCH 3/4] feat(sdk): flows answer --by for relayed decisions Cloud's answer route runs flows answer inside the resumed sandbox; the OS user there is the sandbox, not the person who decided. --by records who. Co-Authored-By: Claude Opus 5 (1M context) --- docs/SURFACE.md | 8 ++++---- packages/sdk/src/cli.ts | 22 +++++++++++++++++----- packages/sdk/src/cli/answer.ts | 4 +++- packages/sdk/tests/cli-answer.test.ts | 4 +++- packages/sdk/tests/human-live.test.ts | 4 ++-- 5 files changed, 29 insertions(+), 13 deletions(-) diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 18b6eefc9..160a2c31d 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -630,7 +630,7 @@ flows check [--watch] [--json] flows run [--json] [--no-spawn] [--data-dir ] flows run [--json] [--no-spawn] [--data-dir ] --input flows resume [--json] [--no-spawn] [--data-dir ] -flows answer [--json] [--no-spawn] [--data-dir ] [--note ] +flows answer [--json] [--no-spawn] [--data-dir ] [--note ] [--by ] flows observer [--data-dir ] ``` @@ -828,9 +828,9 @@ Answer with: flows answer human-2 yes|no Then continue with: flows resume ``` -No process waits. `flows answer yes|no [--note ]` -records the decision as the answer contract `{ answer: boolean, note?, -answeredBy?, at }` — an `event.emit` keyed by the wait id, which the kernel +No process waits. `flows answer yes|no [--note ] +[--by ]` records the decision as the answer contract `{ answer: +boolean, note?, answeredBy?, at }` (`answeredBy` is `--by`, else the OS user) — an `event.emit` keyed by the wait id, which the kernel journals as `wait.completed{human_responded}` and closes the wait once: a second answer is refused (`human_wait_unknown`), as is a wait the run is not asking. `flows resume` then re-runs the body; every step before the gate is diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts index d7f5a9977..c5cd2b4b0 100644 --- a/packages/sdk/src/cli.ts +++ b/packages/sdk/src/cli.ts @@ -66,7 +66,7 @@ type ParsedArgs = | { command: 'check'; json: boolean; watch: boolean; value: string } | { command: 'run'; bucket: string | undefined; reuseFromRunId: string | undefined; localAgent: boolean; dataDir: string; input: string | undefined; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } | { command: 'resume'; localAgent: boolean; dataDir: string; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } - | { command: 'answer'; dataDir: string; json: boolean; spawn: boolean; note: string | undefined; runId: string; waitId: string; answer: boolean } + | { command: 'answer'; dataDir: string; json: boolean; spawn: boolean; note: string | undefined; by: string | undefined; runId: string; waitId: string; answer: boolean } | { command: 'observer'; dataDir: string } | { command: 'hn-monitor'; sub: 'start'; dataDir: string; specPath: string; pollIntervalMs: number | undefined } | { command: 'tick'; sub: 'start'; dataDir: string; specPath: string; scheduleId: string; @@ -95,7 +95,7 @@ const USAGE = [ 'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent] --input ', 'flows tick start --schedule-id --interval-ms [--epoch-ms ] [--max-catch-up ] [--poll-interval-ms ] [--data-dir ] ', 'flows resume [--allow-human-influenced] [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent] ', - 'flows answer [--json] [--no-spawn] [--data-dir ] [--note ] ', + 'flows answer [--json] [--no-spawn] [--data-dir ] [--note ] [--by ] ', 'flows replay [--allow-human-influenced] [--json] [--data-dir ] [--at ]', 'flows observer [--data-dir ]', 'flows hn-monitor start [--data-dir ] [--poll-interval-ms ] ', @@ -148,6 +148,7 @@ export async function runCli( if (parsed.command === 'answer') { const execution = await answerFlow(parsed.runId, parsed.waitId, parsed.answer, parsed.dataDir, { ...(parsed.note === undefined ? {} : { note: parsed.note }), + ...(parsed.by === undefined ? {} : { answeredBy: parsed.by }), daemon: { spawn: parsed.spawn && spawnAllowedByEnv() }, }); emitRunReport(execution, parsed.json, io); @@ -609,9 +610,12 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { } /** - * `flows answer [--json] [--no-spawn] [--data-dir ] [--note ] `. + * `flows answer [--json] [--no-spawn] [--data-dir ] [--note ] [--by ] `. * The answer is a literal `yes`/`no` (also `true`/`false`) so a shell cannot - * hand the kernel an ambiguous word as a decision. + * hand the kernel an ambiguous word as a decision. `--by` records who answered + * when the invoker is relaying a person's decision (Cloud's answer route runs + * this inside the resumed sandbox with the caller's identity); it defaults to + * the OS user. */ function parseAnswerArgs(rest: readonly string[]): ParsedArgs | undefined { let json = false; @@ -619,6 +623,7 @@ function parseAnswerArgs(rest: readonly string[]): ParsedArgs | undefined { let dataDir = DEFAULT_DATA_DIR; let sawDataDir = false; let note: string | undefined; + let by: string | undefined; const positionals: string[] = []; for (let index = 0; index < rest.length; index += 1) { const argument = rest[index]!; @@ -632,6 +637,13 @@ function parseAnswerArgs(rest: readonly string[]): ParsedArgs | undefined { spawn = false; continue; } + if (argument === '--by') { + const value = rest[index + 1]; + if (by !== undefined || value === undefined || value.startsWith('-') || value.trim() === '') return undefined; + by = value; + index += 1; + continue; + } if (argument === '--data-dir') { const value = rest[index + 1]; if (sawDataDir || value === undefined || value.startsWith('-')) return undefined; @@ -654,7 +666,7 @@ function parseAnswerArgs(rest: readonly string[]): ParsedArgs | undefined { const [runId, waitId, word] = positionals as [string, string, string]; const answer = word === 'yes' || word === 'true' ? true : word === 'no' || word === 'false' ? false : undefined; if (answer === undefined) return undefined; - return { command: 'answer', dataDir, json, spawn, note, runId, waitId, answer }; + return { command: 'answer', dataDir, json, spawn, note, by, runId, waitId, answer }; } function parseHnMonitorArgs(rest: readonly string[]): ParsedArgs | undefined { diff --git a/packages/sdk/src/cli/answer.ts b/packages/sdk/src/cli/answer.ts index 2206b36db..0a828834f 100644 --- a/packages/sdk/src/cli/answer.ts +++ b/packages/sdk/src/cli/answer.ts @@ -7,6 +7,8 @@ import { connect, emptyReport, protocolFailure, socketFor, type RunExecution } f export interface AnswerOptions { readonly note?: string; + /** Who answered, when relaying a person's decision; defaults to the OS user. */ + readonly answeredBy?: string; readonly daemon?: EnsureDaemonOptions; } @@ -51,7 +53,7 @@ export async function answerFlow( } const payload = humanAnswerPayload(answer, { ...(options.note === undefined ? {} : { note: options.note }), - answeredBy: safeUsername(), + answeredBy: options.answeredBy ?? safeUsername(), }); const emitted = await client.eventEmit(runId, waitId, payload); if (emitted.matched !== 1) { diff --git a/packages/sdk/tests/cli-answer.test.ts b/packages/sdk/tests/cli-answer.test.ts index ac9af7380..5220a4577 100644 --- a/packages/sdk/tests/cli-answer.test.ts +++ b/packages/sdk/tests/cli-answer.test.ts @@ -20,6 +20,8 @@ describe('flows answer invocation', () => { ['--note without text', ['answer', 'run-1', 'human-1', 'yes', '--note']], ['--note followed by a flag', ['answer', 'run-1', 'human-1', 'yes', '--note', '--json']], ['--data-dir without a path', ['answer', 'run-1', 'human-1', 'yes', '--data-dir']], + ['--by without an identity', ['answer', 'run-1', 'human-1', 'yes', '--by']], + ['--by followed by a flag', ['answer', 'run-1', 'human-1', 'yes', '--by', '--json']], ['a repeated --json', ['answer', '--json', '--json', 'run-1', 'human-1', 'yes']], ['worker flags that mean nothing here', ['answer', 'run-1', 'human-1', 'yes', '--local-agent']], ['--allow-human-influenced', ['answer', 'run-1', 'human-1', 'yes', '--allow-human-influenced']], @@ -27,7 +29,7 @@ describe('flows answer invocation', () => { const output = capture(); expect(await runCli(args, output.io)).toBe(2); expect(output.stderr.join('\n')).toContain('[invalid_invocation]'); - expect(output.stderr.join('\n')).toContain('flows answer [--json] [--no-spawn] [--data-dir ] [--note ] '); + expect(output.stderr.join('\n')).toContain('flows answer [--json] [--no-spawn] [--data-dir ] [--note ] [--by ] '); }); it('refuses a wait id that is not human- before contacting any daemon', async () => { diff --git a/packages/sdk/tests/human-live.test.ts b/packages/sdk/tests/human-live.test.ts index 8a3b11904..c5b3d6972 100644 --- a/packages/sdk/tests/human-live.test.ts +++ b/packages/sdk/tests/human-live.test.ts @@ -107,7 +107,7 @@ describe('f.human against a real daemon', () => { const garbage = f.answer(runId, 'human-2', 'maybe'); expect(garbage.status).toBe(2); - const answered = f.answer(runId, 'human-2', 'yes', '--note', 'looks good'); + const answered = f.answer(runId, 'human-2', 'yes', '--note', 'looks good', '--by', 'khaliq@agent-relay.com'); expect(answered.status, answered.stderr + answered.stdout).toBe(0); expect(JSON.parse(answered.stdout)).toMatchObject({ ok: true, command: 'answer', runId, answer: { waitId: 'human-2', answer: true, note: 'looks good' } }); @@ -127,7 +127,7 @@ describe('f.human against a real daemon', () => { 'step.attempt.started@1', 'wait.human@1', 'wait.completed@1', 'step.attempt.started@2', 'step.completed@2', ]); expect(closed[2]!.payload).toMatchObject({ wait_id: 'human-2', completionReason: 'human_responded', - result: { answer: true, note: 'looks good' } }); + result: { answer: true, note: 'looks good', answeredBy: 'khaliq@agent-relay.com' } }); expect(closed[4]!.payload).toMatchObject({ completionReason: 'success' }); // A resume after completion replays nothing and reports the durable result. const again = f.resume(runId); From 527c09fff3339509ec6650a99b5f54e10f935e07 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Fri, 18 Sep 2026 06:11:44 -0700 Subject: [PATCH 4/4] fix(sdk,kernel): f.human honors a named gate; the kernel attributes and timestamps answers - A `.gate(config)` on f.human is now lowered into the human-N step's verification like every other authored step (it was dropped, so a declared gate was never evaluated). - event.emit refuses a human response without a non-empty answeredBy, journals attribution: client_asserted (the socket authenticated the caller, not the kernel), drops any client at and stamps at_ms from the entry's own clock. The SDK contract follows: clients send { answer, note?, answeredBy }; the body reads at_ms/attribution back. Co-Authored-By: Claude Opus 5 (1M context) --- docs/SURFACE.md | 5 +- kernel/DESIGN.md | 2 +- kernel/relayflowd/src/engine/remote.rs | 35 ++++++++- kernel/relayflowd/src/server/tests.rs | 29 ++++++- packages/sdk/src/authored-flow-executor.ts | 14 +++- packages/sdk/src/authored-human.ts | 46 +++++++---- packages/sdk/src/cli/answer.ts | 9 ++- packages/sdk/tests/authored-human.test.ts | 88 ++++++++++++++++------ packages/sdk/tests/authored-root.test.ts | 2 +- packages/sdk/tests/human-live.test.ts | 6 +- 10 files changed, 183 insertions(+), 53 deletions(-) diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 160a2c31d..cb954e3ac 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -830,7 +830,10 @@ Then continue with: flows resume No process waits. `flows answer yes|no [--note ] [--by ]` records the decision as the answer contract `{ answer: -boolean, note?, answeredBy?, at }` (`answeredBy` is `--by`, else the OS user) — an `event.emit` keyed by the wait id, which the kernel +boolean, note?, answeredBy }` (`answeredBy` is `--by`, else the OS user; the +kernel refuses an unattributed answer, journals `attribution: client_asserted` +because the socket — not the kernel — authenticated the caller, and stamps +`at_ms` from its own clock, dropping any client-supplied time) — an `event.emit` keyed by the wait id, which the kernel journals as `wait.completed{human_responded}` and closes the wait once: a second answer is refused (`human_wait_unknown`), as is a wait the run is not asking. `flows resume` then re-runs the body; every step before the gate is diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 35ae33f94..30b9b8830 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -391,7 +391,7 @@ Minimal verb set for gate 1: | `effect.confirm` | `{run_id, step_id, attempt, idempotency_key, surface_path}` → `{confirmed}` | phase two: the elected attempt made the provider call. Until it lands the election is reclaimable by a later attempt, so a worker that dies mid-writeback loses nothing; only the election holder may confirm | | `step.complete` | `{run_id, step_id, attempt, idempotency_key, completionReason, output, usage, started_pins, end_pins, effects, trajectory_tail}` → `{}` | completes a dispatched step — **also the out-of-band path**: the lease holder reports its actual start/end pins and journaled effect refs; kernel verifies them, runs the gate, and decides the edge | | `step.wait` | `{run_id, step_id, attempt, idempotency_key, wait_id, prompt, requested_of, options?, timeout_at_ms?}` → run outcome | the lease holder parks its attempt on a `wait.human` (§1.5) instead of completing it: no `step.completed`, no iteration charged, lease released, run `parked`. `wait_id` names the question for the life of the run; a reuse is refused. Only the lease holder may call it | -| `event.emit` | `{run_id, event_key, payload}` → `{matched: n}` | satisfies `wait.event` by `event_key`; a human response arrives here too, with `event_key` = the `wait_id`, closing `wait.human` with `completionReason: human_responded` and `result` = the payload. The step folds to `Runnable` and is dispatched as a fresh attempt | +| `event.emit` | `{run_id, event_key, payload}` → `{matched: n}` | satisfies `wait.event` by `event_key`; a human response arrives here too, with `event_key` = the `wait_id`, closing `wait.human` with `completionReason: human_responded`. A human response must carry a non-empty `answeredBy`; the kernel records it as `attribution: client_asserted` (the socket, not the kernel, is the trust boundary), drops any client `at`, and stamps `at_ms` from its own clock. The step folds to `Runnable` and is dispatched as a fresh attempt | | `stream.append` | `{run_id, stream, message}` → `{offset}` | durable channel write; journals `stream.appended` | | `stream.read` | `{run_id, stream, from_offset, limit}` → `{messages, next_offset}` | at-least-once replayable read; committing the consumer offset happens via the reader's step pins, not a verb | | `journal.read` | `{run_id, from_seq, limit}` → `{entries}` | raw journal access — replay, audit, the report step | diff --git a/kernel/relayflowd/src/engine/remote.rs b/kernel/relayflowd/src/engine/remote.rs index 8d51816a0..1d98763c6 100644 --- a/kernel/relayflowd/src/engine/remote.rs +++ b/kernel/relayflowd/src/engine/remote.rs @@ -410,7 +410,38 @@ impl Engine { open.retain(|(wait_id, _, _, _)| wait_id != &done.wait_id); } } + // A human response must say who gave it, and the journal — not the + // client — says when. The kernel cannot verify the identity a daemon + // client asserts (the socket is the trust boundary; Cloud's answer + // route authenticates the caller before it reaches here), so the + // result records the attribution as client-asserted and stamps the + // entry's own clock as `at_ms`, dropping any client-supplied time. + let answers_human = open + .iter() + .any(|(_, _, _, reason)| *reason == WaitCompletionReason::HumanResponded); + if answers_human { + let attributed = payload + .get("answeredBy") + .and_then(Value::as_str) + .is_some_and(|who| !who.trim().is_empty()); + if !attributed { + bail!("a human response to {event_key} must carry a non-empty answeredBy") + } + } for (wait_id, step_id, attempt, completion_reason) in &open { + let now_ms = self.clock.now_ms(); + let result = if *completion_reason == WaitCompletionReason::HumanResponded { + let mut answer = payload.as_object().cloned().unwrap_or_default(); + answer.remove("at"); + answer.insert("at_ms".to_owned(), Value::from(now_ms)); + answer.insert( + "attribution".to_owned(), + Value::from("client_asserted"), + ); + Value::Object(answer) + } else { + payload.clone() + }; self.append( &mut journal, &JournalEntry::new( @@ -418,11 +449,11 @@ impl Engine { run_id, step_id.clone(), *attempt, - self.clock.now_ms(), + now_ms, WaitCompletedPayload { wait_id: wait_id.clone(), completion_reason: *completion_reason, - result: payload.clone(), + result, }, ), )?; diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index feb791279..c960de3c9 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -581,7 +581,8 @@ fn step_wait_parks_the_attempt_and_a_human_answer_redispatches_it() { ); assert_eq!(stray.result.unwrap()["matched"], 0); - let answered = request( + // An unattributed answer cannot close a human wait. + let anonymous = request( data_dir, &hub, 2, @@ -591,6 +592,23 @@ fn step_wait_parks_the_attempt_and_a_human_answer_redispatches_it() { }}) .to_string(), ); + assert!(!anonymous.ok); + assert!( + anonymous.error.unwrap().message.contains("answeredBy"), + "refusal names the missing attribution" + ); + + let answered = request( + data_dir, + &hub, + 2, + &writer, + &json!({"id": "emit", "verb": "event.emit", "params": { + "run_id": run_id, "event_key": "human-1", + "payload": {"answer": true, "answeredBy": "khaliq", "at": "1999-01-01T00:00:00Z"}, + }}) + .to_string(), + ); assert!(answered.ok, "event.emit failed: {:?}", answered.error); assert_eq!(answered.result.unwrap()["matched"], 1); let redispatch = read_frame(&mut worker_reader); @@ -607,6 +625,12 @@ fn step_wait_parks_the_attempt_and_a_human_answer_redispatches_it() { assert_eq!(completed.payload["wait_id"], "human-1"); assert_eq!(completed.payload["completionReason"], "human_responded"); assert_eq!(completed.payload["result"]["answer"], true); + // The journal's clock, not the client's, says when; attribution is + // recorded as what it is. + assert_eq!(completed.payload["result"]["answeredBy"], "khaliq"); + assert_eq!(completed.payload["result"]["attribution"], "client_asserted"); + assert!(completed.payload["result"].get("at").is_none()); + assert_eq!(completed.payload["result"]["at_ms"], completed.at_ms); // Answering again matches nothing: the wait is closed. let repeat = request( data_dir, @@ -614,7 +638,8 @@ fn step_wait_parks_the_attempt_and_a_human_answer_redispatches_it() { 2, &writer, &json!({"id": "emit", "verb": "event.emit", "params": { - "run_id": run_id, "event_key": "human-1", "payload": {"answer": false}, + "run_id": run_id, "event_key": "human-1", + "payload": {"answer": false, "answeredBy": "khaliq"}, }}) .to_string(), ); diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index 48c87dee0..8a59a9e76 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -433,7 +433,11 @@ export async function executeAuthoredFlow( } const id = `human-${nextStep++}`; const to = humanOptions.to; - return trackStep(authoredSteps, new AuthoredFlowOperation( + // Hoisted like `llmOp`/`runOp`: the start closure reads the caller's + // `.gate(config)` at spec-build time, so a named gate on the answer is + // lowered into the `human-N` step's `verification` like any other step's. + let humanOp!: AuthoredFlowOperation; + humanOp = new AuthoredFlowOperation( id, 'human', () => assertOperationAllowed('human', definition.name, requestedCompletion), () => observeStep(id, 'deterministic', async () => { @@ -448,13 +452,15 @@ export async function executeAuthoredFlow( if (recorded === undefined) throw new AuthoredHumanParked({ waitId: id, question, to }, rootRunId); const record = { human: id, to, answer: recorded.answer, ...(recorded.note === undefined ? {} : { note: recorded.note }), - ...(recorded.answeredBy === undefined ? {} : { answeredBy: recorded.answeredBy }) }; + answeredBy: recorded.answeredBy, + ...(recorded.atMs === undefined ? {} : { at: new Date(recorded.atMs).toISOString() }) }; const literal = `'${JSON.stringify(record).replaceAll("'", "'\\''")}'`; - await lowerDeterministic(id, `printf '%s' ${literal}`, false); + await lowerDeterministic(id, `printf '%s' ${literal}`, false, undefined, humanOp.namedGate); return recorded.answer; }, options.onProgress), lifecycle, - )); + ); + return trackStep(authoredSteps, humanOp); }, dispatch() { assertOperationAllowed('dispatch', definition.name, requestedCompletion); diff --git a/packages/sdk/src/authored-human.ts b/packages/sdk/src/authored-human.ts index 510fb33d5..cf3c01eb0 100644 --- a/packages/sdk/src/authored-human.ts +++ b/packages/sdk/src/authored-human.ts @@ -2,20 +2,30 @@ import { AuthoredFlowExecutionError, type AuthoredHumanWait } from './authored-f import type { JournalClient } from './journal-client.js'; /** - * The answer contract for a parked `f.human`. This is the `payload` of the + * The answer contract for a parked `f.human`. + * + * The CLIENT sends `{ answer, note?, answeredBy }` as the `payload` of the * `event.emit` that closes the wait (kernel DESIGN.md §5: `event_key` is the - * `wait_id`) and therefore the `result` of the journaled - * `wait.completed{human_responded}` the resumed body reads back. `flows - * answer` and Cloud's answer route both produce exactly this shape. + * `wait_id`); `answeredBy` is required — the kernel refuses an unattributed + * answer. The KERNEL journals it as the `result` of + * `wait.completed{human_responded}` with two fields of its own: `at_ms`, the + * entry's clock (any client `at` is dropped — the journal says when), and + * `attribution: "client_asserted"`, because the daemon socket, not the kernel, + * is what authenticated whoever said `answeredBy`. `flows answer` and Cloud's + * resumed sandbox both produce the client half; the body reads the kernel half. */ export interface HumanAnswer { readonly answer: boolean; readonly note?: string; - readonly answeredBy?: string; - /** ISO-8601 instant the answer was given. */ - readonly at?: string; + readonly answeredBy: string; + /** Kernel clock at the journaled `wait.completed`, epoch milliseconds. */ + readonly atMs?: number; + /** How `answeredBy` was established. The kernel writes `client_asserted`. */ + readonly attribution?: string; } +export type HumanAnswerPayload = Pick; + /** An open `wait.human` on a run: asked, not yet answered. */ export interface OpenHumanWait extends AuthoredHumanWait { readonly stepId: string; @@ -24,32 +34,36 @@ export interface OpenHumanWait extends AuthoredHumanWait { export const HUMAN_WAIT_ID = /^human-[1-9][0-9]*$/; -export function humanAnswerPayload(answer: boolean, extra: { note?: string; answeredBy?: string } = {}): HumanAnswer { +export function humanAnswerPayload(answer: boolean, extra: { note?: string; answeredBy: string }): HumanAnswerPayload { + if (extra.answeredBy.trim() === '') { + throw new AuthoredFlowExecutionError('human_answer_invalid', 'an answer must say who gave it (answeredBy)'); + } return { answer, ...(extra.note === undefined || extra.note === '' ? {} : { note: extra.note }), - ...(extra.answeredBy === undefined || extra.answeredBy === '' ? {} : { answeredBy: extra.answeredBy }), - at: new Date().toISOString(), + answeredBy: extra.answeredBy, }; } /** Narrow an untrusted journal `result` to the answer contract, or refuse it. */ export function parseHumanAnswer(value: unknown, waitId: string): HumanAnswer { - const record = value as Partial | null; + const record = value as { answer?: unknown; note?: unknown; answeredBy?: unknown; at_ms?: unknown; attribution?: unknown } | null; if (typeof record !== 'object' || record === null || typeof record.answer !== 'boolean' || (record.note !== undefined && typeof record.note !== 'string') - || (record.answeredBy !== undefined && typeof record.answeredBy !== 'string') - || (record.at !== undefined && typeof record.at !== 'string')) { + || typeof record.answeredBy !== 'string' || record.answeredBy.trim() === '' + || (record.at_ms !== undefined && !Number.isSafeInteger(record.at_ms)) + || (record.attribution !== undefined && typeof record.attribution !== 'string')) { throw new AuthoredFlowExecutionError( 'human_answer_invalid', - `the recorded answer to ${waitId} is not { answer: boolean }; answer it again with flows answer`, + `the recorded answer to ${waitId} is not { answer: boolean, answeredBy: string }; answer it again with flows answer`, ); } return { answer: record.answer, ...(record.note === undefined ? {} : { note: record.note }), - ...(record.answeredBy === undefined ? {} : { answeredBy: record.answeredBy }), - ...(record.at === undefined ? {} : { at: record.at }), + answeredBy: record.answeredBy, + ...(record.at_ms === undefined ? {} : { atMs: record.at_ms as number }), + ...(record.attribution === undefined ? {} : { attribution: record.attribution }), }; } diff --git a/packages/sdk/src/cli/answer.ts b/packages/sdk/src/cli/answer.ts index 0a828834f..ecb0bd027 100644 --- a/packages/sdk/src/cli/answer.ts +++ b/packages/sdk/src/cli/answer.ts @@ -51,9 +51,16 @@ export async function answerFlow( + (others.length === 0 ? ' It has no open question.' : ` Open: ${others.join(', ')}.`), }] } }; } + const answeredBy = options.answeredBy ?? safeUsername(); + if (answeredBy === '') { + return { exitCode: 2, report: { ...base, runId, socketPath, diagnostics: [{ + severity: 'refusal', kind: 'human_wait_unknown', + message: 'An answer must say who gave it and this environment has no OS user; pass --by .', + }] } }; + } const payload = humanAnswerPayload(answer, { ...(options.note === undefined ? {} : { note: options.note }), - answeredBy: options.answeredBy ?? safeUsername(), + answeredBy, }); const emitted = await client.eventEmit(runId, waitId, payload); if (emitted.matched !== 1) { diff --git a/packages/sdk/tests/authored-human.test.ts b/packages/sdk/tests/authored-human.test.ts index 35423f46a..2c94e2ae4 100644 --- a/packages/sdk/tests/authored-human.test.ts +++ b/packages/sdk/tests/authored-human.test.ts @@ -18,7 +18,9 @@ describe('f.human lowering', () => { let path: string; let server: Server; let rootEntries: Array> = []; - const started: Array<{ id: string; command: string; admissionKey?: string }> = []; + const started: Array<{ id: string; command: string; admissionKey?: string; gate?: { id: string; command: string } }> = []; + const failedRuns = new Set(); + let gateVerdict: 'pass' | 'fail' = 'pass'; const outputs = new Map(); const stepOf = new Map(); let nextRun = 1; @@ -31,12 +33,21 @@ describe('f.human lowering', () => { const spec = params.spec as { steps: Array<{ id: string; command: string }> }; const step = spec.steps[0]!; const runId = `child-${nextRun++}`; - started.push({ id: step.id, command: step.command, ...(typeof params.admission_key === 'string' ? { admissionKey: params.admission_key } : {}) }); + // A named gate lowers as a second `.gate` step whose command embeds + // the pattern (named-gate-lowering.ts); record it so a test can prove + // the gate reached the spec. + const gateStep = spec.steps[1]; + started.push({ id: step.id, command: step.command, ...(typeof params.admission_key === 'string' ? { admissionKey: params.admission_key } : {}), + ...(gateStep === undefined ? {} : { gate: { id: gateStep.id, command: gateStep.command } }) }); // The fake "executes" printf '%s' '' by unquoting the literal. const printed = /^printf '%s' '((?:[^']|'\\'')*)'$/.exec(step.command)?.[1]?.replaceAll("'\\''", "'") ?? ''; outputs.set(runId, printed); stepOf.set(runId, step.id); - sendResult(ctx, { run_id: runId, status: 'completed', completion_reason: 'success', completed_steps: 1 }); + // The fake cannot run the gate's embedded engine; the test says how it judges. + const failed = gateStep !== undefined && gateVerdict === 'fail'; + if (failed) failedRuns.add(runId); + sendResult(ctx, { run_id: runId, status: failed ? 'failed' : 'completed', + completion_reason: failed ? 'step_failed' : 'success', completed_steps: 1 }); }, 'journal.read': (ctx, params) => { const runId = params.run_id as string; @@ -45,10 +56,11 @@ describe('f.human lowering', () => { sendResult(ctx, { entries: rootEntries.map((entry, index) => ({ seq: index + 1, ...entry })).filter(entry => entry.seq >= from) }); return; } + const failed = failedRuns.has(runId); sendResult(ctx, { entries: [{ entry_type: 'step.completed', step_id: stepOf.get(runId), - payload: { completionReason: 'success', disposition: 'step_done', - output: { exit_code: 0, stdout_tail: outputs.get(runId) ?? '', stderr_tail: '' } }, + payload: { completionReason: failed ? 'verification_failed' : 'success', disposition: 'step_done', + output: failed ? null : { exit_code: 0, stdout_tail: outputs.get(runId) ?? '', stderr_tail: '' } }, }] }); }, 'stream.read': (ctx) => sendResult(ctx, { messages: [], next_offset: 0 }), @@ -60,7 +72,7 @@ describe('f.human lowering', () => { rmSync(path, { force: true }); }); - beforeEach(() => { rootEntries = []; started.length = 0; }); + beforeEach(() => { rootEntries = []; started.length = 0; failedRuns.clear(); gateVerdict = 'pass'; }); async function client(): Promise { const journal = new JournalClient(path, { requestTimeoutMs: 2000 }); @@ -103,7 +115,7 @@ describe('f.human lowering', () => { }); it('continues with the recorded answer, lowered as a memoized human-N step', async () => { - rootEntries = [asked, answered({ answer: true, note: 'lgtm', answeredBy: 'khaliq', at: '2026-09-18T00:00:00Z' })]; + rootEntries = [asked, answered({ answer: true, note: 'lgtm', answeredBy: 'khaliq', at_ms: Date.UTC(2026, 8, 18), attribution: 'client_asserted' })]; const journal = await client(); try { const handle = flow('ask', async f => { @@ -115,13 +127,37 @@ describe('f.human lowering', () => { expect(result.journalSteps.map(step => step.id)).toEqual(['human-1', 'complete-2']); expect(started[0]).toMatchObject({ id: 'human-1' }); expect(started[0]!.admissionKey).toMatch(/^[a-z0-9:_-]+$/i); - expect(JSON.parse(outputs.get('child-1')!)).toEqual({ human: 'human-1', to: 'khaliq', answer: true, note: 'lgtm', answeredBy: 'khaliq' }); + expect(JSON.parse(outputs.get('child-1')!)).toEqual({ human: 'human-1', to: 'khaliq', answer: true, note: 'lgtm', answeredBy: 'khaliq', at: '2026-09-18T00:00:00.000Z' }); expect(await readOpenHumanWaits(journal, ROOT)).toEqual([]); } finally { journal.close(); } }); + it('lowers a named gate on the answer into the human-N step, so a rejected answer can fail it', async () => { + const gate = { type: 'regex_match', pattern: '"answer":true' } as const; + // Approved: the gate is in the lowered spec, passes, and the body continues. + rootEntries = [asked, answered({ answer: true, answeredBy: 'khaliq' })]; + let journal = await client(); + try { + const handle = flow('gated', async f => { await f.human('Ship it?', { to: 'khaliq' }).gate(gate); f.done('success'); }); + const result = await executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT }); + expect(result.completionReason).toBe('success'); + expect(started[0]?.gate?.id).toBe('human-1.gate'); + expect(started[0]?.gate?.command).toContain(JSON.stringify(gate.pattern)); + } finally { journal.close(); } + // Rejected: the same gate is judged on the journaled answer and fails the step. + rootEntries = [asked, answered({ answer: false, answeredBy: 'khaliq' })]; + started.length = 0; + gateVerdict = 'fail'; + journal = await client(); + try { + const handle = flow('gated', async f => { await f.human('Ship it?', { to: 'khaliq' }).gate(gate); f.done('success'); }); + await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toMatchObject({ code: 'step_failed' }); + expect(started[0]?.gate?.id).toBe('human-1.gate'); + } finally { journal.close(); } + }); + it('a negative answer is a value, not a failure', async () => { - rootEntries = [asked, answered({ answer: false })]; + rootEntries = [asked, answered({ answer: false, answeredBy: 'khaliq' })]; const journal = await client(); try { const handle = flow('ask', async f => { @@ -133,7 +169,7 @@ describe('f.human lowering', () => { }); it('asks the second question only after the first is answered, under its own id', async () => { - rootEntries = [asked, answered({ answer: true })]; + rootEntries = [asked, answered({ answer: true, answeredBy: 'khaliq' })]; const journal = await client(); try { const handle = flow('ask-twice', async f => { @@ -147,13 +183,15 @@ describe('f.human lowering', () => { } finally { journal.close(); } }); - it('refuses a recorded answer that is not { answer: boolean }', async () => { - rootEntries = [asked, answered({ answer: 'yes' })]; - const journal = await client(); - try { - const handle = flow('ask', async f => { await f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); - await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toMatchObject({ code: 'human_answer_invalid' }); - } finally { journal.close(); } + it('refuses a recorded answer that is not { answer: boolean, answeredBy }', async () => { + for (const bad of [{ answer: 'yes', answeredBy: 'khaliq' }, { answer: true }, { answer: true, answeredBy: ' ' }]) { + rootEntries = [asked, answered(bad)]; + const journal = await client(); + try { + const handle = flow('ask', async f => { await f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); + await expect(executeAuthoredFlow(handle, journal, undefined, { rootRunId: ROOT })).rejects.toMatchObject({ code: 'human_answer_invalid' }); + } finally { journal.close(); } + } }); it('refuses an empty question or an unnamed recipient before touching the journal', async () => { @@ -175,7 +213,7 @@ describe('f.human lowering', () => { }); it('an unawaited f.human is an unawaited step', async () => { - rootEntries = [asked, answered({ answer: true })]; + rootEntries = [asked, answered({ answer: true, answeredBy: 'khaliq' })]; const journal = await client(); try { const handle = flow('fire-and-forget', async f => { f.human('Ship it?', { to: 'khaliq' }); f.done('success'); }); @@ -185,15 +223,17 @@ describe('f.human lowering', () => { }); describe('the answer contract', () => { - it('is { answer, note?, answeredBy?, at }', () => { + it('the client sends { answer, note?, answeredBy }; the kernel adds at_ms and attribution', () => { const payload = humanAnswerPayload(true, { note: 'ok', answeredBy: 'khaliq' }); - expect(payload).toMatchObject({ answer: true, note: 'ok', answeredBy: 'khaliq' }); - expect(Date.parse(payload.at!)).not.toBeNaN(); - expect(humanAnswerPayload(false, { note: '' })).not.toHaveProperty('note'); - expect(parseHumanAnswer(payload, 'human-1')).toEqual(payload); + expect(payload).toEqual({ answer: true, note: 'ok', answeredBy: 'khaliq' }); + expect(humanAnswerPayload(false, { note: '', answeredBy: 'khaliq' })).not.toHaveProperty('note'); + expect(() => humanAnswerPayload(true, { answeredBy: ' ' })).toThrow(/human_answer_invalid/); + // What the journal hands back: the kernel's clock and its attribution note. + expect(parseHumanAnswer({ ...payload, at_ms: 1_800_000_000_000, attribution: 'client_asserted' }, 'human-1')) + .toEqual({ answer: true, note: 'ok', answeredBy: 'khaliq', atMs: 1_800_000_000_000, attribution: 'client_asserted' }); }); it('refuses every other shape', () => { - for (const bad of [null, 'yes', { answer: 'yes' }, { answer: true, note: 3 }, {}]) { + for (const bad of [null, 'yes', { answer: 'yes', answeredBy: 'k' }, { answer: true, note: 3, answeredBy: 'k' }, {}, { answer: true }, { answer: true, answeredBy: 'k', at_ms: 1.5 }]) { expect(() => parseHumanAnswer(bad, 'human-1')).toThrow(/human_answer_invalid/); } }); diff --git a/packages/sdk/tests/authored-root.test.ts b/packages/sdk/tests/authored-root.test.ts index b790ca0a9..bb493f03b 100644 --- a/packages/sdk/tests/authored-root.test.ts +++ b/packages/sdk/tests/authored-root.test.ts @@ -300,7 +300,7 @@ describe('durable authored root', () => { payload: { wait_id: 'human-1', prompt: 'Ship it?', requested_of: 'khaliq' }, }, { entry_type: 'wait.completed', step_id: 'authored-root', attempt: 1, - payload: { wait_id: 'human-1', completionReason: 'human_responded', result: { answer: false } }, + payload: { wait_id: 'human-1', completionReason: 'human_responded', result: { answer: false, answeredBy: 'khaliq', at_ms: 1, attribution: 'client_asserted' } }, }]; vi.mocked(loadAuthoredFlow).mockResolvedValue(loaded); diff --git a/packages/sdk/tests/human-live.test.ts b/packages/sdk/tests/human-live.test.ts index c5b3d6972..82de1dd54 100644 --- a/packages/sdk/tests/human-live.test.ts +++ b/packages/sdk/tests/human-live.test.ts @@ -127,7 +127,11 @@ describe('f.human against a real daemon', () => { 'step.attempt.started@1', 'wait.human@1', 'wait.completed@1', 'step.attempt.started@2', 'step.completed@2', ]); expect(closed[2]!.payload).toMatchObject({ wait_id: 'human-2', completionReason: 'human_responded', - result: { answer: true, note: 'looks good', answeredBy: 'khaliq@agent-relay.com' } }); + result: { answer: true, note: 'looks good', answeredBy: 'khaliq@agent-relay.com', attribution: 'client_asserted' } }); + // The journal's clock says when; the client never supplies `at`. + const result = closed[2]!.payload.result as Record; + expect(result).not.toHaveProperty('at'); + expect(result.at_ms).toBe((closed[2] as { at_ms?: number }).at_ms); expect(closed[4]!.payload).toMatchObject({ completionReason: 'success' }); // A resume after completion replays nothing and reports the durable result. const again = f.resume(runId);