Parallel Work: The Send API, Map-Reduce, and Fan-out/Fan-in (LangGraph in Production, Module 05)

This is Module 05 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

---

The problem: editing files one at a time

Imagine this issue lands in Forge’s queue:

Feature: add CSV export to report.py, expose the --csv flag in cli.py, and document the new option in README.md — three independent changes across three files.

After Module 4, Forge handles this competently. It triages the issue as feature, the Planner reads the codebase and produces a Plan with target_files = ["tally/report.py", "tally/cli.py", "README.md"], and then… it edits report.py, waits for the model to reply, edits cli.py, waits again, edits README.md, waits a third time. Three sequential calls, three wait cycles, three chances to miss context from the previous edit.

For a real codebase touching 5, 10, or 20 files per feature, sequential editing is not just slow — it is architecturally wrong. Nothing prevents cli.py’s editor from contradicting what report.py’s editor decided. And there is no mechanism to aggregate N partial results into a single, coherent state before the Tester runs.

The Send API is LangGraph’s answer. In this module you’ll give Forge a parallel editing pass: the Planner emits a list of FileTarget objects, a routing function fans them out with Send (one Editor per file, all running in the same superstep), each Editor produces a FileEdit, and a fan-in reducer (operator.add) aggregates all edits into edits before the Tester ever sees them.

---

In this module

You’ll learn:

• How LangGraph’s Pregel/BSP model executes parallel branches and guarantees fan-in coherence without locks (CA5.1)
• How to wire static parallel branches with multiple add_edge calls from one node (CA5.2)
• How to dispatch N workers at runtime with the Send API — from langgraph.types import Send — using a routing function that returns list[Send] (CA5.3, CA5.4)
• How to collect results in shared state with operator.add and the defer=True node flag (CA5.5, CA5.6)
• How to combine a state update and a fan-out atomically with Command(goto=[Send(...), ...]) (CA5.7)

You’ll build: Forge gains a parallel editing pass. The Planner emits targets: list[FileTarget], a routing function fans them out with Send (one Editor per file, same superstep), each Editor produces a FileEdit, and the fan-in reducer (operator.add) aggregates all edits into edits before handing off to the Tester. The write_file tool is added. FileTarget and FileEdit are frozen.

Concepts covered: CA5 — Parallelism & map-reduce (core). This module BUILDS CA5.1–CA5.7 of the LangGraph theory inventory.

Prerequisites: Modules 1–4, the cumulative forge/ package (Planner that emits a Plan with target_files), .env with ANTHROPIC_API_KEY (or the Ollama fallback for $0).

---

Where you are

M1 ✅  Setup, init_chat_model, first StateGraph
M2 ✅  ForgeState, intake node, reducers
M3 ✅  Triage router, test-fix loop, sandbox
M4 ✅  Planner, create_agent, Plan + PlanStep
M5 👉  The Send API, fan-out/fan-in, Editor×N (this module)
M6 ⬜  Persistence: SqliteSaver, threads, resumable runs
M7 ⬜  Human-in-the-loop: interrupt(), plan approval, PR approval
M8–M16 ⬜

Forge before this module: Planner produces a Plan with target_files, then Forge edits one file sequentially in propose_edit, then Tester runs.

Forge after this module: Planner produces Plan → dispatch fans out via Send → Editor×N run in parallel in the same superstep → apply node (with defer=True) waits for all editors and aggregates edits → Tester runs.

One important note on what this module does NOT do: runs still don’t survive a process crash. That requires a checkpointer, which is Module 6. And the fan-out starts immediately after planning — no human approves the plan first. That’s Module 7.

---

The Pregel/BSP model: supersteps and barriers

Before touching any code, you need to internalize the execution model. LangGraph runs on Pregel — a graph computation framework named after Google’s 2010 paper. Pregel uses Bulk Synchronous Parallel (BSP) execution:

1. All currently active nodes execute simultaneously. This is called a superstep.
2. When all active nodes finish, a BSP barrier applies their writes to the shared state atomically.
3. The next superstep begins with the updated state.

This is the guarantee that makes map-reduce safe without locks. During a superstep, the Editor processing report.py and the Editor processing cli.py run concurrently. They cannot read each other’s outputs mid-execution — they just write their results. The barrier then applies those writes via the reducer before any downstream node sees them.

