Build an Ultracode-Like Coding Workflow with LangGraph

TL;DR

• LangGraph is the durable top-level workflow runtime and source of truth. Worker tool loops use LangChain’s create_agent, which is itself built on LangGraph; the design does not depend on Agno.
• Model calls are ordinary graph nodes with typed structured output. LangGraph owns state, dynamic worker fan-out, fan-in, retries, repair rounds, checkpoints, interrupts, and resume.
• Use Send to dispatch one routine worker or several independent workers from a model-generated plan.
• Use reducer-enabled state only for append-only worker and review events. Deterministic code—not a reducer model—must build the candidate manifest and test report.
• Reviewers receive the exact candidate SHA, immutable diff, and raw test artifacts. Approval is bound to that same candidate before promotion.
• PostgreSQL makes graph state durable; it does not provide a run queue or sandbox. Add an API/queue runtime and isolated workspace service.
• For production, enforce idempotency because LangGraph resumes from checkpoints and may re-execute nodes.
• This approximation runs one independent worker wave per round. It does not attempt to execute an arbitrary dependency DAG.

What You Will Learn Here

• How to model an Ultracode-like loop as a StateGraph
• How to generate dynamic workers with Send
• How reducers and fan-in work
• How to preserve immutable success criteria and regression commands
• How to integrate, test, review, repair, and approve one candidate
• How to use PostgreSQL checkpoints and Command(resume=...)
• How to deploy with custom FastAPI or Agent Server
• Which infrastructure sits outside LangGraph

This is Part 3 of the Ultracode Architecture series. Part 1 derives the portable design. Part 2 is an independent Agno/AgentOS alternative. This article assumes you chose the LangGraph ecosystem and does not combine it with Agno.

Why LangGraph Fits This Problem

Ultracode-like work is closer to a state machine than a chat loop:

request
  -> typed plan
  -> dynamic worker wave
  -> candidate integration
  -> deterministic tests
  -> synthesis
  -> verifier + refuter
  -> repair or approval
  -> idempotent promotion

LangGraph supplies:

What LangGraph does not supply is just as important:

• Git worktrees
• process isolation
• command policy
• artifact storage
• authenticated approval identity
• a durable run queue in the OSS library
• tenant quotas and billing

Those remain infrastructure contracts.

Architecture

API / queue runtime
       |
       v
LangGraph thread
       |
       +--> plan node
       |
       +--> Send(worker A)
       +--> Send(worker B)
       +--> Send(worker C)
                |
                v
       deterministic integrate node
                |
                v
       deterministic test node
                |
                v
          synthesis node
           /          \
          v            v
     verifier       refuter
           \          /
            review barrier
               |
       fail --> repair plan
               |
              pass
               |
        interrupt approval
               |
        promote candidate SHA

Scope of the scaffold

The code is intentionally not runnable until you provide the adapter implementations. The graph demonstrates their required postconditions and where they connect.

Project Structure

langgraph-ultracode/
  app/
    schemas.py
    model.py
    workspace.py
    graph.py
    api.py
    bootstrap.py        # Concrete workspace and approval adapters
  tests/
    test_plan.py
    test_resume.py
    test_candidate_binding.py
  pyproject.toml

The API signatures were reviewed on July 15, 2026 against Python 3.10+, LangGraph 1.2.9, LangChain 1.3.13, LangChain OpenAI 1.3.5, and LangGraph Checkpoint PostgreSQL 3.1.0.

uv add "langgraph==1.2.9" "langgraph-checkpoint-postgres==3.1.0" \
  "langchain==1.3.13" "langchain-openai==1.3.5" "psycopg[binary,pool]" \
  pydantic fastapi uvicorn

Per-node and Send timeout APIs require LangGraph 1.2 or later.

Step 1: Define Typed State and Outputs

# app/schemas.py
from typing import Literal

from pydantic import BaseModel, Field, model_validator


class WorkItem(BaseModel):
    id: str
    objective: str
    read_only: bool
    write_globs: list[str] = Field(default_factory=list)
    verification_commands: list[str] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_permissions(self):
        if self.read_only and self.write_globs:
            raise ValueError("read-only tasks cannot declare writes")
        if not self.read_only and not self.write_globs:
            raise ValueError("writable tasks require declared scopes")
        return self


