This is Module 03 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 01 or browse the full syllabus.

Control Flow: Conditional Edges, Command, and Self-Correcting Cycles (LangGraph in Production, Module 03)

---

The Bug Forge Can’t Fix Yet

Here is a real issue against Tally, the expense-tracking CLI we use as Forge’s target codebase:

tally report --month 2026-05 crashes with a KeyError when a category has no expenses that month.

Look at the relevant line in tally/report.py:

for cat in CATEGORIES:
    lines.append(f"  {cat:14s} {totals[cat]:8.2f}")  # BUG: KeyError if cat absent

category_totals returns only the categories that actually appear in that month’s expenses. If no transactions were filed under, say, "transport", the key is absent and the format string explodes.

Forge from Module 02 reads that issue and summarizes it beautifully. Then it stops. The graph goes intake → END. Even if you bolted on an edit node, you would have a second problem: the first patch an LLM generates against a subtle bug often fails one or more edge-case tests. A straight-line graph cannot distinguish a bug from a feature request, and it cannot retry after a failed test run.

Without control flow, a coding agent is just an issue summarizer. This module gives Forge a triage router and a self-correcting loop that keeps trying until the tests are green or a guard stops it — whichever comes first.

---

In This Module

You’ll learn to:

• Route a graph with conditional edges: add_conditional_edges(source, routing_fn, path_map) — CA3.1
• Update state and route in one return using Command(update=, goto=) from langgraph.types — CA3.2/CA3.3
• Build a self-correcting cycle: an edge that returns to an earlier node — CA3.4
• Bound the cycle so it always terminates: the attempts guard and recursion_limit / GraphRecursionError — CA3.5
• Wire linear chains without manual edges using add_sequence — CA3.6

You’ll build: Forge gains a triage router (route = "bug" / "feature" / "chore") and a self-correcting test-fix loop — propose_edit → tester → [fail & attempts<N → fixer → tester]* — bounded by an attempts guard and recursion_limit, with the first real Tally pytest run executing inside a sandbox copy of the repo.

Concepts covered: CA3 — Control flow (core). This module BUILDs CA3.1–CA3.6 of the LangGraph theory inventory.

Prerequisites: Modules 01–02 completed; the cumulative forge/ package (three-field ForgeState + build_graph() skeleton); .env with ANTHROPIC_API_KEY (or the Ollama fallback for $0); pytest~=8 installed so Tally’s test suite runs.

---

Where You Are

• ✅ M01 — Setup, init_chat_model, first StateGraph, meet Forge & Tally
• ✅ M02 — ForgeState, add_messages reducer, build_graph() skeleton: intake → END
• 👉 M03 — Conditional edges, Command, cycles, bounds — triage router + test-fix loop
• ⬜ M04 — Tools (read_file, list_files, search_code), create_agent, structured Plan
• ⬜ M05–M16 — Parallelism, persistence, HITL, time-travel, streaming, multi-agent, deployment, evaluation

Forge before this module: reads an issue, summarizes it, does nothing with the code.

Forge after this module: triages the issue by type, copies Tally into a sandbox, proposes an edit, runs the real pytest suite, and loops through a fixer until the tests go green — bounded by a business guard and the framework’s own recursion backstop.

One boundary to set early: propose_edit and fixer are simple nodes in M03 — they read one known file and rewrite it. The real tools (read_file, list_files, search_code) and the structured Planner that decides which files to touch arrive in Module 04. And nothing Forge does in this module survives a crash: the checkpointer comes in Module 06.

---

Edges, Conditional Edges, and the Triage Router

Normal edges vs conditional edges

A normal edge is unconditional: builder.add_edge("a", "b") means “after node a finishes, always run node b.” The control flow is fixed at compile time.

A conditional edge asks a function to decide at runtime. The call looks like this:

from langgraph.graph import StateGraph, START, END

builder.add_conditional_edges(
    "triage",           # source node
    route_by_type,      # routing function: (state) -> str
    {                   # path map: return value -> destination node key
        "bug":     "propose_edit",
        "feature": "propose_edit",
        "chore":   "propose_edit",
    },
)

The routing function is a pure state reader — it receives the current state and returns a string that must appear in the path map. Its contract is simple:

def route_by_type(state: ForgeState) -> str:
    """Conditional-edge routing function: pure read of state -> next node key."""
    return state["route"]

The routing function does not write to state. It cannot. That restriction is intentional: it keeps routing predictable and the graph renderable. If you need to write state while routing, reach for Command — covered in the next section.

The path map ({"bug": "propose_edit", ...}) has two roles. Functionally, it validates return values and raises early if a routing function returns something unexpected. Visually, it tells LangGraph Studio which edges to draw — without it the graph renderer cannot show the branches.

START and END are special sentinel nodes imported from langgraph.graph. START is the implicit entry point; add_edge(START, "intake") sets the first node the graph executes. END is the terminal: any node that routes to END (via a normal edge or a path-map entry) signals the graph to stop and return the current state.