graph TD
    S1["Superstep 1 — Planner\n(produces targets: list[FileTarget])"]
    B1[/"BSP barrier: writes applied"/]
    S2["Superstep 2 — dispatch routing fn\n(returns list[Send])"]
    B2[/"BSP barrier"/]
    S3["Superstep 3 — Editor × N\n(all run in PARALLEL, same superstep)"]
    B3[/"BSP barrier: operator.add merges all edits"/]
    S4["Superstep 4 — apply node (defer=True)\n(all FileEdits now collected)"]
    B4[/"BSP barrier"/]
    S5["Superstep 5 — Tester\n(pytest on complete sandbox)"]

    S1 --> B1 --> S2 --> B2 --> S3 --> B3 --> S4 --> B4 --> S5

The key guarantee: every Editor in Superstep 3 finishes before the BSP barrier fires. The apply node in Superstep 4 therefore always sees a complete edits list — not a partial one.

Static parallel branches first

Before the dynamic dispatch case, it is worth understanding static parallel branches, because they reveal the underlying BSP model clearly. If you always want exactly two Editors and you know their names at compile time, you can wire them like this:

# Static fan-out: fixed number of branches, N known at build time
builder.add_edge("planner", "editor_report")
builder.add_edge("planner", "editor_cli")
builder.add_edge("editor_report", "apply")
builder.add_edge("editor_cli", "apply")

LangGraph sees that apply has two incoming edges from the same source and knows to wait for both before the next superstep. This is valid, but it only works when N is fixed. If the Planner decides to touch 1 file on a small bug and 7 files on a large feature, static branches break down. You need the number of workers to be a runtime decision.

That is exactly what the Send API solves.

---

The Send API: dynamic fan-out at runtime

The Send API lets a routing function return a list of dispatch instructions — one per worker — computed at runtime from the current state.

The critical import line (and one of the most common v0→v1 migration mistakes):

# v1.x — always import from langgraph.types
from langgraph.types import Send

# ❌ The old v0.x path — this will fail or behave incorrectly in v1.x
# from langgraph.constants import Send

Send("node_name", arg) tells the graph: “dispatch one invocation of node_name, passing arg as its input.” The arg is private to that worker — it is not the global shared state. Each worker gets its own isolated input dict.

Here is the dispatch function from the actual Forge lab code:

# forge/nodes.py — the fan-out routing function (Module 5)
from langgraph.types import Send

def dispatch(state: ForgeState) -> list[Send]:
    """Fan out one editor task per target file (the Send API)."""
    return [
        Send("editor", {"repo_path": state["repo_path"], "issue": state["issue"], "target": t})
        for t in state["targets"]
    ]

Three things to notice:

1. The function signature is (state: ForgeState) -> list[Send]. Any routing function that returns list[Send] triggers dynamic fan-out — LangGraph handles the rest.
2. Each Send passes a private dict to the editor, not the full ForgeState. The editor receives {"repo_path": ..., "issue": ..., "target": t}. Workers are isolated.
3. The number of Send objects equals len(state["targets"]). If the Planner produced three FileTarget objects, three Editors run in parallel. If it produced one, one runs. Zero targets yields zero sends — the apply node still fires, but with an empty edits list (a degenerate but valid case that the tests verify).

Wiring the fan-out in graph.py

In build_graph(), you attach the dispatch routing function as a conditional edge from the planner node:

# forge/graph.py — wiring the fan-out (Module 5)
from langgraph.graph import END, START, StateGraph
from forge import nodes
from forge.state import ForgeState

def build_graph():
    builder = StateGraph(ForgeState)
    builder.add_node("intake",  nodes.intake)
    builder.add_node("triage",  nodes.triage)
    builder.add_node("planner", nodes.planner)
    builder.add_node("editor",  nodes.editor)
    builder.add_node("apply",   nodes.apply_edits, defer=True)  # fan-in gate
    builder.add_node("tester",  nodes.tester)
    builder.add_node("fixer",   nodes.fixer)

    builder.add_edge(START, "intake")
    builder.add_edge("intake", "triage")
    builder.add_conditional_edges(
        "triage", nodes.route_by_type,
        {"bug": "planner", "feature": "planner", "chore": "planner"},
    )
    builder.add_conditional_edges("planner", nodes.dispatch, ["editor"])  # fan-out
    builder.add_edge("editor", "apply")
    builder.add_edge("apply", "tester")
    builder.add_conditional_edges(
        "tester", nodes.route_after_tests,
        {"done": END, "give_up": END, "fixer": "fixer"},
    )
    builder.add_edge("fixer", "tester")
    return builder.compile()

The line builder.add_conditional_edges("planner", nodes.dispatch, ["editor"]) is the key. When dispatch returns [Send("editor", {...}), Send("editor", {...}), ...], LangGraph spawns one invocation of "editor" per Send, all in the same superstep. The ["editor"] list is the set of possible target nodes (for graph compilation validation).