class WorkflowPlan(BaseModel):
    substantive: bool
    reason: str
    success_criteria: list[str] = Field(min_length=1)
    tasks: list[WorkItem] = Field(min_length=1, max_length=4)

    @model_validator(mode="after")
    def validate_wave(self):
        expected = range(2, 5) if self.substantive else range(1, 2)
        if len(self.tasks) not in expected:
            raise ValueError("routine plans need 1 task; substantive plans need 2-4")
        ids = [task.id for task in self.tasks]
        if len(ids) != len(set(ids)):
            raise ValueError("task IDs must be unique")
        return self


class WorkerOutput(BaseModel):
    status: Literal["completed", "blocked"]
    summary: str
    unresolved_risks: list[str] = Field(default_factory=list)


class ReviewOutput(BaseModel):
    candidate_passed: bool
    blocking_issues: list[str] = Field(default_factory=list)
    evidence: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_review(self):
        if self.candidate_passed and self.blocking_issues:
            raise ValueError("passing reviews cannot contain blocking issues")
        return self


class SynthesisOutput(BaseModel):
    summary: str
    acceptance_evidence: list[str] = Field(default_factory=list)
    unresolved_risks: list[str] = Field(default_factory=list)

The graph should store model-owned output separately from infrastructure-owned identity:

class WorkerRecord(BaseModel):
    task_id: str
    round: int
    workspace_id: str
    base_sha: str
    changed_files: list[str]
    artifact_uris: list[str]
    output: WorkerOutput


class Candidate(BaseModel):
    id: str
    thread_id: str
    round: int
    base_sha: str
    parent_candidate_id: str | None
    candidate_sha: str
    task_ids: list[str]
    workspace_ids: list[str]
    changed_files: list[str]
    diff_uri: str


class TestReport(BaseModel):
    candidate_id: str
    candidate_sha: str
    passed: bool
    exit_codes: dict[str, int]
    artifact_uri: str
    artifact_sha256: str

    @model_validator(mode="after")
    def validate_exit_codes(self):
        expected = bool(self.exit_codes) and all(
            code == 0 for code in self.exit_codes.values()
        )
        if self.passed != expected:
            raise ValueError("passed must match deterministic exit codes")
        return self

Step 2: Keep Model Calls Small and Typed

# app/model.py
import json
import os
from typing import TypeVar, cast

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

T = TypeVar("T", bound=BaseModel)


def base_model() -> ChatOpenAI:
    return ChatOpenAI(
        model=os.environ["AGENT_MODEL_ID"],
        max_retries=2,
    )


async def typed_call(
    schema: type[T],
    *,
    role: str,
    payload: dict,
) -> T:
    model = base_model().with_structured_output(
        schema,
        method="function_calling",
    )
    result = await model.ainvoke(
        [
            SystemMessage(content=role),
            HumanMessage(content=json.dumps(payload)),
        ]
    )
    return cast(T, result)

Use a model that supports tool calling and structured output. The worker path specifically requires simultaneous tool use and a final structured response; otherwise select an explicit LangChain ToolStrategy or implement the worker as a dedicated LangGraph tool subgraph. Keep different system contracts for planner, worker, reducer, verifier, and refuter while the runtime code remains deterministic.

Step 3: Inject a Workspace Service

# app/workspace.py
from typing import Protocol

from app.schemas import Candidate, TestReport, WorkItem, WorkerOutput, WorkerRecord


class WorkspaceService(Protocol):
    async def planner_evidence(self, repository: str, base_sha: str) -> dict: ...
    async def create_worker(
        self,
        *,
        repository: str,
        thread_id: str,
        round: int,
        task: WorkItem,
        expected_base_sha: str,
        parent_candidate_id: str | None,
        idempotency_key: str,
    ) -> tuple[str, list]: ...
    async def inspect_and_freeze(
        self,
        workspace_id: str,
        task: WorkItem,
        output: WorkerOutput,
        *,
        round: int,
    ) -> WorkerRecord: ...
    async def terminate_and_reset_for_retry(self, workspace_id: str) -> None: ...
    async def quarantine(self, workspace_id: str, reason: str) -> None: ...
    async def compose(
        self,
        *,
        thread_id: str,
        round: int,
        records: list[WorkerRecord],
        parent_candidate_id: str | None,
    ) -> Candidate: ...
    async def test(self, candidate: Candidate, commands: list[str]) -> TestReport: ...
    async def evidence(self, candidate: Candidate, report: TestReport) -> dict: ...
    async def promote(
        self,
        candidate: Candidate,
        *,
        approval_record_id: str,
        idempotency_key: str,
    ) -> dict: ...
    async def cleanup(self, thread_id: str) -> None: ...

