diff --git a/gateway/platforms/api_server_runs.py b/gateway/platforms/api_server_runs.py index 387263912b..94691c8891 100644 --- a/gateway/platforms/api_server_runs.py +++ b/gateway/platforms/api_server_runs.py @@ -37,10 +37,12 @@ _SUBAGENT_EVENT_KEYS = ( "output_tokens", "reasoning_tokens", "api_calls", "cost_usd", "files_read", "files_written", "output_tail") _SUBAGENT_TEXT_KEYS = ("goal", "summary", "output_tail") -# Terminal usage payload: (wire key, agent attribute), in wire order. +# Terminal usage payload: (wire key, agent attribute), in wire order. Cache reads ride along so a +# cost poller does not book them as full-price input (#102101). _USAGE_FIELDS = ( ("input_tokens", "session_prompt_tokens"), ("output_tokens", "session_completion_tokens"), - ("total_tokens", "session_total_tokens")) + ("total_tokens", "session_total_tokens"), ("cache_read_tokens", "session_cache_read_tokens"), + ("cache_write_tokens", "session_cache_write_tokens")) # Tool-progress event -> SSE payload fields (tool_name, preview, kwargs); key order is wire format. _FIXED_EVENT_FIELDS = { "tool.started": lambda tool, preview, kw: {"tool": tool, "preview": preview}, @@ -625,8 +627,30 @@ async def _handle_runs(self, request: "web.Request", *, _api_server) -> "web.Res return _accepted_response(run_id, "started", gateway_session_key, replayed=False) +def _run_usage(agent) -> Dict[str, int]: + """Terminal ``usage`` payload from the agent's session counters; a missing or non-numeric + counter (test doubles, agents without cache accounting) reads as ``0``.""" + usage = {} + for key, attr in _USAGE_FIELDS: + value = getattr(agent, attr, 0) + usage[key] = int(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else 0 + return usage + + +def _served_runtime(agent) -> Dict[str, str]: + """The ``{provider, model}`` pair that actually served the turn. After a ``fallback_providers`` + switch the agent keeps the fallback runtime until the NEXT turn restores the primary, so when + ``run_conversation()`` returns these attributes name the served pair — the run record's + ``model`` field only echoes the request (#102101). Non-string attributes read as ``""``.""" + pair = {} + for key in ("provider", "model"): + value = getattr(agent, key, "") + pair[key] = value if isinstance(value, str) else "" + return pair + + def _run_agent_sync(self, run: _RunLaunch, agent, approval_notify, *, _api_server): - """Executor-thread body of one run; returns ``(result, usage)``.""" + """Executor-thread body of one run; returns ``(result, usage, served_runtime)``.""" from gateway.session_context import clear_session_vars from gateway.hosted_room_execution_policy import ( RoomExecutionPolicy, bind_room_execution_policy, reset_room_execution_policy) @@ -690,7 +714,7 @@ def _run_agent_sync(self, run: _RunLaunch, agent, approval_notify, *, _api_serve for token, reset in resets: with suppress(Exception): reset(token) - return r, {key: getattr(agent, attr, 0) or 0 for key, attr in _USAGE_FIELDS} + return r, _run_usage(agent), _served_runtime(agent) def _make_approval_notify(self, run: _RunLaunch, *, _api_server) -> Callable[[Dict[str, Any]], None]: @@ -755,7 +779,7 @@ async def _execute_run(self, run: _RunLaunch, *, _api_server) -> None: **run.agent_kwargs) self._active_run_agents[run_id] = agent approval_notify = _make_approval_notify(self, run, _api_server=_api_server) - result, usage = await loop.run_in_executor( + result, usage, served_runtime = await loop.run_in_executor( None, lambda: _run_agent_sync(self, run, agent, approval_notify, _api_server=_api_server)) if not isinstance(result, dict): result = {} @@ -766,7 +790,8 @@ async def _execute_run(self, run: _RunLaunch, *, _api_server) -> None: # Non-retryable client errors (401/400) return failed=True rather than raising. _finish("failed", fields, error=_redact_api_error_text(result.get("error") or "agent run failed")) else: - _finish(status, fields, output=result.get("final_response", ""), usage=usage) + # ``runtime`` rides on both the pollable status and the run.completed event via _finish. + _finish(status, fields, output=result.get("final_response", ""), usage=usage, runtime=served_runtime) except asyncio.CancelledError: _finish("cancelled") raise diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py index 279cab2d5b..adf12472e2 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/tests/gateway/test_api_server_runs.py @@ -408,6 +408,47 @@ class TestRunStatus: assert mock_agent.run_conversation.call_args.kwargs["task_id"] == "space-session" assert status["session_id"] == "space-session" + @pytest.mark.asyncio + async def test_status_completed_run_reports_served_runtime_and_cache_tokens(self, adapter): + """After a fallback_providers switch the run record carries the runtime that actually + served the turn plus cache-read tokens, next to the requested ``model`` (#102101). + + ``agent.provider`` / ``agent.model`` still hold the fallback pair when + ``run_conversation()`` returns: the primary is only restored at the start of the NEXT + turn, so they are the served pair, while the top-level ``model`` echoes the request. + """ + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent") as mock_create: + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "done"} + mock_agent.provider = "openai-codex" + mock_agent.model = "gpt-5.6-luna" + mock_agent.session_prompt_tokens = 100 + mock_agent.session_completion_tokens = 5 + mock_agent.session_total_tokens = 105 + mock_agent.session_cache_read_tokens = 84 + mock_agent.session_cache_write_tokens = 11 + mock_create.return_value = mock_agent + + resp = await cli.post("/v1/runs", json={"input": "hello", "model": "deepseek-v4-pro"}) + run_id = (await resp.json())["run_id"] + + for _ in range(40): + status = await (await cli.get(f"/v1/runs/{run_id}")).json() + if status["status"] == "completed": + break + await asyncio.sleep(0.05) + + assert status["status"] == "completed" + # Top-level model still echoes the request; the served pair is disclosed alongside. + assert status["model"] == "deepseek-v4-pro" + assert status["runtime"] == {"provider": "openai-codex", "model": "gpt-5.6-luna"} + assert status["usage"] == { + "input_tokens": 100, "output_tokens": 5, "total_tokens": 105, + "cache_read_tokens": 84, "cache_write_tokens": 11, + } + # --------------------------------------------------------------------------- # GET /v1/runs/{run_id}/events — SSE event stream @@ -466,6 +507,48 @@ class TestRunEvents: assert "run.completed" in body assert "Hello!" in body + @pytest.mark.asyncio + async def test_completed_event_carries_served_runtime_and_cache_tokens(self, adapter): + """The run.completed SSE event discloses the same served runtime and cache tokens as the + pollable status, so streaming clients get identical cost-attribution data (#102101).""" + import json as _json + + app = _create_runs_app(adapter) + async with TestClient(TestServer(app)) as cli: + with patch.object(adapter, "_create_agent") as mock_create: + mock_agent = MagicMock() + mock_agent.run_conversation.return_value = {"final_response": "served"} + mock_agent.provider = "openai-codex" + mock_agent.model = "gpt-5.6-luna" + mock_agent.session_prompt_tokens = 774050 + mock_agent.session_completion_tokens = 6286 + mock_agent.session_total_tokens = 780336 + mock_agent.session_cache_read_tokens = 650000 + mock_agent.session_cache_write_tokens = 42 + mock_create.return_value = mock_agent + + resp = await cli.post("/v1/runs", json={"input": "hello", "model": "deepseek-v4-pro"}) + run_id = (await resp.json())["run_id"] + + events_resp = await cli.get(f"/v1/runs/{run_id}/events") + assert events_resp.status == 200 + body = await events_resp.text() + + completed = None + for frame in body.split("\n"): + if frame.startswith("data: "): + try: + payload = _json.loads(frame[len("data: "):]) + except ValueError: + continue + if payload.get("event") == "run.completed": + completed = payload + break + assert completed is not None, "run.completed event missing from stream" + assert completed["runtime"] == {"provider": "openai-codex", "model": "gpt-5.6-luna"} + assert completed["usage"]["cache_read_tokens"] == 650000 + assert completed["usage"]["cache_write_tokens"] == 42 + @pytest.mark.asyncio async def test_approval_resolve_all_is_scoped_to_target_run(self, auth_adapter): diff --git a/website/docs/user-guide/features/api-server.md b/website/docs/user-guide/features/api-server.md index 925f055b39..4b77943ade 100644 --- a/website/docs/user-guide/features/api-server.md +++ b/website/docs/user-guide/features/api-server.md @@ -473,10 +473,13 @@ Poll the current run state. This is useful for dashboards that need status witho "session_id": "space-session", "model": "hermes-agent", "output": "Done.", - "usage": {"input_tokens": 50, "output_tokens": 200, "total_tokens": 250} + "usage": {"input_tokens": 50, "output_tokens": 200, "total_tokens": 250, "cache_read_tokens": 40, "cache_write_tokens": 0}, + "runtime": {"provider": "openai", "model": "gpt-5"} } ``` +`model` echoes what the request asked for. On a completed run, `runtime` is the provider/model pair that actually served the turn — after a [fallback provider](fallback-providers.md) switch it names the fallback pair, so a cost-attribution poller books the run to the right provider. `usage.cache_read_tokens` / `usage.cache_write_tokens` are the session's prompt-cache reads and writes, so cached input is not priced as full-price input. The `run.completed` event on the events stream carries the same `usage` and `runtime` fields. + Statuses are retained briefly after terminal states (`completed`, `failed`, `cancelled`, or `interrupted`) for polling and UI reconciliation. When the gateway shuts down while a run is active, the run is persisted as `interrupted` (error `Gateway shutdown interrupted the run.`, terminal event `run.interrupted`) before the agent is asked to stop, so a durable run never survives a restart as `running`; a late result from the interrupted turn cannot overwrite it. ### GET /v1/runs/\{run_id\}/events