---

Fan-in: collecting results with operator.add and defer=True

The reducer channel

Each Editor worker writes one FileEdit into the shared state. The trick is that multiple parallel writes to the same channel need a reducer — otherwise, last-write-wins would mean only one Editor’s output survives.

This is where operator.add comes in. In ForgeState, edits is declared as:

# forge/state.py — ForgeState (Module 5 additions)
import operator
from typing import Annotated
from forge.models import FileEdit, FileTarget, Plan, TestResult

class ForgeState(TypedDict, total=False):
    # --- Module 2 ---
    issue: str
    messages: Annotated[list[AnyMessage], add_messages]
    repo_path: str
    # --- Module 3 ---
    route: str
    attempts: int
    test_result: TestResult | None
    # --- Module 4 ---
    plan: Plan | None
    # --- Module 5 ---
    targets: list[FileTarget]                          # overwrite (fan-out units)
    edits: Annotated[list[FileEdit], operator.add]     # THE aggregation reducer
    # --- Module 7 adds: plan_approved, pull_request ---
    # --- Module 12 adds: review ---

edits is the only field in ForgeState that uses operator.add. Every other field is last-write-wins. When Editor A writes {"edits": [edit_for_report]} and Editor B writes {"edits": [edit_for_cli]}, the BSP barrier applies both writes through the reducer: old_edits + [edit_for_report] + [edit_for_cli]. Both FileEdit objects end up in the list.

The editor node itself is clean and simple:

# forge/nodes.py — the Editor map worker
from forge.models import FileEdit, FileTarget

def editor(state: dict) -> dict:
    """Map worker: edit ONE file. Receives its target via the Send payload."""
    target: FileTarget = state["target"]   # private input, not the global state
    current = (Path(state["repo_path"]) / target.path).read_text()
    model = config.get_model("smart")
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=(
            f"Issue:\n{state['issue']}\n\nEdit file {target.path}: {target.instruction}\n"
            f"Current contents:\n{current}"
        )),
    ])
    edit = FileEdit(
        path=target.path,
        new_content=extract_code(resp.content),
        summary=target.instruction,
    )
    return {"edits": [edit]}   # operator.add concatenates across all editors

Notice that state here is a plain dict, not a ForgeState. Each worker receives only what the Send passed to it. The worker never sees state["targets"] or state["plan"] — just {"repo_path": ..., "issue": ..., "target": <one FileTarget>}.

defer=True: the fan-in gate

Here is the subtle part. After each Editor finishes, LangGraph could theoretically schedule the apply node immediately, with only the edits collected so far. If Editor A finishes first and apply runs before Editor B, you would apply only one edit, the Tester would run on a half-updated codebase, and you would get failures that look completely mysterious.

defer=True prevents this. When you declare:

builder.add_node("apply", nodes.apply_edits, defer=True)

LangGraph holds that node back until all the Send tasks that route into it have finished. Only then does apply get scheduled. It is the explicit fan-in gate.

Without defer=True, the behavior is non-deterministic. The first Editor to finish could trigger apply with a partial edits list. This is one of the most common bugs when first using the Send API.

The apply_edits node then has the complete list:

# forge/nodes.py — the fan-in node (apply_edits, defer=True)
def apply_edits(state: ForgeState) -> dict:
    """Fan-in (defer=True): write every edit once all editors have finished."""
    edits = state.get("edits", [])
    for e in edits:
        sandbox.write_file(state["repo_path"], e.path, e.new_content)
    return {
        "attempts": state.get("attempts", 0) + 1,
        "messages": [AIMessage(content=f"Applied {len(edits)} edit(s).")],
    }

sandbox.write_file writes the full file content to the sandbox copy of Tally. The FileEdit.new_content field always carries the complete file content — not a diff, not a patch. This is a frozen contract. An apply that writes complete files is deterministic and reproducible; an apply that works with diffs needs merge logic and has failure modes.

Here is the full Forge M5 pipeline visualized:

graph LR
    P["Planner\n(Plan → targets: list[FileTarget])"]
    D["dispatch\n(routing fn → list[Send])"]
    E1["Editor\n(tally/report.py)"]
    E2["Editor\n(tally/cli.py)"]
    E3["Editor\n(README.md)"]
    A["apply\n(defer=True — waits for ALL editors)"]
    T["Tester\n(pytest in sandbox)"]

    P --> D
    D -->|"Send('editor', {target: report.py})"| E1
    D -->|"Send('editor', {target: cli.py})"| E2
    D -->|"Send('editor', {target: README.md})"| E3
    E1 -->|"edits += [FileEdit]"| A
    E2 -->|"edits += [FileEdit]"| A
    E3 -->|"edits += [FileEdit]"| A
    A --> T