This service must derive identity from Git and the sandbox—not model output. It validates actual changed files, composes patches, runs commands, hashes artifacts, and promotes only the approved candidate.

The sandbox service owns the hard worker timeout. It must revoke the lease, terminate and reap the process group/container, restore the deterministic workspace, and only then surface a timeout error. This scaffold deliberately avoids LangGraph node timeouts around side-effecting workers because LangGraph 1.2.9 does not wait for cancelled node cleanup before propagating NodeTimeoutError.

Step 4: Build the Graph

# app/graph.py
import asyncio
import hashlib
import json
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy, Send, TimeoutPolicy, interrupt
from langchain.agents import create_agent
from pydantic import ValidationError

from app.model import base_model, typed_call
from app.schemas import (
    Candidate,
    ReviewOutput,
    SynthesisOutput,
    TestReport,
    WorkItem,
    WorkerOutput,
    WorkerRecord,
    WorkflowPlan,
)
from app.workspace import WorkspaceService

MAX_REPAIR_ROUNDS = 2


class State(TypedDict, total=False):
    request: str
    repository: str
    base_sha: str
    thread_id: str
    round: int
    original_success_criteria: list[str]
    baseline_commands: list[str]
    plan: dict
    parent_candidate_id: str | None
    worker_records: Annotated[list[dict], operator.add]
    candidate: dict
    test_report: dict
    synthesis: dict
    reviews: Annotated[list[dict], operator.add]
    approval: dict
    wave_failure: dict | None
    final: dict


class WorkerState(TypedDict):
    request: str
    repository: str
    thread_id: str
    round: int
    task: dict
    expected_base_sha: str
    parent_candidate_id: str | None


def retry_transient(exc: Exception) -> bool:
    return isinstance(exc, ConnectionError)


