# LangGraph

> Two ways to use LangGraph with Pyrula, what each journals, and what reruns after a restart.

Source: https://docs.pyrula.io/agents/integrations/langgraph/

`pyrula init --langgraph` writes a runnable app in the first shape. Anything else is listed under [Not supported](#not-supported).

| Shape | Who owns the run | You write |
|---|---|---|
| [A graph inside a Pyrula run](#a-graph-inside-a-pyrula-run) | Pyrula. The graph is the body of an `@agent` or `@workflow`. | `durable_chat_model(...)` for the model, `ctx.step` around tool effects. |
| [A Pyrula run as a graph node](#a-pyrula-run-as-a-graph-node) | LangGraph. One node submits a Pyrula run and awaits it. | `as_langgraph_node(...)`, a checkpointed graph, a Pyrula worker. |

Both need `pip install "pyrula[agents,langgraph]"` (`langgraph>=1.0.0`, `langchain-core>=0.3.67`); the first also needs `"pyrula[agents,langchain]"`.

## A graph inside a Pyrula run

Wrap the chat model in `durable_chat_model` and build the graph in the agent body. The run's journal is the only checkpoint. After a restart the body runs again from the top, and the provider is called only for the model call that had not finished.

```python
import asyncio
import operator
from typing import Annotated, Any, TypedDict

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool as lc_tool
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode

from pyrula.agents import AgentContext, agent
from pyrula.agents.integrations import durable_chat_model
from pyrula.agents.testing import AgentRunner

provider_calls: list[int] = []
charged: list[str] = []


class ScriptedChat(BaseChatModel):
    """Stands in for a provider: asks for a charge, then answers."""

    @property
    def _llm_type(self) -> str:
        return "scripted"

    def bind_tools(self, tools, **kwargs):
        return self.bind(tools=[{"name": t.name} for t in tools], **kwargs)

    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
        provider_calls.append(len(messages))
        if not any(m.type == "tool" for m in messages):
            reply = AIMessage(
                content="",
                tool_calls=[{"id": "c1", "name": "charge", "args": {"order": "A-1"}}],
            )
        else:
            reply = AIMessage(content="charged A-1")
        return ChatResult(generations=[ChatGeneration(message=reply)])

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        raise NotImplementedError("durable_chat_model journals the async path only")


class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]


@agent(name="docs_graph_agent", timeout=60)
async def support(ctx: AgentContext, message: str) -> str:
    @lc_tool
    async def charge(order: str) -> str:
        """Charge an order."""
        # The graph runs this tool itself, so the effect goes through ctx.step to run once.
        return await ctx.step(f"charge:{order}", lambda: charged.append(order) or "ok")

    model = durable_chat_model(ScriptedChat()).bind_tools([charge])

    async def call_model(state: State) -> dict[str, Any]:
        return {"messages": [await model.ainvoke(state["messages"])]}

    def route(state: State) -> str:
        return "tools" if state["messages"][-1].tool_calls else END

    graph = StateGraph(State)
    graph.add_node("model", call_model)
    graph.add_node("tools", ToolNode([charge]))
    graph.add_edge(START, "model")
    graph.add_conditional_edges("model", route, ["tools", END])
    graph.add_edge("tools", "model")

    out = await graph.compile().ainvoke({"messages": [HumanMessage(content=message)]})
    return out["messages"][-1].content


async def main() -> None:
    runner = AgentRunner(support)
    # The worker dies as the second model call starts.
    runner.inject_crash_at("llm_call", nth=2, when="pending")
    await runner.run(message="charge A-1")
    events = await runner.run(message="charge A-1")

    assert events[-1].payload["result"] == "charged A-1"
    assert provider_calls == [1, 3]  # the first call was not made again
    assert charged == ["A-1"]  # the charge was not made again


asyncio.run(main())
```
### What is journaled

| Part of the graph | After a restart |
|---|---|
| Model call through `durable_chat_model` | Answered from the journal. Async only (`ainvoke`, `astream`); `invoke` raises `UnjournaledSurface`. |
| The call a worker died inside | Made again. |
| A tool the graph runs (`ToolNode`) | Runs again. Make it idempotent, or run the effect in `ctx.step`. |
| `ctx.step` inside a node or tool | Answered from the journal, unless its result was not yet recorded. |
| Routing and state | Recomputed. The path must follow from the recorded responses. |
| A model call a spend cap refuses | Stops the run (`budget_exceeded`), also from a `ToolNode` tool. |
| A pause or cancel | Stops the run at the next model call. |
| `await ctx.interrupt(...)` in a node | Parks the run until answered. |

Each live call writes `llm:done` from `usage_metadata` and counts toward `AgentLimits`, so a cycling graph stops there rather than at LangGraph's recursion limit. Pass `model_id=` when the model has no `model_name`, so a cost cap can price it.

A call is identified by the wrapper's `name` and its position among that name's calls. On replay the messages, `tools`, `tool_choice` and `response_format` must match the recording, or the run ends with `replay_input_changed`. Give parallel branches that call models their own `name=`. `astream` on the model journals the whole response as one call.

## A Pyrula run as a graph node

`as_langgraph_node` makes one node a durable Pyrula run: it submits the run to the store and awaits the result, and a worker on the same store executes it.

```python
from pyrula.agents.integrations import as_langgraph_node

graph.add_node("provision", as_langgraph_node("provision_infra", store=store, node_key="provision"))
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "t-1"}}
await app.ainvoke({"input": {"user_id": 7}}, config)
# After the graph process restarts, LangGraph's own resume re-attaches the same run:
await app.ainvoke(None, config)
```
A runnable version with an in-process worker is [`examples/langgraph_durable_node.py`](https://github.com/pyrula-io/pyrula/tree/main/examples/langgraph_durable_node.py).

The node reads `state["input"]` and returns the result under `"output"` (`input_key`, `output_key`). Other options are `timeout=` (default: wait forever), `poll_interval=0.5` and `on_interrupt=` to replace the `ctx.interrupt` bridge. The graph needs a checkpointer, because the run id is `<thread_id>::<LangGraph task>::<node_key>`.

| Event | What happens |
|---|---|
| Graph process dies while the node waits | `ainvoke(None, config)` re-attaches the run already submitted. Sending the original input again starts a second run. |
| Worker dies mid-run | Another worker recovers it; recorded steps are read back. |
| Workflow calls `ctx.interrupt` | The node raises LangGraph's `interrupt()`; `Command(resume=...)` answers it from any process. |
| Several questions in a row | Each `Command(resume=...)` answers the open one. A question after a timed-out interrupt, or in a run that continued as new, raises `RuntimeError`. Cancel the suspended run before starting over. |
| Workflow fails or is cancelled | The node raises `DurableInvocationError` or `DurableInvocationCancelled`. |

### One run per LangGraph task

The run id includes the LangGraph task. LangGraph keeps the task across its own resume of a node and gives every pass and every new `ainvoke` a new one.

- A cycle through the node runs the workflow once per pass; two `ainvoke` calls on one `thread_id` each run it.
- A failed run stays failed: every `ainvoke(None, config)` and `RetryPolicy` attempt re-attaches and raises the same error. Put retries inside the workflow.
- A re-attach with different input raises `ValueError`.

## Not supported

- A checkpointer on a graph inside a Pyrula run. The saver restores old messages and the run ends with `replay_input_changed`.
- LangGraph's `interrupt()` inside a Pyrula run. Use `ctx.interrupt`.
- `with_fallbacks` or `with_retry` around `durable_chat_model`. Use `RetryPolicy` on the node (it does not retry an `as_langgraph_node` run).
- Streaming tokens from a node.
- Wrappers beyond `durable_chat_model`, `as_langgraph_node` and `from_langchain_tool`. A plain LangChain tool reruns on resume unless it runs in `ctx.step`.
- Answering a node's question outside the graph. Answer through `Command(resume=...)`.
- `invoke` from a thread. Use the async path.

See the [guarantee matrix](https://docs.pyrula.io/agents/integrations/guarantees.md) and [what's durable, and what isn't](https://docs.pyrula.io/agents/integrations/durable-vs-not.md).