Here is the triage section of Forge’s graph as it stands after Module 03:

flowchart LR
    START --> intake
    intake --> triage
    triage -->|route=bug| propose_edit
    triage -->|route=feature| propose_edit
    triage -->|route=chore| propose_edit
    propose_edit --> tester

The triage node itself writes the route field into state, then the conditional edge reads it:

def triage(state: ForgeState) -> dict:
    """Classify the issue into a route the graph can branch on."""
    model = config.get_model("fast")
    resp = model.invoke(
        [SystemMessage(content=TRIAGE_SYSTEM), HumanMessage(content=state["issue"])]
    )
    text = resp.content.lower()
    route = next((r for r in ("bug", "feature", "chore") if r in text), "bug")
    return {"route": route, "messages": [resp]}

Notice the fallback to "bug" when the model returns something unexpected. In production you would log that case; in this lab it is enough to fail gracefully.

---

One Return to Rule Them All: Command(update, goto)

The conditional-edge pattern has one limitation: the routing function cannot write state. But sometimes the same node that decides where to go next also needs to update state atomically. Consider a node that must increment an attempts counter AND choose the next branch. You would need to write attempts in the node body and route from a separate function — two places to maintain.

Command solves this by letting a node return both an update and a destination in a single object:

from langgraph.types import Command
from typing import Literal

def route_after_tests_command(state: ForgeState) -> Command[Literal["tester", "fixer", "give_up"]]:
    """Write state AND route in one return — the Command pattern."""
    tr = state["test_result"]
    if tr.passed:
        return Command(goto="done")
    if state.get("attempts", 0) >= MAX_ATTEMPTS:
        return Command(update={"attempts": state["attempts"]}, goto="give_up")
    return Command(
        update={"attempts": state.get("attempts", 0) + 1},
        goto="fixer",
    )

The type annotation Command[Literal["tester", "fixer", "give_up"]] tells the graph renderer which outgoing edges this node has — it serves the same documentation role as a path map.

When to use Command vs a conditional edge:

Situation	Idiom
Routing is a pure read of state	Conditional edge + routing function
The node must write state AND route in one move	Command(update=, goto=)
Multiple distinct exit conditions each need different state writes	Command — keeps side-effect and jump atomic

Use a conditional edge when routing is a pure read of state; reach for Command when the node must write and route in one move — it keeps the side-effect and the jump atomic.

In the Module 03 lab, route_after_tests is implemented as a plain routing function (the conditional-edge version). The Command variant is left as “Try it yourself” exercise (c) — they are equivalent in what they do, but Command is the idiomatic choice when the routing node must also mutate state.

One freshness note: import Command from langgraph.types, not from langgraph.constants. The langgraph.constants path is the v0.x location and no longer works in LangGraph 1.x.

---

Cycles: the Self-Correcting Test-Fix Loop

What a cycle is

A cycle in a LangGraph graph is simply an edge that points back to a node that has already executed. There is nothing magical about it: builder.add_edge("fixer", "tester") creates a cycle because tester runs before fixer and now runs again after it. LangGraph’s Pregel execution model handles re-entry naturally — each superstep the graph evaluates all nodes whose incoming edges are satisfied.

Cycles are not bugs. For a coding agent, they are the whole point: you fix, you test, you fix again. The bug is leaving a cycle unbounded — more on that in the next section.

The Tester node: deterministic, not an agent tool

Before showing the loop, one design decision deserves explicit treatment.

run_tests (the Tester node) is a deterministic node — a plain Python function that shells out to pytest. It is not a @tool that a language model calls. This distinction matters:

• The test result is the observable, API-free signal of the whole lab. Whether the tests pass is a fact, not a model judgment.
• Making it a tool would make the signal non-deterministic (the model decides when and whether to invoke it) and would burn API credits for zero benefit.
• As a node it is always invoked, always after propose_edit and after fixer. The control structure is in the graph, not in the model’s tool-use loop.

The sandbox: always a copy

Forge executes code. That means it cannot work on the live Tally repo — one bad edit and the test infrastructure itself could be broken. The sandbox (forge/sandbox.py) exists for this reason: it copies Tally into a fresh tempdir on every run, applies edits to that copy, and runs pytest against the copy. The original is never touched.

def make_sandbox(repo_src: str | Path) -> str:
    """Copy the target repo into a fresh temp dir; return the path Forge may edit."""
    root = Path(tempfile.mkdtemp(prefix="forge-"))
    dst = root / Path(repo_src).name
    shutil.copytree(repo_src, dst)
    return str(dst)

This is the L8 decision from the course architecture: Forge executes code → sandbox is mandatory. The sandbox gets hardened in Module 09 (timeouts, network isolation, secrets scrubbing). For now, a simple shutil.copytree is enough.

Parsing the pytest output: TestResult

