fix: define HTTP compaction outcomes for external agents

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.

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

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

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(

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")]
 

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