E1, E2, and E3 run in the same superstep. The BSP barrier fires when all three complete. The operator.add reducer merges their FileEdit outputs. Only then does apply run.

---

Command(goto=[Send, …]): atomic update + fan-out

There is a cleaner, more advanced pattern that merges the planner and dispatch steps. In the lab code, planner is a regular node that writes targets to state, and then dispatch is a separate routing function. But what if you want to update shared state AND start the fan-out in one atomic step?

Command — imported from langgraph.types — lets a node return both a state update and a goto instruction at once:

# Advanced pattern: atomic state update + fan-out in one return
from langgraph.types import Command, Send
from forge.models import FileTarget

def planner_with_dispatch(state: ForgeState) -> Command:
    plan = agents.plan_change(state["repo_path"], state["issue"])
    targets = [
        FileTarget(
            path=f,
            instruction=next(
                (s.description for s in plan.steps if s.target_file == f),
                plan.summary,
            ),
        )
        for f in plan.target_files
    ]
    return Command(
        update={"plan": plan, "targets": targets},   # written to shared state
        goto=[Send("editor", {"repo_path": state["repo_path"],
                              "issue": state["issue"],
                              "target": t}) for t in targets],
    )

Why use this form? A routing function that returns list[Send] is a pure reader — it cannot write to shared state. If you need the fan-out decision and the state update to be atomic (the plan and targets fields appear in the same checkpoint as the Send dispatch), Command(update={...}, goto=[Send(...)]) is the right tool.

One distinction matters here: this Command is Command(goto=[Send(...)]) — the dispatch form. A different variant, Command(resume=), is the HITL form used to resume a graph after an interrupt(). That pattern is Module 7. The two forms look similar but serve completely different purposes.

---

The frozen contracts

Module 5 introduces two frozen Pydantic models. Reproduce them exactly in every module from here forward — no extra fields, no renames:

# forge/models.py — frozen at Module 5

class FileTarget(BaseModel):
    """One unit of map work: a file to edit + what to do."""
    path: str
    instruction: str


class FileEdit(BaseModel):
    """The map result, reduced via operator.add.

    Carries the COMPLETE new file content (not a diff) so the fan-in apply
    is deterministic and reproducible.
    """
    path: str
    new_content: str   # complete file content — NEVER a diff or partial patch
    summary: str
    applied: bool = False

FileTarget is the unit of map — one per file the Planner wants to edit. FileEdit is the map result — one per Editor worker. The operator.add reducer accumulates them into ForgeState["edits"]. The applied flag starts as False and can be set to True by apply_edits if you want traceability (the current lab leaves it False for simplicity but the field is there for future modules).

The write_file tool, added to forge/tools.py in this module, gives agents a direct way to write files. The Editor in the lab does not use it (it returns a FileEdit and lets apply_edits do the writing via sandbox.write_file), but it exists for agent-driven scenarios:

# forge/tools.py — write_file (Module 5 addition)
def make_write_tool(repo_path: str | Path):
    root = Path(repo_path)

    @tool
    def write_file(path: str, content: str) -> str:
        """Overwrite a file in the repository with new content."""
        p = root / path
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content)
        return f"wrote {path} ({len(content)} bytes)"

    return write_file

---

Common misconception — two to know

Misconception 1: Send imports from langgraph.constants. This was the v0.x path. In v1.x, the canonical import is:

from langgraph.types import Send    # ✅ v1.x
# from langgraph.constants import Send  # ❌ v0.x — deprecated and removed

All runtime types — Send, Command, interrupt, RetryPolicy, CachePolicy, StateSnapshot — import from langgraph.types in v1.x.

Misconception 2: operator.add makes the order of FileEdit in edits deterministic. It does not. The BSP barrier applies writes in the order editors finish, which can vary between runs. For Forge this is fine — each FileEdit targets a distinct file, so the application order does not matter. But if two Editors write to the same file, the edits list could contain two conflicting FileEdit objects for that path, and the last one applied wins. Two workers targeting the same file is a design error; the Planner’s job is to ensure target_files has no duplicates.

Misconception 3: defer=True is optional. Without it, apply_edits can fire after the first Editor completes, leaving the rest of edits empty or incomplete. Tests fail for inexplicable reasons. defer=True is mandatory on any fan-in node.

---

Hands-on lab: Build it

Goal: give Forge a parallel editing pass and verify it with a real pytest run on Tally.

Observable result (offline):