The TestResult schema is frozen at Module 03 — it will not change as later modules add more fields to ForgeState and models.py. Here is the frozen definition:

class TestResult(BaseModel):
    """The verdict of one sandboxed pytest run — frozen at Module 3."""

    passed: bool
    total: int
    failures: list[str]       # ids of the failing tests
    output_tail: str          # last 15 lines of pytest stdout+stderr

Four fields, nothing more. Do not add severity, duration, or score — they are not in the spec and will cause downstream modules to fail contract checks.

The parser in sandbox.run_tests is simple but robust enough for the Tally fixture:

def run_tests(repo_path: str | Path, timeout: int = 120) -> TestResult:
    """Run the target package's pytest suite in a subprocess and parse the result."""
    repo = Path(repo_path)
    root, pkg = repo.parent, repo.name
    proc = subprocess.run(
        [sys.executable, "-m", "pytest", pkg, "-q", "--no-header", "-p", "no:cacheprovider"],
        cwd=root, capture_output=True, text=True, timeout=timeout,
    )
    out = proc.stdout + proc.stderr
    failures = [ln.split()[1].split("::")[-1] for ln in out.splitlines() if ln.startswith("FAILED")]
    m_pass = re.search(r"(\d+) passed", out)
    m_fail = re.search(r"(\d+) failed", out)
    total = (int(m_pass.group(1)) if m_pass else 0) + (int(m_fail.group(1)) if m_fail else 0)
    return TestResult(
        passed=proc.returncode == 0,
        total=total,
        failures=failures,
        output_tail="\n".join(out.splitlines()[-15:]),
    )

The key detail: passed is based on returncode, not on string parsing. pytest exits 0 on full pass and non-zero otherwise. Everything else is informational.

The complete test-fix loop

Here is the full graph structure after Module 03:

flowchart TD
    START --> intake
    intake --> triage
    triage -->|"route=bug"| propose_edit
    triage -->|"route=feature"| propose_edit
    triage -->|"route=chore"| propose_edit
    propose_edit --> tester
    tester -->|"passed=True"| done["END (done)"]
    tester -->|"attempts >= MAX"| give_up["END (give_up)"]
    tester -->|"passed=False & attempts < MAX"| fixer
    fixer --> tester

The cycle is tester → fixer → tester. It closes when either condition exits the loop: green tests route to END, exhausted budget routes to END via give_up.

The routing function that controls the cycle reads both test_result and attempts:

def route_after_tests(state: ForgeState) -> str:
    """Cycle control: pass -> done; budget spent -> give_up; otherwise -> fixer."""
    if state["test_result"].passed:
        return "done"
    if state.get("attempts", 0) >= MAX_ATTEMPTS:
        return "give_up"
    return "fixer"

And the fixer node feeds the failing test output back to the model so each retry has more information than the last:

def fixer(state: ForgeState) -> dict:
    """Re-edit using the failing test output until green or the budget runs out."""
    model = config.get_model("smart")
    current = _target(state).read_text()
    tr = state["test_result"]
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=(
            f"Issue:\n{state['issue']}\n\nThe tests still fail:\n{tr.output_tail}\n\n"
            f"File {TARGET_FILE}:\n{current}"
        )),
    ])
    sandbox.write_file(state["repo_path"], TARGET_FILE, extract_code(resp.content))
    return {"messages": [resp], "attempts": state.get("attempts", 0) + 1}

fixer increments attempts so the routing function can count passes correctly. propose_edit also increments attempts on its first call (the first attempt), so after two total calls — one from propose_edit and one from fixer — attempts is 2.

The wiring in build_graph():

builder.add_node("propose_edit", nodes.propose_edit)
builder.add_node("tester", nodes.tester)
builder.add_node("fixer", nodes.fixer)

builder.add_edge("propose_edit", "tester")
builder.add_conditional_edges(
    "tester", nodes.route_after_tests,
    {"done": END, "give_up": END, "fixer": "fixer"},
)
builder.add_edge("fixer", "tester")  # <- the back edge that creates the cycle

add_edge("fixer", "tester") is the one line that creates the cycle. The rest is standard.

---

Making Loops Terminate: attempts, recursion_limit, and GraphRecursionError

An unbounded cycle is an infinite loop and, in the context of a coding agent, an infinite API bill. LangGraph gives you two independent lines of defense, and you need both.

The business guard: attempts

attempts in ForgeState is a plain integer counter. route_after_tests checks it:

MAX_ATTEMPTS = 3  # defined at the top of nodes.py

def route_after_tests(state: ForgeState) -> str:
    if state["test_result"].passed:
        return "done"
    if state.get("attempts", 0) >= MAX_ATTEMPTS:
        return "give_up"
    return "fixer"

This is the intended exit path for a loop that exhausts its budget. When attempts reaches MAX_ATTEMPTS, the graph routes to give_up → END cleanly. No exception, no drama. The user gets a partial result with the last TestResult in state.

