From 4d5c16f0160974cefcfe33febd2a2f24d235931f Mon Sep 17 00:00:00 2001 From: Ashpreet Date: Sun, 11 Oct 2026 23:03:35 +0100 Subject: [PATCH] fix: define HTTP compaction outcomes for external agents --- cookbook/frameworks/reliability/README.md | 17 ++++ cookbook/frameworks/reliability/TEST_LOG.md | 15 ++++ libs/agno/agno/agents/codex/agent.py | 8 +- libs/agno/agno/os/routers/agents/router.py | 45 +++++++++- libs/agno/tests/unit/agents/test_codex.py | 3 +- .../unit/os/routers/test_compaction_route.py | 88 +++++++++++++++++++ 6 files changed, 169 insertions(+), 7 deletions(-) diff --git a/cookbook/frameworks/reliability/README.md b/cookbook/frameworks/reliability/README.md index 47379b8ef0..392ef9ae48 100644 --- a/cookbook/frameworks/reliability/README.md +++ b/cookbook/frameworks/reliability/README.md @@ -130,3 +130,20 @@ docker compose -f cookbook/frameworks/reliability/compose.yaml down See [TEST_LOG.md](TEST_LOG.md) for recorded results. Do not interpret an unexecuted scenario or a green process exit without evidence as a pass. + +## HTTP compaction + +`POST /agents/{agent_id}/sessions/{session_id}/compact` checks session ownership +before attempting compaction. Native Agno agents return their compaction result. +Codex returns the same JSON keys (`status`, `message`, `compacted`, `record`, +`metrics`), with `record` and `metrics` null: its compaction changes the local native +thread. `no_history` means no resumable native thread is available on this replica; +it does not mean the database has no run history. Route the request to a replica +with that thread's files. A busy session returns 409; a timeout returns 504 and +does not guarantee the native operation stopped, so do not blindly resubmit. + +Claude returns 501 from this endpoint. Send `/compact` as the message to the normal +Claude run endpoint instead. This uses Claude's native command and session-store +transcript synchronization. Automatic compaction and successful native commands +must be verified independently of HTTP status codes. Compaction is not a durable +background job and callers must coordinate it with other turns on the session. diff --git a/cookbook/frameworks/reliability/TEST_LOG.md b/cookbook/frameworks/reliability/TEST_LOG.md index 6ea46f734f..765c1c8add 100644 --- a/cookbook/frameworks/reliability/TEST_LOG.md +++ b/cookbook/frameworks/reliability/TEST_LOG.md @@ -268,3 +268,18 @@ Suggested wording: > need idempotency. Turns within a session should be serialized. Claude credential > refresh, metrics, step replay and the pending agent-retry changes still need > release acceptance before we promise the full reliability contract. + +### HTTP compaction contract regression + +**Status:** PASS + +**Description:** `libs/agno/tests/unit/os/routers/test_compaction_route.py` checks +native response compatibility, Claude's explicit 501, Codex boolean serialization, +handled busy/timeouts, and session visibility before capability checks. The Codex +adapter suite verifies native compaction notifications, missing rollouts and busy +sessions using a scripted SDK fixture. + +**Result:** 129 passed, 5 provider-inapplicable cases skipped. Deterministic tests; +this does not establish live-provider compaction or cross-replica mutual exclusion. + +--- diff --git a/libs/agno/agno/agents/codex/agent.py b/libs/agno/agno/agents/codex/agent.py index 7bd5ec7d5f..c0046e5df2 100644 --- a/libs/agno/agno/agents/codex/agent.py +++ b/libs/agno/agno/agents/codex/agent.py @@ -314,7 +314,7 @@ class CodexAgent(BaseExternalAgent): 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. + running for the session raises SessionBusyError. """ sdk = _sdk() session = await self.aget_session(session_id, user_id) @@ -346,8 +346,10 @@ class CodexAgent(BaseExternalAgent): run = cls._run_in_flight(session) if run is not None: status = getattr(run.status, "value", run.status) - raise RuntimeError( - f"Cannot compact session {session_id}: run {run.run_id} is {status}. Wait for it to finish." + raise SessionBusyError( + f"Cannot compact session {session_id}: run {run.run_id} is {status}. Wait for it to finish.", + run_id=run.run_id, + session_id=session_id, ) @staticmethod diff --git a/libs/agno/agno/os/routers/agents/router.py b/libs/agno/agno/os/routers/agents/router.py index 877351351b..01d7b80b52 100644 --- a/libs/agno/agno/os/routers/agents/router.py +++ b/libs/agno/agno/os/routers/agents/router.py @@ -29,6 +29,7 @@ from agno.exceptions import ( OutputCheckError, RunNotContinuableError, RunNotFoundError, + SessionBusyError, ) from agno.media import Audio, Image, Video from agno.media import File as FileMedia @@ -76,6 +77,7 @@ from agno.os.schema import ( ) from agno.os.settings import AgnoAPISettings from agno.os.utils import ( + AgnoHTTPException, adraft_preview_identity, afinalize_continue_stream, allow_draft_preview, @@ -1875,7 +1877,7 @@ def get_agent_router( operation_id="compact_agent_session", summary="Compact Agent Session", description=( - "Fold this session's older history into a summary now, without waiting for the " + "For native Agno agents, fold this session's older history into a summary without waiting for the " "context to reach ``compact_at_tokens``. The stored transcript is never modified - " "compaction shortens what is sent to the model, not the record.\n\n" "A compaction can legitimately decline, which is reported rather than raised. Check " @@ -1888,7 +1890,13 @@ def get_agent_router( "safe cut point\n" "- ``no_history`` - the session exists but has no stored history yet\n" "- ``not_enabled`` - compaction is not configured on this agent\n" - "- ``summary_failed`` - the summarizer returned nothing" + "- ``summary_failed`` - the summarizer returned nothing\n\n" + "Codex compacts its local native thread and returns the same response keys, with no " + "Agno compaction record. If native thread files are unavailable, status is ``no_history``; " + "stored Agno run history alone cannot compact that thread. Claude does not implement " + "this endpoint: it returns 501. Submit ``/compact`` through the Claude run endpoint instead. " + "A busy Codex session returns 409; a compaction timeout returns 504 and does not prove " + "that the native operation stopped." ), responses={ 200: { @@ -1929,6 +1937,9 @@ def get_agent_router( }, }, 404: {"description": "Agent or session not found", "model": NotFoundResponse}, + 409: {"description": "Session has an in-flight run"}, + 501: {"description": "Agent does not implement HTTP compaction"}, + 504: {"description": "Native compaction did not finish before the timeout"}, }, dependencies=[Depends(require_resource_access("agents", "run", "agent_id"))], ) @@ -1978,7 +1989,35 @@ def get_agent_router( if await agent.aget_session(session_id=session_id, user_id=effective_user_id) is None: # type: ignore[union-attr] raise HTTPException(status_code=404, detail=SESSION_NOT_FOUND) - result = await agent.acompact(session_id=session_id, user_id=effective_user_id) # type: ignore[union-attr] + compact = getattr(agent, "acompact", None) + if not callable(compact): + raise HTTPException( + status_code=501, + detail="This agent does not implement HTTP compaction. For Claude, submit /compact as a run input.", + ) + try: + result = await compact(session_id=session_id, user_id=effective_user_id) + except SessionBusyError as exc: + raise AgnoHTTPException(exc) from exc + except TimeoutError as exc: + raise HTTPException( + status_code=504, + detail="Compaction timed out. The native operation may still be running; check before retrying.", + ) from exc + + # Codex reports native thread compaction as a bool; it does not produce an Agno + # history-fold record. Do not infer success from the existence of stored run history. + if isinstance(result, bool): + from agno.compaction.types import CompactionResult, CompactionStatus + + return CompactionResult( + status=CompactionStatus.COMPACTED if result else CompactionStatus.NO_HISTORY, + message=( + "Compacted the native thread on this replica." + if result + else "No resumable native thread is available on this replica. Stored run history is unchanged." + ), + ).to_dict() return result.to_dict() @router.get( diff --git a/libs/agno/tests/unit/agents/test_codex.py b/libs/agno/tests/unit/agents/test_codex.py index 4f227050f4..7bd04a1031 100644 --- a/libs/agno/tests/unit/agents/test_codex.py +++ b/libs/agno/tests/unit/agents/test_codex.py @@ -1229,6 +1229,7 @@ def test_compact_propagates_other_resume_failures(fake_client, tmp_db, monkeypat def test_compact_refuses_while_a_run_is_in_flight(fake_client, tmp_db): + from agno.exceptions import SessionBusyError from agno.run.agent import RunOutput from agno.session.agent import AgentSession @@ -1243,7 +1244,7 @@ def test_compact_refuses_while_a_run_is_in_flight(fake_client, tmp_db): return session agent.aget_session = aget_session # type: ignore[method-assign] - with pytest.raises(RuntimeError, match="r1 is RUNNING"): + with pytest.raises(SessionBusyError, 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")] diff --git a/libs/agno/tests/unit/os/routers/test_compaction_route.py b/libs/agno/tests/unit/os/routers/test_compaction_route.py index ee10c93be5..15f2818246 100644 --- a/libs/agno/tests/unit/os/routers/test_compaction_route.py +++ b/libs/agno/tests/unit/os/routers/test_compaction_route.py @@ -96,3 +96,91 @@ def test_response_shape_is_stable(client: TestClient): body = client.post("/agents/a1/sessions/s1/compact").json() assert set(body) == {"status", "message", "compacted", "record", "metrics"} + + +@pytest.fixture(params=["claude", "codex"]) +def external_compaction(request, monkeypatch): + from unittest.mock import AsyncMock + + from agno.agents.claude import ClaudeAgent + from agno.agents.codex import CodexAgent + from agno.db.in_memory import InMemoryDb + + db = InMemoryDb() + db.upsert_session(AgentSession(session_id="s1", agent_id="external", user_id="owner", runs=[])) + db.upsert_session(AgentSession(session_id="other", agent_id="different", user_id="owner", runs=[])) + cls = ClaudeAgent if request.param == "claude" else CodexAgent + agent = cls(id="external", db=db) + run = AsyncMock(side_effect=AssertionError("Compaction must not execute a model turn")) + monkeypatch.setattr(agent, "arun", run) + app = AgentOS(agents=[agent]).get_app() + + @app.middleware("http") + async def scope(request, call_next): + request.state.user_id = request.headers.get("x-test-user", "owner") + request.state.user_isolation_enabled = True + return await call_next(request) + + with TestClient(app, raise_server_exceptions=False) as client: + yield request.param, agent, client, run + run.assert_not_called() + + +def test_external_missing_session_is_404_before_capability_check(external_compaction): + _, _, client, _ = external_compaction + response = client.post("/agents/external/sessions/missing/compact") + assert response.status_code == 404, response.text + + +def test_external_compaction_cannot_access_other_users_session(external_compaction): + _, _, client, _ = external_compaction + response = client.post("/agents/external/sessions/s1/compact", headers={"x-test-user": "stranger"}) + assert response.status_code == 404, response.text + + +def test_external_compaction_other_component_is_404(external_compaction): + _, _, client, _ = external_compaction + response = client.post("/agents/external/sessions/other/compact") + assert response.status_code == 404, response.text + + +def test_claude_http_compaction_is_explicitly_unsupported(external_compaction): + provider, _, client, _ = external_compaction + if provider != "claude": + pytest.skip("Claude capability contract") + response = client.post("/agents/external/sessions/s1/compact") + assert response.status_code == 501, response.text + assert "/compact as a run input" in response.json()["detail"] + + +@pytest.mark.parametrize("result", [True, False]) +def test_codex_http_compaction_serializes_native_result(external_compaction, monkeypatch, result): + from unittest.mock import AsyncMock + + provider, agent, client, _ = external_compaction + if provider != "codex": + pytest.skip("Codex capability contract") + compact = AsyncMock(return_value=result) + monkeypatch.setattr(agent, "acompact", compact) + response = client.post("/agents/external/sessions/s1/compact") + assert response.status_code == 200, response.text + body = response.json() + assert body["compacted"] is result + assert body["status"] == ("compacted" if result else "no_history") + assert body["record"] is None and body["metrics"] is None + compact.assert_awaited_once_with(session_id="s1", user_id="owner") + + +@pytest.mark.parametrize("failure,code", [("busy", 409), ("timeout", 504)]) +def test_codex_http_compaction_errors_are_handled(external_compaction, monkeypatch, failure, code): + from unittest.mock import AsyncMock + + from agno.exceptions import SessionBusyError + + provider, agent, client, _ = external_compaction + if provider != "codex": + pytest.skip("Codex capability contract") + error = SessionBusyError("run is active", session_id="s1") if failure == "busy" else TimeoutError("native timeout") + monkeypatch.setattr(agent, "acompact", AsyncMock(side_effect=error)) + response = client.post("/agents/external/sessions/s1/compact") + assert response.status_code == code, response.text