fix(agent): size the step budget for research turns and fail usefully

A turn died in production with "Recursion limit of 25 reached without
hitting a stop condition", killing the SSE stream mid-answer.

25 does not mean 25 tool calls. The limit counts super-steps, and the
middleware chain makes one model-call-plus-tool round trip cost four of
them (model -> TodoListMiddleware.after_model ->
CopilotKitMiddleware.after_model -> tools), plus ~3 fixed for entry and
exit. The real budget was about five tool calls per turn. The budget also
resets on resume, so this was one turn wanting a sixth call, not a long
conversation accumulating.

That was the right size when the limit was lowered from 100 to 25 in
4ac1a7a for a triage-only bot. The agent now loads 85 tools and gets
asked research questions -- "find how many PRs shipped in CPK related to
channels, categorize them, chart it" answered from 212 PRs and must have
landed right at the cap; the next comparable question went over.

Raise the default to 60 (about fourteen tool calls) and read it from
AGENT_RECURSION_LIMIT so it can be tuned without a deploy, rejecting a
value too low to finish a single tool call.

Raising the ceiling makes a runaway turn more expensive, so also stop
losing the failure. GraphRecursionError escaped as an ASGI error and the
Channel could only answer with its generic "I hit an error", forty
seconds in, with no hint that a narrower question would work. Catch it,
end the run the way the protocol expects, say what happened and what to
try, and log the tool trail so the next occurrence is diagnosable from
the logs instead of from Slack.

Author: jerelvelarde

agent/agent.py

@@ -1,6 +1,7 @@
 """OpenTag's triage-first Deep Agent."""
 
 import os
+from collections.abc import Mapping
 from pathlib import Path
 
 from copilotkit import CopilotKitMiddleware
@@ -30,6 +31,19 @@
 )
 VALID_VERBOSITY_LEVELS = frozenset({"low", "medium", "high"})
 
+# Super-steps the graph may take in one turn. This is NOT a tool-call budget:
+# the middleware chain means one model-call-plus-tool round trip costs four
+# super-steps (model -> TodoListMiddleware.after_model ->
+# CopilotKitMiddleware.after_model -> tools), plus ~3 fixed for entry and exit.
+# So the usable budget is roughly (limit - 3) / 4 tool calls -- the old value of
+# 25 bought about five, which a GitHub or Linear research question exhausts
+# mid-answer. 60 buys about fourteen.
+DEFAULT_RECURSION_LIMIT = 60
+
+# Below this the agent cannot complete a single tool call and every research
+# turn dies; treat it as a misconfiguration rather than a tuning choice.
+MIN_RECURSION_LIMIT = 8
+
 # Deep Agents adds shell execution and a general-purpose delegation tool by
 # default. OpenTag has no sandbox for execute, and delegating routine turns to
 # another agent adds latency without improving triage.