attempts counts fix passes (cycles through the loop). It is business logic.

The framework backstop: recursion_limit and GraphRecursionError

recursion_limit is a LangGraph runtime parameter — a count of supersteps (complete graph passes) allowed in a single invoke or stream call. The default is 25. You pass it as part of the config dict:

graph.invoke(
    {"issue": "...", "repo_path": repo},
    {"recursion_limit": 25},
)

If the graph executes more supersteps than recursion_limit allows, LangGraph raises GraphRecursionError from langgraph.errors:

from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke(inputs, {"recursion_limit": 4})
except GraphRecursionError as exc:
    print(f"Graph hit recursion limit: {exc}")
    # recover or report

recursion_limit is the safety net, not the primary control. If your loop has a correct attempts guard, you should never hit GraphRecursionError in normal operation. If you do hit it in a well-guarded loop, either your recursion_limit is set too low relative to the expected number of non-loop supersteps, or your guard has a bug.

Guard	What it counts	Role
attempts	Fix passes (loop iterations)	Business logic — the intended exit
recursion_limit	Total supersteps in the run	Framework safety net — raises GraphRecursionError

The distinction is real and matters for debugging. A graph that hits GraphRecursionError on a run with 3 fix passes and a 25-superstep limit probably has more supersteps per cycle than you expected (intake + triage + propose_edit + tester + fixer + tester is already 6). Count your supersteps, not just your loop iterations.

add_sequence: linear chains without manual edges

When a section of your graph is a straight line, builder.add_sequence([n1, n2, n3]) wires it without individual add_edge calls:

# Equivalent to:
#   builder.add_edge("intake", "triage")
#   builder.add_edge("triage", "propose_edit")
builder.add_sequence(["intake", "triage", "propose_edit"])

add_sequence is not magic — it just calls add_edge in a loop. Use it for the straight runs in your graph; use explicit add_edge and add_conditional_edges for branches and cycles. It also works with START and END as endpoints:

builder.add_sequence([START, "intake", "triage"])

The Module 03 lab wires the straight sections manually for clarity, but add_sequence is the right tool for boilerplate linear chains in larger graphs.

---

⚠️ Common misconceptions

“GraphRecursionError means my graph has a bug.” Not necessarily. GraphRecursionError is the framework’s safety net — it fires when a run exceeds recursion_limit supersteps (default 25). If your loop has a correct attempts guard, the right fix is to check that the guard is actually reachable and (if needed) raise recursion_limit for graphs with longer expected paths. The cycle itself is not the bug; an unbounded or mis-guarded cycle is.

“A routing function can write state.” It cannot. A conditional edge routing function receives state and returns a node name — that is the entire contract. If you need to update state while routing, use Command(update=..., goto=...) instead. These two idioms are not interchangeable: the routing function is read-only by design, and Command is write-and-route in one move.

Freshness trap — importing Command from langgraph.constants: the v0.x path is from langgraph.constants import Command. In LangGraph v1.x the correct path is from langgraph.types import Command. The same applies to Send (not exercised in this module — that is Module 05’s topic). Using the v0.x import raises an ImportError in v1.x.

---

Hands-on Lab: Build It

Objective: Extend the forge/ package from Module 02 with a triage router and a self-correcting test-fix loop. The observable signal: pytest against Tally runs for real — no API call required.

End state you are building toward:

triage      → route=bug
propose_edit→ edited tally/report.py (attempt 1)
run_tests   → 1 failed  [test_report_handles_empty_category]
route       → fixer (attempts=1 < 3)
fixer       → edited tally/report.py (attempt 2)
run_tests   → 12 passed, 0 failed  ✅
route       → done (tests green)
DONE: TestResult(passed=True, total=12, failures=[], ...)

Step 1 — Carry forward Module 02

Copy your module-02/ directory to module-03/ byte-for-byte. Do not rewrite anything. The forge package currently has: __init__.py, config.py (frozen), state.py (three fields), graph.py (build_graph() with intake → END), nodes.py (intake node).

Step 2 — Create forge/models.py

This file holds Pydantic contracts for Forge. Only TestResult lives here in Module 03; Plan, FileTarget, FileEdit, Review, and PullRequest arrive in later modules.

from __future__ import annotations
from pydantic import BaseModel

class TestResult(BaseModel):
    """The verdict of one sandboxed pytest run — frozen at Module 3."""

    __test__ = False  # this is a domain model, not a pytest test class

    passed: bool
    total: int
    failures: list[str]       # ids of the failing tests
    output_tail: str          # last 15 lines of pytest stdout+stderr

The __test__ = False dunder prevents pytest from collecting TestResult as a test class when it scans your package. Without it, pytest will warn (or error on strict setups).

Step 3 — Create forge/sandbox.py

The sandbox provides three operations: copy the repo, write a file into the copy, and run tests on the copy.

