mirror of
https://github.com/agno-agi/agno.git
synced 2026-10-12 02:10:54 +00:00
feat: compaction cookbooks for Claude and Codex, CodexAgent.compact (#10924)
## Summary Native context compaction for the external agent adapters, verified end to end and documented with runnable cookbooks. **Claude (`ClaudeAgent`)** needs no library change. `/compact` is an ordinary run input, and a `PreCompact` hook passed through `options_kwargs` sees every compaction. With transcript storage on, the compaction is mirrored into `agno_transcripts` like every other transcript line: a `compact_boundary` system entry followed by the `isCompactSummary` user entry. A later turn on any replica resumes from the database and continues from that summary. - `cookbook/frameworks/claude-agent-sdk/compaction.py`: seeds a fact, adds three turns including a long document, sends `/compact`, prints the two rows the compaction wrote, then resumes the session from a second agent instance with a different working directory. That instance answers from the summary. **Codex (`CodexAgent`)** has no slash command; compaction is an app-server request. This PR adds `CodexAgent.compact` / `acompact(session_id, user_id=None, timeout=120.0)`: - Resolves the Codex thread behind the Agno session (returns `False` when none exists yet, or when its rollout is gone: the stored id is then forgotten, as a run would). - Resumes the thread through the high-level `AsyncCodex.thread_resume` with the same kwargs a run uses, so the SDK applies its wire translation (`full-access` to `danger-full-access`, approval mode to approval policy, camelCase keys). Only `thread/compact/start` goes to the low-level app-server client behind it, which also owns the global notification queue. The first version resumed on the low-level client with the untranslated kwargs; the app-server rejected `sandbox="full-access"` and silently dropped instructions and approval mode (review finding from @ysolanky, replicated against 0.161.0). - Waits for the thread status to go active and then idle on the global notification queue. The `thread/compacted` notification is routed to a per-turn queue the caller never registered, so thread status is the reliable signal; the `thread/compacted` branch is kept as forward compatibility and commented as such. - Refuses with `RuntimeError` while a run on the session is pending or running, since compaction opens its own app-server on the thread and two processes would append to one rollout. Documented in the docstring and README. - Raises the builtin `TimeoutError` if the thread does not go idle within `timeout` (mapped from `asyncio.TimeoutError` for Python 3.10); the client is closed either way. - The sync wrapper uses `run_coroutine_sync`, now a shared helper in `agents/base.py` that `BaseExternalAgent.run` uses too. - `cookbook/frameworks/codex/codex_compaction.py`: same scenario, calling `acompact` and continuing from the summary. Both READMEs gain a compaction section and both TEST_LOGs record the live runs. ## 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 - [ ] 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 **Verification** - Both cookbooks run live: claude-sonnet-4-6 with SQLite transcript storage, and gpt-5.6-luna with openai-codex 0.161.0. In both, the seeded fact was recalled after compaction. - Claude: 35 transcript rows before `/compact`, 45 after; the hook fired with trigger `manual`. - Codex: `acompact` returned `False` before the first run and `True` after; the thread went idle again in about eight seconds and the rollout gained a `compacted` entry. - Unit tests: eight tests in `libs/agno/tests/unit/agents/test_codex.py` against fake high-level and low-level clients: no thread, resume and wait sequence, resume kwargs identical to a run's (`sandbox="full-access"`, instructions, approval mode, no start-only keys), missing rollout forgets the id, other resume failures propagate, in-flight run refused, the compacted notification, timeout closing the client. Agents suite passes. - Live: an agent with `sandbox="full-access"`, `instructions` and `approval_mode="deny_all"` compacted and recalled the fact; a nonexistent thread id returned False and was removed from the session. - The Claude cookbook reads transcripts through `db.get_transcript_entries` instead of raw SQL. **Notes** - A conversation of one turn returns "Not enough messages to compact"; the cookbooks add turns first. - The `/compact` run records an empty assistant message in the Agno run output. Harmless, and a candidate for a small follow-up. - Automatic compaction (triggered by context pressure) uses the same path on both sides; only the manual trigger is exercised here. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5.1 <[email protected]> Co-authored-by: Yash Pratap Solanky <[email protected]>
This commit is contained in:
Kaustubh
co-authored by
Claude Fable 5.1
Yash Pratap Solanky
Claude Fable 5.1
Yash Pratap Solanky
parent
7479103782
commit
e39a56fb7e
@@ -47,6 +47,16 @@ The `continue_from/` folder has one cookbook per scenario: continue a finished r
|
||||
|
||||
Requires a database with transcript storage, and only runs recorded with it can be continued. Files the agent changed after the checkpoint are not rewound. `regenerate` and HITL `requirements` are not supported.
|
||||
|
||||
## Compaction
|
||||
|
||||
Claude Code compacts a long conversation into a summary on its own when the context window fills, or when asked with `/compact`. Through `ClaudeAgent` the slash command is an ordinary run input, and a `PreCompact` hook passed in `options_kwargs` sees every compaction. With transcript storage the compaction lands in `agno_transcripts` like any other line: a `compact_boundary` system entry followed by the summary. A later turn on any replica resumes from the database and continues from that summary.
|
||||
|
||||
```bash
|
||||
.venvs/demo/bin/python cookbook/frameworks/claude-agent-sdk/compaction.py
|
||||
```
|
||||
|
||||
A very short conversation returns "Not enough messages to compact"; the cookbook adds a few turns first.
|
||||
|
||||
## Metrics
|
||||
|
||||
Every run reports the tokens, cache usage and cost Claude Code returned for the turn on `RunOutput.metrics`, with one entry per model in `metrics.details["model"]` (Claude Code uses a small helper model next to the main one, and subagents may use others). Agno adds wall-clock `duration` and, when streaming, `time_to_first_token`. The `RunCompleted` event carries the same metrics. Completed runs are added to `session_data["session_metrics"]`, which AgentOS reads for the sessions list, the session view and the metrics page. Like the native Anthropic model, `input_tokens` excludes the cached prefix; cached tokens are in `cache_read_tokens` and `cache_write_tokens`.
|
||||
|
||||
@@ -24,6 +24,26 @@
|
||||
|
||||
## 2026-10-09
|
||||
|
||||
### compaction.py (transcript read API)
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Reran after replacing the cookbook's raw SQL on `agno_transcripts` with `db.get_transcript_entries(framework, project_key, session_id)`, keyed by the Claude session id stored on the Agno session.
|
||||
|
||||
**Result:** 29 rows before `/compact`, 39 after; the `compact_boundary` and summary entries were found at positions 33 and 34; replica B recalled the fact.
|
||||
|
||||
---
|
||||
|
||||
### compaction.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Seeds a fact, adds three turns including a 1200-line filler document, sends `/compact` as the run input with a `PreCompact` hook registered through `options_kwargs`, prints the transcript rows the compaction produced, then resumes the session from a second agent instance with a different working directory.
|
||||
|
||||
**Result:** 35 transcript rows before, 45 after. The hook fired with trigger `manual`. Rows 39 and 40 were the `compact_boundary` system entry and the `isCompactSummary` user entry. Replica B resumed from the database alone and answered `tangerine-walrus-88` from the summary. A conversation of one turn returns "Not enough messages to compact", which is why the cookbook adds turns first.
|
||||
|
||||
---
|
||||
|
||||
### continue_from.py (finished runs always fork)
|
||||
|
||||
**Status:** PASS
|
||||
@@ -131,7 +151,6 @@
|
||||
|
||||
---
|
||||
|
||||
|
||||
### Transcript mirror failure regression (2026-10-08)
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
"""
|
||||
Compact a Claude conversation and keep going from the database
|
||||
==============================================================
|
||||
Claude Code compacts a long conversation into a summary, on its own when
|
||||
the context window fills up or when asked with /compact. Through Agno the
|
||||
slash command is just a run input, and with transcript storage the
|
||||
compaction is written to the agno_transcripts table like every other
|
||||
transcript line: a compact_boundary system entry followed by the summary.
|
||||
|
||||
Scenario: seed a fact, add a few turns including a long document, send
|
||||
/compact, print the transcript rows the compaction produced, then start a
|
||||
second agent instance with a different working directory. It resumes from
|
||||
the database alone and answers from the summary.
|
||||
|
||||
Requirements:
|
||||
pip install claude-agent-sdk
|
||||
A database with transcript storage (SQLite or PostgreSQL).
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/claude-agent-sdk/compaction.py
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from claude_agent_sdk import HookMatcher
|
||||
|
||||
from agno.agents.claude import ClaudeAgent
|
||||
from agno.agents.claude.session_store import AgnoSessionStore
|
||||
from agno.db.sqlite import SqliteDb
|
||||
from agno.run.base import RunStatus
|
||||
|
||||
SESSION_ID = "compaction-demo"
|
||||
FACT = "The release codename is tangerine-walrus-88."
|
||||
|
||||
compactions = []
|
||||
|
||||
|
||||
async def on_pre_compact(input_data, tool_use_id, context):
|
||||
"""Claude Code hook: fires before each compaction with trigger 'manual' or 'auto'."""
|
||||
compactions.append(input_data.get("trigger"))
|
||||
return {}
|
||||
|
||||
|
||||
def compaction_rows(agent: ClaudeAgent):
|
||||
"""Transcript entries written by the compaction: the boundary marker and the summary.
|
||||
|
||||
Transcripts are read back through the database's transcript API, keyed by the Claude
|
||||
session id the agent stored on the Agno session.
|
||||
"""
|
||||
session = agent.get_session(SESSION_ID)
|
||||
sdk_session_id = (
|
||||
(session.session_data or {}).get("claude_sdk_session_id") if session else None
|
||||
)
|
||||
if not sdk_session_id:
|
||||
return 0, []
|
||||
entries = agent.db.get_transcript_entries(
|
||||
framework=AgnoSessionStore.framework,
|
||||
project_key=agent.project_key,
|
||||
session_id=sdk_session_id,
|
||||
)
|
||||
found = []
|
||||
for position, entry in enumerate(entries, start=1):
|
||||
if entry.get("type") == "system" and entry.get("subtype") == "compact_boundary":
|
||||
found.append((position, "compact_boundary", str(entry.get("content"))))
|
||||
elif entry.get("type") == "user" and entry.get("isCompactSummary"):
|
||||
summary = entry.get("message", {}).get("content", "")
|
||||
found.append(
|
||||
(position, "compact_summary", str(summary)[:160].replace("\n", " "))
|
||||
)
|
||||
return len(entries), found
|
||||
|
||||
|
||||
async def main():
|
||||
with tempfile.TemporaryDirectory(prefix="agno-compaction-") as root:
|
||||
db_file = str(Path(root) / "runs.db")
|
||||
|
||||
def make_agent(name: str) -> ClaudeAgent:
|
||||
workdir = Path(root) / name
|
||||
workdir.mkdir()
|
||||
return ClaudeAgent(
|
||||
id="compaction-demo",
|
||||
model="claude-sonnet-4-6",
|
||||
db=SqliteDb(db_file=db_file),
|
||||
cwd=str(workdir),
|
||||
allowed_tools=[],
|
||||
max_turns=2,
|
||||
max_budget_usd=0.5,
|
||||
options_kwargs={
|
||||
"hooks": {"PreCompact": [HookMatcher(hooks=[on_pre_compact])]}
|
||||
},
|
||||
)
|
||||
|
||||
replica_a = make_agent("replica-a")
|
||||
|
||||
# 1. Build up a conversation. A very short one cannot be compacted.
|
||||
run = await replica_a.arun(
|
||||
f"Remember this: {FACT} Reply with just OK.", session_id=SESSION_ID
|
||||
)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
document = " ".join(
|
||||
f"Line {i}: the quick brown fox jumps over the lazy dog number {i}."
|
||||
for i in range(1200)
|
||||
)
|
||||
for prompt in [
|
||||
"Keep this document in mind and reply with just OK: " + document,
|
||||
"Name three primary colors, comma separated.",
|
||||
"What is 17 times 3? Reply with just the number.",
|
||||
]:
|
||||
run = await replica_a.arun(prompt, session_id=SESSION_ID)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
total, found = compaction_rows(replica_a)
|
||||
print(f"Before /compact: {total} transcript rows, {len(found)} compaction rows")
|
||||
|
||||
# 2. Ask Claude Code to compact. The slash command is an ordinary run input.
|
||||
run = await replica_a.arun("/compact", session_id=SESSION_ID)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
assert compactions == ["manual"], compactions
|
||||
total, found = compaction_rows(replica_a)
|
||||
print(
|
||||
f"After /compact: {total} transcript rows, PreCompact hook saw {compactions}"
|
||||
)
|
||||
for position, kind, text in found:
|
||||
print(f" row {position} {kind}: {text}")
|
||||
assert [kind for _, kind, _ in found] == ["compact_boundary", "compact_summary"]
|
||||
|
||||
# 3. A second instance with a different working directory has no local transcript.
|
||||
# It resumes from the database and answers from the compacted summary.
|
||||
replica_b = make_agent("replica-b")
|
||||
run = await replica_b.arun(
|
||||
"What is the release codename? Reply with just the codename.",
|
||||
session_id=SESSION_ID,
|
||||
)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
print(f"Replica B after compaction: {run.content}")
|
||||
assert "tangerine-walrus-88" in (run.content or "")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -36,6 +36,8 @@ CodexAgent(name="Codex").login_api_key("sk-...")
|
||||
- `codex_retries.py` — retry failed runs with exponential backoff. The cookbook injects a transient failure so the retry can be watched resuming the failed attempt's thread, then shows exhaustion and the default of no retries. Limits and permanent errors are not retried
|
||||
- `codex_agentos.py` — serve Codex through AgentOS
|
||||
- `codex_session_agentos.py` — same with SQLite-backed sessions
|
||||
- `codex_compaction.py` — compact the Codex thread behind a session with `CodexAgent.acompact` (a full-access agent with instructions and an approval mode), continue from the summary, then the guards: compaction refuses while a background run on the session is in flight, and a stored thread id whose rollout is gone is forgotten
|
||||
|
||||
- `codex_metrics.py` — token usage per run on `RunOutput.metrics` and session totals for AgentOS
|
||||
- `codex_metrics_agentos.py` — the same through the AgentOS API (`--verify` prints run, session, sessions list and `/metrics`)
|
||||
|
||||
|
||||
@@ -102,33 +102,23 @@ Tested 2026-10-08 with openai-codex 0.161.0 (bundled Codex CLI 0.161.0), model g
|
||||
|
||||
---
|
||||
|
||||
### codex_retries.py
|
||||
### codex_compaction.py
|
||||
|
||||
**Status:** PASS (2026-10-10, openai-codex 0.161.0, gpt-5.6-luna)
|
||||
**Status:** PASS (2026-10-09, openai-codex 0.161.0, gpt-5.6-luna)
|
||||
|
||||
**Description:** Ran live with the demo environment. The cookbook injects the failure itself: the first attempt is cut off with a simulated transient error at the turn's completion notification. Four cases with `retries=2`, `delay_between_retries=1`, `exponential_backoff=True`: one failure then success (non-streaming), one failure then success (streaming), more failures than retries, and the default `retries=0`.
|
||||
**Description:** Seeds a fact, adds three turns including a 1200-line filler document, calls `CodexAgent.acompact(session_id)`, then asks for the fact on the same session. `acompact` returns False before any thread exists, then resumes the thread on the low-level client, sends `thread/compact/start` and waits for the thread status to go active and back to idle on the global notification queue.
|
||||
|
||||
**Result:** Non-streaming: two attempts, the retry resumed the same Codex thread and answered `51`. Streaming: two attempts, stream closed with RunCompleted. Exhaustion: three attempts with 1s then 2s backoff, run ended in ERROR with the last error. Default: one attempt, ERROR.
|
||||
**Result:** `acompact` returned False before the first run and True after; the thread went idle again in about eight seconds and the rollout gained a `compacted` entry. The next turn answered `tangerine-walrus-88`. The `thread/compacted` notification itself is routed to a per-turn queue the caller never registered, which is why the adapter waits on thread status instead.
|
||||
|
||||
---
|
||||
|
||||
### codex_retries.py (review follow-up)
|
||||
### codex_compaction.py (review fixes, cookbook extended)
|
||||
|
||||
**Status:** PASS (2026-10-10, openai-codex 0.161.0, gpt-5.6-luna)
|
||||
|
||||
**Description:** Reran after the base-class changes from the review (retry warning event on streamed retries; tool calls of failed non-streamed attempts kept). A failed Codex turn now leaves its tool calls in `run_state["tools"]` before raising.
|
||||
**Description:** Review found that `acompact` resumed the thread on the low-level client with the adapter's high-level kwargs, which are sent verbatim: `sandbox="full-access"` was rejected (`unknown variant full-access`) and instructions and approval mode were silently dropped. Replicated with a probe against the app-server. The resume now goes through `AsyncCodex.thread_resume` exactly as runs do, and only `thread/compact/start` uses the low-level client. Also: a missing rollout returns False and forgets the stored thread id; compaction refuses while a run on the session is pending or running; the wait maps `asyncio.TimeoutError` to the builtin.
|
||||
|
||||
**Result:** All four cookbook cases pass as before: retry resumed the same thread, streaming closed with RunCompleted, exhaustion after three attempts, default of no retries.
|
||||
|
||||
---
|
||||
|
||||
### codex_retries.py (tool calls kept across a mid-turn failure)
|
||||
|
||||
**Status:** PASS (2026-10-10, openai-codex 0.161.0, gpt-5.6-luna)
|
||||
|
||||
**Description:** The non-streaming path now consumes the turn's notification stream itself and records each tool call as it completes, instead of the SDK's `handle.run()`, which returns items only when the turn ends. Live check: a prompt that runs one shell command, with a simulated transport failure injected before `turn/completed` on the first attempt, `retries=1`.
|
||||
|
||||
**Result:** Non-streamed and streamed: two attempts, completed, and the stored run holds the shell call from each attempt (two tools, two tool messages). Before the change the non-streamed run lost the first attempt's call. The four cookbook cases, `codex_tools.py` and `codex_structured_output.py` pass unchanged.
|
||||
**Result:** The cookbook now uses `sandbox="full-access"`, `instructions` and `approval_mode="deny_all"`, the configuration the first version could not compact. It compacted after three turns and recalled the fact on the next turn. With a background run pending on the session, `acompact` raised `RuntimeError` naming the run; after the run finished it compacted again. Pointing the session at a nonexistent thread id returned False with a warning and removed the id.
|
||||
|
||||
### codex_metrics.py
|
||||
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
"""
|
||||
Compact a Codex thread behind an Agno session
|
||||
=============================================
|
||||
Codex keeps each conversation in its own thread and compacts it when the
|
||||
context window fills up. CodexAgent.compact / acompact asks the app-server
|
||||
to do that now for the thread behind an Agno session, replacing the older
|
||||
turns with a summary. The next run on the session continues from the summary.
|
||||
|
||||
The agent here uses a full-access sandbox, developer instructions and an
|
||||
approval mode. Compaction resumes the thread with the same options as a run,
|
||||
so all of them apply to the resumed thread.
|
||||
|
||||
Scenario:
|
||||
1. Seed a fact and add a few turns including a long document.
|
||||
2. Compact, then ask for the fact again on the same session.
|
||||
3. The guards: compaction refuses while a background run on the session is
|
||||
still in flight, and returns False when the stored thread no longer exists.
|
||||
The last step logs a warning on purpose: the adapter reports whenever it
|
||||
drops a thread reference it can no longer resume.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_compaction.py
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import tempfile
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.db.sqlite import SqliteDb
|
||||
from agno.run.base import RunStatus
|
||||
|
||||
SESSION_ID = "codex-compaction-demo"
|
||||
FACT = "The release codename is tangerine-walrus-88."
|
||||
FINAL = {RunStatus.completed, RunStatus.error, RunStatus.cancelled}
|
||||
|
||||
|
||||
async def wait_for_run(agent: CodexAgent, run_id: str, timeout: float = 180.0):
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
while loop.time() < deadline:
|
||||
stored = await agent.aget_run_output(run_id, SESSION_ID)
|
||||
if stored is not None and stored.status in FINAL:
|
||||
return stored
|
||||
await asyncio.sleep(1.0)
|
||||
raise TimeoutError(f"run {run_id} did not finish in {timeout}s")
|
||||
|
||||
|
||||
async def main():
|
||||
with tempfile.TemporaryDirectory(prefix="agno-codex-compaction-") as workdir:
|
||||
agent = CodexAgent(
|
||||
id="codex-compaction",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="full-access",
|
||||
approval_mode="deny_all",
|
||||
instructions="Answer in as few words as possible.",
|
||||
db=SqliteDb(db_file=f"{workdir}/runs.db"),
|
||||
cwd=workdir,
|
||||
)
|
||||
|
||||
# No thread exists before the first run, so there is nothing to compact yet.
|
||||
assert await agent.acompact(SESSION_ID) is False
|
||||
|
||||
# 1. Build up a conversation.
|
||||
run = await agent.arun(
|
||||
f"Remember this: {FACT} Reply with just OK.", session_id=SESSION_ID
|
||||
)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
document = " ".join(
|
||||
f"Line {i}: the quick brown fox jumps over the lazy dog number {i}."
|
||||
for i in range(1200)
|
||||
)
|
||||
for prompt in [
|
||||
"Keep this document in mind and reply with just OK: " + document,
|
||||
"Name three primary colors, comma separated.",
|
||||
"What is 17 times 3? Reply with just the number.",
|
||||
]:
|
||||
run = await agent.arun(prompt, session_id=SESSION_ID)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
print(f"Turn: {run.content!r}")
|
||||
|
||||
# 2. Compact the thread. This resumes it in the app-server with the agent's sandbox,
|
||||
# instructions and approval mode, requests the compaction and waits for the thread
|
||||
# to be idle again. The next turn continues from the summary.
|
||||
compacted = await agent.acompact(SESSION_ID)
|
||||
print(f"Compacted: {compacted}")
|
||||
assert compacted is True
|
||||
run = await agent.arun(
|
||||
"What is the release codename? Reply with just the codename.",
|
||||
session_id=SESSION_ID,
|
||||
)
|
||||
assert run.status == RunStatus.completed, run.content
|
||||
print(f"After compaction: {run.content}")
|
||||
assert "tangerine-walrus-88" in (run.content or "")
|
||||
|
||||
# 3a. Compaction opens its own app-server on the thread, so it refuses while a run on
|
||||
# the session is in flight; two processes must not append to one rollout.
|
||||
pending = await agent.arun(
|
||||
"Reply with just OK.", session_id=SESSION_ID, background=True
|
||||
)
|
||||
try:
|
||||
await agent.acompact(SESSION_ID)
|
||||
raise AssertionError("compaction should refuse while a run is in flight")
|
||||
except RuntimeError as error:
|
||||
print(f"While a run is in flight: {error}")
|
||||
finished = await wait_for_run(agent, pending.run_id)
|
||||
assert finished.status == RunStatus.completed, finished.content
|
||||
assert await agent.acompact(SESSION_ID) is True
|
||||
print("After the run finished: compacted again")
|
||||
|
||||
# 3b. A stored thread id whose rollout no longer exists is forgotten, as a run would
|
||||
# do, and compaction reports there was nothing to compact. The adapter logs a
|
||||
# warning when it drops a thread reference; the one printed next is expected.
|
||||
print(
|
||||
"Pointing the session at a thread that does not exist; the next warning is expected:"
|
||||
)
|
||||
session = await agent.aget_session(SESSION_ID)
|
||||
session.session_data["codex_thread_id"] = "01a121ff-0000-7000-8000-000000000000"
|
||||
await agent.aupsert_session(session)
|
||||
assert await agent.acompact(SESSION_ID) is False
|
||||
session = await agent.aget_session(SESSION_ID)
|
||||
assert "codex_thread_id" not in (session.session_data or {})
|
||||
print("Stale thread id: compaction returned False and the id was forgotten")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -4,7 +4,7 @@ from contextlib import asynccontextmanager, suppress
|
||||
from contextvars import ContextVar
|
||||
from dataclasses import dataclass
|
||||
from time import time
|
||||
from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, List, Optional, Sequence, Union
|
||||
from typing import TYPE_CHECKING, Any, AsyncIterator, Awaitable, Dict, Iterator, List, Optional, Sequence, Union
|
||||
from uuid import uuid4
|
||||
|
||||
from agno.db.base import AsyncBaseDb, BaseDb, SessionType
|
||||
@@ -61,6 +61,25 @@ _live_handles: Dict[str, _LiveHandle] = {}
|
||||
_handle_owner: ContextVar[Optional[object]] = ContextVar("agno_external_handle_owner", default=None)
|
||||
|
||||
|
||||
def run_coroutine_sync(coro: Awaitable[Any]) -> Any:
|
||||
"""Run a coroutine from sync code.
|
||||
|
||||
On a thread with a running event loop (a notebook, a sync call made inside an async
|
||||
server) the coroutine runs on a worker thread with its own loop; otherwise asyncio.run
|
||||
is used directly. Shared by the sync wrappers of every external agent.
|
||||
"""
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
loop = None
|
||||
if loop is not None and loop.is_running():
|
||||
import concurrent.futures
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor() as pool:
|
||||
return pool.submit(asyncio.run, coro).result() # type: ignore[arg-type]
|
||||
return asyncio.run(coro) # type: ignore[arg-type]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExternalRunWarningEvent(CustomEvent):
|
||||
"""Nonfatal adapter warning emitted to streaming consumers."""
|
||||
@@ -216,23 +235,7 @@ class BaseExternalAgent:
|
||||
raise ValueError("Use arun(background=True) on a persistent event loop")
|
||||
if stream:
|
||||
return self._run_stream(input, session_id=session_id, user_id=user_id, **kwargs)
|
||||
else:
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
loop = None
|
||||
|
||||
if loop and loop.is_running():
|
||||
import concurrent.futures
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor() as pool:
|
||||
result = pool.submit(
|
||||
asyncio.run,
|
||||
self._arun_non_stream(input, session_id=session_id, user_id=user_id, **kwargs),
|
||||
).result()
|
||||
return result
|
||||
else:
|
||||
return asyncio.run(self._arun_non_stream(input, session_id=session_id, user_id=user_id, **kwargs))
|
||||
return run_coroutine_sync(self._arun_non_stream(input, session_id=session_id, user_id=user_id, **kwargs))
|
||||
|
||||
def print_response(
|
||||
self,
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
import asyncio
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from importlib import import_module
|
||||
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Set, Tuple
|
||||
from uuid import uuid4
|
||||
|
||||
from agno.agents.base import BaseExternalAgent, ExternalRunMetricsEvent, ExternalRunResult
|
||||
from agno.agents.base import BaseExternalAgent, ExternalRunMetricsEvent, ExternalRunResult, run_coroutine_sync
|
||||
from agno.exceptions import RunCancelledException
|
||||
from agno.metrics import ModelMetrics, RunMetrics
|
||||
from agno.models.response import ToolExecution
|
||||
from agno.run.base import RunStatus
|
||||
from agno.run.agent import (
|
||||
RunContentEvent,
|
||||
RunOutputEvent,
|
||||
@@ -37,6 +39,7 @@ _SANDBOX_ALIASES: Dict[str, str] = {
|
||||
"danger_full_access": "full-access",
|
||||
}
|
||||
|
||||
|
||||
# thread_start-only options that thread_resume does not accept.
|
||||
_START_ONLY_KEYS = {"ephemeral", "service_name", "session_start_source", "thread_source"}
|
||||
|
||||
@@ -216,6 +219,109 @@ class CodexAgent(BaseExternalAgent):
|
||||
async with sdk.AsyncCodex(self._codex_config(sdk)) as codex:
|
||||
await codex.login_api_key(api_key)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Context compaction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def compact(self, session_id: str, user_id: Optional[str] = None, timeout: float = 120.0) -> bool:
|
||||
"""Sync version of acompact."""
|
||||
return run_coroutine_sync(self.acompact(session_id, user_id=user_id, timeout=timeout))
|
||||
|
||||
async def acompact(self, session_id: str, user_id: Optional[str] = None, timeout: float = 120.0) -> bool:
|
||||
"""Compact the Codex thread behind an Agno session.
|
||||
|
||||
Codex keeps the conversation in its own thread. Compaction asks the app-server to
|
||||
replace the older turns with a summary so later turns fit the context window; the
|
||||
next run on the session continues from that summary. The thread, its rollout and the
|
||||
summary stay on the machine running Codex.
|
||||
|
||||
Returns False when the session has no Codex thread, or its thread can no longer be
|
||||
resumed (the stored id is then forgotten, as a run would). Waits until the thread is
|
||||
idle again and raises TimeoutError if that takes longer than timeout seconds.
|
||||
|
||||
Compaction opens its own app-server on the thread, so it must not overlap a run on the
|
||||
same session: two processes would append to one rollout. A run recorded as pending or
|
||||
running for the session raises RuntimeError.
|
||||
"""
|
||||
sdk = _sdk()
|
||||
session = await self.aget_session(session_id, user_id)
|
||||
thread_id = self._get_thread_id(session, session_id)
|
||||
if not thread_id:
|
||||
return False
|
||||
self._check_no_run_in_flight(session, session_id)
|
||||
async with self._new_client() as codex:
|
||||
# Resume through the high-level client: it translates the adapter's options
|
||||
# (sandbox names, approval mode, instructions) into the wire parameters that the
|
||||
# low-level thread_resume would otherwise send verbatim.
|
||||
try:
|
||||
thread = await codex.thread_resume(thread_id, **self._thread_kwargs(sdk, resume=True))
|
||||
except Exception as e:
|
||||
if not self._is_missing_thread(e):
|
||||
raise
|
||||
log_warning(f"Codex: cannot compact thread {thread_id} for session {session_id}: {e}. Forgetting it.")
|
||||
self._forget_thread(session, session_id)
|
||||
if session is not None and self.db is not None:
|
||||
await self.aupsert_session(session)
|
||||
return False
|
||||
await thread.compact()
|
||||
await self._await_thread_idle(self._app_server_client(codex), thread_id, timeout)
|
||||
log_debug(f"Codex: compacted thread {thread_id} for session {session_id}")
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _check_no_run_in_flight(session: Any, session_id: str) -> None:
|
||||
for run in getattr(session, "runs", None) or []:
|
||||
status = getattr(run.status, "value", run.status)
|
||||
if status in (RunStatus.pending.value, RunStatus.running.value):
|
||||
raise RuntimeError(
|
||||
f"Cannot compact session {session_id}: run {run.run_id} is {status}. Wait for it to finish."
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _app_server_client(codex: Any) -> Any:
|
||||
"""The low-level client behind an AsyncCodex, the only way to read the global
|
||||
notification queue: the SDK has no public signal for when compaction finishes."""
|
||||
client = getattr(codex, "_client", None)
|
||||
if client is None or not hasattr(client, "next_notification"):
|
||||
raise RuntimeError("This openai-codex version does not expose app-server notifications; cannot compact")
|
||||
return client
|
||||
|
||||
@staticmethod
|
||||
async def _await_thread_idle(client: Any, thread_id: str, timeout: float) -> None:
|
||||
"""Wait for the compaction to finish: the thread goes active, then idle again.
|
||||
|
||||
The app-server reports thread/compacted on a per-turn queue the caller never
|
||||
registered, so the thread status notifications on the global queue are used instead.
|
||||
"""
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
active = False
|
||||
while True:
|
||||
remaining = deadline - loop.time()
|
||||
if remaining <= 0:
|
||||
raise TimeoutError(f"Codex thread {thread_id} did not finish compacting within {timeout}s")
|
||||
try:
|
||||
notification = await asyncio.wait_for(client.next_notification(), timeout=remaining)
|
||||
except asyncio.TimeoutError as e:
|
||||
# asyncio.TimeoutError is only an alias of the builtin from Python 3.11.
|
||||
raise TimeoutError(f"Codex thread {thread_id} did not finish compacting within {timeout}s") from e
|
||||
method = getattr(notification, "method", "")
|
||||
payload = getattr(notification, "payload", None)
|
||||
if getattr(payload, "thread_id", thread_id) != thread_id:
|
||||
continue
|
||||
if method == "thread/compacted":
|
||||
# Today the app-server routes this to a per-turn queue the caller never
|
||||
# registered, so it is never seen here; kept in case that routing changes.
|
||||
return
|
||||
if method != "thread/status/changed":
|
||||
continue
|
||||
status = getattr(payload, "status", None)
|
||||
kind = getattr(getattr(status, "root", status), "type", None)
|
||||
if kind == "active":
|
||||
active = True
|
||||
elif kind == "idle" and active:
|
||||
return
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SDK option builders
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -97,19 +97,26 @@ class FakeThread:
|
||||
self._state.calls.append({"op": "run", "thread_id": self.id, "prompt": prompt, "kwargs": kwargs})
|
||||
return SimpleNamespace(final_response=self._state.final_response, items=[], status="completed", usage=None)
|
||||
|
||||
async def compact(self) -> Any:
|
||||
self._state.calls.append({"op": "thread_compact", "thread_id": self.id})
|
||||
return SimpleNamespace()
|
||||
|
||||
|
||||
class FakeAsyncCodex:
|
||||
state: FakeState
|
||||
# Low-level app-server client class, set by the compaction tests (AsyncCodex._client).
|
||||
client_class: Any = None
|
||||
|
||||
def __init__(self, config: Any = None) -> None:
|
||||
self.config = config
|
||||
self.state.calls.append({"op": "client", "config": config})
|
||||
self._client = self.client_class(config) if self.client_class is not None else None
|
||||
|
||||
async def __aenter__(self) -> "FakeAsyncCodex":
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc: Any) -> None:
|
||||
return None
|
||||
self.state.calls.append({"op": "client_exit"})
|
||||
|
||||
async def thread_start(self, **kwargs: Any) -> FakeThread:
|
||||
self.state.thread_counter += 1
|
||||
@@ -794,6 +801,161 @@ def test_retry_skips_rejected_rpc_requests(fake_sdk, monkeypatch, code, retried)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Compaction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _status_changed(thread_id: str, kind: str) -> Any:
|
||||
status = SimpleNamespace(root=SimpleNamespace(type=kind))
|
||||
return SimpleNamespace(method="thread/status/changed", payload=SimpleNamespace(thread_id=thread_id, status=status))
|
||||
|
||||
|
||||
class FakeAsyncCodexClient:
|
||||
"""Low-level app-server client: only the global notification queue."""
|
||||
|
||||
state: FakeState
|
||||
events: List[Any] = []
|
||||
hang: bool = False
|
||||
|
||||
def __init__(self, config: Any = None) -> None:
|
||||
self.config = config
|
||||
self._events = iter(self.events)
|
||||
|
||||
async def next_notification(self) -> Any:
|
||||
if self.hang:
|
||||
await asyncio.sleep(10)
|
||||
return next(self._events)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_client(fake_sdk, monkeypatch) -> type:
|
||||
FakeAsyncCodexClient.state = fake_sdk
|
||||
FakeAsyncCodexClient.events = []
|
||||
FakeAsyncCodexClient.hang = False
|
||||
monkeypatch.setattr(FakeAsyncCodex, "client_class", FakeAsyncCodexClient)
|
||||
return FakeAsyncCodexClient
|
||||
|
||||
|
||||
def _idle_after_active(thread_id: str = "thread-1") -> List[Any]:
|
||||
return [_status_changed(thread_id, "active"), _status_changed(thread_id, "idle")]
|
||||
|
||||
|
||||
def _started_session(fake_sdk, tmp_db, **agent_kwargs) -> CodexAgent:
|
||||
fake_sdk.notifications = [_delta("m1", "ok"), _turn_completed()]
|
||||
agent = CodexAgent(name="Codex", id="codex", db=tmp_db, **agent_kwargs)
|
||||
_collect(agent, "first", session_id="s1")
|
||||
return agent
|
||||
|
||||
|
||||
def test_compact_without_a_thread_is_a_no_op(fake_client, tmp_db):
|
||||
agent = CodexAgent(name="Codex", id="codex", db=tmp_db)
|
||||
assert agent.compact("s1") is False
|
||||
assert not [c for c in fake_client.state.calls if c["op"] == "thread_compact"]
|
||||
|
||||
|
||||
def test_compact_resumes_the_session_thread_and_waits_for_idle(fake_client, tmp_db):
|
||||
agent = _started_session(fake_client.state, tmp_db, model="gpt-5.6-luna", cwd="/work")
|
||||
fake_client.events = [
|
||||
_status_changed("thread-9", "active"), # another thread, ignored
|
||||
_status_changed("thread-1", "idle"), # idle before compaction starts, ignored
|
||||
_status_changed("thread-1", "active"),
|
||||
SimpleNamespace(method="mcpServer/startupStatus/updated", payload=SimpleNamespace()),
|
||||
_status_changed("thread-1", "idle"),
|
||||
SimpleNamespace(method="should-not-be-read", payload=SimpleNamespace()),
|
||||
]
|
||||
|
||||
assert agent.compact("s1") is True
|
||||
|
||||
wanted = {"thread_resume", "thread_compact", "client_exit"}
|
||||
ops = [c["op"] for c in fake_client.state.calls if c["op"] in wanted]
|
||||
assert ops[-3:] == ["thread_resume", "thread_compact", "client_exit"]
|
||||
resume = [c for c in fake_client.state.calls if c["op"] == "thread_resume"][-1]
|
||||
assert resume["thread_id"] == "thread-1"
|
||||
assert resume["kwargs"]["model"] == "gpt-5.6-luna" and resume["kwargs"]["cwd"] == "/work"
|
||||
assert next(c for c in fake_client.state.calls if c["op"] == "thread_compact")["thread_id"] == "thread-1"
|
||||
|
||||
|
||||
def test_compact_resumes_with_the_same_translated_options_as_a_run(fake_client, tmp_db):
|
||||
"""The resume goes through the high-level client, which maps these to the wire format
|
||||
(full-access -> danger-full-access, approval_mode -> approval policy, camelCase keys).
|
||||
The low-level client would send the dict verbatim and the app-server would reject it."""
|
||||
agent = _started_session(
|
||||
fake_client.state,
|
||||
tmp_db,
|
||||
sandbox="full-access",
|
||||
instructions="Always answer in French.",
|
||||
approval_mode="deny_all",
|
||||
ephemeral=False,
|
||||
)
|
||||
fake_client.events = _idle_after_active()
|
||||
|
||||
assert agent.compact("s1") is True
|
||||
|
||||
[compact_resume] = [c for c in fake_client.state.calls if c["op"] == "thread_resume"]
|
||||
assert compact_resume["kwargs"]["sandbox"] is Sandbox.full_access
|
||||
assert compact_resume["kwargs"]["developer_instructions"] == "Always answer in French."
|
||||
assert compact_resume["kwargs"]["approval_mode"] is ApprovalMode.deny_all
|
||||
assert "ephemeral" not in compact_resume["kwargs"], "thread_start-only options are not sent on resume"
|
||||
# Exactly what a run's resume sends, so the SDK applies the same wire translation.
|
||||
assert compact_resume["kwargs"] == agent._thread_kwargs(codex_module._sdk(), resume=True)
|
||||
|
||||
|
||||
def test_compact_forgets_a_thread_whose_rollout_is_gone(fake_client, tmp_db):
|
||||
agent = _started_session(fake_client.state, tmp_db)
|
||||
fake_client.state.unresumable.add("thread-1")
|
||||
|
||||
assert agent.compact("s1") is False
|
||||
|
||||
assert not [c for c in fake_client.state.calls if c["op"] == "thread_compact"]
|
||||
assert "codex_thread_id" not in (agent.get_session("s1").session_data or {})
|
||||
|
||||
|
||||
def test_compact_propagates_other_resume_failures(fake_client, tmp_db, monkeypatch):
|
||||
agent = _started_session(fake_client.state, tmp_db)
|
||||
|
||||
async def busy_resume(self, thread_id, **kwargs):
|
||||
raise RuntimeError("app-server busy")
|
||||
|
||||
monkeypatch.setattr(FakeAsyncCodex, "thread_resume", busy_resume)
|
||||
with pytest.raises(RuntimeError, match="busy"):
|
||||
agent.compact("s1")
|
||||
assert agent.get_session("s1").session_data["codex_thread_id"] == "thread-1"
|
||||
|
||||
|
||||
def test_compact_refuses_while_a_run_is_in_flight(fake_client, tmp_db):
|
||||
from agno.run.agent import RunOutput
|
||||
from agno.session.agent import AgentSession
|
||||
|
||||
agent = CodexAgent(name="Codex", id="codex", db=tmp_db)
|
||||
session = AgentSession(
|
||||
session_id="s1",
|
||||
session_data={"codex_thread_id": "thread-1"},
|
||||
runs=[RunOutput(run_id="r1", status=RunStatus.running)],
|
||||
)
|
||||
|
||||
async def aget_session(session_id, user_id=None):
|
||||
return session
|
||||
|
||||
agent.aget_session = aget_session # type: ignore[method-assign]
|
||||
with pytest.raises(RuntimeError, match="r1 is RUNNING"):
|
||||
agent.compact("s1")
|
||||
assert not [c for c in fake_client.state.calls if c["op"] in ("thread_resume", "thread_compact")]
|
||||
|
||||
|
||||
def test_compact_accepts_the_compacted_notification(fake_client, tmp_db):
|
||||
agent = _started_session(fake_client.state, tmp_db)
|
||||
fake_client.events = [SimpleNamespace(method="thread/compacted", payload=SimpleNamespace(thread_id="thread-1"))]
|
||||
assert asyncio.run(agent.acompact("s1")) is True
|
||||
|
||||
|
||||
def test_compact_times_out_and_closes_the_client(fake_client, tmp_db):
|
||||
agent = _started_session(fake_client.state, tmp_db)
|
||||
fake_client.hang = True
|
||||
with pytest.raises(TimeoutError, match="did not finish compacting"):
|
||||
agent.compact("s1", timeout=0.05)
|
||||
assert [c["op"] for c in fake_client.state.calls][-1] == "client_exit"
|
||||
|
||||
|
||||
# Metrics
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
Reference in New Issue
Block a user