#moonbitlang/workflow

#Setup and package map

moon add moonbitlang/workflow

import {
  "moonbitlang/workflow",
  "moonbitlang/async",
}

supported_targets = "native+wasm"

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"]

#Calls and limits

#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.
• 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 and cancellation raise through and cancel the task group. They do not produce a journalled outcome. Usage from an interrupted call is not added to the core counters; adapters that need it during teardown can use spawn.ContractProgress.

#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=@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.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=@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)
}

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

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

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"]

#Observability

#Plugging in an engine

#Hosting and CLI adapters

#Examples and development

• simplify.mbtx runs a simplification sweep, choosing a hosted runner when configured or a standalone OpenSeek runner.
• compose.mbtx routes calls across multiple engines.

just check
just test
just tidy