#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; try_agent folds it into a Result for fan-out sites. Forgetting to choose a failure policy is a compile error, never a silent null.
• 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.
• 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,
  })
}

///|
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 => {
    wf.try_agent(
      kind="judge",
      "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 "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 = wf.try_agent("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(_) =>
        wf.try_agent(
          "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(
    "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("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,
})

///|
let runner = @workflow.Runner(call => {
  // spawn something, await it, and account honestly:
  Finished(value=report_json, attempt={
    subrun_id,
    steps_used,
    prompt_tokens,
    completion_tokens,
  })
})