from __future__ import annotations
import re, shutil, subprocess, sys, tempfile
from pathlib import Path
from forge.models import TestResult

def make_sandbox(repo_src: str | Path) -> str:
    """Copy the target repo into a fresh temp dir; return the editable path."""
    root = Path(tempfile.mkdtemp(prefix="forge-"))
    dst = root / Path(repo_src).name
    shutil.copytree(repo_src, dst)
    return str(dst)

def write_file(repo_path: str | Path, rel: str, content: str) -> str:
    """Write a file inside the sandbox (relative to the repo root)."""
    p = Path(repo_path) / rel
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content)
    return str(p)

def run_tests(repo_path: str | Path, timeout: int = 120) -> TestResult:
    """Run pytest in a subprocess and parse the result into a TestResult."""
    repo = Path(repo_path)
    root, pkg = repo.parent, repo.name
    proc = subprocess.run(
        [sys.executable, "-m", "pytest", pkg, "-q", "--no-header", "-p", "no:cacheprovider"],
        cwd=root, capture_output=True, text=True, timeout=timeout,
    )
    out = proc.stdout + proc.stderr
    failures = [ln.split()[1].split("::")[-1] for ln in out.splitlines() if ln.startswith("FAILED")]
    m_pass = re.search(r"(\d+) passed", out)
    m_fail = re.search(r"(\d+) failed", out)
    total = (int(m_pass.group(1)) if m_pass else 0) + (int(m_fail.group(1)) if m_fail else 0)
    return TestResult(
        passed=proc.returncode == 0,
        total=total,
        failures=failures,
        output_tail="\n".join(out.splitlines()[-15:]),
    )

run_tests is not a LangGraph node — it is a plain helper that the tester node calls. This keeps the subprocess logic testable in isolation.

Security note: this sandbox is intentionally minimal. Module 09 adds timeouts per test, network isolation, and secrets scrubbing. Do not run make_sandbox on untrusted repos without those guards.

Step 4 — Extend forge/state.py

Add three fields to ForgeState — by addition only. The Module 02 fields (issue, messages, repo_path) are unchanged and must stay at the top.

from forge.models import TestResult

class ForgeState(TypedDict, total=False):
    # --- Module 2 ---
    issue: str
    messages: Annotated[list[AnyMessage], add_messages]
    repo_path: str
    # --- Module 3 ---
    route: str                    # triage: "bug" | "feature" | "chore"
    attempts: int                 # test-fix loop guard
    test_result: TestResult | None  # latest pytest verdict
    # --- Module 4 adds: plan ---
    # --- Module 5 adds: targets, edits ---
    # --- Module 7 adds: plan_approved, pull_request ---
    # --- Module 12 adds: review ---

The comments are not decorative — they are the module-introduction calendar, enforced by the contract in 06-FIL-ROUGE-SPEC.md §2.1. Fields introduced in later modules must not appear here before their module ships.

Step 5 — Add nodes to forge/nodes.py

Add five new node functions to the existing file. intake from Module 02 stays unchanged.

MAX_ATTEMPTS = 3
TARGET_FILE = "report.py"  # M3 works on one known file; multi-file planning is M4-M5

def triage(state: ForgeState) -> dict:
    """Classify the issue — write 'route' into state."""
    model = config.get_model("fast")
    resp = model.invoke([
        SystemMessage(content=TRIAGE_SYSTEM),
        HumanMessage(content=state["issue"]),
    ])
    text = resp.content.lower()
    route = next((r for r in ("bug", "feature", "chore") if r in text), "bug")
    return {"route": route, "messages": [resp]}

def route_by_type(state: ForgeState) -> str:
    """Routing function for the triage conditional edge — pure read."""
    return state["route"]

def propose_edit(state: ForgeState) -> dict:
    """First attempt: rewrite the target file to address the issue."""
    model = config.get_model("smart")
    current = (Path(state["repo_path"]) / TARGET_FILE).read_text()
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=f"Issue:\n{state['issue']}\n\nFile {TARGET_FILE}:\n{current}"),
    ])
    sandbox.write_file(state["repo_path"], TARGET_FILE, extract_code(resp.content))
    return {"messages": [resp], "attempts": state.get("attempts", 0) + 1}

def tester(state: ForgeState) -> dict:
    """Deterministic node: run the sandboxed pytest suite."""
    return {"test_result": sandbox.run_tests(state["repo_path"])}

def route_after_tests(state: ForgeState) -> str:
    """Cycle control routing function."""
    if state["test_result"].passed:
        return "done"
    if state.get("attempts", 0) >= MAX_ATTEMPTS:
        return "give_up"
    return "fixer"

