Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 12 additions & 7 deletions docs/EVENT-AWAIT.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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.
46 changes: 45 additions & 1 deletion docs/SURFACE.md
Original file line number Diff line number Diff line change
Expand Up @@ -630,6 +630,7 @@ flows check [--watch] [--json] <flow.yaml|spec.json>
flows run [--json] [--no-spawn] [--data-dir <dir>] <flow.yaml|spec.json>
flows run [--json] [--no-spawn] [--data-dir <dir>] <flow.ts> --input <inline-json-or-file>
flows resume [--json] [--no-spawn] [--data-dir <dir>] <run-id>
flows answer [--json] [--no-spawn] [--data-dir <dir>] [--note <text>] [--by <identity>] <run-id> <wait-id> <yes|no>
flows observer [--data-dir <dir>]
```

Expand Down Expand Up @@ -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
Expand All @@ -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 "<run-id>" is waiting for khaliq to answer human-2: "Ship this?\n…"
Answer with: flows answer <run-id> human-2 yes|no
Then continue with: flows resume <run-id>
```

No process waits. `flows answer <run-id> <wait-id> yes|no [--note <text>]
[--by <identity>]` 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
Expand Down
31 changes: 17 additions & 14 deletions examples/social-post-pipeline/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 "<run-id>" is waiting for khaliq to answer human-5: "Ready to publish …"
# Answer with: flows answer <run-id> human-5 yes|no
flows answer <run-id> human-5 yes --note "copy approved"
flows resume --local-agent <run-id>
```

- `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.
15 changes: 7 additions & 8 deletions examples/social-post-pipeline/social-post-pipeline.flow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 <run> 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";

Expand Down Expand Up @@ -89,7 +87,8 @@ export default flow<SocialPostInput>(
`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");
},
Expand Down
6 changes: 5 additions & 1 deletion kernel/DESIGN.md
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down Expand Up @@ -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 |
Expand Down
34 changes: 30 additions & 4 deletions kernel/relayflowd/src/engine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand Down Expand Up @@ -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)]
Expand Down Expand Up @@ -643,9 +659,19 @@ impl<C: Clock> Engine<C> {

fn open_run(&self, run_id: &str) -> Result<SqliteJournal> {
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()))
}

Expand Down
Loading
Loading