def build_graph(checkpointer, workspaces: WorkspaceService, approvals):
    async def plan_node(state: State) -> dict:
        repo = await workspaces.planner_evidence(
            state["repository"],
            state["base_sha"],
        )
        plan = await typed_call(
            WorkflowPlan,
            role=(
                "Plan one routine task or 2-4 independent tasks. "
                "Declare write scopes and deterministic verification commands."
            ),
            payload={"request": state["request"], "repository": repo},
        )
        return {
            "plan": plan.model_dump(),
            "round": 0,
            "original_success_criteria": plan.success_criteria,
            "baseline_commands": sorted(
                {
                    command
                    for task in plan.tasks
                    for command in task.verification_commands
                }
            ),
            "parent_candidate_id": None,
        }

    def dispatch(state: State):
        plan = WorkflowPlan.model_validate(state["plan"])
        expected_base_sha = (
            state["base_sha"]
            if state["round"] == 0
            else Candidate.model_validate(state["candidate"]).candidate_sha
        )
        return [
            Send(
                "worker",
                {
                    "request": state["request"],
                    "repository": state["repository"],
                    "thread_id": state["thread_id"],
                    "round": state["round"],
                    "task": task.model_dump(),
                    "expected_base_sha": expected_base_sha,
                    "parent_candidate_id": state["parent_candidate_id"],
                },
            )
            for task in plan.tasks
        ]

    async def worker_node(state: WorkerState) -> dict:
        task = WorkItem.model_validate(state["task"])
        workspace_id = None
        try:
            workspace_id, tools = await workspaces.create_worker(
                repository=state["repository"],
                thread_id=state["thread_id"],
                round=state["round"],
                task=task,
                expected_base_sha=state["expected_base_sha"],
                parent_candidate_id=state["parent_candidate_id"],
                idempotency_key=(
                    f'{state["thread_id"]}:{state["round"]}:{task.id}:workspace'
                ),
            )
            coding_agent = create_agent(
                model=base_model(),
                tools=tools,
                response_format=WorkerOutput,
                checkpointer=False,
                system_prompt=(
                    "Complete only the assigned coding task. "
                    "Use only the provided scoped workspace tools."
                ),
            )
            agent_result = await coding_agent.ainvoke(
                {
                    "messages": [
                        {
                            "role": "user",
                            "content": json.dumps(
                                {
                                    "request": state["request"],
                                    "task": task.model_dump(),
                                    "workspace_id": workspace_id,
                                }
                            ),
                        }
                    ]
                }
            )
            output = WorkerOutput.model_validate(
                agent_result["structured_response"]
            )
            record = await workspaces.inspect_and_freeze(
                workspace_id,
                task,
                output,
                round=state["round"],
            )
            return {"worker_records": [record.model_dump()]}
        except asyncio.CancelledError:
            if workspace_id:
                await asyncio.shield(
                    workspaces.terminate_and_reset_for_retry(workspace_id)
                )
            raise
        except Exception as exc:
            if workspace_id:
                if retry_transient(exc):
                    await workspaces.terminate_and_reset_for_retry(workspace_id)
                else:
                    await workspaces.quarantine(workspace_id, str(exc))
            raise

    async def integrate_node(state: State) -> dict:
        plan = WorkflowPlan.model_validate(state["plan"])
        records = [
            WorkerRecord.model_validate(item)
            for item in state["worker_records"]
            if item["round"] == state["round"]
        ]
        expected = {task.id for task in plan.tasks}
        actual = [record.task_id for record in records]
        if (
            len(actual) != len(expected)
            or set(actual) != expected
            or any(record.output.status != "completed" for record in records)
        ):
            return {
                "wave_failure": {
                    "kind": "incomplete_worker_wave",
                    "expected_task_ids": sorted(expected),
                    "actual_task_ids": actual,
                }
            }

        candidate = await workspaces.compose(
            thread_id=state["thread_id"],
            round=state["round"],
            records=records,
            parent_candidate_id=state["parent_candidate_id"],
        )
        expected_base_sha = (
            state["base_sha"]
            if state["round"] == 0
            else Candidate.model_validate(state["candidate"]).candidate_sha
        )
        if (
            candidate.thread_id != state["thread_id"]
            or candidate.round != state["round"]
            or set(candidate.task_ids) != expected
            or set(candidate.workspace_ids)
            != {record.workspace_id for record in records}
            or candidate.parent_candidate_id != state["parent_candidate_id"]
            or candidate.base_sha != expected_base_sha
            or any(record.base_sha != expected_base_sha for record in records)
            or len(candidate.task_ids) != len(set(candidate.task_ids))
            or len(candidate.workspace_ids) != len(set(candidate.workspace_ids))
        ):
            raise RuntimeError("candidate lineage does not match the plan")

        commands = sorted(
            set(state["baseline_commands"])
            | {
                command
                for task in plan.tasks
                for command in task.verification_commands
            }
        )
        report = await workspaces.test(candidate, commands)
        if (
            report.candidate_id != candidate.id
            or report.candidate_sha != candidate.candidate_sha
        ):
            raise RuntimeError("test report is not bound to candidate")
        return {
            "candidate": candidate.model_dump(),
            "test_report": report.model_dump(),
            "wave_failure": None,
        }

    def route_after_integrate(state: State) -> str:
        return "failed" if state.get("wave_failure") else "synthesize"

    async def synthesize_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        report = TestReport.model_validate(state["test_report"])
        evidence = await workspaces.evidence(candidate, report)
        synthesis = await typed_call(
            SynthesisOutput,
            role=(
                "Summarize whether the candidate satisfies the criteria. "
                "Do not hide failed tests or unresolved risks."
            ),
            payload={
                "criteria": state["original_success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "evidence": evidence,
            },
        )
        return {"synthesis": synthesis.model_dump()}

    async def review_node(state: State, role: str) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        report = TestReport.model_validate(state["test_report"])
        evidence = await workspaces.evidence(candidate, report)
        review = await typed_call(
            ReviewOutput,
            role=(
                "Verify every criterion using immutable evidence."
                if role == "verifier"
                else "Try to falsify the exact candidate with a blocking example."
            ),
            payload={
                "candidate_sha": candidate.candidate_sha,
                "criteria": state["original_success_criteria"],
                "test_report": report.model_dump(),
                "evidence": evidence,
            },
        )
        return {
            "reviews": [
                {
                    "role": role,
                    "round": state["round"],
                    "candidate_sha": candidate.candidate_sha,
                    **review.model_dump(),
                }
            ]
        }

    async def verifier(state: State) -> dict:
        return await review_node(state, "verifier")

    async def refuter(state: State) -> dict:
        return await review_node(state, "refuter")

    def barrier(_: State) -> dict:
        return {}

    def route_review(state: State) -> str:
        candidate = Candidate.model_validate(state["candidate"])
        report = TestReport.model_validate(state["test_report"])
        current = {
            item["role"]: item
            for item in state["reviews"]
            if (
                item["round"] == state["round"]
                and item["candidate_sha"] == candidate.candidate_sha
            )
        }
        passed = (
            report.passed
            and set(current) == {"verifier", "refuter"}
            and all(item["candidate_passed"] for item in current.values())
            and all(not item["blocking_issues"] for item in current.values())
        )
        if passed:
            return "approval" if candidate.changed_files else "report"
        if state["round"] < MAX_REPAIR_ROUNDS:
            return "repair"
        return "failed"

    async def repair_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        repo = await workspaces.planner_evidence(
            state["repository"],
            candidate.candidate_sha,
        )
        repair = await typed_call(
            WorkflowPlan,
            role=(
                "Create one non-empty bounded repair wave. "
                "Do not change the supplied success criteria."
            ),
            payload={
                "request": state["request"],
                "success_criteria": state["original_success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": state["test_report"],
                "reviews": [
                    item
                    for item in state["reviews"]
                    if item["round"] == state["round"]
                ],
                "repository": repo,
            },
        )
        repair = repair.model_copy(
            update={"success_criteria": state["original_success_criteria"]}
        )
        return {
            "plan": repair.model_dump(),
            "round": state["round"] + 1,
            "parent_candidate_id": candidate.id,
        }

    def approval_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        binding_digest = hashlib.sha256(
            f'{state["thread_id"]}:{candidate.id}:{candidate.candidate_sha}'.encode()
        ).hexdigest()
        approval = interrupt(
            {
                "kind": "merge",
                "thread_id": state["thread_id"],
                "candidate_id": candidate.id,
                "candidate_sha": candidate.candidate_sha,
                "round": state["round"],
                "binding_digest": binding_digest,
                "test_report": state["test_report"],
            }
        )
        return {"approval": approval}

    def route_approval(state: State) -> str:
        return "promote" if state["approval"]["approved"] else "failed"

    async def promote_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        approval_id = await approvals.verify(
            state["approval"],
            expected_thread_id=state["thread_id"],
            expected_candidate_id=candidate.id,
            expected_candidate_sha=candidate.candidate_sha,
            expected_round=state["round"],
            expected_action="merge",
            expected_binding_digest=hashlib.sha256(
                f'{state["thread_id"]}:{candidate.id}:{candidate.candidate_sha}'.encode()
            ).hexdigest(),
        )
        result = await workspaces.promote(
            candidate,
            approval_record_id=approval_id,
            idempotency_key=f'{state["thread_id"]}:promote:{candidate.candidate_sha}',
        )
        return {"final": {"status": "completed", "result": result}}

    def report_node(state: State) -> dict:
        return {
            "final": {
                "status": "completed",
                "kind": "report",
                "candidate_sha": state["candidate"]["candidate_sha"],
                "synthesis": state["synthesis"],
                "test_report": state["test_report"],
            }
        }

    def failed_node(state: State) -> dict:
        return {
            "final": {
                "status": "failed",
                "round": state["round"],
                "candidate": state.get("candidate"),
                "wave_failure": state.get("wave_failure"),
            }
        }

    async def cleanup_node(state: State) -> dict:
        await workspaces.cleanup(state["thread_id"])
        return {}

    graph = StateGraph(State)
    graph.add_node(
        "plan",
        plan_node,
        retry_policy=RetryPolicy(
            max_attempts=2,
            retry_on=lambda exc: isinstance(exc, ValidationError),
        ),
        timeout=TimeoutPolicy(run_timeout=180),
    )
    graph.add_node(
        "worker",
        worker_node,
        retry_policy=RetryPolicy(
            max_attempts=2,
            retry_on=retry_transient,
        ),
    )
    graph.add_node("integrate", integrate_node)
    graph.add_node("synthesize", synthesize_node)
    graph.add_node("verify", verifier)
    graph.add_node("refute", refuter)
    graph.add_node("barrier", barrier)
    graph.add_node("repair", repair_node)
    graph.add_node("approval", approval_node)
    graph.add_node("promote", promote_node)
    graph.add_node("report", report_node)
    graph.add_node("failed", failed_node)
    graph.add_node("cleanup", cleanup_node)

    graph.add_edge(START, "plan")
    graph.add_conditional_edges("plan", dispatch)
    graph.add_conditional_edges("repair", dispatch)
    graph.add_edge("worker", "integrate")
    graph.add_conditional_edges("integrate", route_after_integrate)
    graph.add_edge("synthesize", "verify")
    graph.add_edge("synthesize", "refute")
    graph.add_edge(["verify", "refute"], "barrier")
    graph.add_conditional_edges("barrier", route_review)
    graph.add_conditional_edges("approval", route_approval)
    graph.add_edge("promote", "cleanup")
    graph.add_edge("report", "cleanup")
    graph.add_edge("failed", "cleanup")
    graph.add_edge("cleanup", END)

    return graph.compile(checkpointer=checkpointer)

The graph is the one durable top-level scheduler and source of truth. The LangChain worker helper runs an ephemeral tool loop, but it cannot pick arbitrary workspaces or bypass the candidate/test/approval chain.

Step 5: Add Durable Persistence

Use InMemorySaver only for tests. Keep the PostgreSQL saver open for the API process lifetime:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

from app.bootstrap import approvals, workspaces
from app.graph import build_graph


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncPostgresSaver.from_conn_string(
        os.environ["LANGGRAPH_DATABASE_URL"]
    ) as checkpointer:
        app.state.graph = build_graph(checkpointer, workspaces, approvals)
        yield


app = FastAPI(lifespan=lifespan)

The snippet is appropriate for low traffic. AsyncPostgresSaver 3.1.0 uses an instance-level lock around cursor operations, so attaching a connection pool does not remove all saver-level serialization. Benchmark checkpoint throughput and scale graph workers/saver instances deliberately, or use Agent Server’s managed runtime topology.

Run await checkpointer.setup() once in a migration job. Enable LANGGRAPH_STRICT_MSGPACK=true or configure an explicit allowed-module list for checkpoint deserialization.

Every independent coding job needs a new, tenant-owned thread_id; all invocations and approval resumes for that job reuse it:

result = await app.state.graph.ainvoke(
    {
        "request": "Audit all routes for missing authorization.",
        "repository": "github.com/acme/api",
        "base_sha": "abc123",
        "thread_id": thread_id,
        "round": 0,
        "worker_records": [],
        "reviews": [],
    },
    config={"configurable": {"thread_id": thread_id}},
)

Step 6: Resume Approval

Load the pending interrupt from the graph for the authenticated tenant. Persist an approval record bound to its candidate and binding digest, then resume:

from langgraph.types import Command


approval = await approvals.create_from_interrupt(
    pending_interrupt,
    approved=True,
    approved_by=authenticated_user.id,
)

result = await graph.ainvoke(
    Command(resume=approval.model_dump()),
    config={"configurable": {"thread_id": thread_id}},
)

Do not trust a candidate SHA supplied directly by the client. Load it from the pending thread state and verify ownership.

Deployment Choices

Custom FastAPI

Use FastAPI when you want full ownership of:

• run creation
• tenant authorization
• streaming
• queueing
• cancellation
• approval endpoints
• retention

This is the smallest OSS design, but the most application work.

LangSmith Agent Server

Agent Server adds a durable run queue, persistence, streaming, deployment tooling, and optional split API/queue workers. Single-host execution remains the default low-traffic topology.

For a standalone self-hosted deployment, current documentation requires:

• PostgreSQL 14+
• Redis 5+ or Valkey 8+
• DATABASE_URI and REDIS_URI
• LANGSMITH_API_KEY
• LANGGRAPH_CLOUD_LICENSE_KEY
• license-verification egress unless using the documented air-gapped path
• langgraph.json declaring dependencies and an exported graph/factory
• at least one continuously running queue worker

Standalone servers must be long-lived; the docs prohibit serverless scale-to-zero because it can cause task loss.

When Agent Server owns deployment, let it inject the production checkpointer and store rather than compiling the deployed graph with your own saver.

Infrastructure Requirements

Client / IDE / CI
       |
       v
API + tenant authorization
       |
       v
run queue / graph workers
       |
       +--> PostgreSQL checkpoints
       +--> cancellation and lease signals
       +--> immutable artifact storage
       +--> approval records
       |
       v
sandbox worker pool
  |        |        |
worker   worker   worker
  |        |        |
isolated repository workspaces
       |
       v
Git provider + CI

Redis is optional in a small custom runtime. It becomes useful for cross-replica signaling, queues, cancellation, and leases. It is required by the documented standalone Agent Server architecture.

Durable Execution Rules

LangGraph checkpoints at super-step boundaries. Nodes may run again after recovery.

Use idempotency keys for:

workspace creation
artifact writes
candidate composition
test runs
approval records
promotion
notifications

Use:

{thread_id}:{round}:{task_id}:{operation}

Keep non-idempotent side effects after approval interrupts or wrap them in task/idempotency infrastructure.

Failure Semantics

The graph’s terminal cleanup node handles normal completion and modeled failure. The API/queue supervisor must handle unhandled exceptions and orphan cleanup.

Evaluation

Compare:

1. One model/tool loop
2. A simple sequential LangGraph
3. This dynamic LangGraph

Use:

• routine tasks that should create one worker
• parallel read-only audits
• disjoint code changes
• overlap conflicts
• test failures and repairs
• approval rejection
• restart/resume
• cancellation during shell execution

Track accepted-task rate, trigger precision, policy violations, repair count, cost per accepted task, reviewer catch rate, human review time, and duplicate side effects after resume.

Evaluate paths, not only final answers. A correct answer produced through an unauthorized write or duplicated promotion is still a failed run.

What This Scaffold Leaves to You

You still need:

• a concrete worktree/sandbox implementation
• scoped coding tools
• authenticated approval storage
• API routes and thread authorization
• queueing and backpressure
• artifact retention
• process-group cancellation
• deployment manifests
• scenario tests

The graph shows where those contracts connect. It does not turn a local checkout into a security sandbox.

What It Does Not Copy

It does not copy Claude Code’s:

• private trigger heuristic
• generated JavaScript scripts
• context assembly
• provider effort controls
• workflow UI
• convergence internals

It approximates the portable loop with explicit, testable application code.

Final Recommendation

Start with:

• one graph
• one persistent checkpointer
• one workspace service
• four workers maximum
• two repair rounds
• deterministic tests
• two independent reviews
• approval bound to candidate SHA

The operating rule is:

LangGraph controls state and scheduling. Models propose. Sandbox infrastructure executes. Deterministic code decides what may be promoted.

Related Reading

• Claude Code Ultracode Architecture Blueprint for Cursor and Custom Agents
• Build an Ultracode-Like Coding Workflow with Agno and AgentOS
• 12-Factor Agents in Practice: LangChain/LangGraph and Agno/AgentOS
• How to Stream LangChain and LangGraph into the AI SDK
• Secure Agentic Code Generation

Sources

• LangChain Docs, “Agents”, accessed July 15, 2026.
• LangChain Docs, “Structured output”, accessed July 15, 2026.
• LangGraph Docs, “Graph API overview”, accessed July 15, 2026.
• LangGraph Docs, “Use the Graph API”, accessed July 15, 2026.
• LangGraph Docs, “Workflows and agents”, accessed July 15, 2026.
• LangGraph Docs, “Persistence”, accessed July 15, 2026.
• LangGraph Docs, “Fault tolerance”, accessed July 15, 2026.
• LangGraph Docs, “Interrupts”, accessed July 15, 2026.
• LangGraph Docs, “Subgraphs”, accessed July 15, 2026.
• LangGraph Docs, “Event streaming”, accessed July 15, 2026.
• LangGraph, “PostgreSQL checkpointer README”, accessed July 15, 2026.
• LangSmith Docs, “Agent Server”, accessed July 15, 2026.
• LangSmith Docs, “Self-host standalone servers”, accessed July 15, 2026.
• LangSmith Docs, “Self-hosted dependency versions”, accessed July 15, 2026.
• PyPI, langgraph 1.2.9, July 10, 2026.
• PyPI, langchain 1.3.13, accessed July 15, 2026.
• PyPI, langchain-openai 1.3.5, accessed July 15, 2026.
• PyPI, langgraph-checkpoint-postgres 3.1.0, accessed July 15, 2026.