def fixer(state: ForgeState) -> dict:
    """Re-edit using the failing test output; increment attempts."""
    model = config.get_model("smart")
    current = (Path(state["repo_path"]) / TARGET_FILE).read_text()
    tr = state["test_result"]
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=(
            f"Issue:\n{state['issue']}\n\nTests still fail:\n{tr.output_tail}\n\n"
            f"File {TARGET_FILE}:\n{current}"
        )),
    ])
    sandbox.write_file(state["repo_path"], TARGET_FILE, extract_code(resp.content))
    return {"messages": [resp], "attempts": state.get("attempts", 0) + 1}

tester is the Tester node. Notice it calls sandbox.run_tests directly — there is no LLM call, no tool dispatch, no model judgment. The test result is a fact.

Step 6 — Rewire forge/graph.py

Replace build_graph() with the full Module 03 version:

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("propose_edit", nodes.propose_edit)
    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": "propose_edit", "feature": "propose_edit", "chore": "propose_edit"},
    )
    builder.add_edge("propose_edit", "tester")
    builder.add_conditional_edges(
        "tester", nodes.route_after_tests,
        {"done": END, "give_up": END, "fixer": "fixer"},
    )
    builder.add_edge("fixer", "tester")  # the back edge — creates the cycle
    return builder.compile()

builder.compile() validates the graph: it checks that every path-map key has a corresponding node, that START has an outgoing edge, and that there are no orphan nodes. If you misspell a node name or forget a path-map entry, the error surfaces here, not at runtime.

Step 7 — Run the tests

The test suite for this module verifies three things without any API key:

1. The sandbox logic (copies the repo, runs real pytest, parses TestResult)
2. The cycle-control logic (route_after_tests under all three conditions)
3. The full pipeline with a fake model that fails on the first attempt and succeeds on the second

# From tests/test_pipeline.py — the integration test
def test_fix_loop_makes_tally_green(monkeypatch, make_fake_model):
    tally_src = Path(__file__).resolve().parents[1] / "tally"
    repo = sandbox.make_sandbox(tally_src)
    sandbox.write_file(repo, "tests/test_issue_empty_category.py", REPRO_TEST)

    buggy = (Path(repo) / "report.py").read_text()
    fixed = buggy.replace("totals[cat]", "totals.get(cat, 0.0)")

    fake = make_fake_model([
        "report.py raises KeyError when a category has no expenses.",  # intake
        "bug",                                                          # triage
        buggy,                                                          # propose_edit (fails)
        fixed,                                                          # fixer (passes)
    ])
    monkeypatch.setattr(config, "get_model", lambda tier="fast", **k: fake)

    out = build_graph().invoke(
        {"issue": "tally report --month crashes with a KeyError", "repo_path": repo},
        {"recursion_limit": 25},
    )
    assert out["test_result"].passed is True
    assert out["attempts"] == 2
    assert "totals.get(cat" in (Path(repo) / "report.py").read_text()

The fake model replays canned responses in order. The test verifies that after two attempts the graph reaches passed=True, that attempts is correctly set to 2, and that the sandbox file has been modified.

Run with:

cd module-03
uv run pytest tests/ -v

All tests pass offline. Set FORGE_LIVE_TESTS=1 to run a real model call against the actual Tally bug (~$0.03 with Claude Haiku at June 2026 pricing).

Try it yourself

(a) Add a fourth route "docs" to the triage path map and wire it to a new docs_stub node that just writes {"route": "docs"} and routes to END. Verify that an issue like "update the README" correctly exits through the docs path.

(b) Lower recursion_limit to 4 and run build_graph().invoke(...) with three fix passes expected. Observe the GraphRecursionError. Catch it with from langgraph.errors import GraphRecursionError and print a friendly message.

(c) Replace the route_after_tests conditional edge with a Command-returning node. Verify the behavior is identical. Notice that the path-map argument to add_conditional_edges is no longer needed — the Command type annotation serves that purpose.

---

In Production

The test-fix loop pattern is powerful and has some sharp edges when it leaves the lab.

Cap the retries. In production, set MAX_ATTEMPTS at 2–3, not higher. Beyond that the model is usually stuck on the same incorrect mental model of the code, and each additional pass burns real money. The attempts guard is a cost control lever as much as a correctness one. A loop that runs 5 times and still fails has almost certainly identified a problem that needs a human, not another LLM call.

Every fix pass is a billed API call. With smart-tier models, three fix passes on a 200-line file can easily run $0.10–$0.20. At scale across dozens of open issues, that adds up fast. Instrument how often the fixer runs in your telemetry.

Observe which test fails and on which attempt. The test_result.failures list and output_tail are in state after every tester run. In Module 11, get_stream_writer lets you stream these per-superstep so operators can see progress in real time. For now, log them at minimum.

Flaky tests will make the loop re-trigger unnecessarily. A test that occasionally fails due to timing, a temp file race, or network jitter will cause route_after_tests to return "fixer" even when the code is correct. Wrapping the Tester node in a RetryPolicy for transient failures is the Module 09 solution — leave a comment in your code so you remember where to come back.

