<!-- Sema documentation — langgraph-orchestrator-worker
     Sema is a new AI-native, neurosymbolic programming language most models
     have not been trained on. Full corpus: https://sema.49.12.246.95.sslip.io/llms-full.txt
     Install language support: https://sema.49.12.246.95.sslip.io/install-skill/ -->

# langgraph-orchestrator-worker

> The langgraph-orchestrator-worker worked example.

> The langgraph-orchestrator-worker worked example.

Run it from `sema/`:

```bash
sema check examples/langgraph-orchestrator-worker
SEMA_STRICT=1 sema run examples/langgraph-orchestrator-worker
sema assure examples/langgraph-orchestrator-worker --grade silver
```

## Source

### `src/main.sema`

```sema
"""LangGraph-parity orchestrator-worker flow over durable native Sema circuits.

The pinned reference (`reference/oracle.py`, langgraph==1.2.9 +
langchain-core==1.4.9) EXECUTES LangGraph: reducer-backed state, dynamic
`Send` worker dispatch, checkpointer persistence, node caching, one retried
worker whose failed write rolls back, a typed approval interrupt/resume, and
replay/fork time travel. This fixture expresses the same scenario matrix with
native agents and circuits; `sema-runtime`'s `langgraph_parity` test compares
canonical final state, canonical worker event sequences, and counters.

`SCENARIO` selects one lane per process (all spellings are 8 bytes so the
parity test can swap scenarios and resume runs without moving source spans):

- `matrix00` (default): fan-out at 1/3/16/64 workers, retry-with-rollback,
  and the approval-request phase — one canonical JSON line each.
- `fanout01|fanout03|fanout16|fanout64`, `retry000`: single lanes. A resumed
  `fanout03` run is the replay lane (all worker leaves reused, zero model
  calls); resuming it after the planner's `Methods:`->`Results:` edit is the
  fork lane (exactly one leaf re-executes).
- `approve0`/`approve1`: approval request, then resume-with-decision on the
  same run id (all leaves reused). Sema has no durable typed interrupt value
  yet, so the decision arrives as typed circuit input on resume; the runtime
  gap is classified in the example README.
"""

assure silver

TOPIC = "Reliable agent workflows"
SCENARIO = "matrix00"

def plan_sections(topic: str, count: int) -> list[str] !{}:
    if count == 3:
        return ["Context: " + topic, "Methods: " + topic, "Risks: " + topic]
    mut sections = []
    for index in range(count):
        sections.append("Part " + str(index) + ": " + topic)
    return sections

@provides("agent.execute")
def scripted_executor(packet: dict[str, any]) -> str !{}:
    section = packet["inputs"]["section"]
    if packet["agent"] == "flaky_writer" and packet["inputs"]["attempt"] == 0 and section.startswith("Methods"):
        return ""
    return "drafted:" + section

agent section_writer(section: str) -> str by scripted_model:
    sem "Write exactly the assigned report section without reordering it"
    budget model_calls=1, tokens=64
    ensure len(result) > 0

agent flaky_writer(section: str, attempt: int) -> str by scripted_model:
    sem "Write the assigned section; the flaky section fails its first attempt"
    budget model_calls=1, tokens=64
    ensure len(result) > 0

circuit write_report(topic: str, count: int) -> list[str] !{model.invoke}:
    budget agents=64, spawn_depth=0, model_calls=64, tokens=16384
    sections = plan_sections(topic, count)
    return parallel [section_writer(section) for section in sections]

circuit write_report_retry(topic: str) -> dict[str, any] !{model.invoke, agent.spawn}:
    budget agents=4, spawn_depth=1, model_calls=4, tokens=1024
    sections = plan_sections(topic, 3)
    mut report = []
    mut failed = 0
    for section in sections:
        outcome = (spawn flaky_writer(section, 0)).join()
        match outcome:
            case Ok(text):
                report.append(text)
            case Err(_):
                failed = failed + 1
                report.append(flaky_writer(section, 1))
    return {"report": report, "failed": failed}

circuit approved_report(topic: str, approved: bool, note: str) -> dict[str, any] !{model.invoke}:
    budget agents=3, spawn_depth=0, model_calls=3, tokens=1024
    sections = plan_sections(topic, 3)
    report = parallel [section_writer(section) for section in sections]
    summary = "approve " + str(len(report)) + " sections"
    if approved:
        return {"status": "approved", "report": report, "note": note, "summary": summary}
    return {"status": "awaiting_approval", "report": report, "summary": summary}

def canonical(scenario: str, plan: list[str], sections: list[str], approval: any, interrupt: any, executed: int, reused: int, failed: int) -> str !{}:
    return json.dumps({
        "scenario": scenario,
        "final": {"topic": TOPIC, "plan": plan, "sections": sections, "approval": approval},
        "interrupt": interrupt,
        "counters": {"executed": executed, "reused": reused, "failed": failed, "model_calls": executed + failed},
    })

def run_fanout(count: int) -> str !{model.invoke}:
    mut calls = 0
    mut report = []
    with meter as usage:
        report = write_report(TOPIC, count)
        calls = usage.total_calls
    return canonical("fanout_" + str(count), plan_sections(TOPIC, count), report, None, None, calls, count - calls, 0)

def run_retry() -> str !{model.invoke, agent.spawn}:
    mut calls = 0
    mut outcome = {}
    with meter as usage:
        outcome = write_report_retry(TOPIC)
        calls = usage.total_calls
    failed = outcome["failed"]
    return canonical("retry_rollback", plan_sections(TOPIC, 3), outcome["report"], None, None, calls - failed, 0, failed)

def approval_value(approved: bool, note: str) -> any !{}:
    if approved:
        return {"approved": approved, "note": note}
    return "pending"

def interrupt_value(approved: bool, summary: str) -> any !{}:
    if approved:
        return None
    return {"summary": summary}

def run_approval(approved: bool) -> str !{model.invoke}:
    mut calls = 0
    mut outcome = {}
    with meter as usage:
        if approved:
            outcome = approved_report(TOPIC, true, "ship it")
        else:
            outcome = approved_report(TOPIC, false, "")
        calls = usage.total_calls
    mut note = ""
    if approved:
        note = outcome["note"]
    interrupt = interrupt_value(approved, outcome["summary"])
    return canonical("approval", plan_sections(TOPIC, 3), outcome["report"], approval_value(approved, note), interrupt, calls, 3 - calls, 0)

test "planner preserves canonical section order":
    check plan_sections("Sema", 3) == ["Context: Sema", "Methods: Sema", "Risks: Sema"]

test "planner scales to dynamic worker counts":
    check len(plan_sections("Sema", 64)) == 64
    check plan_sections("Sema", 2) == ["Part 0: Sema", "Part 1: Sema"]

test "scripted executor drafts deterministically and fails the flaky first attempt":
    check scripted_executor({"agent": "section_writer", "inputs": {"section": "Context: X"}}) == "drafted:Context: X"
    check scripted_executor({"agent": "flaky_writer", "inputs": {"section": "Methods: X", "attempt": 0}}) == ""
    check scripted_executor({"agent": "flaky_writer", "inputs": {"section": "Methods: X", "attempt": 1}}) == "drafted:Methods: X"

def run_single(name: str) -> str !{model.invoke, agent.spawn}:
    if name == "fanout01":
        return run_fanout(1)
    if name == "fanout03":
        return run_fanout(3)
    if name == "fanout16":
        return run_fanout(16)
    if name == "fanout64":
        return run_fanout(64)
    if name == "retry000":
        return run_retry()
    if name == "approve0":
        return run_approval(false)
    return run_approval(true)

def main() -> str !{model.invoke, agent.spawn}:
    require SCENARIO in ["matrix00", "fanout01", "fanout03", "fanout16", "fanout64", "retry000", "approve0", "approve1"]
    mut lines = []
    if SCENARIO == "matrix00":
        lines = [run_fanout(1), run_fanout(3), run_fanout(16), run_fanout(64), run_retry(), run_approval(false)]
    else:
        lines = [run_single(SCENARIO)]
    output = "\n".join(lines)
    print(output)
    return output
```

