langgraph-orchestrator-worker

The langgraph-orchestrator-worker worked example.

Run it from sema/:

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

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

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

Parameters

name	type
topic	str
count	int

Returns list[str]

Effects !{}

def scripted_executor

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

Parameters

name	type
packet	dict[str, any]

Returns str

Effects !{}

agent section_writer

agent section_writer(section: str) -> str

Parameters

name	type
section	str

Returns str

agent flaky_writer

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

Parameters

name	type
section	str
attempt	int

Returns str

circuit write_report

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

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

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

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

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

Parameters

name	type
count	int

Returns str

Effects !{model.invoke}

def run_retry

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

Returns str

Effects !{model.invoke, agent.spawn}

def approval_value

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

Parameters

name	type
approved	bool
note	str

Returns any

Effects !{}

def interrupt_value

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

Parameters

name	type
approved	bool
summary	str

Returns any

Effects !{}

def run_approval

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

Parameters

name	type
approved	bool

Returns str

Effects !{model.invoke}

def run_single

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

Parameters

name	type
name	str

Returns str

Effects !{model.invoke, agent.spawn}

def main

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

Returns str

Effects !{model.invoke, agent.spawn}