@@ -56,6 +70,25 @@ def _validated_openai_setting(
     return value
 
 
+def recursion_limit(env: Mapping[str, str] = os.environ) -> int:
+    """Read AGENT_RECURSION_LIMIT, or fall back to the tuned default."""
+    raw = env.get("AGENT_RECURSION_LIMIT")
+    if raw is None or not raw.strip():
+        return DEFAULT_RECURSION_LIMIT
+    try:
+        value = int(raw)
+    except ValueError as error:
+        raise RuntimeError(
+            f'Invalid AGENT_RECURSION_LIMIT: "{raw}" — must be an integer'
+        ) from error
+    if value < MIN_RECURSION_LIMIT:
+        raise RuntimeError(
+            f"Invalid AGENT_RECURSION_LIMIT: {value} — must be at least "
+            f"{MIN_RECURSION_LIMIT}, or the agent cannot finish a tool call"
+        )
+    return value
+
+
 def build_agent():
     """Build the OpenTag triage graph."""
     api_key = os.environ.get("OPENAI_API_KEY")
@@ -111,4 +144,10 @@ def build_agent():
     print(f"[AGENT] internal-source tools: {len(internal_tools)}")
     print(f"[AGENT] Main tools: {[t.name for t in main_tools]}")
 
-    return agent_graph.with_config({"recursion_limit": 25})
+    limit = recursion_limit()
+    print(
+        f"[AGENT] recursion limit: {limit} "
+        f"(~{(limit - 3) // 4} tool calls per turn)"
+    )
+
+    return agent_graph.with_config({"recursion_limit": limit})

agent/main.py

@@ -5,11 +5,11 @@
 import sys
 
 from ag_ui_langgraph import add_langgraph_fastapi_endpoint
-from copilotkit import LangGraphAGUIAgent
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from agent import build_agent
+from step_budget import StepBudgetAwareAgent
 
 app = FastAPI(
     title="OpenTag Agent",
@@ -63,7 +63,7 @@ def local_server_port(env: Mapping[str, str] = os.environ) -> int:
     agent_graph = build_agent()
     add_langgraph_fastapi_endpoint(
         app=app,
-        agent=LangGraphAGUIAgent(
+        agent=StepBudgetAwareAgent(
             name=AGENT_NAME,
             description=AGENT_DESCRIPTION,
             graph=agent_graph,

agent/pyproject.toml

@@ -25,6 +25,7 @@ py-modules = [
   "agent",
   "internal_sources",
   "main",
+  "step_budget",
   "tools",
   "write_confirmation",
 ]

agent/step_budget.py

@@ -0,0 +1,78 @@
+"""Turn a blown step budget into an answer instead of a broken stream.
+
+`GraphRecursionError` is raised from inside LangGraph's Pregel loop, which by
+then is being consumed by the AG-UI SSE response. Uncaught, it escapes as an
+ASGI error mid-stream: the client sees a truncated run, and the Channel can
+only fall back to its generic "I hit an error" reply -- forty seconds after the
+user asked, with no hint that a narrower question would have worked.
+
+This wrapper ends the run the way the protocol expects (a text message, then
+RUN_FINISHED) and says what actually happened.
+"""
+
+import logging
+import uuid
+from typing import Any, AsyncGenerator
+
+from ag_ui.core import (
+    RunFinishedEvent,
+    TextMessageContentEvent,
+    TextMessageEndEvent,
+    TextMessageStartEvent,
+)
+from copilotkit import LangGraphAGUIAgent
+from langgraph.errors import GraphRecursionError
+
+logger = logging.getLogger(__name__)
+
+STEP_BUDGET_MESSAGE = (
+    "That one ran past my step budget, so I stopped before finishing — "
+    "nothing was left half-written. Narrowing it usually does the trick: "
+    "name a single repo, a date range, or ask for the counts first and the "
+    "breakdown after."
+)
+
+
+class StepBudgetAwareAgent(LangGraphAGUIAgent):
+    """A LangGraph AG-UI agent that reports hitting its recursion limit."""
+
+    async def run(self, input: Any) -> AsyncGenerator[Any, None]:
+        try:
+            async for event in super().run(input):
+                yield event
+        except GraphRecursionError:
+            thread_id = getattr(input, "thread_id", "") or ""
+            run_id = getattr(input, "run_id", "") or ""
+            # The trajectory is the part worth having in the logs: without it
+            # the next occurrence is only diagnosable by reading Slack.
+            logger.error(
+                {
+                    "error": "Agent exhausted its step budget",
+                    "context": {
+                        "component": "step_budget",
+                        "thread_id": thread_id,
+                        "run_id": run_id,
+                        "tool_calls": _tool_call_trail(input),
+                    },
+                }
+            )
+
+            message_id = str(uuid.uuid4())
+            yield TextMessageStartEvent(message_id=message_id, role="assistant")
+            yield TextMessageContentEvent(
+                message_id=message_id, delta=STEP_BUDGET_MESSAGE
+            )
+            yield TextMessageEndEvent(message_id=message_id)
+            yield RunFinishedEvent(thread_id=thread_id, run_id=run_id)
+
+
+def _tool_call_trail(input: Any) -> list[str]:
+    """Names of the tools the run had already called, oldest first."""
+    names: list[str] = []
+    for message in getattr(input, "messages", None) or []:
+        for call in getattr(message, "tool_calls", None) or []:
+            name = getattr(call, "function", None)
+            name = getattr(name, "name", None) or getattr(call, "name", None)
+            if name:
+                names.append(str(name))
+    return names

agent/tests/test_agent_configuration.py

@@ -97,9 +97,49 @@ def test_build_agent_rejects_invalid_openai_tuning(monkeypatch, name, value):
 
 
 def test_build_agent_bounds_graph_recursion(monkeypatch):
+    monkeypatch.delenv("AGENT_RECURSION_LIMIT", raising=False)
     _, captured = build_with_captured_configuration(monkeypatch)
 
-    assert captured["config"] == {"recursion_limit": 25}
+    assert captured["config"] == {
+        "recursion_limit": agent_mod.DEFAULT_RECURSION_LIMIT
+    }
+
+
+def test_the_default_step_budget_covers_a_multi_step_research_turn():
+    """One tool round trip costs four super-steps, plus ~3 fixed per run.
+
+    The default has to leave room for a real research question -- the old
+    value of 25 allowed about five tool calls, which a GitHub search plus a
+    chart exhausts mid-answer.
+    """
+    tool_calls = (agent_mod.DEFAULT_RECURSION_LIMIT - 3) // 4
+
+    assert tool_calls >= 12
+
+
+def test_recursion_limit_is_configurable(monkeypatch):
+    monkeypatch.setenv("AGENT_RECURSION_LIMIT", "120")
+    _, captured = build_with_captured_configuration(monkeypatch)
+
+    assert captured["config"] == {"recursion_limit": 120}
+
+
+def test_a_blank_recursion_limit_falls_back_to_the_default():
+    assert (
+        agent_mod.recursion_limit({"AGENT_RECURSION_LIMIT": "   "})
+        == agent_mod.DEFAULT_RECURSION_LIMIT
+    )
+    assert agent_mod.recursion_limit({}) == agent_mod.DEFAULT_RECURSION_LIMIT
+
+
+def test_a_nonsense_recursion_limit_is_rejected():
+    with pytest.raises(RuntimeError, match="AGENT_RECURSION_LIMIT"):
+        agent_mod.recursion_limit({"AGENT_RECURSION_LIMIT": "lots"})
+
+
+def test_a_recursion_limit_too_low_to_finish_a_tool_call_is_rejected():
+    with pytest.raises(RuntimeError, match="at least"):
+        agent_mod.recursion_limit({"AGENT_RECURSION_LIMIT": "4"})
 
 
 def test_build_agent_refreshes_current_date_before_each_model_call(monkeypatch):

agent/tests/test_step_budget.py

@@ -0,0 +1,106 @@
+import asyncio
+
+import pytest
+from ag_ui.core import EventType
+from langgraph.errors import GraphRecursionError
+from step_budget import STEP_BUDGET_MESSAGE, StepBudgetAwareAgent
+
+
+class FakeInput:
+    def __init__(self, messages=None):
+        self.thread_id = "thread-1"
+        self.run_id = "run-1"
+        self.messages = messages or []
+
+
+class FakeCall:
+    def __init__(self, name):
+        self.name = name
+
+
+class FakeMessage:
+    def __init__(self, names):
+        self.tool_calls = [FakeCall(n) for n in names]
+
+
+def drain(agent, run_input):
+    async def collect():
+        return [event async for event in agent.run(run_input)]
+
+    return asyncio.run(collect())
+
+
+def run_with(monkeypatch, events, error=None):
+    """Drive StepBudgetAwareAgent.run over a stubbed parent generator."""
+    agent = StepBudgetAwareAgent.__new__(StepBudgetAwareAgent)
+
+    async def fake_parent(_self, _input):
+        for event in events:
+            yield event
+        if error is not None:
+            raise error
+
+    monkeypatch.setattr(
+        "step_budget.LangGraphAGUIAgent.run", fake_parent, raising=True
+    )
+    return drain(agent, FakeInput([FakeMessage(["search_pull_requests"])]))
+
+
+def test_events_pass_through_untouched_when_the_run_succeeds(monkeypatch):
+    events = run_with(monkeypatch, ["a", "b"])
+
+    assert events == ["a", "b"]
+
+
+def test_a_blown_step_budget_becomes_an_answer_not_a_broken_stream(monkeypatch):
+    events = run_with(
+        monkeypatch, ["a"], GraphRecursionError("Recursion limit of 60 reached")
+    )
+
+    # Whatever the run produced before the limit survives, then the run is
+    # closed the way the protocol expects rather than dying mid-stream.
+    assert events[0] == "a"
+    types = [getattr(e, "type", None) for e in events[1:]]
+    assert types == [
+        EventType.TEXT_MESSAGE_START,
+        EventType.TEXT_MESSAGE_CONTENT,
+        EventType.TEXT_MESSAGE_END,
+        EventType.RUN_FINISHED,
+    ]
+
+
+def test_the_step_budget_message_tells_the_user_what_to_do(monkeypatch):
+    events = run_with(monkeypatch, [], GraphRecursionError("boom"))
+    content = next(
+        e for e in events if getattr(e, "type", None) == EventType.TEXT_MESSAGE_CONTENT
+    )
+
+    assert content.delta == STEP_BUDGET_MESSAGE
+    assert "narrow" in content.delta.lower()
+    # The user needs to know the write side is clean, not half-applied.
+    assert "half-written" in content.delta
+
+
+def test_the_closing_events_carry_the_run_identity(monkeypatch):
+    events = run_with(monkeypatch, [], GraphRecursionError("boom"))
+    finished = events[-1]
+    start = events[0]
+    end = events[2]
+
+    assert finished.thread_id == "thread-1"
+    assert finished.run_id == "run-1"
+    # One message, opened and closed under the same id.
+    assert start.message_id == end.message_id
+
+
+def test_other_failures_still_propagate(monkeypatch):
+    with pytest.raises(ValueError):
+        run_with(monkeypatch, [], ValueError("something else"))
+
+
+def test_the_tool_trail_is_logged_for_diagnosis(monkeypatch, caplog):
+    run_with(monkeypatch, [], GraphRecursionError("boom"))
+
+    logged = [r for r in caplog.records if "step_budget" in str(r.msg)]
+    assert logged, "the failure must be diagnosable from the logs alone"
+    assert "search_pull_requests" in str(logged[0].msg)