## Reflected API

# `main`

LangGraph-parity orchestrator-worker flow over durable native Sema circuits.

The pinned reference (`reference/oracle.py`, langgraph==1.2.9 +
langchain-core==1.4.9) EXECUTES LangGraph: reducer-backed state, dynamic
`Send` worker dispatch, checkpointer persistence, node caching, one retried
worker whose failed write rolls back, a typed approval interrupt/resume, and
replay/fork time travel. This fixture expresses the same scenario matrix with
native agents and circuits; `sema-runtime`'s `langgraph_parity` test compares
canonical final state, canonical worker event sequences, and counters.

`SCENARIO` selects one lane per process (all spellings are 8 bytes so the
parity test can swap scenarios and resume runs without moving source spans):

- `matrix00` (default): fan-out at 1/3/16/64 workers, retry-with-rollback,
  and the approval-request phase — one canonical JSON line each.
- `fanout01|fanout03|fanout16|fanout64`, `retry000`: single lanes. A resumed
  `fanout03` run is the replay lane (all worker leaves reused, zero model
  calls); resuming it after the planner's `Methods:`-&gt;`Results:` edit is the
  fork lane (exactly one leaf re-executes).
- `approve0`/`approve1`: approval request, then resume-with-decision on the
  same run id (all leaves reused). Sema has no durable typed interrupt value
  yet, so the decision arrives as typed circuit input on resume; the runtime
  gap is classified in the example README.

