# moonbitlang/workflow

Write a multi-agent workflow as an ordinary async MoonBit program. Share
concurrency and call limits across its agents, choose how failures affect
each stage, and resume completed work from a journal. The program's loops
and branches determine what runs; there is no separate graph to declare.

The core package never spawns anything. Its one seam is `Runner`: a
single async function from `AgentCall` to `AgentOutcome`. Engines plug
in from the outside — through the `spawn` sub-package's child-contract
implementation for out-of-process engines, or any in-process function;
tests plug in fakes, which is why everything below runs hermetically.

## Setup and package map

Add the module to a consuming project:

```sh
moon add moonbitlang/workflow
```

Import the packages you use in that project's `moon.pkg`. For the examples
below, which also use the async runtime:

```moonbit nocheck
import {
  "moonbitlang/workflow",
  "moonbitlang/async",
}

supported_targets = "native+wasm"
```

Run with `moon run --target native <package>` or `--target wasm`. Every
package in this module supports these two targets. Blocks marked `mbt check`
in these READMEs are compiled and run by `moon test`; shell commands and
`nocheck` snippets are usage instructions, not automatic agent launches.

| Package | Use it for |
| --- | --- |
| [`workflow`](README.mbt.md) | Calls, typed outcomes, concurrency, retries, replay, and accounting. |
| [`workflow/spawn`](spawn/README.mbt.md) | Running a process that speaks the child contract. |
| [`workflow/hosted`](hosted/README.mbt.md) | Running under a host that supplies child ids, launch limits, and journal paths. |
| [`workflow/shim`](shim/README.mbt.md) | Building adapters from a foreign CLI's output to the child contract. |
| [`workflow/shim/claude`](shim/claude/README.mbt.md) | Executable adapter for Claude Code. |
| [`workflow/shim/codex`](shim/codex/README.mbt.md) | Executable adapter for Codex. |
| [`workflow/viz`](viz/README.mbt.md) | Rendering a journal as a standalone HTML report. |
| [`workflow/examples/scout`](examples/scout/README.mbt.md) | A small executable demonstrating process calls and journal replay. |

```mermaid
flowchart TD
  Program["Your async program"] --> Core["workflow<br/>calls and policies"]
  Core <--> Journal["Journal<br/>resolved outcomes"]
  Core --> Runner["Runner"]
  Runner --> Fake["In-process engine<br/>or test fake"]
  Runner --> Spawn["spawn<br/>child contract"]
  Host["Host handoff"] --> Hosted["hosted<br/>ids and sidecar"]
  Hosted --> Runner
  Spawn --> Engine["Contract-speaking<br/>engine"]
  Spawn --> Shim["shim/claude<br/>or shim/codex"]
  Shim --> CLI["Agent CLI"]
  Journal --> Viz["viz: HTML report"]
```

The arrows show runtime handoffs, not package imports. `hosted` and `spawn`
depend on the core; the core has no dependency on a particular engine.

## Calls and limits

Construct `Workflow(runner=...)` once and share it across concurrent tasks.

| Setting or method | Meaning |
| --- | --- |
| `max_concurrent` | Maximum simultaneous runner invocations; default 8, minimum 1. |
| `max_calls` | Maximum live runner invocations in this workflow; omitted means unlimited, 0 allows replay only. |
| `replay_scope` | Extra identity string, default empty. Set it when work depends on a repository revision or configuration absent from the input. |
| `agent(prompt~, kind~)` | Sends `{query, hints?}`; the label defaults to a shortened prompt. |
| `agent_call(kind~, input~, label~)` | Sends the exact JSON shape a kind requires. |
| `agent_as[T : FromJson](...)` | Calls `agent`, then decodes the complete report as `T`. |
| `calls_made()` / `calls_replayed()` | Live runner invocations / outcomes served from prior journal entries. |
| `tokens_spent()` / `cost_usd()` | Usage from resolved live outcomes in this run. Historical replay usage is excluded. |

