#moonbitlang/workflow

#The failure model

• 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.
• Cost is never lost. 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 is not failure. Engine bugs and cancellation raise through and cancel the task group; the launch allowance only counts agents that actually launched — a call cancelled while queued costs nothing.

#A workflow, end to end

///|
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={
    attempt_id: "sr-\{call.label}",
    steps_used: 3,
    prompt_tokens: 70,
    completion_tokens: 30,
    cost_usd: None,
  })
}

///|
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.quorum(results, need=2)
    .filter(v => v is { "confirmed": True, .. })
    .length()
  assert_eq(confirmed, 2)
  assert_eq(wf.calls_made(), 3)
  assert_eq(wf.tokens_spent(), 300)
}

///|
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={
          attempt_id: "sr-\{call.label}",
          steps_used: 1,
          prompt_tokens: 40,
          completion_tokens: 10,
          cost_usd: None,
        })
      }
    }),
  )
  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)
}

///|
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={
    attempt_id: "sr-\{call.label}",
    steps_used: 1,
    prompt_tokens: 10,
    completion_tokens: 5,
    cost_usd: None,
  })
}

///|
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)
}

///|
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)
}

///|
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

///|
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)
}

#Observability

#Plugging in an engine

///|
let runner = @spawn.contract_runner(launch=_ => {
  command: "my-engine",
  args: [],
  cwd: None,
  extra_env: None,
  deadline_ms: None,
})

#Running inside a host

///|
async fn main {
  guard @hosted.context() is Some(ctx) else { return }
  ctx.run(wf => {
    wf.phase("survey")
    let found = @workflow.parallel([
      () => {
        @workflow.attempt(() => wf.agent(prompt="where is X?", kind="explore"))
      },
      () => {
        @workflow.attempt(() => wf.agent(prompt="who calls Y?", kind="explore"))
      },
    ])
    for answer in @workflow.collect_ok(found) {
      println(answer.stringify())
    }
  })
}

#Claude Code and Codex as engines

///|
let runner = @spawn.contract_runner(launch=call => {
  command: "moonx",
  // read-only by default; `worker` calls (or --writable) may edit
  args: [
    "moonbitlang/workflow/shim/claude", "--model", "claude-sonnet-5", "--", "--max-budget-usd",
    "2",
  ],
  cwd: None,
  extra_env: None,
  deadline_ms: None,
})

just shims   # _build/native/debug/build/shim/{claude,codex}/*.exe

///|
let runner = @workflow.Runner(call => {
  // spawn something, await it, and account honestly:
  Finished(value=report_json, attempt={
    attempt_id,
    steps_used,
    prompt_tokens,
    completion_tokens,
    cost_usd: None, // `Double?` has no default: name it, even when unknown
  })
})