A step limit bounds how many calls an agent makes. It does not bound the duration of one call or the tokens that call uses. Here you add per-tool timeouts, a model-node timeout and a soft token budget to a code-review graph.
In a bounded Gemini run on October 23, 2026, the model proposed fetch_job_log(job_id=815). Native ToolNode executed that request against the local CI fixture: its three-second timeout interrupted an eight-second delay. The timeout became an error ToolMessage. The probe’s separate static breakpoint then paused the graph after tools, before any model follow-up; the timeout itself did not interrupt or pause the graph. This is the actual tool result:
The five saved Gemini comparisons elsewhere in this lesson are supplied historical recordings. They are separate from this bounded observation and from the current offline tests.
When you need this. The review assistant answers in a chat channel, and a CI log fetch can outlast the caller’s wait. Bound tool and model calls separately, and check accumulated usage before requesting more tools. A timed-out log fetch should leave the CI result unknown rather than invent a failed job.
1. Time out each tool call
Download the complete lab, extract it and open a terminal in agent-budgets/. The README lists its runtime and recorded examples.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txtUse the extracted lab and its virtual environment. ci.py stands in for a CI service: each job has a fixed delay in place of network time, and the integration job of #421 takes eight seconds.
"""Tools over a stub CI service. Each job has a fixed delay that stands in for network time."""
import asyncio
from langchain_core.tools import tool
TOOL_TIMEOUT = 3.0 # seconds one tool call may take
JOBS = {
812: {"pr": 415, "name": "unit-tests", "delay": 0.2,
"log": "FAILED tests/test_client.py::test_default_timeout - AssertionError: expected 10, got 30"},
815: {"pr": 421, "name": "integration", "delay": 8.0,
"log": "ERROR: postgres container did not become healthy within 120s"},
}
@tool
async def list_failed_jobs(pr: int) -> list[dict]:
"""List the failed CI jobs of a pull request, by its number."""
async with asyncio.timeout(TOOL_TIMEOUT):
await asyncio.sleep(0.1)
return [{"job_id": i, "name": j["name"]} for i, j in JOBS.items() if j["pr"] == pr]
@tool
async def fetch_job_log(job_id: int) -> str:
"""Fetch the end of one CI job's log."""
async with asyncio.timeout(TOOL_TIMEOUT):
job = JOBS.get(job_id)
if job is None:
return f"No job {job_id}."
await asyncio.sleep(job["delay"])
return job["log"]Each tool wraps its work in asyncio.timeout(TOOL_TIMEOUT). When the time runs out, the pending await is cancelled and TimeoutError is raised inside the tool. The timeout lives in the tool, not on the tools node, for a reason: one model reply can ask for several calls, and ToolNode runs them concurrently. A per-call timeout cancels only the slow call; a node-level timeout would discard the results of the calls that finished.
2. Turn the timeout into a tool message
A timeout inside a tool is only half the job; now we decide what the model hears about it.
"""The code-review agent triages CI, with a timeout on every tool and model call and a token budget."""
import asyncio
import operator
import os
import sys
from typing import Annotated, Literal
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_ollama import ChatOllama
from langgraph.errors import NodeError, NodeTimeoutError
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.types import Command, TimeoutPolicy
from typing_extensions import TypedDict
import ci
TOOLS = [ci.list_failed_jobs, ci.fetch_job_log]
SYSTEM = """You help a developer work through the team's code-review queue. To explain a CI failure,
list the pull request's failed jobs, fetch the log of each, and explain the failure in two sentences.
If a tool fails, say what you could not check."""
MODEL = os.environ.get("REVIEW_MODEL", "gemini-3.8-flash") # REVIEW_MODEL=gemma4:e2b runs it on Ollama
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
tokens: Annotated[int, operator.add] # sum of valid reported token counts
usage_known: bool # false when the latest reply has missing or invalid usage
stop_reason: str
def tool_timed_out(error: TimeoutError) -> str:
"""Turn a tool timeout into a tool message, so the model can answer without it."""
return (f"The tool call did not finish within {ci.TOOL_TIMEOUT:g}s, so its result is unknown. "
"This says nothing about the CI job itself. Do not retry.")
def model_failed(state: State, error: NodeError) -> Command:
"""The model node timed out (or raised): end the run in a named state instead of an exception."""
return Command(update={"usage_known": False, "stop_reason": "model_timeout" if isinstance(error.error, NodeTimeoutError) else "model_error",
"messages": [AIMessage(f"Stopped: {error.error}")]}, goto=END)
def chat_model():
if MODEL.startswith("gemini"):
return ChatGoogleGenerativeAI(model=MODEL, max_retries=1) # one client retry inside the node timeout
return ChatOllama(model=MODEL, temperature=0, reasoning=False)
def build_graph(model=None, *, token_budget=4000, model_timeout=60.0):
llm = (model or chat_model()).bind_tools(TOOLS)
async def agent(state: State) -> dict:
reply = await llm.ainvoke([SystemMessage(SYSTEM), *state["messages"]])
usage = reply.usage_metadata
total = usage.get("total_tokens") if isinstance(usage, dict) else None
known = isinstance(total, int) and not isinstance(total, bool) and total >= 0
update = {"messages": [reply], "usage_known": known}
if known:
update["tokens"] = total
return update
def route(state: State) -> Literal["tools", "stop", "__end__"]:
if not state["usage_known"]:
return "stop"
if not state["messages"][-1].tool_calls:
return END
return "stop" if state["tokens"] >= token_budget else "tools"
def stop(state: State) -> dict:
used = state["tokens"]
if not state["usage_known"]:
return {"stop_reason": "missing_usage", "messages": [AIMessage(
f"Stopped: model usage is missing or invalid; the run total is unknown. "
f"The known subtotal from reported calls is {used} tokens.")]}
return {"stop_reason": "token_budget",
"messages": [AIMessage(f"Stopped: {used} of {token_budget} tokens used.")]}
graph = StateGraph(State)
graph.add_node("agent", agent, timeout=TimeoutPolicy(run_timeout=model_timeout), error_handler=model_failed)
graph.add_node("tools", ToolNode(TOOLS, handle_tool_errors=tool_timed_out))
graph.add_node("stop", stop)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route)
graph.add_edge("tools", "agent")
graph.add_edge("stop", END)
return graph.compile()
async def triage(graph, question: str) -> dict:
return await graph.ainvoke({"messages": [HumanMessage(question)], "tokens": 0}, {"recursion_limit": 12})
if __name__ == "__main__":
result = asyncio.run(triage(build_graph(), " ".join(sys.argv[1:]) or "Why is CI failing on pull request 415?"))
label = "tokens" if result.get("usage_known", False) else "known subtotal; run total unknown"
print(result.get("stop_reason", "answered"), result["tokens"], label, "|", result["messages"][-1].text)ToolNode(TOOLS, handle_tool_errors=tool_timed_out) passes the handler a function. ToolNode reads the type hint of its error parameter, TimeoutError, and catches only that type. The handler’s return value becomes the ToolMessage for the call, so the model learns that the result is unknown and can still answer. Any other exception from a tool, such as a TypeError from a bug, still stops the run, because hiding bugs from the caller is worse than failing.
The message says the result is unknown, not that the job failed. A timeout is a fact about the call; letting the model read it as a fact about the CI job would turn a slow service into a wrong answer.
A slow tool. In the supplied historical comparison, the log-fetch step for #421 lasted 3 seconds and returned a timeout; the recorded model reply said it could not check the log. asyncio.timeout in the tool, and a typed handle_tool_errors handler. Check: test_a_slow_tool_becomes_a_tool_message_and_the_run_goes_on.
A bug in a tool. A TypeError from a bad job record is not a timeout, so it stops the run. Catch only the exception types that describe a slow or unavailable service. Check: test_other_tool_errors_are_not_hidden_from_the_caller.
3. Bound the model call with TimeoutPolicy
add_node(..., timeout=TimeoutPolicy(run_timeout=model_timeout)) limits each run of the agent node. When the limit passes, LangGraph cancels the node and raises NodeTimeoutError. error_handler=model_failed catches it: LangGraph runs the handler as a node of its own, named __error_handler__agent, with a NodeError that wraps the original exception. The handler returns a Command that writes a stop reason and goes to END, so the caller gets a state, not a traceback.
ChatGoogleGenerativeAI(max_retries=1) is the normal graph factory setting. The client can retry failed requests inside the node, and those retries use the same node timeout. The separate bounded probe below injects a model with max_retries=0 and an external operator that forbids retries.
A slow model call. In the supplied historical comparison, a 0.5-second limit cancelled the first call: “Node 'agent' exceeded its run timeout of 0.500s (elapsed: 0.501s).” TimeoutPolicy on the node and an error_handler that ends the run. Check: test_a_slow_model_call_ends_the_run_at_its_timeout.
4. Spend a token budget
tokens is a state field with an operator.add reducer, so each agent run adds a valid reported total_tokens value from its reply. It is the known subtotal, not an estimate for calls whose usage was not reported. route checks the total after each model call. If the reply asks for more tools and the budget is spent, the run goes to stop instead of running them.
A spent budget. In the supplied historical comparison, a budget of 500 tokens stopped the run before fetching the log, at 632 tokens. Check the total in the router before more tools run. Check: test_the_token_budget_stops_the_run_before_more_tools_run.
One call overshoots the budget. In the supplied historical comparison with a budget of 1000 tokens, the last call took the run to 1649. Treat this threshold as a soft limit: a single call can exceed it. Bound requests separately and inspect reported usage; this router does not guarantee provider billing. Check: test_the_budget_is_checked_between_calls_so_one_call_can_overshoot_it.
Missing or invalid model usage. If total_tokens is absent, negative or not an integer, usage_known becomes false and the router stops with missing_usage before another tool or model call. It keeps the subtotal from earlier reported calls; the run total is unknown. An unreported call is not counted as measured zero. The same check applies to a final answer. A model timeout or error also makes the run total unknown; earlier reported tokens remain a known subtotal. Check: test_missing_or_invalid_usage_stops_before_tools_and_keeps_known_subtotal, test_missing_usage_on_an_answer_is_not_reported_as_a_measured_zero.
A timeout does not undo work. A cancelled tool call may have reached the service, and a write may have happened; the retries lesson deals with that. Here both tools only read, which is why the timeout handler can tell the model not to retry.
5. Read the saved comparisons and run the lab
Follow the lab setup, then try the existing positive command before recording the comparisons:
python graph.py "Why is CI failing on pull request 415?"record.py makes fresh model requests and overwrites runs.json; use a disposable copy. Its supplied historical output below contains five scenarios: a normal answer, the slow log fetch, a budget of 500 tokens, a budget of 1000, and a model timeout of 0.5 seconds.
python record.py== answered: Why is CI failing on pull request 415? limits={}
outcome: answered tokens: 1675
== slow-tool: Why is CI failing on pull request 421? limits={}
outcome: answered tokens: 1821
== budget-stopped: Why is CI failing on pull request 415? limits={'token_budget': 500}
outcome: token_budget tokens: 632
== budget-overshoot: Why is CI failing on pull request 415? limits={'token_budget': 1000}
outcome: answered tokens: 1649
== slow-model: Why is CI failing on pull request 415? limits={'model_timeout': 0.5}
outcome: model_timeout tokens: 0This preserved historical output labels the timed-out model call as zero tokens. No usage was returned for that call: zero is only the known reported subtotal, and the run total is unknown. The current graph and recorder mark that distinction explicitly; these saved results are not a fresh measurement of the changed policy.
Read a recorded tool timeout
Slow tool timeout. The recorded model proposed fetch_job_log(job_id=815). The fixture delays that job for eight seconds; the three-second tool timeout returned an error ToolMessage with the same call ID. A static breakpoint stopped the graph after tools, so the run contains no model follow-up answer. This example isolates the tool timeout; the separate tests exercise model timeouts and token-budget stops.
The tool call did not finish within 3s, so its result is unknown. This says nothing about the CI job itself. Do not retry.Inspect the saved events
[
{
"event": "task",
"question": "The local CI fixture says pull request #421 has failed job 815. Fetch the log for job 815 directly with fetch_job_log. Do not retry a tool, edit files, or contact a real CI service. This observation pauses after tools.",
"policy": {
"lesson": "04",
"model": "gemini-3.8-flash",
"max_model_calls": 1,
"max_output_tokens": 4096,
"thinking_level": "low",
"http_timeout_seconds": 45,
"max_retries": 0,
"vertexai": false,
"graph_model_timeout_seconds": 60,
"interrupt_after": [
"tools"
]
}
},
{
"event": "model_request",
"ordinal": 1,
"policy": {
"lesson": "04",
"model": "gemini-3.8-flash",
"max_model_calls": 1,
"max_output_tokens": 4096,
"thinking_level": "low",
"http_timeout_seconds": 45,
"max_retries": 0,
"vertexai": false,
"graph_model_timeout_seconds": 60,
"interrupt_after": [
"tools"
]
}
},
{
"event": "model_response",
"ordinal": 1,
"message": {
"type": "ai",
"text": "",
"tool_calls": [
{
"name": "fetch_job_log",
"args": {
"job_id": 815
},
"id": "call_97095"
}
],
"usage": {
"input_tokens": 224,
"output_tokens": 179,
"total_tokens": 403,
"input_token_details": {
"cache_read": 0
},
"output_token_details": {
"reasoning": 159
}
},
"finish_reason": "STOP"
}
},
{
"event": "graph_step",
"node": "agent",
"messages": [
{
"type": "ai",
"text": "",
"tool_calls": [
{
"name": "fetch_job_log",
"args": {
"job_id": 815
},
"id": "call_97095"
}
],
"usage": {
"input_tokens": 224,
"output_tokens": 179,
"total_tokens": 403,
"input_token_details": {
"cache_read": 0
},
"output_token_details": {
"reasoning": 159
}
},
"finish_reason": "STOP"
}
]
},
{
"event": "graph_step",
"node": "tools",
"messages": [
{
"type": "tool",
"text": "The tool call did not finish within 3s, so its result is unknown. This says nothing about the CI job itself. Do not retry.",
"name": "fetch_job_log",
"tool_call_id": "call_97095",
"status": "error"
}
]
},
{
"event": "static_breakpoint",
"after": "tools"
},
{
"event": "completion",
"status": "observed_native_tool_timeout_before_followup",
"classification": "accepted tool proposal and native timeout observed; model provenance requires external HTTP receipts; no follow-up model answer",
"accepted_tool_proposal": true,
"node_path": [
"agent",
"tools"
],
"model_calls": 1,
"tool_timeout_seconds": 3.0,
"fixture_job_delay_seconds": 8.0,
"tools": [
{
"type": "tool",
"text": "The tool call did not finish within 3s, so its result is unknown. This says nothing about the CI job itself. Do not retry.",
"name": "fetch_job_log",
"tool_call_id": "call_97095",
"status": "error"
}
],
"reported_token_subtotal": 403,
"usage_known": true,
"elapsed_seconds": 6.4699207919766195,
"no_automatic_rerun": true,
"no_followup_model_call": true
}
]Incomplete preflight. An earlier attempt saved one token-count response, then stopped with DecodingError while reconstructing the response. It made no generation request or tool call. The local invocation counter counts an attempted graph call, not completed inference.
How it works
Each limit sits where its cost is decided. The tool timeout is inside the call, because only the call knows when its I/O started. The model timeout is on the node, because a model call has no I/O your code controls. The token budget is in the router, because it needs the total from state, and state is only updated between steps.
That last point explains the overshoot. This graph checks the budget after a call returns; its router cannot interrupt a call already in progress. With a budget of 1000, the run had used 698 tokens after two model calls, so it went on; the third call alone used 951, and the run ended at 1649. Reasoning tokens make each call’s size hard to predict: 441 of that last call’s tokens were reasoning, which the request does not show.
Test the limits without a model
test_graph.py uses a scripted chat model with a configurable delay and fixed token counts, so each limit can be exercised without provider requests.
python -m pytest -v test_graph.py test_live_probe.py test_run_evidence.pytest_graph.py::test_a_slow_tool_becomes_a_tool_message_and_the_run_goes_on PASSED
test_graph.py::test_a_fast_tool_returns_its_result PASSED
test_graph.py::test_the_token_budget_stops_the_run_before_more_tools_run PASSED
test_graph.py::test_the_budget_is_checked_between_calls_so_one_call_can_overshoot_it PASSED
test_graph.py::test_a_slow_model_call_ends_the_run_at_its_timeout PASSED
test_graph.py::test_other_tool_errors_are_not_hidden_from_the_caller PASSED
============================== 6 passed ===============================The saved output reports 6 passing tests. The current graph tests add missing, malformed and explicitly zero usage cases. test_live_probe.py uses scripted replies to exercise the real async graph, static breakpoint and incomplete outcomes; it makes no provider requests. test_run_evidence.py checks the saved genuine and incomplete records. These checks are separate from the genuine generation request above.
Read the test and recording scripts
"""Tests for the CI triage graph's limits. A scripted chat model replaces Gemini, so no network or API key is needed."""
import asyncio
import time
import pytest
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, ToolMessage
import ci
import graph
class ScriptedChat(GenericFakeChatModel):
delay: float = 0.0
def bind_tools(self, tools, **kwargs):
return self
async def _agenerate(self, *args, **kwargs):
await asyncio.sleep(self.delay)
return self._generate(*args, **kwargs)
def reply(tokens, **call):
calls = [{"name": call.pop("name"), "args": call, "id": f"c{tokens}"}] if call else []
return AIMessage("" if calls else "answer", tool_calls=calls,
usage_metadata={"input_tokens": tokens, "output_tokens": 0, "total_tokens": tokens})
def triage(messages, delay=0.0, **limits):
model = ScriptedChat(messages=iter(messages), delay=delay)
return asyncio.run(graph.triage(graph.build_graph(model, **limits), "Why is CI failing on pull request 421?"))
def test_a_slow_tool_becomes_a_tool_message_and_the_run_goes_on(monkeypatch):
monkeypatch.setattr(ci, "TOOL_TIMEOUT", 0.2)
started = time.perf_counter()
result = triage([reply(100, name="fetch_job_log", job_id=815), reply(100)])
tool = next(m for m in result["messages"] if isinstance(m, ToolMessage))
assert "did not finish within 0.2s" in tool.content
assert result["messages"][-1].content == "answer"
assert time.perf_counter() - started < 2 # the 8 s call was cancelled
def test_a_fast_tool_returns_its_result():
result = triage([reply(100, name="fetch_job_log", job_id=812), reply(100)])
tool = next(m for m in result["messages"] if isinstance(m, ToolMessage))
assert tool.content.startswith("FAILED tests/test_client.py")
def test_the_token_budget_stops_the_run_before_more_tools_run():
result = triage([reply(500, name="list_failed_jobs", pr=415),
reply(400, name="fetch_job_log", job_id=812)], token_budget=800)
assert result["stop_reason"] == "token_budget" and result["tokens"] == 900
assert sum(isinstance(m, ToolMessage) for m in result["messages"]) == 1
def test_the_budget_is_checked_between_calls_so_one_call_can_overshoot_it():
result = triage([reply(500, name="list_failed_jobs", pr=415), reply(700)], token_budget=800)
assert "stop_reason" not in result and result["tokens"] == 1200
def test_a_slow_model_call_ends_the_run_at_its_timeout():
started = time.perf_counter()
result = triage([reply(100)], delay=5.0, model_timeout=0.3)
assert result["stop_reason"] == "model_timeout"
assert "exceeded its run timeout of 0.300s" in result["messages"][-1].content
assert time.perf_counter() - started < 2
def test_other_tool_errors_are_not_hidden_from_the_caller(monkeypatch):
monkeypatch.setitem(ci.JOBS, 900, {"pr": 415, "name": "lint", "delay": "soon", "log": ""})
with pytest.raises(TypeError):
triage([reply(100, name="fetch_job_log", job_id=900), reply(100)])
@pytest.mark.parametrize("usage", [None, {}, {"total_tokens": None}, {"total_tokens": "100"},
{"total_tokens": -1}, {"total_tokens": True}, {"total_tokens": 1.5}])
def test_missing_or_invalid_usage_stops_before_tools_and_keeps_known_subtotal(usage):
# model_copy lets this adapter-boundary test represent malformed provider metadata.
unknown = reply(1, name="fetch_job_log", job_id=812).model_copy(update={"usage_metadata": usage})
result = triage([reply(120, name="list_failed_jobs", pr=415), unknown])
assert result["stop_reason"] == "missing_usage"
assert result["usage_known"] is False and result["tokens"] == 120
assert sum(isinstance(m, ToolMessage) for m in result["messages"]) == 1
assert "120" in result["messages"][-1].content and "known subtotal" in result["messages"][-1].content
def test_missing_usage_on_an_answer_is_not_reported_as_a_measured_zero():
unknown = AIMessage("An answer without provider usage.")
result = triage([unknown])
assert result["stop_reason"] == "missing_usage"
assert result["usage_known"] is False
assert "known subtotal" in result["messages"][-1].content
assert "unknown" in result["messages"][-1].content
def test_reported_zero_tokens_is_distinct_from_missing_usage():
result = triage([reply(0)])
assert result["usage_known"] is True and result["tokens"] == 0
assert "stop_reason" not in result
def test_a_timeout_after_reported_usage_marks_the_run_total_unknown():
class SlowSecondReply(ScriptedChat):
calls: int = 0
async def _agenerate(self, *args, **kwargs):
self.calls += 1
if self.calls == 2:
await asyncio.sleep(5)
return self._generate(*args, **kwargs)
model = SlowSecondReply(messages=iter([reply(120, name="list_failed_jobs", pr=415)]))
result = asyncio.run(graph.triage(graph.build_graph(model, model_timeout=0.3), "Check CI."))
assert result["stop_reason"] == "model_timeout"
assert result["tokens"] == 120 and result["usage_known"] is False
assert sum(isinstance(m, ToolMessage) for m in result["messages"]) == 1"""Run the CI triage agent on Gemini under each limit and save what happened."""
import asyncio
import json
import time
from langchain_core.messages import HumanMessage, ToolMessage
from graph import MODEL, build_graph, chat_model
SCENARIOS = [
# name, question, graph limits
("answered", "Why is CI failing on pull request 415?", {}),
("slow-tool", "Why is CI failing on pull request 421?", {}),
("budget-stopped", "Why is CI failing on pull request 415?", {"token_budget": 500}),
("budget-overshoot", "Why is CI failing on pull request 415?", {"token_budget": 1000}),
("slow-model", "Why is CI failing on pull request 415?", {"model_timeout": 0.5}),
]
async def run(name, question, limits) -> dict:
graph = build_graph(chat_model(), **limits)
state = {"messages": [HumanMessage(question)], "tokens": 0}
steps, outcome, mark = [], "answered", time.perf_counter()
async for kind, data in graph.astream(state, {"recursion_limit": 12}, stream_mode=["updates", "values"]):
if kind == "values":
state = data
continue
node, delta = next(iter(data.items()))
now = time.perf_counter()
step = {"node": node, "seconds": round(now - mark, 1)}
mark = now
if node == "tools":
step["results"] = [m.content for m in delta["messages"] if isinstance(m, ToolMessage)]
if node == "agent":
usage = delta["messages"][-1].usage_metadata or {}
step["usage_known"] = delta["usage_known"]
if delta["usage_known"]:
step["tokens"] = delta["tokens"]
details = usage.get("output_token_details") if isinstance(usage, dict) else None
step["reasoning"] = details.get("reasoning") if isinstance(details, dict) else None
steps.append(step)
outcome = delta.get("stop_reason", outcome)
last = state["messages"][-1]
return {"scenario": name, "question": question, "limits": limits,
"outcome": outcome, "tokens": state["tokens"], "usage_known": state.get("usage_known", False), "steps": steps,
"final": getattr(last, "text", last)}
async def main() -> None:
runs = [await run(*s) for s in SCENARIOS]
with open("runs.json", "w") as f:
json.dump({"model": MODEL, "runs": runs}, f, indent=1)
for r in runs:
print(f"== {r['scenario']}: {r['question']} limits={r['limits']}")
for step in r["steps"]:
detail = (f"{step['tokens']} tokens ({step['reasoning']} reasoning)" if "tokens" in step else
"model usage unavailable" if step.get("usage_known") is False else
" | ".join(x[:70] for x in step.get("results", [])))
print(f" {step['node']:<6} {step['seconds']:>5}s {detail}")
label = "tokens" if r["usage_known"] else "known subtotal; run total unknown"
print(f" outcome: {r['outcome']} {label}: {r['tokens']}")
print(f" final: {r['final'][:200]}")
if __name__ == "__main__":
asyncio.run(main()){
"scenario": "budget-stopped",
"question": "Why is CI failing on pull request 415?",
"limits": {
"token_budget": 500
},
"outcome": "token_budget",
"tokens": 632,
"steps": [
{
"node": "agent",
"seconds": 2.2,
"tokens": 278,
"reasoning": 82
},
{
"node": "tools",
"seconds": 0.1,
"results": [
"[{\"job_id\": 812, \"name\": \"unit-tests\"}]"
]
},
{
"node": "agent",
"seconds": 2.2,
"tokens": 354,
"reasoning": 27
},
{
"node": "stop",
"seconds": 0
}
],
"final": "Stopped: 632 of 500 tokens used."
}Read the explanation
The router checks reported token usage after each call. It stops further tool rounds once the threshold is reached, but one call can overshoot it. The separate recursion limit caps graph steps; this token threshold is not a hard call-count limit.
What’s next
- Checkpoint a LangGraph workflow and resume it after a crash: limits end a run cleanly; checkpoints let a run that died continue.
References
Official sources checked October 21, 2026, against langgraph 1.2.16 and langchain-core 1.6.12.
- LangGraph: Fault tolerance:
TimeoutPolicy,run_timeoutand nodeerror_handler. - LangGraph: Workflows and agents:
ToolNodeand the agent loop. - LangGraph: Graph API: reducers and
Command. - LangChain: ChatGoogleGenerativeAI:
max_retriesand usage metadata. - Python: asyncio.timeout.
- How
ToolNodepicks the exception types from the handler’s type hints is from the installed langgraph-prebuilt 1.1.1 source (_infer_handled_types).