The call allowance counts entry into the runner, even if it refuses work or
cannot spawn a process. Queue cancellation and replay consume no allowance.
`max_steps` and `schema` are requests to the runner; the core does not enforce
an engine's step limit or JSON Schema. `agent_as` independently decodes the
returned value. There is no core token budget or wall deadline; process
deadlines belong to `spawn` and `hosted`.

## The failure model

Everything else follows from three decisions:

- **Failure is data, in a typed channel.** An agent that produced no
  usable report raises `WorkflowError::AgentFailed` with a typed
  `AgentFailure` cause; `attempt` folds it into a `Result` for fan-out
  sites. A failure is never a silent `null`: it either raises
  `AgentFailed` or lands in `attempt`'s `Result`.
- **Resolved failures retain their usage.** `AgentOutcome` carries the attempt's spend on
  BOTH arms — a timed-out child's tokens land in `tokens_spent()` just
  like a success's. `attempt=None` means no launch was ever tried.
  Engines that price their work report it too: `cost_usd()` sums their
  figures, and stays a floor when an engine in the mix prices nothing.
- **Cancellation propagates.** Engine bugs `raise` through; cancellation
  is a signal that bypasses `catch` and unwinds through `defer`/`errdefer`.
  Either way the task group is cancelled and no journalled outcome is
  produced. Usage from an interrupted call is not added to the core counters:
  a child reports its spend in its result, which a cancelled call never
  reads, so the hosted sidecar marks such a launch `unaccounted`.

## A workflow, end to end

One phase fans three verifiers out over a finding, a quorum policy
requires at least 2 of the 3 CALLS to succeed (agreement between the
returned verdicts is then the script's own judgment, as the filter below
shows), and every launch is capped by the shared semaphore:

```mbt check
///|
async fn verdict_runner(call : @workflow.AgentCall) -> @workflow.AgentOutcome {
  @async.pause()
  let verdict : Json = if call.input is { "query": String(q), .. } &&
    q.contains("lens=repro") {
    { "confirmed": false }
  } else {
    { "confirmed": true }
  }
  Finished(
    value=verdict,
    attempt=@workflow.AgentAttempt(
      attempt_id="sr-\{call.label}",
      steps_used=3,
      prompt_tokens=70,
      completion_tokens=30,
    ),
  )
}

///|
async test "fan out three lenses, gate on a 2-of-3 quorum" {
  let wf = @workflow.Workflow(
    runner=@workflow.Runner(verdict_runner),
    max_concurrent=4,
    max_calls=16,
  )
  wf.phase("Verify")
  let results = @workflow.fan_out(["correctness", "security", "repro"], lens => {
    @workflow.attempt(() => {
      wf.agent(
        kind="judge",
        prompt="Judge the finding through lens=\{lens}: real?",
        label="verify:\{lens}",
      )
    })
  })
  let confirmed = @workflow.collect_ok(results, min_ok=2)
    .filter(v => v is { "confirmed": True, .. })
    .length()
  assert_eq(confirmed, 2)
  assert_eq(wf.calls_made(), 3)
  assert_eq(wf.tokens_spent(), 300)
}
```

`fan_out` gives every item its own `Result` slot — one lost verifier
never poisons its siblings, and the tokens they spent stay spent. When
later work is worthless without ALL of a stage, give `parallel` thunks
that raise instead (bare `wf.agent` calls, no `attempt`): the first failure
cancels every sibling still in flight. The policies are one identifier each:
`all_ok` and `collect_ok(min_ok?)`. Both fan-outs are one task group: a raise
inside one cancels the rest and unwinds the whole stage, so nothing outlives
it.

| Policy | When to use it |
| --- | --- |
| `parallel` / `fan_out` with `attempt` | Keep a `Result` for each input, in input order. Typed workflow failures stay in their own slots. |
| `parallel` with raising thunks | Every branch is required; a raise cancels siblings still running. |
| `all_ok` | Require all already-collected results to succeed; raises the first error in array order. |
| `collect_ok(min_ok=0)` | Keep successes, optionally requiring a minimum count (a quorum); it does not compare answers for agreement. |