$ uv run pytest tests/ -q
....................................       <-- all M1–M5 offline smoke tests pass
14 passed, 0 failed

Observable result (live, FORGE_LIVE_TESTS=1):

triage     → route=bug
planner    → Plan(target_files=["report.py", "cli.py"], risk=low)
dispatch   → 2 Editors dispatched via Send API
editor[0]  → FileEdit(path="report.py", applied=False)
editor[1]  → FileEdit(path="cli.py",    applied=False)
apply      → 2 FileEdits collected (operator.add), writing to sandbox...
tester     → 14 passed, 0 failed  ✅

Step 1: Freeze the new contracts in models.py

Open forge/models.py and add FileTarget and FileEdit below the existing Plan class. These are frozen — reproduce them exactly:

# forge/models.py — Module 5 additions (after Plan/PlanStep)
class FileTarget(BaseModel):
    """One unit of map work: a file to edit + what to do — frozen at Module 5."""
    path: str
    instruction: str


class FileEdit(BaseModel):
    """The map result, reduced via operator.add — frozen at Module 5.

    new_content carries the COMPLETE file — apply is deterministic, not diff-based.
    """
    path: str
    new_content: str
    summary: str
    applied: bool = False

TestResult, PlanStep, and Plan remain unchanged.

Step 2: Extend ForgeState with two new fields

In forge/state.py, add the two Module 5 fields after plan:

# forge/state.py — Module 5 fields (added after plan: Plan | None)
from forge.models import FileEdit, FileTarget, Plan, TestResult

# --- Module 5 ---
targets: list[FileTarget]                          # map units (one per file to edit)
edits: Annotated[list[FileEdit], operator.add]     # THE aggregation reducer (fan-in)
# --- Module 7 adds: plan_approved, pull_request ---
# --- Module 12 adds: review ---

edits is the only field with operator.add. Do not add this reducer to any other field.

Step 3: Add write_file to tools.py

In forge/tools.py, add the make_write_tool factory after make_read_tools:

# forge/tools.py — write_file (Module 5)
def make_write_tool(repo_path: str | Path):
    root = Path(repo_path)

    @tool
    def write_file(path: str, content: str) -> str:
        """Overwrite a file in the repository with new content."""
        p = root / path
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content)
        return f"wrote {path} ({len(content)} bytes)"

    return write_file

Step 4: Add the fan-out and fan-in nodes to nodes.py

Three additions to forge/nodes.py:

Update planner to produce targets (the Module 4 planner only wrote plan):

# forge/nodes.py — planner now populates targets (Module 5 update)
def planner(state: ForgeState) -> dict:
    """Explore the repo (create_agent + read tools) and produce a Plan + map targets."""
    plan = agents.plan_change(state["repo_path"], state["issue"])
    targets = [
        FileTarget(
            path=f,
            instruction=next(
                (s.description for s in plan.steps if s.target_file == f),
                plan.summary,
            ),
        )
        for f in plan.target_files
    ]
    msg = AIMessage(content=f"Planned {len(targets)} edit(s): {plan.summary}")
    return {"plan": plan, "targets": targets, "messages": [msg]}

Add the dispatch routing function:

# forge/nodes.py — dispatch (the fan-out routing function, Module 5)
from langgraph.types import Send   # ALWAYS langgraph.types, never langgraph.constants

def dispatch(state: ForgeState) -> list[Send]:
    """Fan out one editor task per target file (the Send API)."""
    return [
        Send("editor", {"repo_path": state["repo_path"], "issue": state["issue"], "target": t})
        for t in state["targets"]
    ]

Add the Editor worker and apply_edits fan-in:

# forge/nodes.py — Editor (map worker) and apply_edits (fan-in, defer=True)
def editor(state: dict) -> dict:
    """Map worker: edit ONE file. Receives its target via the Send payload."""
    target: FileTarget = state["target"]
    current = (Path(state["repo_path"]) / target.path).read_text()
    model = config.get_model("smart")
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=(
            f"Issue:\n{state['issue']}\n\nEdit file {target.path}: {target.instruction}\n"
            f"Current contents:\n{current}"
        )),
    ])
    edit = FileEdit(
        path=target.path,
        new_content=extract_code(resp.content),
        summary=target.instruction,
    )
    return {"edits": [edit]}   # operator.add concatenates across all editors


def apply_edits(state: ForgeState) -> dict:
    """Fan-in (defer=True): write every edit once all editors have finished."""
    edits = state.get("edits", [])
    for e in edits:
        sandbox.write_file(state["repo_path"], e.path, e.new_content)
    return {
        "attempts": state.get("attempts", 0) + 1,
        "messages": [AIMessage(content=f"Applied {len(edits)} edit(s).")],
    }

