ag-ui-protocol / ag-ui-protocol/ag-ui
LangGraphAgent: an exception inside the run leaves the stream with no terminal event
- Linguagem predominante
- Python
- Estrelas
- 15.9k
- Forks
- 1.4k
- Merge médio
- 1d 17h
- PRs com merge (30d)
- 163
Descrição
**Versions:** `ag-ui-langgraph` `main` @ `0.0.43` (also reproduces on the published `0.0.42`), `ag-ui-protocol` `0.1.10`, `langgraph` 1.x, Python 3.12.
## Expected
Per `concepts/events.mdx`:
> The `RunStarted` and either `RunFinished` or `RunError` events are mandatory, forming the boundaries of an agent run.
> Every run terminates with either `RunFinished` or `RunError`.
A run that fails should therefore surface a terminal `RUN_ERROR` on the stream.
## Actual
`_handle_stream_events` emits `RUN_ERROR` in exactly one place — `if event["event"] == "error"`, for an upstream error *event* out of `astream_events` (`agent.py:336-352`). The method's own structure is `try: … finally: self.active_run = None`; there is no `except`.
So anything *raised* inside the run propagates straight out of the async generator: a provider fault, a checkpointer failure, `GraphRecursionError` from Pregel, an exception from user middleware. The generator dies, the caller's `async for` stops, and the stream ends with no terminal event of any kind.
## Repro
```python
from unittest.mock import AsyncMock, MagicMock
from ag_ui.core import RunAgentInput, UserMessage
from ag_ui_langgraph.agent import LangGraphAgent
from langgraph.graph.state import CompiledStateGraph
graph = MagicMock(spec=CompiledStateGraph)
graph.config_specs, graph.nodes = [], {}
state = MagicMock(values={"messages": []}, tasks=[], next=[], metadata={"writes": {}})
graph.aget_state = AsyncMock(return_value=state)
async def exploding_stream():
if False:
yield None
raise RuntimeError("provider exploded")
graph.astream_events = MagicMock(return_value=exploding_stream())
agent = LangGraphAgent(name="t", graph=graph)
seen = []
async for event in agent.run(
RunAgentInput(
thread_id="t1", run_id="run-1",
messages=[UserMessage(id="u1", role="user", content="hi")],
tools=[], context=[], state={}, forwarded_props={},
)
):
seen.append(event.type)
```
## Observed
```
events = [RUN_STARTED, ...] # no terminal event
RUN_ERROR in stream = false
run() raised = true (RuntimeError: provider exploded)
```
## Impact
A client rendering from the event stream sees a run that simply stops — no error state, no terminal. `verifyEvents` cannot catch it: it rejects illegal events that are *sent*, and a stream that ends sends nothing to reject (#2300).
Downstream this is papered over rather than fixed. `@copilotkit/runtime` notices the missing terminal in `finalizeRunEvents` and synthesizes one, but with what it can know from the outside:
```
message: "Run ended without emitting a terminal event"
code: "INCOMPLETE_STREAM"
```
The adapter is the only layer that knows the actual cause, and it is the layer that drops it.
`GraphRecursionError` makes this concrete: the graph's own step limit is a normal, expected outcome, and today it reaches the user as a protocol diagnostic rather than "the run hit its step limit".
This is the same defect class reported for the Mastra integration in #2416.
## Proposed fix
Wrap the stream body and report the failure as `RUN_ERROR` **before** re-raising, so the protocol boundary is honored without changing what callers see:
- callers that treat a raised failure as a failure keep working unchanged — the exception still propagates, and `TestAgetStateMidStreamError` in `test_subgraph_streaming.py` keeps passing;
- a run that already emitted a terminal event does not get a second one (`@ag-ui/client` rejects the second, #1892);
- `CancelledError` is a `BaseException`, so it never reaches the handler — a caller that walked away is not owed an error event.
I have the patch and three tests ready (+53 lines in `agent.py`, no behavior removed; the full Python suite passes at 400). Happy to open the PR — per `CONTRIBUTING.md` I am filing this first and will wait to be assigned.
Guia de contribuição
Avaliação
Esta issue ainda não foi avaliada.