`all_ok` and `collect_ok` inspect completed results; they do not
cancel work early. `attempt` captures only `WorkflowError`, including
`AgentFailed`, `CallBudgetExhausted`, and `QuorumNotReached`.

A model that answered off-schema or ran out of steps can be worth asking
again. `retry` wraps a raising step with `max_retry` extra attempts (default
1), spaced by `backoff` (default `Immediate`). Its default `worth_retrying`
policy retries `AgentFailed` except `Skipped`; it does not retry a spent
call budget or unmet quorum. A custom `retriable` callback can change that
typed-error policy. Engine bugs and cancellation always propagate.

Put `attempt` around `retry`, rather than passing `retry` a step that already
returns `Result`. Each live retry uses a slot and call allowance, and its
resolved outcome is journalled. On resume, an earlier successful report can
be replayed:

```mbt check
///|
async test "retry until the engine actually answers" {
  let flaky : Ref[Int] = { val: 0, }
  let wf = @workflow.Workflow(
    runner=@workflow.Runner(call => {
      @async.pause()
      flaky.val += 1
      if flaky.val < 3 {
        DidNotFinish(failure=NoReport, attempt=None)
      } else {
        Finished(
          value={ "attempt": flaky.val },
          attempt=@workflow.AgentAttempt(
            attempt_id="sr-\{call.label}",
            steps_used=1,
            prompt_tokens=40,
            completion_tokens=10,
          ),
        )
      }
    }),
  )
  let report = @workflow.retry(
    () => wf.agent(prompt="Name the worst bug in spawn/", kind="judge"),
    max_retry=2,
  )
  assert_eq(report, { "attempt": 3 })
  assert_eq(wf.calls_made(), 3)
}
```

`agent` builds the explore input shape — `{query, hints?}` — because that
is what most calls are. A kind with its own shape (the contract's
`review` and `worker`) goes through `agent_call`, which sends the exact
JSON it is given; that input IS the replay identity, so an engine
encoding its own kinds gets replay for free. `attempt` folds the typed
error channel around ANY raising step, so those calls reach a fan-out
without a `try_` twin per entry point:

```mbt check
///|
async fn slice_runner(call : @workflow.AgentCall) -> @workflow.AgentOutcome {
  @async.pause()
  guard call.input is { "task": String(task), .. } else {
    return DidNotFinish(failure=Failed("not a worker input"), attempt=None)
  }
  Finished(
    value={ "status": "done", "task": task },
    attempt=@workflow.AgentAttempt(
      attempt_id="sr-\{call.label}",
      steps_used=1,
      prompt_tokens=10,
      completion_tokens=5,
    ),
  )
}

///|
async test "fan a worker kind out through its own input shape" {
  let wf = @workflow.Workflow(runner=@workflow.Runner(slice_runner))
  let slices = ["rename the seam", "widen the ceiling"]
  let done = @workflow.fan_out(slices, task => {
    @workflow.attempt(() => {
      wf.agent_call(
        kind="worker",
        input={ "task": task, "worker_root": "/w" },
        label="worker:\{task}",
      )
    })
  })
  assert_eq(@workflow.all_ok(done).length(), 2)
}
```

When the script wants a TYPE rather than JSON, decode at the boundary:
`agent_as` runs the same call and turns a report that does not satisfy
the type into a typed `AgentFailed(Failed("report rejected: …"))` carrying
the decoder's path. A `schema` rides the request so an engine
that can constrain its model to the shape does (the Claude and Codex
shims do); decoding still runs, because the engine is not trusted to
validate, and the schema is part of the call's replay identity:

```mbt check
///|
struct Confirmation {
  confirmed : Bool
} derive(FromJson)

///|
async test "decode the report at the boundary" {
  let wf = @workflow.Workflow(runner=@workflow.Runner(verdict_runner))
  let verdict : Confirmation = wf.agent_as(
    prompt="Judge the finding through lens=security: real?",
    kind="judge",
    schema={
      "type": "object",
      "properties": { "confirmed": { "type": "boolean" } },
      "required": ["confirmed"],
    },
  )
  assert_true(verdict.confirmed)
}
```

Multi-stage pipelines are just function composition inside the fan-out —
stages need no barrier between them, so composing them per-item IS the
pipeline:

```mbt check
///|
async test "find then verify, with no barrier between the stages" {
  let wf = @workflow.Workflow(runner=@workflow.Runner(verdict_runner))
  let verified = @workflow.fan_out(["pkg/a", "pkg/b"], target => {
    let finding = @workflow.attempt(() => {
      wf.agent(prompt="Find the worst bug in \{target}", kind="judge")
    })
    match finding {
      // Each finding proceeds to verification the moment ITS finder
      // returns — b's finder may still be running while a verifies.
      Ok(_) =>
        @workflow.attempt(() => {
          wf.agent(
            prompt="Adversarially verify the finding in \{target}",
            kind="judge",
          )
        })
      Err(error) => Err(error)
    }
  })
  assert_eq(@workflow.all_ok(verified).length(), 2)
}
```

## Replay: crash, resume, and pay only for new work

Every live outcome is appended to the `Journal` — the call's WORK
identity (kind, input, max_steps, scope, schema — never the display label) plus
the lossless outcome, plus attribution for readers: the phase the call
was issued under and the wall-clock window it ran in, as milliseconds
since the epoch. Attribution never takes part in replay matching.
Re-running the same program against the same journal
replays successes for free, keeps a human's `Skipped` refusal standing,
and re-attempts other failures — getting past those is what resume is
for:

```mbt check
///|
async test "the second generation replays instead of re-paying" {
  let journal = @workflow.Journal::in_memory()
  let wf1 = @workflow.Workflow(
    runner=@workflow.Runner(verdict_runner),
    journal~,
  )
  let first = wf1.agent(
    prompt="Judge the finding through lens=security: real?",
    kind="judge",
  )
  assert_eq(wf1.tokens_spent(), 100)

  // Same program, next generation: served from the journal — no launch,
  // no slot, no fresh spend.
  let wf2 = @workflow.Workflow(
    runner=@workflow.Runner(verdict_runner),
    journal=@workflow.Journal::in_memory(prior=journal.recorded()),
  )
  assert_eq(
    wf2.agent(
      prompt="Judge the finding through lens=security: real?",
      kind="judge",
    ),
    first,
  )
  assert_eq(wf2.calls_made(), 0)
  assert_eq(wf2.calls_replayed(), 1)
  assert_eq(wf2.tokens_spent(), 0)
}
```

File-backed journals (`@workflow.Journal::load(path)`) are append-only JSONL,
accumulated across generations. A malformed final line is dropped and the
file repaired; corruption anywhere else raises `JournalCorrupted`. Repair
preserves the healthy bytes without re-encoding them. Identical concurrent
calls are treated as intentional samples, such as three identical
verifiers, and consume separate prior entries.

```mermaid
flowchart TD
  Call["Call identity: kind, input, max_steps, scope, schema"] --> Candidate{"Prior replay candidate?"}
  Candidate -->|Success first, then Skipped| Validate{"Runner accepts candidate?"}
  Validate -->|Serve| Replay["Return or raise without a launch"]
  Validate -->|Rerun| Candidate
  Candidate -->|None left| Slot["Acquire concurrency slot and check call budget"]
  Slot --> Live["Invoke runner"]
  Live -->|Resolved outcome| Record["Account usage and append journal entry"]
  Live -->|Cancellation or infrastructure error| Raise["Re-raise; no outcome entry"]
  Record --> Result["Return report or raise AgentFailed"]
```

Replay is from prior entries only: two identical calls in the same live run
still execute twice. Among matching prior entries, successes are considered
in file order before recorded `Skipped` refusals. Other failures are not
replay candidates. Labels, phases, and timestamps do not affect identity.

