ag-ui-protocol / ag-ui-protocol/ag-ui

LangGraphAgent: an exception inside the run leaves the stream with no terminal event

Đang mở
#2,531 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
15.9k
Fork
1.4k
Merge trung bình
1 ngày 17 giờ
Pull request đã merge (30 ngày)
163

Mô tả

**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.

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.