# `def plan_sections`

```sema
def plan_sections(topic: str, count: int) -> list[str] !{}
```

**Parameters**

| name | type |
|---|---|
| `topic` | `str` |
| `count` | `int` |

**Returns** `list[str]`

**Effects** `!{}`

# `def scripted_executor`

```sema
def scripted_executor(packet: dict[str, any]) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `packet` | `dict[str, any]` |

**Returns** `str`

**Effects** `!{}`

# `agent section_writer`

```sema
agent section_writer(section: str) -> str
```

**Parameters**

| name | type |
|---|---|
| `section` | `str` |

**Returns** `str`

# `agent flaky_writer`

```sema
agent flaky_writer(section: str, attempt: int) -> str
```

**Parameters**

| name | type |
|---|---|
| `section` | `str` |
| `attempt` | `int` |

**Returns** `str`

# `circuit write_report`

```sema
circuit write_report(topic: str, count: int) -> list[str] !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `topic` | `str` |
| `count` | `int` |

**Returns** `list[str]`

**Effects** `!{model.invoke}`

# `circuit write_report_retry`

```sema
circuit write_report_retry(topic: str) -> dict[str, any] !{model.invoke, agent.spawn}
```

**Parameters**

| name | type |
|---|---|
| `topic` | `str` |

**Returns** `dict[str, any]`

**Effects** `!{model.invoke, agent.spawn}`

# `circuit approved_report`

```sema
circuit approved_report(topic: str, approved: bool, note: str) -> dict[str, any] !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `topic` | `str` |
| `approved` | `bool` |
| `note` | `str` |

**Returns** `dict[str, any]`

**Effects** `!{model.invoke}`

# `def canonical`

```sema
def canonical(scenario: str, plan: list[str], sections: list[str], approval: any, interrupt: any, executed: int, reused: int, failed: int) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `scenario` | `str` |
| `plan` | `list[str]` |
| `sections` | `list[str]` |
| `approval` | `any` |
| `interrupt` | `any` |
| `executed` | `int` |
| `reused` | `int` |
| `failed` | `int` |

**Returns** `str`

**Effects** `!{}`

# `def run_fanout`

```sema
def run_fanout(count: int) -> str !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `count` | `int` |

**Returns** `str`

**Effects** `!{model.invoke}`

# `def run_retry`

```sema
def run_retry() -> str !{model.invoke, agent.spawn}
```

**Returns** `str`

**Effects** `!{model.invoke, agent.spawn}`

# `def approval_value`

```sema
def approval_value(approved: bool, note: str) -> any !{}
```

**Parameters**

| name | type |
|---|---|
| `approved` | `bool` |
| `note` | `str` |

**Returns** `any`

**Effects** `!{}`

# `def interrupt_value`

```sema
def interrupt_value(approved: bool, summary: str) -> any !{}
```

**Parameters**

| name | type |
|---|---|
| `approved` | `bool` |
| `summary` | `str` |

**Returns** `any`

**Effects** `!{}`

# `def run_approval`

```sema
def run_approval(approved: bool) -> str !{model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `approved` | `bool` |

**Returns** `str`

**Effects** `!{model.invoke}`

# `def run_single`

```sema
def run_single(name: str) -> str !{model.invoke, agent.spawn}
```

**Parameters**

| name | type |
|---|---|
| `name` | `str` |

**Returns** `str`

**Effects** `!{model.invoke, agent.spawn}`

# `def main`

```sema
def main() -> str !{model.invoke, agent.spawn}
```

**Returns** `str`

**Effects** `!{model.invoke, agent.spawn}`