Use one writing workflow process per journal path. Appends are serialized
within a journal instance, not across processes. `Journal::load` is suitable
for resuming a stopped writer, not watching an active one; use
[`viz`](viz/README.mbt.md) for a reader that never repairs the source file.
A complete final entry missing only its newline is retained and the newline
is restored. A malformed tail is removed. Each stored entry has the
versioned envelope `{"v":1,"e":...}`. `prior()` and `recorded()` return copies
of the loaded entries and this instance's new outcomes respectively.

`agent_as` decodes after the raw report has been recorded. If decoding fails,
the journal still holds a `Finished` raw report, and resume can reproduce
the same decode error. Change the input, schema, or scope when the task's
meaning changes, or reject an unsuitable candidate in `validate_replay`.
The journal does not make arbitrary side effects exactly once: a process
can finish work and crash before its outcome is durably recorded.

A v1 line that carries no entry is metadata: replay skips it and repair
preserves it, so a tool can annotate a journal, and a declared plan has a
place to live when one arrives. Labels follow the convention
`stage:instance` (`verify:claude:3`, `survey:journal`): readers group the
instances of one stage by the prefix before the first colon.

## Observability

`@workflow.Workflow(on_event=...)` narrates the run: `PhaseStarted` and an
`AgentStarted`/`AgentFinished` bracket that balances on EVERY path —
success, typed failure, cancellation (`Interrupted`), and infrastructure
error (`Errored`) — plus `AgentReplayed` for journal hits. Purely
observational: no control flow rides on events.

## Plugging in an engine

A `Runner` wraps one async `AgentCall -> AgentOutcome` function. Return
`Finished(value~, attempt~)` for a report or
`DidNotFinish(failure~, attempt~)` for an expected unsuccessful call.
Use `attempt=None` when there is no attempt to account for. Propagate
cancellation and unexpected infrastructure errors.

The checked examples above use in-process runners. To run a process, use
[`spawn.contract_runner`](spawn/README.mbt.md): its `launch` callback returns
an executable and argv naming `{result_file}`, and the package handles the
child contract, accounting, deadlines, and teardown: the child writes one
result file there, as OpenSeek's `openseek run` and this module's shims do. A `Runner::invoke` call is useful for
routing one runner to another; it does not itself add workflow limits,
journalling, or replay.

For reports that name mutable resources, supply `validate_replay` when
constructing the runner. The validator receives the current call and a
candidate outcome. `Serve(outcome)` accepts it, possibly with refreshed
resource handles. `Rerun` consumes that candidate and tries the next match;
if none is acceptable, the workflow runs live. A validation error rolls
back the candidate claim and propagates. Set `replay_scope` to distinguish
work whose meaning depends on a revision, model, or configuration outside
its input; the workflow does not add those coordinates automatically.

## Hosting and CLI adapters

Choose [`hosted`](hosted/README.mbt.md) when a parent supplies the executable,
reserved child ids, and output paths in `WORKFLOW_HOST`. `ctx.run` creates
the workflow and attaches its journal and launch sidecar. The host owns
reservation allocation; this handoff does not implement a sandbox.
[Host handoff](docs/host-handoff.md) defines the configuration protocol.

The two executable adapters make existing agent CLIs speak the
[child contract](docs/child-contract.md); launch one as
`moonx moonbitlang/workflow/shim/claude --result-file {result_file} …`:

| Adapter | Default tool policy | Step definition | Report |
| --- | --- | --- | --- |
| [`shim/claude`](shim/claude/README.mbt.md) | Disallows listed editing and shell tools. | Distinct assistant message ids. | `{answer, engine, session_id?, num_turns?, cost_usd?}` |
| [`shim/codex`](shim/codex/README.mbt.md) | Requests the `read-only` sandbox. | Completed items excluding reasoning and notices. | `{answer, engine, thread_id?, notices?}` |

