LangGraph

pyrula init --langgraph writes a runnable app in the first shape. Anything else is listed under Not supported.

Shape	Who owns the run	You write
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	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.

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.

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.

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 and what’s durable, and what isn’t.