The sandbox is minimal today. make_sandbox uses shutil.copytree with no resource limits. A pathological input could generate large files or spawn long-running subprocesses. Module 09 hardens this: per-test timeouts, subprocess resource limits, network isolation, and secrets scrubbing. Do not run M03’s sandbox on untrusted code in a multi-tenant environment.

None of this survives a crash yet. If the Python process dies mid-loop, the graph state is gone. Module 06 adds a checkpointer (SqliteSaver) and thread IDs so Forge can resume from the last completed superstep. The sandbox temp directory would also need to be preserved across runs — that is part of the M06 story.

---

Mastery Corner

What to really understand here

CA3.1 — Conditional edges and routing functions. A conditional edge calls a routing function that reads state and returns a string key. The key is looked up in the path map to find the next node. The routing function cannot write state — it is a pure read. This is where most newcomers get confused: they expect to do work in the routing function and find they cannot.

CA3.2 — Command(update, goto). When a node needs to update state and decide routing atomically, return Command(update={...}, goto="next_node"). The type annotation Command[Literal["a", "b"]] replaces the path map for graph-rendering purposes.

CA3.3 — START, END, and entrypoints. add_edge(START, "first_node") designates the first node executed. A path-map entry of END (or add_edge("node", END)) designates an exit. Every compiled graph must have at least one path from START to END.

CA3.4 — Cycles. A back edge (add_edge("fixer", "tester") where tester is upstream) creates a cycle. Cycles are legitimate; they are the mechanism for retry loops, approval loops, and any pattern that needs to revisit a node.

CA3.5 — Two bounds, two roles. attempts is your business logic exit (clean, logged, graceful). recursion_limit is the framework’s safety net (raises GraphRecursionError if violated). You need both: attempts because you want to handle the exhausted-budget case yourself; recursion_limit because you want a backstop if the business logic has a bug.

CA3.6 — add_sequence. For straight-line sections of a graph, builder.add_sequence([n1, n2, n3]) is the ergonomic alternative to repeated add_edge calls. It does not create any magic — just chains edges in order.

---

Quiz

Q1. You are building a node that must increment an attempts counter AND choose the next node to execute, all in one operation. Which idiom do you use?

• (A) A routing function passed to add_conditional_edges — routing functions can update state via return {"attempts": ...}
• (B) Command(update={"attempts": new_val}, goto="next_node") returned from the node
• (C) Two separate nodes: one that increments attempts, one that returns the route
• (D) add_sequence with a routing hook parameter

Q2. A graph runs and raises GraphRecursionError after approximately 25 steps. What is most likely happening, and what is the correct fix?

• (A) The GraphRecursionError is a known LangGraph bug — downgrade to v0.x to avoid it
• (B) The cycle has no termination guard; add an attempts guard in the routing function and optionally raise recursion_limit if the graph has many non-loop supersteps
• (C) The path map is missing an entry; LangGraph internally loops trying to resolve the route
• (D) recursion_limit must always be set to 100 or higher to prevent this error

Q3. What is the difference between recursion_limit and the attempts field in ForgeState?

• (A) They count the same thing — recursion_limit is just a runtime alias for the attempts field
• (B) recursion_limit counts supersteps (total graph passes in a run); attempts counts fix-loop iterations (business logic). recursion_limit is a last-resort backstop; attempts is the intended exit
• (C) attempts is a LangGraph framework concept; recursion_limit is a user-defined variable in ForgeState
• (D) recursion_limit counts nodes visited; attempts counts LLM calls

Q4. In the Forge architecture, run_tests (the Tester node) is implemented as a deterministic subprocess call rather than as a @tool for the agent. Why?

• (A) @tool is not supported inside a StateGraph — it can only be used with create_agent
• (B) The test result is an objective fact (the test suite either passes or it does not) and should not depend on a language model deciding when or whether to invoke it. A deterministic node is always invoked and produces a verifiable signal without API cost
• (C) Using @tool for pytest would require network access, which is blocked in the sandbox
• (D) @tool functions cannot return structured output like TestResult — they can only return strings

Q5. The route_after_tests routing function reads state["test_result"], but what happens if the tester node has not yet run (for example, if you accidentally wire route_after_tests as the first conditional edge)?

• (A) LangGraph initializes all state fields to None by default, so state["test_result"] is None and the function returns "fixer" silently
• (B) state["test_result"] raises a KeyError (or returns None with total=False), because ForgeState uses total=False — the field is absent until the tester node writes it. The routing function must guard with state.get("test_result") before accessing .passed
• (C) GraphRecursionError is raised because the graph detects an out-of-order node execution
• (D) The StateGraph compiler enforces node ordering and prevents this configuration at compile time

---

Answers

A1: (B). A routing function passed to add_conditional_edges is read-only by design — it can return a string key but cannot update state. Command(update=..., goto=...) is the correct idiom for a node that must do both. Option (C) would work but breaks the atomicity requirement.