Both select their write-capable policy for `kind="worker"` or `--writable`.
Their `max_steps` enforcement is reactive: an observed step over the limit
is counted before the CLI is stopped. Both accept a schema for the inner
`answer`; `agent_as` decodes the complete report wrapper, so its result type
must include that wrapper. The adapter READMEs describe the exact options,
usage limitations, and recorded CLI versions.

The [`shim`](shim/README.mbt.md) library provides shared framing and process
lifetime helpers for adding another CLI dialect. No framework package
provisions worktrees or integrates worker edits; a controller that needs
those operations must implement them.

## Examples and development

[`examples/scout`](examples/scout/README.mbt.md) is a small local executable
for keyless probes and journal replay. The `.mbtx` scripts in `examples/`
are larger standalone programs:

- [`simplify.mbtx`](examples/simplify.mbtx) runs a simplification sweep,
  choosing a hosted runner when configured or a standalone OpenSeek runner.
- [`compose.mbtx`](examples/compose.mbtx) routes calls across multiple engines.
- [`review-simplify.mbtx`](examples/review-simplify.mbtx) asks Codex and
  Claude to review every package, without a host. It groups both reports
  by package (or package batch) in `simplification-review.md`, keeping engine attribution and
  failures visible. Suggestions are not applied or treated as consensus.
  It defaults to two concurrent calls across both engines, through the
  published shims, with a fresh journal per run.

```sh
moonx ./examples/review-simplify.mbtx --list
moonx ./examples/review-simplify.mbtx --jobs 2
# Codex-only sweep of another module, amortizing CLI startup across four packages.
moonx ./examples/review-simplify.mbtx --repo ../openseek --engines codex --batch-size 4
moonx ./examples/review-simplify.mbtx --engines claude shim/codex
# Models are optional; omitted settings inherit each CLI's configuration.
moonx ./examples/review-simplify.mbtx --codex-model MODEL --claude-model MODEL
# Extra CLI arguments are JSON arrays, preserving each argument verbatim.
moonx ./examples/review-simplify.mbtx --codex-args '["-c","model_reasoning_effort=\"high\""]' --claude-args '["--effort","high"]'
```

`--batch-size` defaults to 1; larger values share one review call across several
packages. Increase `--steps` / `--deadline-ms` for larger batches as needed.
Discovery lists excluded nested modules; review each separately with `--repo`.
Each package still receives its own section, with suggestions categorized for
separate commits. The prompt checks comprehensions and conditional array spreads
where clearer, preserving evaluation order, side effects, and error timing.
Failed calls remain in the report and produce a nonzero exit status.

`--codex-shim` and `--claude-shim` accept locally built native executables;
`--out` selects the combined report path. Additional CLI arguments can
select profiles/settings, but must preserve JSON output and the read-only
policy. Claude's default shim disables Bash as well as editing tools, so
its reviewer uses file-reading tools. Review steps have different meanings
across engines; `--steps` is not a common model-call or token budget.

Those scripts use the published module without a version pin. They are not
part of `moon test`; run one with `moonx ./<script.mbtx>` (or
`moon run <script.mbtx> --`).
Running a model-backed example requires its engines and credentials.

From the repository root, the normal validation sequence is:

```sh
just check
just test
just tidy
```

`check` type-checks native and wasm with warnings denied and checks formatting.
`test` runs unit, process-contract, and checked README examples on both
targets. `tidy` regenerates package interfaces and formats source; review the
`pkg.generated.mbti` diff when changing APIs.

The core implementation is split by responsibility:
[`workflow.mbt`](workflow.mbt) handles calls and accounting,
[`combinators.mbt`](combinators.mbt) supplies concurrency and failure policies,
[`runner.mbt`](runner.mbt) defines the engine boundary,
[`journal.mbt`](journal.mbt) implements persistence and replay matching, and
[`types.mbt`](types.mbt) defines calls, attempts, and failures. The generated
[`package interface`](pkg.generated.mbti) lists all public signatures.