mirror of
https://github.com/agno-agi/agno.git
synced 2026-10-11 22:09:58 +00:00
feat: store run events for external agents (#10957)
## Summary
External agents (`ClaudeAgent`, `CodexAgent` and every other
`BaseExternalAgent`) did not keep the events of a run. A finished run in
AgentOS showed its content and tools, but
`/agents/{id}/runs/{run_id}/resume` had nothing to replay once the live
event stream no longer held the run, and the run view had no event
timeline. Native agents keep events through `store_events` and
`events_to_skip`; this adds the same to the external base.
Supersedes #7841 (same idea, written against the base class before
retries, continuation and metrics were added, and conflicting with
`main` since).
Before:
<img width="1772" height="166" alt="image"
src="https://github.com/user-attachments/assets/a4ac1909-145d-4e60-9bf5-7f88375e89ec"
/>
After:
<img width="1984" height="538" alt="image"
src="https://github.com/user-attachments/assets/b85c2ad6-60f2-480a-b285-c1f88ef7b60d"
/>
**What changes**
- `BaseExternalAgent` takes `store_events` (default `False`) and
`events_to_skip` (default `[RunEvent.run_content]`), with the same names
and defaults as `Agent`. With `store_events=True`, a streamed run keeps
on `RunOutput.events` the start event, every event the adapter yielded
that is not skipped (tool calls, custom warning events), the `retry`
warning events emitted between attempts, and the terminal event
(`RunCompleted`, `RunError` or `RunCancelled`). Content deltas are
skipped by default because the run already holds the full content.
- The terminal event is built before the run is persisted, so the stored
run ends with it, as native agents do. Non-streamed runs store no
events, as for `Agent`.
- The started tool-call event is no longer rewritten when the tool
completes: the base keeps a copy of the started event's tool for its
merge, so a stored `ToolCallStarted` shows no result and
`ToolCallCompleted` does.
- AgentOS sets `store_events = True` on every external agent it serves,
next to the database injection, which is what it already does for native
agents.
- Stored events keep the `event_index` the event stream stamped on them
when the run went through the queue worker, so a client's
`last_event_index` applies on a replay from the database. Runs streamed
inline through `/runs` are unstamped and replay from position 0, the
same as native agents on that path.
**Cookbooks and docs**
- `cookbook/frameworks/claude-agent-sdk/claude_stored_events.py` and
`cookbook/frameworks/codex/codex_stored_events.py`: a direct streamed
run read back from SQLite with its stored events, then the agent served
through AgentOS in-process, a streamed run over HTTP, the run dropped
from the live event stream, and `/resume` replaying it from the
database.
- README sections on stored events for both harnesses; the
background-runs note that stored-event replay was a follow-up now points
at it.
## Type of change
- [ ] Bug fix
- [x] New feature
- [ ] Breaking change
- [ ] Improvement
- [ ] Model update
- [ ] Other:
---
## Checklist
- [x] Code complies with style guidelines
- [x] Ran format/validation scripts (`./scripts/format.sh` and
`./scripts/validate.sh`)
- [x] Self-review completed
- [x] Documentation updated (comments, docstrings)
- [x] Examples and guides: Relevant cookbook examples have been included
or updated (if applicable)
- [x] Tested in clean environment
- [x] Tests added/updated (if applicable)
### Duplicate and AI-Generated PR Check
- [x] I have searched existing [open pull
requests](https://github.com/agno-agi/agno/pulls) and confirmed that no
other PR already addresses this issue
- [x] If a similar PR exists, I have explained below why this PR is a
better approach
- [x] Check if this PR was entirely AI-generated (by Copilot, Claude
Code, Cursor, etc.)
---
## Additional Notes
**Relation to #7841:** that PR added the same two fields and collected
events in `_arun_stream`, but `_arun_stream` has since been rewritten
for retries, continuation and metrics, and the branch is 600+ commits
behind `main` with conflicts. This PR is a fresh implementation of the
same idea on the current base; credit to @anuragts for the original.
**Verified live** (claude-sonnet-4-6; openai-codex 0.161.0 with
gpt-5.6-luna): both cookbooks pass. Claude streamed 6 events and stored
4 (RunStarted, ToolCallStarted, ToolCallCompleted, RunCompleted); Codex
streamed 16 and stored 4. In both, AgentOS turned `store_events` on by
itself and `/resume` replayed the four stored events from the database
with status COMPLETED after the live buffer had forgotten the run.
**Unit tests** (`test_external_store_events.py`): off by default; stored
kinds and persistence round-trip without content deltas;
`events_to_skip` honoured; error runs end with `RunError`; a retried run
keeps both attempts' tool events with the `retry` warning between them;
a run cancelled during backoff ends with `RunCancelled`; non-streamed
runs store none; AgentOS sets the flag; `/resume` replays from the
database on the inline path; a queued run stores stamped indices and a
`last_event_index` floor applies on the database replay. Agents suite
plus the replay-fallback and queue-worker tests pass (396).
This commit is contained in:
@@ -2,6 +2,7 @@ import asyncio
|
||||
import json
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from contextvars import ContextVar
|
||||
from copy import copy
|
||||
from dataclasses import dataclass
|
||||
from time import time
|
||||
from typing import TYPE_CHECKING, Any, AsyncIterator, Awaitable, Dict, Iterator, List, Optional, Sequence, Union
|
||||
@@ -144,6 +145,12 @@ class BaseExternalAgent:
|
||||
framework: str = "external"
|
||||
markdown: bool = True
|
||||
db: Optional[Union[BaseDb, AsyncBaseDb]] = None
|
||||
# Keep the streamed events on the persisted run (RunOutput.events), as Agent does, so a
|
||||
# finished run can be shown and replayed from the database. AgentOS turns this on.
|
||||
store_events: bool = False
|
||||
# Events not kept when store_events is on. Defaults to RunContent: the deltas add up to
|
||||
# the run's content, which is stored anyway.
|
||||
events_to_skip: Optional[List[RunEvent]] = None
|
||||
# Number of times to retry a failed run
|
||||
retries: int = 0
|
||||
# Delay between retries (in seconds)
|
||||
@@ -156,6 +163,8 @@ class BaseExternalAgent:
|
||||
|
||||
if self.id is None:
|
||||
self.id = generate_id_from_name(self.name)
|
||||
if self.events_to_skip is None:
|
||||
self.events_to_skip = [RunEvent.run_content]
|
||||
|
||||
def get_id(self) -> str:
|
||||
"""Return the agent ID, guaranteed non-None after __post_init__."""
|
||||
@@ -1262,7 +1271,18 @@ class BaseExternalAgent:
|
||||
history = (
|
||||
self._get_history_from_session(session, exclude_run_id=run_id) if session and not continuation else None
|
||||
)
|
||||
yield RunStartedEvent(run_id=run_id, agent_id=self.get_id(), agent_name=self.name or "", session_id=session_id)
|
||||
stored_events: Optional[List[Any]] = [] if self.store_events else None
|
||||
skipped = {event.value for event in self.events_to_skip or []}
|
||||
|
||||
def keep(event: Any) -> Any:
|
||||
# Record an event on the run before it is yielded, unless the agent skips its kind.
|
||||
if stored_events is not None and getattr(event, "event", None) not in skipped:
|
||||
stored_events.append(event)
|
||||
return event
|
||||
|
||||
yield keep(
|
||||
RunStartedEvent(run_id=run_id, agent_id=self.get_id(), agent_name=self.name or "", session_id=session_id)
|
||||
)
|
||||
accumulated_content = ""
|
||||
warnings: List[Dict[str, Any]] = []
|
||||
tools: Dict[str, ToolExecution] = {}
|
||||
@@ -1296,11 +1316,11 @@ class BaseExternalAgent:
|
||||
elif isinstance(event, (ToolCallStartedEvent, ToolCallCompletedEvent)) and event.tool:
|
||||
key = event.tool.tool_call_id or str(uuid4())
|
||||
if key not in tools:
|
||||
tools[key] = event.tool
|
||||
tools[key] = copy(event.tool)
|
||||
elif isinstance(event, ToolCallCompletedEvent):
|
||||
tools[key].result = event.tool.result
|
||||
tools[key].tool_call_error = event.tool.tool_call_error
|
||||
yield event
|
||||
yield keep(event)
|
||||
break
|
||||
except RunCancelledException:
|
||||
raise
|
||||
@@ -1311,7 +1331,7 @@ class BaseExternalAgent:
|
||||
# the retry streams a fresh answer, and keep it in the stored run's warnings.
|
||||
warning = self._retry_warning(attempt, num_attempts, error)
|
||||
warnings.append(warning)
|
||||
yield ExternalRunWarningEvent(run_id=run_id, agent_id=self.get_id(), warning=warning)
|
||||
yield keep(ExternalRunWarningEvent(run_id=run_id, agent_id=self.get_id(), warning=warning))
|
||||
await self._await_retry(run_id, attempt, num_attempts, error)
|
||||
except RunCancelledException:
|
||||
status = RunStatus.cancelled
|
||||
@@ -1336,8 +1356,6 @@ class BaseExternalAgent:
|
||||
if attempts_made > 1:
|
||||
run.metadata = {**(run.metadata or {}), "attempts": attempts_made}
|
||||
self._finish_run_output(run, run_state, continuation)
|
||||
if session is not None:
|
||||
await self._apersist_run_in_session(session, run)
|
||||
fields: Dict[str, Any] = dict(
|
||||
run_id=run_id,
|
||||
session_id=session_id,
|
||||
@@ -1345,14 +1363,23 @@ class BaseExternalAgent:
|
||||
agent_name=self.name or "",
|
||||
content=run.content,
|
||||
)
|
||||
# The terminal event is built before the run is persisted so the stored run ends with it,
|
||||
# which is what a replay from the database needs to know the run is over.
|
||||
terminal: Any
|
||||
if status == RunStatus.cancelled:
|
||||
yield RunCancelledEvent(
|
||||
terminal = RunCancelledEvent(
|
||||
run_id=run_id, session_id=session_id, agent_id=self.get_id(), reason="Run cancelled"
|
||||
)
|
||||
elif run_error is not None:
|
||||
yield RunErrorEvent(**fields, error_type=error_type_of(run_error))
|
||||
terminal = RunErrorEvent(**fields, error_type=error_type_of(run_error))
|
||||
else:
|
||||
yield RunCompletedEvent(**fields, metrics=run.metrics)
|
||||
terminal = RunCompletedEvent(**fields, metrics=run.metrics)
|
||||
keep(terminal)
|
||||
if stored_events is not None:
|
||||
run.events = list(stored_events)
|
||||
if session is not None:
|
||||
await self._apersist_run_in_session(session, run)
|
||||
yield terminal
|
||||
if yield_run_output:
|
||||
yield run
|
||||
|
||||
|
||||
@@ -963,6 +963,7 @@ class AgentOS:
|
||||
if isinstance(entry, BaseExternalAgent):
|
||||
if self.db is not None and entry.db is None:
|
||||
entry.db = self.db
|
||||
entry.store_events = True
|
||||
|
||||
if not self._agents:
|
||||
return
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
"""Stored events on external agent runs, mirroring Agent.store_events."""
|
||||
|
||||
import asyncio
|
||||
from typing import Any, List
|
||||
|
||||
import pytest
|
||||
|
||||
from agno.db.sqlite import SqliteDb
|
||||
from agno.run.agent import RunEvent, RunOutput
|
||||
from agno.run.base import RunStatus
|
||||
from tests.unit.agents.test_external_retries import FlakyAgent
|
||||
|
||||
|
||||
def _stream(agent: FlakyAgent, **kwargs: Any) -> List[Any]:
|
||||
async def consume():
|
||||
return [event async for event in agent._arun_stream("go", yield_run_output=True, **kwargs)]
|
||||
|
||||
return asyncio.run(consume())
|
||||
|
||||
|
||||
def _kinds(run: RunOutput) -> List[str]:
|
||||
return [event.event for event in run.events or []]
|
||||
|
||||
|
||||
def test_events_are_not_stored_by_default(tmp_path):
|
||||
agent = FlakyAgent(id="flaky", db=SqliteDb(db_file=str(tmp_path / "runs.db")), failures=0)
|
||||
run = _stream(agent, session_id="s")[-1]
|
||||
assert agent.store_events is False and run.events is None
|
||||
assert not agent.get_run_output(run.run_id, "s").events, "deserialized as an empty list, as for Agent"
|
||||
|
||||
|
||||
def test_streamed_events_are_stored_on_the_run_without_content_deltas(tmp_path):
|
||||
agent = FlakyAgent(id="flaky", db=SqliteDb(db_file=str(tmp_path / "runs.db")), failures=0, store_events=True)
|
||||
events = _stream(agent, session_id="s")
|
||||
run = events[-1]
|
||||
assert _kinds(run) == ["RunStarted", "ToolCallStarted", "ToolCallCompleted", "RunCompleted"]
|
||||
assert "RunContent" in [e.event for e in events[:-1]], "content deltas are streamed"
|
||||
assert "RunContent" not in _kinds(run), "but not stored; the run keeps the full content instead"
|
||||
assert run.events[0] is events[0], "the stored RunStarted is the streamed one"
|
||||
stored = agent.get_run_output(run.run_id, "s")
|
||||
assert _kinds(stored) == _kinds(run), "events survive persistence"
|
||||
assert stored.events[-1].content == "partial 1 answer 1"
|
||||
assert stored.events[2].tool.result == "ok"
|
||||
assert stored.events[1].tool.result is None, "the started event is not rewritten when the tool completes"
|
||||
|
||||
|
||||
def test_events_to_skip_can_be_changed():
|
||||
agent = FlakyAgent(id="flaky", failures=0, store_events=True, events_to_skip=[])
|
||||
run = _stream(agent)[-1]
|
||||
assert _kinds(run) == [
|
||||
"RunStarted",
|
||||
"RunContent",
|
||||
"ToolCallStarted",
|
||||
"ToolCallCompleted",
|
||||
"RunContent",
|
||||
"RunCompleted",
|
||||
]
|
||||
agent = FlakyAgent(id="flaky", failures=0, store_events=True, events_to_skip=[RunEvent.tool_call_started])
|
||||
kinds = _kinds(_stream(agent)[-1])
|
||||
assert "ToolCallStarted" not in kinds and "RunContent" in kinds
|
||||
|
||||
|
||||
def test_error_and_retry_events_are_stored():
|
||||
failing = FlakyAgent(id="flaky", failures=5, store_events=True)
|
||||
run = _stream(failing)[-1]
|
||||
assert run.status == RunStatus.error
|
||||
assert _kinds(run) == ["RunStarted", "ToolCallStarted", "ToolCallCompleted", "RunError"]
|
||||
assert run.events[-1].content == "attempt 1 failed"
|
||||
|
||||
retried = FlakyAgent(id="flaky", retries=1, failures=1, store_events=True)
|
||||
run = _stream(retried)[-1]
|
||||
kinds = _kinds(run)
|
||||
assert kinds[0] == "RunStarted" and kinds[-1] == "RunCompleted"
|
||||
assert kinds.count("ToolCallCompleted") == 2, "both attempts' tool events are kept"
|
||||
retry_events = [e for e in run.events if e.event == "CustomEvent"]
|
||||
assert len(retry_events) == 1 and retry_events[0].warning["type"] == "retry"
|
||||
assert kinds.index("CustomEvent") == 3, "the retry warning sits between the attempts"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cancelled_runs_store_the_cancel_event():
|
||||
agent = FlakyAgent(id="flaky", retries=3, failures=5, delay_between_retries=30, store_events=True)
|
||||
|
||||
async def run() -> RunOutput:
|
||||
events = [e async for e in agent._arun_stream("go", run_id="r1", yield_run_output=True)]
|
||||
return events[-1]
|
||||
|
||||
task = asyncio.create_task(run())
|
||||
while not agent.attempts:
|
||||
await asyncio.sleep(0.01)
|
||||
await agent.acancel_run("r1")
|
||||
out = await asyncio.wait_for(task, 3)
|
||||
assert out.status == RunStatus.cancelled and _kinds(out)[-1] == "RunCancelled"
|
||||
|
||||
|
||||
def test_non_streaming_runs_store_no_events(tmp_path):
|
||||
agent = FlakyAgent(id="flaky", db=SqliteDb(db_file=str(tmp_path / "runs.db")), failures=0, store_events=True)
|
||||
run = agent.run("go", session_id="s")
|
||||
assert run.status == RunStatus.completed and run.events is None
|
||||
|
||||
|
||||
def test_agentos_turns_event_storage_on(tmp_path):
|
||||
from agno.os import AgentOS
|
||||
|
||||
db = SqliteDb(db_file=str(tmp_path / "runs.db"))
|
||||
agent = FlakyAgent(id="flaky", failures=0)
|
||||
AgentOS(agents=[agent], db=db).get_app()
|
||||
assert agent.store_events is True and agent.db is db
|
||||
|
||||
|
||||
def _sse_payloads(text: str) -> List[dict]:
|
||||
import json
|
||||
|
||||
return [json.loads(line.split("data: ", 1)[1]) for line in text.split("\n") if line.startswith("data: ")]
|
||||
|
||||
|
||||
def test_agentos_resume_replays_a_finished_run_from_the_database(tmp_path):
|
||||
"""PATH 3 of /resume: the run is not in the live event stream, so the stored events are replayed."""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from agno.os import AgentOS
|
||||
|
||||
db = SqliteDb(db_file=str(tmp_path / "runs.db"))
|
||||
agent = FlakyAgent(id="flaky", failures=0)
|
||||
app = AgentOS(agents=[agent], db=db).get_app()
|
||||
client = TestClient(app, raise_server_exceptions=False)
|
||||
|
||||
resp = client.post("/agents/flaky/runs", data={"message": "go", "stream": "true", "session_id": "s"})
|
||||
assert resp.status_code == 200
|
||||
live = _sse_payloads(resp.text)
|
||||
run_id = live[0]["run_id"]
|
||||
assert [p["event"] for p in live] == [
|
||||
"RunStarted",
|
||||
"RunContent",
|
||||
"ToolCallStarted",
|
||||
"ToolCallCompleted",
|
||||
"RunContent",
|
||||
"RunCompleted",
|
||||
]
|
||||
stored = agent.get_run_output(run_id, "s")
|
||||
assert _kinds(stored) == ["RunStarted", "ToolCallStarted", "ToolCallCompleted", "RunCompleted"]
|
||||
|
||||
resp = client.post(f"/agents/flaky/runs/{run_id}/resume", data={"session_id": "s"})
|
||||
assert resp.status_code == 200
|
||||
replay = _sse_payloads(resp.text)
|
||||
assert replay[0]["event"] == "replay" and replay[0]["total_events"] == 4
|
||||
assert [p["event"] for p in replay[1:]] == ["RunStarted", "ToolCallStarted", "ToolCallCompleted", "RunCompleted"]
|
||||
assert [p["event_index"] for p in replay[1:]] == [0, 1, 2, 3], "the inline stream does not stamp indices"
|
||||
assert replay[-1]["content"] == "partial 1 answer 1"
|
||||
assert replay[3]["tool"]["result"] == "ok"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_background_runs_store_stream_indices_for_replay_floors(tmp_path):
|
||||
"""Queued runs publish through the event stream, which stamps each stored event with its real index, so
|
||||
a client's last_event_index applies to the replay once the buffer has forgotten the run."""
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from agno.os import AgentOS
|
||||
from agno.os.event_streams import get_event_stream
|
||||
|
||||
from ._queue_harness import run_through_queue
|
||||
|
||||
db = SqliteDb(db_file=str(tmp_path / "runs.db"))
|
||||
agent = FlakyAgent(id="flaky", failures=0)
|
||||
app = AgentOS(agents=[agent], db=db).get_app()
|
||||
job = await run_through_queue(agent, stream=True, max_attempts=1)
|
||||
assert job["status"] == "completed"
|
||||
|
||||
stored = await agent.aget_run_output(job["id"], "s")
|
||||
assert _kinds(stored) == ["RunStarted", "ToolCallStarted", "ToolCallCompleted", "RunCompleted"]
|
||||
indices = [e.event_index for e in stored.events]
|
||||
assert indices[:3] == [0, 2, 3], "RunContent took index 1 on the stream but is not stored"
|
||||
assert indices[3] is None, "the terminal event is persisted before it is published, as for Agent"
|
||||
|
||||
await get_event_stream().cleanup_run(job["id"])
|
||||
client = TestClient(app, raise_server_exceptions=False)
|
||||
resp = client.post(f"/agents/flaky/runs/{job['id']}/resume", data={"session_id": "s", "last_event_index": "2"})
|
||||
assert resp.status_code == 200
|
||||
tail = _sse_payloads(resp.text)
|
||||
assert [(p["event"], p["event_index"]) for p in tail[1:]] == [("ToolCallCompleted", 3), ("RunCompleted", 3)]
|
||||
Reference in New Issue
Block a user