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..cb954e3ac 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 ] [--by ] 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,49 @@ 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 ] +[--by ]` records the decision as the answer contract `{ answer: +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 +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..30b9b8830 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`. 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.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..1d98763c6 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,69 @@ 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); + } + } + // 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) in &open { + 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( @@ -318,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: WaitCompletionReason::EventReceived, - result: payload.clone(), + completion_reason: *completion_reason, + result, }, ), )?; 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..c960de3c9 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -500,6 +500,152 @@ 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); + + // An unattributed answer cannot close a human wait. + let anonymous = 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!(!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); + 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); + // 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, + &hub, + 2, + &writer, + &json!({"id": "emit", "verb": "event.emit", "params": { + "run_id": run_id, "event_key": "human-1", + "payload": {"answer": false, "answeredBy": "khaliq"}, + }}) + .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..8a59a9e76 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,59 @@ 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; + // 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 () => { + 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 }), + 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, 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 new file mode 100644 index 000000000..cf3c01eb0 --- /dev/null +++ b/packages/sdk/src/authored-human.ts @@ -0,0 +1,150 @@ +import { AuthoredFlowExecutionError, type AuthoredHumanWait } from './authored-flow-error.js'; +import type { JournalClient } from './journal-client.js'; + +/** + * 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`); `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; + /** 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; + readonly attempt: number; +} + +export const HUMAN_WAIT_ID = /^human-[1-9][0-9]*$/; + +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 }), + 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 { 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') + || 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, answeredBy: string }; answer it again with flows answer`, + ); + } + return { + answer: record.answer, + ...(record.note === undefined ? {} : { note: record.note }), + answeredBy: record.answeredBy, + ...(record.at_ms === undefined ? {} : { atMs: record.at_ms as number }), + ...(record.attribution === undefined ? {} : { attribution: record.attribution }), + }; +} + +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..c5cd2b4b0 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; 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; @@ -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 ] [--by ] ', '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,15 @@ 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 }), + ...(parsed.by === undefined ? {} : { answeredBy: parsed.by }), + 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 +491,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 +609,66 @@ 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 ] [--by ] `. + * The answer is a literal `yes`/`no` (also `true`/`false`) so a shell cannot + * 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; + let spawn = true; + 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]!; + if (argument === '--json') { + if (json) return undefined; + json = true; + continue; + } + if (argument === '--no-spawn') { + if (!spawn) return 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; + 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, by, runId, waitId, answer }; +} + function parseHnMonitorArgs(rest: readonly string[]): ParsedArgs | undefined { const sub = rest[0]; if (sub !== 'start') return undefined; @@ -810,6 +883,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..ecb0bd027 --- /dev/null +++ b/packages/sdk/src/cli/answer.ts @@ -0,0 +1,98 @@ +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; + /** Who answered, when relaying a person's decision; defaults to the OS user. */ + readonly answeredBy?: 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 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, + }); + 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..2c94e2ae4 --- /dev/null +++ b/packages/sdk/tests/authored-human.test.ts @@ -0,0 +1,240 @@ +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; 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; + + 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++}`; + // 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); + // 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; + 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; + } + const failed = failedRuns.has(runId); + sendResult(ctx, { entries: [{ + entry_type: 'step.completed', step_id: stepOf.get(runId), + 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 }), + }); + }); + + afterAll(async () => { + await new Promise((resolve) => server.close(() => resolve())); + rmSync(path, { force: true }); + }); + + beforeEach(() => { rootEntries = []; started.length = 0; failedRuns.clear(); gateVerdict = 'pass'; }); + + 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_ms: Date.UTC(2026, 8, 18), attribution: 'client_asserted' })]; + 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', 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, answeredBy: 'khaliq' })]; + 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, answeredBy: 'khaliq' })]; + 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, 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 () => { + 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, answeredBy: 'khaliq' })]; + 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('the client sends { answer, note?, answeredBy }; the kernel adds at_ms and attribution', () => { + const payload = humanAnswerPayload(true, { note: 'ok', answeredBy: 'khaliq' }); + 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', 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-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..bb493f03b 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, answeredBy: 'khaliq', at_ms: 1, attribution: 'client_asserted' } }, + }]; + 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/cli-answer.test.ts b/packages/sdk/tests/cli-answer.test.ts new file mode 100644 index 000000000..5220a4577 --- /dev/null +++ b/packages/sdk/tests/cli-answer.test.ts @@ -0,0 +1,58 @@ +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']], + ['--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']], + ])('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 ] [--by ] '); + }); + + 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'); + }); +}); diff --git a/packages/sdk/tests/human-live.test.ts b/packages/sdk/tests/human-live.test.ts new file mode 100644 index 000000000..82de1dd54 --- /dev/null +++ b/packages/sdk/tests/human-live.test.ts @@ -0,0 +1,170 @@ +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', '--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' } }); + 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', 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); + 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;