Step 5: Rewire graph.py

Replace build_graph() with the Module 5 version:

# forge/graph.py — Module 5 full wiring
from langgraph.graph import END, START, StateGraph
from forge import nodes
from forge.state import ForgeState

def build_graph():
    builder = StateGraph(ForgeState)
    builder.add_node("intake",  nodes.intake)
    builder.add_node("triage",  nodes.triage)
    builder.add_node("planner", nodes.planner)
    builder.add_node("editor",  nodes.editor)
    builder.add_node("apply",   nodes.apply_edits, defer=True)   # fan-in gate
    builder.add_node("tester",  nodes.tester)
    builder.add_node("fixer",   nodes.fixer)

    builder.add_edge(START, "intake")
    builder.add_edge("intake", "triage")
    builder.add_conditional_edges(
        "triage", nodes.route_by_type,
        {"bug": "planner", "feature": "planner", "chore": "planner"},
    )
    builder.add_conditional_edges("planner", nodes.dispatch, ["editor"])
    builder.add_edge("editor", "apply")
    builder.add_edge("apply", "tester")
    builder.add_conditional_edges(
        "tester", nodes.route_after_tests,
        {"done": END, "give_up": END, "fixer": "fixer"},
    )
    builder.add_edge("fixer", "tester")
    return builder.compile()

Step 6: Verify with the offline test suite

The test suite exercises both the unit-level Send contracts and an end-to-end integration with a real Tally pytest run. Run it offline (no API key required):

uv run pytest tests/ -q

The integration test in test_pipeline.py is worth reading in full. It uses a RoutedFakeModel — a fake chat model that returns different canned responses based on what is in the prompt, not the order of calls. This is necessary because parallel nodes race for the model, and a sequential replay model would be unreliable:

# tests/test_pipeline.py — the parallel map-reduce integration test
def test_parallel_map_reduce_fixes_bug(monkeypatch, make_routed_model):
    repo = sandbox.make_sandbox(tally_src)
    sandbox.write_file(repo, "tests/test_issue_empty_category.py", REPRO_TEST)

    # route canned replies by prompt content, not call order
    routed = make_routed_model([
        ("restate", "report.py raises KeyError on empty categories."),
        ("Classify", "bug"),
        ("Edit file report.py", fixed_report),
        ("Edit file cli.py", current_cli),
    ], default="# noop\n")
    monkeypatch.setattr(config, "get_model", lambda tier="fast", **k: routed)

    out = build_graph().invoke(
        {"issue": "tally report --month crashes on empty category", "repo_path": repo},
        {"recursion_limit": 25},
    )
    assert out["test_result"].passed is True
    assert len(out["edits"]) == 2
    assert {e.path for e in out["edits"]} == {"report.py", "cli.py"}

The test injects a real bug (a KeyError in report.py when a category has no expenses), introduces the buggy test file into the sandbox, runs the full graph, and asserts that the real Tally pytest suite passes at the end. No mocking of the pytest runner — it actually runs.

Module 5 note: M5 has no entirely new files in forge/. Every change is an additive extension of an existing file. config.py and sandbox.py are inherited unchanged from M1 and M3 respectively.

Try it yourself

1. Degenerate case: pass an issue to a manually constructed ForgeState with targets = []. Verify that dispatch returns an empty list[Send], apply_edits runs with edits = [], and the Tally pytest suite still passes (nothing was changed, so the existing tests should be green).
2. Three-file fan-out: modify the plan fixture in test_pipeline.py to include a third target file. Confirm len(out["edits"]) == 3.
3. Command form: rewrite the planner node to return a Command(update={...}, goto=[Send(...)]) instead of a plain dict, and remove the separate dispatch routing function from graph.py. Verify the tests still pass.

---

In production

The parallel editing pattern works for small fan-outs out of the box. When Forge touches 3 files, 3 Editors run concurrently and each makes one model call. The total wall-clock time approaches the cost of one sequential call — a real speedup.

At scale, the ceiling is your provider’s rate limit, not LangGraph. If the Planner decides to touch 50 files, that is 50 simultaneous requests to your model provider. Most tiers will throttle or 429 you immediately. In production I cap the fan-out at 5–10 concurrent Editors and batch the remaining targets into follow-up supersteps. That ceiling is a business decision, not a LangGraph one.

Order of FileEdit in edits: operator.add concatenates in the order writes arrive at the BSP barrier. For Forge, where every Editor targets a distinct file, order does not affect correctness. But your apply_edits implementation should be robust enough to handle any order — do not rely on edits[0] being the “most important” file.