A2: (B). GraphRecursionError is not a bug — it is the framework’s safety net doing its job. The correct response is to add a business-logic guard (attempts) and ensure the routing function has a path to END. Raising recursion_limit (if the graph has many non-loop steps) is also sometimes necessary, but it is not a substitute for the guard.

A3: (B). recursion_limit is a runtime parameter that counts total supersteps across the entire run. attempts is a field in ForgeState that counts fix-loop iterations. They operate independently: a graph could hit recursion_limit even if attempts has not reached MAX_ATTEMPTS (if there are many non-loop supersteps), and attempts reaching MAX_ATTEMPTS correctly exits the loop without touching recursion_limit.

A4: (B). The key insight is determinism and cost. Whether the tests pass is a fact, not a judgment — there is no reason to involve a language model in that determination. A deterministic node is predictable (always runs, always returns a structured result), cheaper (no API call), and produces the observable signal that makes the lab verifiable without credentials.

A5: (B). ForgeState is declared with total=False (TypedDict optional fields), which means test_result simply does not exist in the dict until tester runs. Accessing state["test_result"] on a state where the key was never written raises a KeyError. The safe pattern is state.get("test_result") and checking for None before accessing .passed. The graph compiler does not enforce execution order — it is your responsibility to ensure the routing function is only reachable after the node that writes the field it reads.

---

Traps to avoid

Trap 1: Treating GraphRecursionError as a framework bug. It is not. If your graph raises it, the first question is whether your cycle has a reachable termination guard. Add attempts logic (or equivalent), verify it is actually reachable in the routing function, and only then consider adjusting recursion_limit.

Trap 2 (v0→v1 freshness): Importing Command or Send from langgraph.constants. In LangGraph v0.x, Command and Send lived in langgraph.constants. In v1.x (the version this course uses), the canonical import is:

from langgraph.types import Command
from langgraph.types import Send  # covered in Module 05

The langgraph.constants path no longer exists in 1.x. If you see this in existing code or tutorials, it is a v0.x artifact and will raise ImportError on a current install.

Trap 3: Making run_tests (the Tester) a @tool. This is tempting if you have a background in ReAct-style agents where every external call is a tool. Resist it. The Tester runs after every edit, unconditionally, as a graph node. Turning it into a tool hands control over test invocation to the language model — which may decide not to test, or test at the wrong time. The test suite is your ground truth. Keep it deterministic and always-on.

---

Key Takeaways

• A conditional edge calls a routing function (state) -> str that reads state and returns a node key looked up in a path map. The routing function cannot write state.
• Command(update=, goto=) (from langgraph.types) lets a node update state and route in a single return — use it when the write and the routing decision belong together.
• A cycle is an edge that points back to an earlier node. It is legitimate; all it needs is a termination guard.
• attempts (business logic) and recursion_limit (framework safety net, raises GraphRecursionError) are two independent bounds with distinct roles. You need both.
• run_tests (the Tester node) is a deterministic subprocess, not an agent tool. The test result is a fact; no LLM should decide when to invoke it.
• Forge works on a sandbox copy of Tally (shutil.copytree into a tempdir) — never on the original repo. This is non-negotiable when an agent executes code.
• builder.add_sequence([n1, n2, n3]) wires linear chains without manual add_edge calls — use it for straight runs, explicit edges for branches and cycles.

---

What’s Next

Forge can now loop and self-correct, but its edit node is a blunt instrument: it reads one hardcoded file and rewrites the whole thing. It has no idea what files the codebase contains, where the relevant code lives, or how to plan a multi-file change.

Module 04 — Tools and the Agent Loop gives Forge real tools (read_file, list_files, search_code), wraps them in a create_agent agent, and adds a Planner node that emits a structured Plan telling the rest of the pipeline exactly which files to touch and why.

---

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-03
• Previous: Module 02 — StateGraph Fundamentals: State, Reducers, Nodes, and Edges
• Next: Module 04 — Tools and the Agent Loop: ToolNode, create_agent, and Structured Output
• Course index: LangGraph in Production

---

References

1. LangGraph documentation — Conditional edges and routing: langchain-ai.github.io/langgraph/concepts/low_level/#conditional-edges (as of June 2026)
2. LangGraph documentation — Command and Send types: langchain-ai.github.io/langgraph/concepts/low_level/#command (as of June 2026)
3. LangGraph documentation — Recursion limit and GraphRecursionError: langchain-ai.github.io/langgraph/concepts/low_level/#recursion-limit (as of June 2026)
4. LangGraph documentation — Why LangGraph: cycles vs DAGs: langchain-ai.github.io/langgraph/concepts/why-langgraph/ (as of June 2026)
5. LangGraph langgraph.types module reference (Command, Send, interrupt): langchain-ai.github.io/langgraph/reference/types/ (as of June 2026)
6. Lab repo module-03: github.com/dupuis1212/langgraph-prod-labs/tree/main/module-03