diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 687ef02575..01e09b7534 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -3010,6 +3010,7 @@ class _StreamingCall: tool_calls = _ToolCallAccumulator() tool_calls_acc = tool_calls.acc finish_reason = model_name = usage_obj = None + response_id = upstream_provider = None # the provider's own id / serving upstream, from the chunks role = "assistant" _diag = self._new_diag() self._writer_token = self._attempt_request_client = self._attempt_stream_response = None @@ -3055,6 +3056,10 @@ class _StreamingCall: continue if hasattr(chunk, "model") and chunk.model: model_name = chunk.model + if response_id is None and isinstance(getattr(chunk, "id", None), str) and chunk.id: + response_id = chunk.id + if upstream_provider is None and isinstance(getattr(chunk, "provider", None), str) and chunk.provider: + upstream_provider = chunk.provider # OpenRouter stamps who served if not chunk.choices: usage, finish_reason = self._choiceless_chunk(chunk, finish_reason) usage_obj = usage or usage_obj @@ -3109,7 +3114,8 @@ class _StreamingCall: if stream.final_response is not None: return self._adopt_final_response(stream.final_response) return self._finish_chat_stream(stream, role, content_parts, reasoning_parts, tool_calls_acc, - finish_reason, model_name, usage_obj, flush_pending=_flush_pending_stream_text) + finish_reason, model_name, usage_obj, flush_pending=_flush_pending_stream_text, + response_id=response_id, upstream_provider=upstream_provider) def _adopt_final_response(self, final_response): """Adapter returned a completed response for ``stream=True``: switch the @@ -3158,7 +3164,7 @@ class _StreamingCall: return mock_tool_calls or None, has_truncated_tool_args def _finish_chat_stream(self, stream, role, content_parts, reasoning_parts, tool_calls_acc, finish_reason, - model_name, usage_obj, *, flush_pending): + model_name, usage_obj, *, flush_pending, response_id=None, upstream_provider=None): """Assemble the non-streaming-shaped response after the chunk loop. A stream ending with no finish_reason is a drop, not a completion: return a partial-stream stub so the loop fails fast instead of executing empty @@ -3193,7 +3199,10 @@ class _StreamingCall: raise provider_stream_error flush_pending() message = SimpleNamespace(role=role, content=full_content, tool_calls=mock_tool_calls, reasoning_content=full_reasoning) - return SimpleNamespace(id="stream-" + str(uuid.uuid4()), model=model_name, usage=usage_obj, + # The provider's id when the chunks carried one (chatcmpl-/gen-...): it is what a provider needs to + # look a request up. Fabricated only when the stream never sent one. + return SimpleNamespace(id=response_id or ("stream-" + str(uuid.uuid4())), model=model_name, usage=usage_obj, + provider=upstream_provider, choices=[SimpleNamespace(index=0, message=message, finish_reason=effective_finish_reason)]) # ── anthropic_messages wire ───────────────────────────────────────── diff --git a/agent/turn_usage.py b/agent/turn_usage.py index a9eb006196..4dfac5ca37 100644 --- a/agent/turn_usage.py +++ b/agent/turn_usage.py @@ -179,11 +179,22 @@ def record_response_usage( _cache_pct = "" if canonical_usage.cache_read_tokens and prompt_tokens: _cache_pct = f" cache={canonical_usage.cache_read_tokens}/{prompt_tokens} ({100*canonical_usage.cache_read_tokens/prompt_tokens:.0f}%)" + # write= is the money (cache writes cost 50x a read); id= is what a provider needs to look the + # request up; upstream= is who actually served it when the route reports that (OpenRouter's + # `provider`). Diagnosing the 1,393-agent run's cache misses took a DB join and a live probe + # because none of the three were on this line. + if canonical_usage.cache_write_tokens: + _cache_pct += f" write={canonical_usage.cache_write_tokens}" + _rid = getattr(response, "id", None) + _ident = f" id={_rid}" if isinstance(_rid, str) and _rid else "" + _upstream = getattr(response, "provider", None) + if isinstance(_upstream, str) and _upstream: + _ident += f" upstream={_upstream}" logger.info( - "API call #%d: model=%s provider=%s in=%d out=%d total=%d latency=%.1fs%s", + "API call #%d: model=%s provider=%s in=%d out=%d total=%d latency=%.1fs%s%s", agent.session_api_calls, agent.model, agent.provider or "unknown", prompt_tokens, completion_tokens, total_tokens, - api_duration, _cache_pct, + api_duration, _cache_pct, _ident, ) # nous.anthropic_wire=auto: the session's wire is decided once, from this first response. if agent.session_api_calls == 1 and (agent.provider or "") == "nous": diff --git a/evals/postmortem/forensics/logcalls.py b/evals/postmortem/forensics/logcalls.py index c8457bd9c3..fa5be57125 100644 --- a/evals/postmortem/forensics/logcalls.py +++ b/evals/postmortem/forensics/logcalls.py @@ -2,7 +2,7 @@ Hermes logs one line per API call:: - ... INFO [] agent.conversation_loop: API call #N: model=... in= out= total=... latency=..s cache=/ + ... INFO [] agent.conversation_loop: API call #N: model=... in= out= total=... latency=..s cache=/ (pct) [write=] [id=] [upstream=] Given the rotated logs, this reproduces: prompt-size distribution, cache hit-ratio buckets, the "plateau" signature of a broken cache prefix (hit count stuck at the previous call's breakpoint while @@ -30,6 +30,11 @@ _LINE = re.compile( ) +_WRITE = re.compile(r" write=(\d+)") +_ID = re.compile(r" id=(\S+)") +_UPSTREAM = re.compile(r" upstream=(.+?)(?: [a-z_]+=|$)") + + def parse_logs(paths: List[str], sids: set) -> List[Dict[str, Any]]: calls: List[Dict[str, Any]] = [] for path in paths: @@ -41,9 +46,15 @@ def parse_logs(paths: List[str], sids: set) -> List[Dict[str, Any]]: for line in fh: m = _LINE.match(line) if m and m.group(2) in sids: - calls.append({"ts": m.group(1), "sid": m.group(2), "n": int(m.group(3)), "model": m.group(4), - "inp": int(m.group(5)), "out": int(m.group(6)), "lat": float(m.group(7)), - "hit": int(m.group(8))}) + rec = {"ts": m.group(1), "sid": m.group(2), "n": int(m.group(3)), "model": m.group(4), + "inp": int(m.group(5)), "out": int(m.group(6)), "lat": float(m.group(7)), + "hit": int(m.group(8))} + # Newer lines (post 2026-09) also carry write= / id= / upstream=; optional. + for key, pat in (("write", _WRITE), ("id", _ID), ("upstream", _UPSTREAM)): + mm = pat.search(line) + if mm: + rec[key] = int(mm.group(1)) if key == "write" else mm.group(1) + calls.append(rec) return calls diff --git a/tests/agent/test_turn_usage_log_line.py b/tests/agent/test_turn_usage_log_line.py new file mode 100644 index 0000000000..878463056f --- /dev/null +++ b/tests/agent/test_turn_usage_log_line.py @@ -0,0 +1,66 @@ +"""The per-call log line carries what a cost or cache investigation needs: the cache write count, the +provider's response id, and the serving upstream when the route reports one. Existing parsers read a +prefix of the line, so the fields are appended and optional.""" +import logging +from types import SimpleNamespace + + +def _agent(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from run_agent import AIAgent + return AIAgent(api_key="k", base_url="https://inference-api.nousresearch.com/v1", provider="nous", + api_mode="chat_completions", model="anthropic/claude-fable-5.1", session_id="t", platform="cli", + quiet_mode=True, skip_context_files=True, skip_memory=True, save_trajectories=False, enabled_toolsets=["file"]) + + +def _usage(read, write, prompt): + return SimpleNamespace(prompt_tokens=prompt, completion_tokens=7, total_tokens=prompt + 7, + prompt_tokens_details=SimpleNamespace(cached_tokens=read, cache_write_tokens=write), + completion_tokens_details=None) + + +def _line(agent, caplog, resp): + from agent import turn_usage + caplog.clear() + with caplog.at_level(logging.INFO, logger="agent.turn_usage"): + turn_usage.record_response_usage(agent, resp, messages=[{"role": "user", "content": "hi"}], api_call_count=1, + api_duration=0.2, compression_attempts=0, max_compression_attempts=3) + return next(r.getMessage() for r in caplog.records if r.getMessage().startswith("API call #")) + + +def test_write_id_and_upstream_are_on_the_line(tmp_path, monkeypatch, caplog): + a = _agent(tmp_path, monkeypatch) + try: + line = _line(a, caplog, SimpleNamespace(usage=_usage(34_283, 28_604, 62_889), id="gen-1788636728-qMa1SbYZcwjrvUzJuF1g", + provider="Anthropic", model="anthropic/claude-fable-5.1")) + finally: + a.close() + assert " cache=34283/62889 (55%)" in line + assert " write=28604" in line + assert " id=gen-1788636728-qMa1SbYZcwjrvUzJuF1g" in line + assert " upstream=Anthropic" in line + # the pre-existing prefix is unchanged, so older parsers keep matching + assert line.startswith("API call #1: model=anthropic/claude-fable-5.1 provider=nous in=62889 out=7 total=62896 latency=0.2s") + + +def test_fields_are_omitted_when_absent(tmp_path, monkeypatch, caplog): + a = _agent(tmp_path, monkeypatch) + try: + line = _line(a, caplog, SimpleNamespace(usage=_usage(0, 0, 100), id=None, model="anthropic/claude-fable-5.1")) + finally: + a.close() + assert "write=" not in line and "id=" not in line and "upstream=" not in line + + +def test_forensics_parser_reads_the_new_fields(tmp_path): + from evals.postmortem.forensics.logcalls import parse_logs + log = tmp_path / "agent.log" + log.write_text( + "2026-09-06 19:32:08,549 INFO [s1] agent.conversation_loop: API call #3: model=anthropic/claude-fable-5.1 provider=nous " + "in=62889 out=7 total=62896 latency=0.2s cache=34283/62889 (55%) write=28604 id=gen-1788636728-qMa1 upstream=Claude Platform on AWS\n" + "2026-09-06 19:32:09,549 INFO [s1] agent.conversation_loop: API call #4: model=anthropic/claude-fable-5.1 provider=nous " + "in=91079 out=7 total=91086 latency=0.2s cache=62887/91079 (69%)\n", encoding="utf-8") + calls = parse_logs([str(log)], {"s1"}) + assert [c["n"] for c in calls] == [3, 4] + assert calls[0]["write"] == 28604 and calls[0]["id"] == "gen-1788636728-qMa1" and calls[0]["upstream"] == "Claude Platform on AWS" + assert "write" not in calls[1] and "id" not in calls[1]