Observability: right now you know how many Editors ran and whether the final test suite passed. You do not know per-Editor latency, which Editor produced the most tokens, or whether one Editor silently returned garbage. Module 11 addresses this with get_stream_writer, where each Editor can emit progress events mid-execution.

Cost: each Editor is one call to the smart tier. On a 5-file fan-out at 2026 Claude Sonnet pricing, you are paying roughly 5× the single-editor cost but getting 5× the parallelism. The trade-off is latency versus cost — parallel editing is faster, sequential editing is cheaper, and the right answer depends on your use case.

Durability: none of these in-flight Editor runs survive a process kill. The whole run is lost. Module 6 adds a SqliteSaver checkpointer so Forge can resume from the last completed superstep after a crash.

Human approval: in Forge M5, the plan is dispatched to Editors immediately after the Planner finishes. There is no human checkpoint between planning and editing. Module 7 introduces interrupt() so a human can review the Plan and the targets before the fan-out fires.

---

Mastery corner

What to really understand here

The Send API is LangGraph’s mechanism for runtime-determined parallelism. The number of workers is not fixed at graph build time — it is computed from the state at execution time. This is the key distinction from static add_edge fan-out.

The BSP barrier is what makes fan-in safe without locks. All workers in a superstep complete before any downstream node sees their writes. The defer=True flag on the fan-in node is the explicit version of this guarantee — it tells LangGraph: “even if some workers finish early, hold this node until the last one is done.”

operator.add on edits is a reducer, not a merger. Each write appends to the existing list. The order of appending depends on barrier timing, not Send order.

Command(goto=[Send(...)]) is the dispatch form — used to combine a state update and a fan-out atomically in one node return. Command(resume=...) is the HITL form — used after an interrupt() to resume execution. These look similar but serve different purposes; mixing them up is a common interview mistake.

Mastery questions

Q1. The Planner node in your graph produces a Plan with 4 target_files. You wire the graph with:

builder.add_edge("planner", "editor_a")
builder.add_edge("planner", "editor_b")

What happens when Forge receives an issue that requires editing 3 files instead of 2?

A) LangGraph dynamically creates a third editor branch at runtime. B) The graph errors at compile time because the edge count does not match target_files. C) Only editor_a and editor_b run; the third file is silently skipped. D) The Planner node raises GraphRecursionError.

---

Q2. You add defer=True to your fan-in node and run a test. Then a colleague removes defer=True to “simplify the code.” What is the most likely observable symptom in a 3-file fan-out?

A) The graph raises ValueError because defer is a required parameter. B) All three Editors run but the results are merged in reverse order. C) The fan-in node sometimes fires with only 1 or 2 FileEdit objects, causing downstream tests to fail on unmodified files. D) The BSP barrier fires twice instead of once, applying some edits twice.

---

Q3. A developer asks you to review this import:

from langgraph.constants import Send

What is wrong with it, and what is the correct form?

A) Nothing is wrong; langgraph.constants is the canonical location for Send in both v0 and v1. B) Send should be imported from langgraph.graph alongside StateGraph. C) This is the v0.x path, removed in v1.x. The correct import is from langgraph.types import Send. D) Send is not a class; it should be imported as a function from langgraph.prebuilt.

---

Q4. You define the edits channel in ForgeState as:

edits: list[FileEdit]   # missing Annotated + operator.add

Two parallel Editors both write to edits. What happens at the BSP barrier?

A) LangGraph detects the conflict and raises ReducerConflict. B) The channel uses last-write-wins: only one Editor’s FileEdit survives. C) Both FileEdit objects are concatenated because LangGraph defaults to operator.add for lists. D) The graph pauses and asks the user to resolve the conflict.

---

Q5. You need a single node to both write {"plan": plan, "targets": targets} to shared state AND fan-out to Editor workers in the same atomic step. Which pattern achieves this?

A) Use a routing function that returns list[Send] and also mutates state in-place before returning. B) Use Command(update={"plan": plan, "targets": targets}, goto=[Send("editor", {"target": t}) for t in targets]) returned from the node. C) Add a second conditional edge from the same node alongside add_conditional_edges. D) Use builder.add_sequence(["planner", "dispatch"]) to merge the two steps.

---

Answers:

Q1 — C. Static add_edge fan-out is fixed at compile time. Two edges create exactly two branches. A third file is not dispatched; no error is raised. The Send API (dispatch returning list[Send]) is the correct approach when the branch count varies at runtime.

Q2 — C. Without defer=True, the apply node is eligible to run as soon as any single Editor finishes. If Editor A finishes first, apply fires with edits = [edit_for_A], writes one file, then the Tester runs on a partially-updated codebase. The other two files are never written, so those tests fail. The failure looks like a model output problem, but the root cause is a missing defer=True.

Q3 — C. from langgraph.constants import Send is the v0.x path. In LangGraph v1.x, all runtime types — Send, Command, interrupt, RetryPolicy, CachePolicy, StateSnapshot — live in langgraph.types. The v0.x location is deprecated and removed in v1.x.

Q4 — B. Without Annotated[list[FileEdit], operator.add], the channel is last-write-wins. Whichever Editor’s write arrives at the BSP barrier last overwrites the other. You end up with exactly one FileEdit in edits, not two. The other Editor’s work is silently discarded.

Q5 — B. A routing function that returns list[Send] is a pure reader — it cannot write to shared state. Command(update={...}, goto=[...]) is the correct form: the update dict is applied to shared state and the goto list triggers the fan-out atomically in one node return. Options A and C are incorrect (routing functions are pure; duplicate conditional edges do not merge state writes). Option D references a non-existent API.

---

Traps to avoid

Trap 1 (v0→v1 freshness): importing Send from langgraph.constants. The v1.x canonical import is from langgraph.types import Send. The same applies to Command, interrupt, RetryPolicy, CachePolicy, and StateSnapshot. If your IDE autocompletes langgraph.constants.Send or you are copying from a pre-2025 tutorial, you are on the wrong path. It may not error immediately (the old module may still re-export the symbol), but you are importing a deprecated path that will break.

Trap 2: omitting defer=True on the fan-in node. This is the most common bug when first using the Send API. The symptom is intermittent test failures where some files appear unmodified even though the Editors ran. The root cause is apply_edits firing before all Editors have written to edits. Fix: builder.add_node("apply", apply_edits, defer=True).

Trap 3: two Editors targeting the same file. operator.add concatenates all FileEdit objects without checking for duplicate paths. If two Editors both produce a FileEdit for report.py, you end up with two entries in edits for the same path. apply_edits will write the second one last, silently overwriting the first. The Planner’s job is to produce a target_files list with no duplicates. If your planning logic could produce duplicates, add a deduplication step before building targets.

---

Key takeaways

• The Pregel/BSP model guarantees that all branches in a superstep complete before the barrier applies their writes — fan-in coherence without locks.
• from langgraph.types import Send (never langgraph.constants); Send("node", private_arg) dispatches a worker with an isolated private input, not the global state.
• A routing function that returns list[Send] = dynamic fan-out — the number of workers is decided at runtime from the current state.
• edits: Annotated[list[FileEdit], operator.add] is the only aggregation reducer in ForgeState; every other field is last-write-wins. operator.add concatenates each Editor’s write at the barrier.
• builder.add_node("apply", fn, defer=True) waits for all Send-dispatched workers before scheduling the node. Without defer=True, the fan-in fires early with partial results.
• Command(update={...}, goto=[Send(...)]) combines a state write and a fan-out in one atomic return — the dispatch form of Command, distinct from the HITL form Command(resume=).
• FileEdit.new_content carries the complete file content (not a diff) — apply is deterministic, reproducible, and conflict-free for distinct target files.

---

What’s next

Forge now edits files in parallel — but kill the process between two supersteps and the whole run is lost. Module 6 adds a SqliteSaver checkpointer and thread IDs so Forge survives a crash and resumes exactly where it left off.

---

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

---

Links:

• Lab code for this module: github.com/dupuis1212/langgraph-prod-labs/tree/main/module-05
• Previous module (M4): Tools and the Agent Loop — create_agent, ToolNode, Structured Output
• Next module (M6): Persistence — Checkpointers, Threads, and Resumable Runs
• Full course index: LangGraph in Production

---

References

1. LangGraph Docs — How to create map-reduce branches for parallel execution: langchain-ai.github.io/langgraph/how-tos/map-reduce/ (as of June 2026)
2. LangGraph API Reference — langgraph.types (Send, Command, interrupt): langchain-ai.github.io/langgraph/reference/types/ (as of June 2026)
3. LangGraph API Reference — StateGraph.add_node (defer parameter): langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node (as of June 2026)
4. LangGraph Conceptual Guide — The Send API: langchain-ai.github.io/langgraph/concepts/low_level/#send (as of June 2026)
5. Malewicz et al., “Pregel: A System for Large-Scale Graph Processing,” SIGMOD 2010 — the BSP execution model underlying LangGraph’s superstep/barrier semantics.
6. Lab repository for this module: github.com/dupuis1212/langgraph-prod-labs/tree/main/module-05