From 3369938a157804c43b63f2a9a0a4c87bbaca1b49 Mon Sep 17 00:00:00 2001 From: KT Date: Tue, 18 Aug 2026 15:05:37 +0800 Subject: [PATCH 1/2] fix(*): report codex token usage and redact logged credentials Four gaps found while regression-testing the v0.1.11 fix list on main. None of them is on that list; the one item still owed from it (doctor not naming the resolved everos binary) is the last of the four. 1. openai_codex reported zero tokens for every turn. The backend does send usage, on `response.completed` -- an event the consumer already handled for its finish_reason and read nothing else from. With usage left empty the per-turn summary suppressed itself (it renders only when in/out tokens exceed zero) and token budgeting saw nothing. The cost half stays absent by design: a plan-billed provider has no per-token price. 2. httpx logs its request line at INFO, and the Telegram Bot API keys every route on /bot/, so a gateway run wrote 15 working bot tokens into its rotating log file -- the file users attach to bug reports. All three sinks now redact. The bot id survives; only the secret half is masked, and ordinary URLs, ports and paths are left alone so the log stays debuggable. This is the message-body counterpart to the existing diagnose=False, which already kept tracebacks from serializing secrets. 3. Migration notices repeated on every load_config. The strips rewrite the in-memory copy only, so an unmigrated file re-emits them forever, and a gateway loads once per cron fire. Deduped per process, next to the existing _warned_paths precedent. A record the logger would drop is not counted as told: the gateway loads its config before it installs a sink, and counting that dropped first line would have turned the noise into silence. 4. doctor now names the everos binary it resolved. Which one raven picked was invisible from every command, and the resolver prefers the interpreter's own directory -- the detail that matters when PATH holds another environment's copy. Verification: 234 tests across the affected modules, diff coverage 95.45% against the 90% gate, and a real gateway run confirming zero token occurrences and one line per migration. Co-authored-by: Claude (claude-opus-5) --- raven/cli/_log_file.py | 11 ++- raven/cli/doctor_commands.py | 9 ++ raven/config/loader.py | 29 +++++- raven/plugin/memory/everos/_server.py | 13 +++ raven/providers/openai_codex_provider.py | 40 ++++++-- raven/utils/log_redaction.py | 78 +++++++++++++++ tests/test_cli_doctor_commands.py | 58 +++++++++++ tests/test_config_loader.py | 63 ++++++++++++ tests/test_log_redaction.py | 119 +++++++++++++++++++++++ tests/test_openai_codex_provider.py | 92 +++++++++++++++++- 10 files changed, 499 insertions(+), 13 deletions(-) create mode 100644 raven/utils/log_redaction.py create mode 100644 tests/test_log_redaction.py diff --git a/raven/cli/_log_file.py b/raven/cli/_log_file.py index ef82706e..59903cec 100644 --- a/raven/cli/_log_file.py +++ b/raven/cli/_log_file.py @@ -21,6 +21,7 @@ from pathlib import Path from raven.config.paths import get_logs_dir +from raven.utils.log_redaction import combine_filters, redacting_filter def redirect_loguru_to_file( @@ -44,13 +45,17 @@ def redirect_loguru_to_file( log_path = get_logs_dir() / filename + # Every sink redacts: the file is what gets attached to a bug report, and + # the terminal is what gets pasted into one. + file_filter = combine_filters(redacting_filter, record_filter) + logger.remove() logger.add( str(log_path), level=file_level, rotation=rotation, retention=retention, - filter=record_filter, + filter=file_filter, enqueue=True, # thread-safe writes from channel threads + asyncio # diagnose=True would annotate tracebacks with local variable values, # writing secrets (API tokens, etc.) into a persisted, retained file. @@ -58,9 +63,9 @@ def redirect_loguru_to_file( diagnose=False, ) if terminal_level is not None: - logger.add(sys.stderr, level=terminal_level) + logger.add(sys.stderr, level=terminal_level, filter=redacting_filter) if os.environ.get("RAVEN_CLI_DEBUG"): - logger.add(sys.stderr, level="DEBUG") + logger.add(sys.stderr, level="DEBUG", filter=redacting_filter) _intercept_stdlib_logging(logger) _strip_tty_stream_handlers() diff --git a/raven/cli/doctor_commands.py b/raven/cli/doctor_commands.py index 7078f23a..f4f59b46 100644 --- a/raven/cli/doctor_commands.py +++ b/raven/cli/doctor_commands.py @@ -74,6 +74,7 @@ class MemoryInfo: backend: Optional[str] = None root: Optional[str] = None + binary: Optional[str] = None owned: bool = True address: Optional[str] = None server_running: bool = False @@ -255,7 +256,10 @@ def _probe_memory(config: "RavenConfig") -> MemoryInfo: info.capabilities = dict(report.capabilities) if info.owned: + from raven.plugin.memory.everos._server import everos_binary_path + info.root = str(everos_root()) + info.binary = everos_binary_path() info.configured = [s for s in (*REQUIRED_SECTIONS, *DEGRADING_SECTIONS) if everos_role_configured(s)] # Recall quality is decided by the embedding role in the user-level # everos.toml: with it recall matches meaning, without it only keywords. @@ -304,6 +308,11 @@ def _render_memory_capabilities(memory: MemoryInfo) -> None: return if memory.root: console.print(f" Memories: {memory.root}") + if memory.owned: + if memory.binary: + console.print(f" Binary: {memory.binary}") + else: + console.print(" Binary: [yellow]not found[/yellow] (searched next to the interpreter, then PATH)") if not memory.owned: console.print( " [dim]Managed by you -- Raven reads it at the address below and never writes,\n" diff --git a/raven/config/loader.py b/raven/config/loader.py index 740aaa8d..d972b4fc 100644 --- a/raven/config/loader.py +++ b/raven/config/loader.py @@ -53,6 +53,33 @@ # load_config calls (status/doctor load more than once) warn only once. _warned_paths: set[str] = set() +# Migration lines already logged in this process. The strips below rewrite the +# in-memory copy only, so an unmigrated file re-emits every line on every load +# -- and a running gateway loads once per cron fire, which turned a one-time +# notice into steady log noise. +_logged_migrations: set[str] = set() + + +class _MigrationLog: + """Logger proxy emitting each distinct migration line once per process.""" + + def __init__(self, inner: logging.Logger) -> None: + self._inner = inner + + def info(self, message: str, *args: Any) -> None: + # A record the logger would drop was never told to anyone. The gateway + # loads its config before it installs a sink, so counting that first + # dropped line as already-logged would silence the copy that reaches + # the log file -- turning the noise this dedup fixes into silence. + if not self._inner.isEnabledFor(logging.INFO): + return + rendered = message % args if args else message + if rendered in _logged_migrations: + return + _logged_migrations.add(rendered) + self._inner.info(message, *args) + + # User-facing lines produced by a migration that actually changed something, # waiting to be printed by whichever CLI entry point owns the terminal. # Migrations run inside the loader, which has no console of its own and whose @@ -362,7 +389,7 @@ def _migrate_config(data: dict, *, pop_extension_keys: bool = True, run_stamped: """ import logging as _logging - _log = _logging.getLogger(__name__) + _log = _MigrationLog(_logging.getLogger(__name__)) if run_stamped: _migrate_legacy_context_window(data, notify=True) diff --git a/raven/plugin/memory/everos/_server.py b/raven/plugin/memory/everos/_server.py index de59a928..05c61d34 100644 --- a/raven/plugin/memory/everos/_server.py +++ b/raven/plugin/memory/everos/_server.py @@ -133,6 +133,19 @@ def _require_llm_configured() -> None: ) +def everos_binary_path() -> str | None: + """The everos the spawner would run, or ``None`` when there is none. + + ``doctor`` reports this: which binary raven resolved is invisible + otherwise, and it is the first thing to check when a memory install that + looks configured still will not start. + """ + try: + return _everos_executable() + except EverosBinaryMissingError: + return None + + def _everos_executable() -> str: """Locate the everos CLI, preferring the one installed alongside raven. diff --git a/raven/providers/openai_codex_provider.py b/raven/providers/openai_codex_provider.py index 2f86b5df..e287ef54 100644 --- a/raven/providers/openai_codex_provider.py +++ b/raven/providers/openai_codex_provider.py @@ -94,20 +94,21 @@ async def chat( timeout = self.generation.timeout try: try: - content, tool_calls, finish_reason = await _request_codex( + content, tool_calls, finish_reason, usage = await _request_codex( url, headers, body, verify=True, timeout=timeout ) except Exception as e: if "CERTIFICATE_VERIFY_FAILED" not in str(e): raise logger.warning("SSL certificate verification failed for Codex API; retrying with verify=False") - content, tool_calls, finish_reason = await _request_codex( + content, tool_calls, finish_reason, usage = await _request_codex( url, headers, body, verify=False, timeout=timeout ) return LLMResponse( content=content, tool_calls=tool_calls, finish_reason=finish_reason, + usage=usage, ) except Exception as e: classification = self.classify_error(e) @@ -151,7 +152,7 @@ async def _request_codex( body: dict[str, Any], verify: bool, timeout: float, -) -> tuple[str, list[ToolCallRequest], str]: +) -> tuple[str, list[ToolCallRequest], str, dict[str, int]]: async with httpx.AsyncClient(timeout=timeout, verify=verify) as client: async with client.stream("POST", url, headers=headers, json=body) as response: if response.status_code != 200: @@ -162,6 +163,27 @@ async def _request_codex( return await _consume_sse(response, timeout) +def _convert_usage(raw: Any) -> dict[str, int]: + """Map a Responses API usage block onto the keys raven's loop reads. + + ``input_tokens`` counts cached tokens as well -- the same total-prompt + convention OpenRouter/LiteLLM use, which ``_build_usage_snapshot`` already + normalizes to fresh-only, so it passes through as ``prompt_tokens`` as-is. + """ + if not isinstance(raw, dict): + return {} + details = raw.get("input_tokens_details") + if not isinstance(details, dict): + details = {} + return { + "prompt_tokens": int(raw.get("input_tokens") or 0), + "completion_tokens": int(raw.get("output_tokens") or 0), + "total_tokens": int(raw.get("total_tokens") or 0), + "cache_read_input_tokens": int(details.get("cached_tokens") or 0), + "cache_creation_input_tokens": int(details.get("cache_write_tokens") or 0), + } + + def _convert_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]: """Convert OpenAI function-calling schema to Codex flat format.""" converted: list[dict[str, Any]] = [] @@ -357,11 +379,14 @@ async def _iter_sse(response: httpx.Response, timeout: float) -> AsyncGenerator[ buffer.append(line) -async def _consume_sse(response: httpx.Response, timeout: float) -> tuple[str, list[ToolCallRequest], str]: +async def _consume_sse( + response: httpx.Response, timeout: float +) -> tuple[str, list[ToolCallRequest], str, dict[str, int]]: content = "" tool_calls: list[ToolCallRequest] = [] tool_call_buffers: dict[str, dict[str, Any]] = {} finish_reason = "stop" + usage: dict[str, int] = {} async for event in _iter_sse(response, timeout): event_type = event.get("type") @@ -416,8 +441,9 @@ async def _consume_sse(response: httpx.Response, timeout: float) -> tuple[str, l ) ) elif event_type == "response.completed": - status = (event.get("response") or {}).get("status") - finish_reason = _map_finish_reason(status) + completed = event.get("response") or {} + finish_reason = _map_finish_reason(completed.get("status")) + usage = _convert_usage(completed.get("usage")) elif event_type in {"error", "response.failed"}: # The code is the retry signal: classify_error buckets by message # substring, and "server_is_overloaded" is what turns a dead-end @@ -432,7 +458,7 @@ async def _consume_sse(response: httpx.Response, timeout: float) -> tuple[str, l detail = ": ".join(str(part) for part in (code, message) if part) raise RuntimeError(f"Codex response failed: {detail}" if detail else "Codex response failed") - return content, tool_calls, finish_reason + return content, tool_calls, finish_reason, usage _FINISH_REASON_MAP = {"completed": "stop", "incomplete": "length", "failed": "error", "cancelled": "error"} diff --git a/raven/utils/log_redaction.py b/raven/utils/log_redaction.py new file mode 100644 index 00000000..a0306cdd --- /dev/null +++ b/raven/utils/log_redaction.py @@ -0,0 +1,78 @@ +"""Strip credentials out of log records before they reach a sink. + +Some upstreams put the credential in the URL itself -- the Telegram Bot API +keys every route on ``/bot/``, and several providers take an API key as +a query parameter -- so any library that logs its request line (httpx does, at +INFO) writes a working credential into a persisted, retained file. The gateway +log is the one users attach to a bug report, which is exactly the wrong place +for it. + +This is the message-body counterpart to ``diagnose=False`` in +:mod:`raven.cli._log_file`, which already keeps tracebacks from serializing +locals holding secrets. +""" + +from __future__ import annotations + +import re +from typing import Any + +REDACTED = "" + +_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = ( + # Telegram: /bot:/. The numeric id stays -- it identifies the + # bot for debugging and is not the secret half. + (re.compile(r"/bot(\d{5,}):[A-Za-z0-9_-]{15,}"), rf"/bot\1:{REDACTED}"), + # Credential passed as a query parameter (Gemini's ?key=, and friends). + ( + re.compile(r"([?&](?:api[-_]?key|access[-_]?token|auth[-_]?token|token|key|secret)=)[^&\s\"']+", re.I), + rf"\1{REDACTED}", + ), + # Basic-auth credentials embedded in a URL. + (re.compile(r"(://)[^/\s:@]+:[^/\s@]+@"), rf"\1{REDACTED}@"), + # Authorization: Bearer . + (re.compile(r"(Bearer\s+)[A-Za-z0-9._~+/=-]{12,}", re.I), rf"\1{REDACTED}"), + # Bare vendor-prefixed keys that appear outside any URL. + (re.compile(r"\b(sk-[A-Za-z0-9_-]{12,}|xox[abprs]-[A-Za-z0-9-]{10,}|gh[pousr]_[A-Za-z0-9]{20,})"), REDACTED), +) + + +def redact(text: str) -> str: + """Return ``text`` with every known credential shape masked.""" + for pattern, replacement in _PATTERNS: + text = pattern.sub(replacement, text) + return text + + +def redacting_filter(record: Any) -> bool: + """Loguru sink filter that rewrites the record in place, always keeping it. + + Loguru formats a record *after* its filters run, so mutating + ``record["message"]`` here is what reaches every sink. + """ + message = record.get("message") + if isinstance(message, str): + record["message"] = redact(message) + return True + + +def combine_filters(*filters: Any) -> Any: + """Chain sink filters, dropping the record as soon as one rejects it. + + The redacting filter must run even when a caller supplied its own + noise-dropping filter, so the two are composed rather than one replacing + the other. + """ + active = [f for f in filters if f is not None] + if not active: + return None + if len(active) == 1: + return active[0] + + def _chained(record: Any) -> bool: + return all(f(record) for f in active) + + return _chained + + +__all__ = ["REDACTED", "combine_filters", "redact", "redacting_filter"] diff --git a/tests/test_cli_doctor_commands.py b/tests/test_cli_doctor_commands.py index 802fa9a5..9f852c4a 100644 --- a/tests/test_cli_doctor_commands.py +++ b/tests/test_cli_doctor_commands.py @@ -664,3 +664,61 @@ def test_a_role_the_server_says_nothing_about_is_not_claimed_either_way( assert "rerank" not in info.configured assert "rerank" not in info.unbuilt + + +def test_doctor_reports_which_everos_binary_it_resolved(healthy_config: Path, no_memory_server, tmp_path) -> None: + """Which everos raven resolved was invisible from every command, so a + memory install that looks configured but will not start had nothing to + check. The resolver prefers the interpreter's own directory, which is the + detail that matters when PATH holds a different environment's copy.""" + from raven.config import update_everos as ue + from raven.plugin.memory.everos import _server + + _configured(no_memory_server, "llm") + _capabilities(no_memory_server, llm=True) + no_memory_server.setattr(ue, "everos_root", lambda: tmp_path / "mem-root") + no_memory_server.setattr(ue, "everos_owned", lambda: True) + no_memory_server.setattr(_server, "_everos_executable", lambda: "/opt/venv/bin/everos") + + r = runner.invoke(app, ["doctor"]) + + assert r.exit_code == 0, r.stdout + assert "Binary:" in r.stdout + assert "everos" in r.stdout + + +def test_doctor_says_when_no_everos_binary_resolves(healthy_config: Path, no_memory_server, tmp_path) -> None: + """A missing binary is reported, not raised: doctor's job is to describe a + broken install rather than fail on it.""" + from raven.config import update_everos as ue + from raven.plugin.memory.everos import _server + + _configured(no_memory_server, "llm") + _capabilities(no_memory_server, llm=True) + no_memory_server.setattr(ue, "everos_root", lambda: tmp_path / "mem-root") + no_memory_server.setattr(ue, "everos_owned", lambda: True) + + def _missing() -> str: + raise _server.EverosBinaryMissingError("everos not found next to /x/bin or on PATH.") + + no_memory_server.setattr(_server, "_everos_executable", _missing) + + r = runner.invoke(app, ["doctor"]) + + assert r.exit_code == 0, r.stdout + assert "Binary:" in r.stdout + assert "not found" in r.stdout + + +def test_doctor_omits_the_binary_line_for_a_server_raven_does_not_run(healthy_config: Path, no_memory_server) -> None: + """Raven never spawns a server it does not own, so which binary it would + have used is not a fact about that install.""" + from raven.config import update_everos as ue + + _configured(no_memory_server, "llm") + _capabilities(no_memory_server, llm=True) + no_memory_server.setattr(ue, "everos_owned", lambda: False) + + r = runner.invoke(app, ["doctor"]) + + assert "Binary:" not in r.stdout diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py index 9b523835..84705bc8 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -481,3 +481,66 @@ def _spy(self: Path, *args: object, **kwargs: object) -> int: loader.Path.write_text = original # type: ignore[method-assign] assert any(name.startswith("config.json.migrating.") and name.endswith(str(os.getpid())) for name in seen) + + +def test_migration_logs_once_per_process(caplog) -> None: + """The strips rewrite the in-memory copy only, so an unmigrated file + re-emits every migration line on every load. A gateway loads once per cron + fire, which turned a one-time notice into steady log noise.""" + import copy + import logging + + from raven.config.loader import _logged_migrations, _migrate_config + + _logged_migrations.clear() + raw = {"cron": {"forwardChannels": []}} + + with caplog.at_level(logging.INFO, logger="raven.config.loader"): + _migrate_config(copy.deepcopy(raw)) + _migrate_config(copy.deepcopy(raw)) + + lines = [r.getMessage() for r in caplog.records if "cron.forwardChannels" in r.getMessage()] + assert len(lines) == 1, lines + + +def test_migration_dedup_is_per_distinct_line(caplog) -> None: + """Two different retired keys are two different notices; deduping by + rendered message must not collapse them into one.""" + import logging + + from raven.config.loader import _logged_migrations, _migrate_config + + _logged_migrations.clear() + + with caplog.at_level(logging.INFO, logger="raven.config.loader"): + _migrate_config({"cron": {"forwardChannels": []}, "sentinel": {"monitors": []}}) + + dropped = [r.getMessage() for r in caplog.records if r.getMessage().startswith("Migrated: dropped")] + assert len(dropped) == 2, dropped + assert len(dropped) == len(set(dropped)) + + +def test_migration_notice_survives_a_load_before_the_sink_exists(caplog) -> None: + """The gateway loads its config before it installs a log sink, so the first + load's INFO records are dropped. Counting a dropped line as already-logged + silences the only copy that would have reached the log file.""" + import copy + import logging + + from raven.config.loader import _logged_migrations, _migrate_config + + _logged_migrations.clear() + raw = {"cron": {"forwardChannels": []}} + + log = logging.getLogger("raven.config.loader") + log.setLevel(logging.WARNING) # no sink yet: INFO goes nowhere + try: + _migrate_config(copy.deepcopy(raw)) + finally: + log.setLevel(logging.NOTSET) + + with caplog.at_level(logging.INFO, logger="raven.config.loader"): + _migrate_config(copy.deepcopy(raw)) + + lines = [r.getMessage() for r in caplog.records if "cron.forwardChannels" in r.getMessage()] + assert len(lines) == 1, "a load whose records were dropped must not consume the notice" diff --git a/tests/test_log_redaction.py b/tests/test_log_redaction.py new file mode 100644 index 00000000..b1b9534a --- /dev/null +++ b/tests/test_log_redaction.py @@ -0,0 +1,119 @@ +"""Credential redaction for log records. + +The regression these pin: a gateway run wrote 15 working Telegram bot tokens +into its rotating log file, because httpx logs the request line at INFO and the +Bot API puts the token in the URL path. +""" + +from __future__ import annotations + +import pytest + +from raven.utils.log_redaction import REDACTED, combine_filters, redact, redacting_filter + +TELEGRAM_LINE = 'HTTP Request: POST https://api.telegram.org/bot8642349359:AAFUU84G42yL-ONi3gk_RtdEprx__Mup0c/getMe "HTTP/1.1 200 OK"' + + +def test_telegram_token_is_masked_but_the_bot_id_survives(): + """The numeric id identifies which bot for debugging; only the secret half + is a credential.""" + out = redact(TELEGRAM_LINE) + assert "AAFUU84G42yL-ONi3gk_RtdEprx__Mup0c" not in out + assert "8642349359" in out + assert REDACTED in out + + +@pytest.mark.parametrize( + "line, secret", + [ + ("GET https://x.googleapis.com/v1/models?key=AIzaSyC7xR2mQ1abcdef", "AIzaSyC7xR2mQ1abcdef"), + ("POST https://h/v1?api_key=abcdef123456&z=1", "abcdef123456"), + ("POST https://h/v1?access_token=tok_abcdef123456", "tok_abcdef123456"), + ("connecting to https://user:hunter2pass@example.com/x", "hunter2pass"), + ("Authorization: Bearer sk-or-v1-722b6e7d2ac1bc73", "sk-or-v1-722b6e7d2ac1bc73"), + ("using key sk-proj-AbCdEf0123456789 now", "sk-proj-AbCdEf0123456789"), + ("slack token xoxb-1234567890-abcdefghij", "xoxb-1234567890-abcdefghij"), + ("token ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345", "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345"), + ], +) +def test_credential_shapes_are_masked(line, secret): + assert secret not in redact(line) + + +@pytest.mark.parametrize( + "line", + [ + "HTTP Request: GET https://api.telegram.org/health 200 OK", + "Cron: executing job 'regr' (5a3bb214)", + "Uvicorn running on http://localhost:18791", + "loaded 5 skills from https://skillhub.evermind.ai/openapi/v1/skills?limit=20", + "path is /Users/admin/.raven/logs/gateway.log", + ], +) +def test_ordinary_lines_are_left_alone(line): + """Over-redaction is its own failure: a log that masks ports, paths and + plain URLs stops being usable for debugging.""" + assert redact(line) == line + + +def test_filter_rewrites_the_record_and_keeps_it(): + record = {"message": TELEGRAM_LINE} + assert redacting_filter(record) is True + assert "AAFUU84G42yL-ONi3gk_RtdEprx__Mup0c" not in record["message"] + + +def test_filter_tolerates_a_non_string_message(): + record = {"message": None} + assert redacting_filter(record) is True + + +def test_combine_filters_semantics(): + assert combine_filters(None, None) is None + assert combine_filters(redacting_filter, None) is redacting_filter + + combined = combine_filters(redacting_filter, lambda r: "drop" not in r["message"]) + kept = {"message": TELEGRAM_LINE} + assert combined(kept) is True + assert REDACTED in kept["message"], "redaction runs even when composed" + assert combined({"message": "please drop this"}) is False + + +def test_gateway_log_file_never_receives_a_live_token(tmp_path, monkeypatch): + """End-to-end through the real sink wiring: what lands on disk is what a + user attaches to a bug report.""" + from loguru import logger + + import raven.cli._log_file as log_file + + monkeypatch.setattr(log_file, "get_logs_dir", lambda: tmp_path) + try: + path = log_file.redirect_loguru_to_file("probe.log", terminal_level=None) + logger.info(TELEGRAM_LINE) + logger.complete() + written = path.read_text() + finally: + logger.remove() + + assert "AAFUU84G42yL-ONi3gk_RtdEprx__Mup0c" not in written + assert REDACTED in written + + +def test_the_debug_stderr_sink_redacts_too(tmp_path, monkeypatch, capsys): + """RAVEN_CLI_DEBUG adds a second sink. Debugging a channel is exactly when + the request line gets read aloud, so that sink needs the same filter.""" + from loguru import logger + + import raven.cli._log_file as log_file + + monkeypatch.setattr(log_file, "get_logs_dir", lambda: tmp_path) + monkeypatch.setenv("RAVEN_CLI_DEBUG", "1") + try: + log_file.redirect_loguru_to_file("probe.log", terminal_level=None) + logger.info(TELEGRAM_LINE) + logger.complete() + finally: + logger.remove() + + err = capsys.readouterr().err + assert "AAFUU84G42yL-ONi3gk_RtdEprx__Mup0c" not in err + assert REDACTED in err diff --git a/tests/test_openai_codex_provider.py b/tests/test_openai_codex_provider.py index d34c13c5..66d6c789 100644 --- a/tests/test_openai_codex_provider.py +++ b/tests/test_openai_codex_provider.py @@ -20,6 +20,7 @@ _consume_sse, _convert_messages, _convert_tool_output, + _convert_usage, _friendly_error, _iter_sse, ) @@ -405,10 +406,97 @@ def stream(arguments: str) -> "_EndingStream": completed = {"type": "response.completed", "response": {"status": "completed"}} return _EndingStream([f"data: {json.dumps(done)}", "", f"data: {json.dumps(completed)}", ""]) - _, whole, _ = await _consume_sse(stream('{"path": "a.py", "content": "done"}'), timeout=1.0) + _, whole, _, _ = await _consume_sse(stream('{"path": "a.py", "content": "done"}'), timeout=1.0) assert whole[0].run_meta is None - _, cut, _ = await _consume_sse(stream('{"path": "a.py", "content": "import ran'), timeout=1.0) + _, cut, _, _ = await _consume_sse(stream('{"path": "a.py", "content": "import ran'), timeout=1.0) assert cut[0].run_meta is not None assert cut[0].run_meta.arguments_repaired is True assert cut[0].arguments["content"] == "import ran", "repaired, not stuffed into a raw blob" + + +# --- usage reporting ------------------------------------------------------- + + +def test_convert_usage_maps_the_responses_shape_onto_loop_keys(): + """`input_tokens` counts cached tokens, matching the OpenRouter/LiteLLM + convention `_build_usage_snapshot` normalizes -- so it passes through as + `prompt_tokens` rather than being pre-subtracted here.""" + mapped = _convert_usage( + { + "input_tokens": 1200, + "input_tokens_details": {"cached_tokens": 800, "cache_write_tokens": 100}, + "output_tokens": 42, + "output_tokens_details": {"reasoning_tokens": 9}, + "total_tokens": 1242, + } + ) + assert mapped == { + "prompt_tokens": 1200, + "completion_tokens": 42, + "total_tokens": 1242, + "cache_read_input_tokens": 800, + "cache_creation_input_tokens": 100, + } + + +@pytest.mark.parametrize("raw", [None, "nope", {}, {"input_tokens": None}]) +def test_convert_usage_survives_a_missing_or_malformed_block(raw): + """A usage block that never arrives must not break the turn.""" + assert isinstance(_convert_usage(raw), dict) + + +@pytest.mark.asyncio +async def test_consume_sse_carries_usage_off_the_completed_event(): + """The backend reports usage only on `response.completed`. Ignoring that + event's usage left every codex turn reporting zero tokens, which silently + disabled the per-turn cost summary and starved token budgeting.""" + + class _EndingStream: + def __init__(self, lines: list[str]) -> None: + self._lines = lines + + async def aiter_lines(self): + for line in self._lines: + yield line + + completed = { + "type": "response.completed", + "response": { + "status": "completed", + "usage": { + "input_tokens": 11, + "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0}, + "output_tokens": 9, + "total_tokens": 20, + }, + }, + } + text = {"type": "response.output_text.delta", "delta": "hi"} + stream = _EndingStream([f"data: {json.dumps(text)}", "", f"data: {json.dumps(completed)}", ""]) + + content, _tool_calls, finish_reason, usage = await _consume_sse(stream, timeout=1.0) + assert content == "hi" + assert finish_reason == "stop" + assert usage["prompt_tokens"] == 11 + assert usage["completion_tokens"] == 9 + + +@pytest.mark.asyncio +async def test_consume_sse_without_a_completed_event_reports_no_usage(): + """Absent usage stays empty rather than becoming a fabricated zero-cost + reading that looks like a real measurement.""" + + class _EndingStream: + def __init__(self, lines: list[str]) -> None: + self._lines = lines + + async def aiter_lines(self): + for line in self._lines: + yield line + + text = {"type": "response.output_text.delta", "delta": "hi"} + stream = _EndingStream([f"data: {json.dumps(text)}", ""]) + + _content, _tool_calls, _finish, usage = await _consume_sse(stream, timeout=1.0) + assert usage == {} From 3c80956e33bd6b94989a4c51b56ab5a0c2bbecb5 Mon Sep 17 00:00:00 2001 From: KT Date: Tue, 18 Aug 2026 15:37:02 +0800 Subject: [PATCH 2/2] fix(config): dedupe the context-window migration notice The dedup in the parent commit covered the migrations that log through _migrate_config's proxy, but _migrate_legacy_context_window logs from its own function with a raw logger and kept repeating. One command walks that migration twice -- load_config's own read, then the persist pass re-reading the raw file -- and the stamp that would stop a third walk is best effort: _write_migration_version swallows OSError, so a config directory that cannot take the stamp leaves the line firing on every load. That is the same repeat the parent commit set out to end. The proxy is now a module-level instance shared by both paths. The added tests cover the twice-per-command walk and the stamped pass; the parent commit's tests reached neither, since they all ran with run_stamped defaulted off. Reported by the PR reviewer on #347. Co-authored-by: Claude (claude-opus-5) --- raven/config/loader.py | 11 +++++++--- tests/test_config_loader.py | 42 +++++++++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/raven/config/loader.py b/raven/config/loader.py index d972b4fc..3795a4fe 100644 --- a/raven/config/loader.py +++ b/raven/config/loader.py @@ -80,6 +80,12 @@ def info(self, message: str, *args: Any) -> None: self._inner.info(message, *args) +# Module-level: the context-window migration logs from its own function, and it +# runs twice per command (load_config's read plus the persist pass re-reading +# the raw file), so it needs the same dedup the in-loader migrations get. +_migration_log = _MigrationLog(logging.getLogger(__name__)) + + # User-facing lines produced by a migration that actually changed something, # waiting to be printed by whichever CLI entry point owns the terminal. # Migrations run inside the loader, which has no console of its own and whose @@ -220,7 +226,7 @@ def _migrate_legacy_context_window(data: dict[str, Any], *, notify: bool = False if defaults.get(legacy_key) == LEGACY_CONTEXT_WINDOW_TOKENS: defaults.pop(legacy_key) changed = True - logging.getLogger(__name__).info( + _migration_log.info( "Migrated: dropped agents.defaults.%s (the retired 65536 default)", legacy_key, ) @@ -387,9 +393,8 @@ def _migrate_config(data: dict, *, pop_extension_keys: bool = True, run_stamped: the config path -- the one that can read the watermark and write it back -- opts in; the shims below are idempotent and always run. """ - import logging as _logging - _log = _MigrationLog(_logging.getLogger(__name__)) + _log = _migration_log if run_stamped: _migrate_legacy_context_window(data, notify=True) diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py index 84705bc8..406953c8 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -544,3 +544,45 @@ def test_migration_notice_survives_a_load_before_the_sink_exists(caplog) -> None lines = [r.getMessage() for r in caplog.records if "cron.forwardChannels" in r.getMessage()] assert len(lines) == 1, "a load whose records were dropped must not consume the notice" + + +def test_context_window_migration_logs_once_across_both_passes(caplog) -> None: + """This migration logs from its own function, and one command walks it + twice: load_config's read, then the persist pass re-reading the raw file. + Its stamp is best-effort (_write_migration_version swallows OSError), so a + config dir that cannot take the stamp leaves it re-running on every load -- + the same repeat this dedup exists to stop.""" + import copy + import logging + + from raven.config.loader import _logged_migrations, _migrate_legacy_context_window + + _logged_migrations.clear() + raw = {"agents": {"defaults": {"contextWindowTokens": 65536}}} + + with caplog.at_level(logging.INFO, logger="raven.config.loader"): + assert _migrate_legacy_context_window(copy.deepcopy(raw), notify=True) is True + assert _migrate_legacy_context_window(copy.deepcopy(raw)) is True + + lines = [r.getMessage() for r in caplog.records if "contextWindowTokens" in r.getMessage()] + assert len(lines) == 1, lines + + +def test_stamped_migration_pass_dedupes_too(caplog) -> None: + """The stamped path runs a migration the unstamped one does not, so it needs + its own coverage: deduping only what `run_stamped=False` reaches would leave + that line repeating.""" + import copy + import logging + + from raven.config.loader import _logged_migrations, _migrate_config + + _logged_migrations.clear() + raw = {"agents": {"defaults": {"contextWindowTokens": 65536}}} + + with caplog.at_level(logging.INFO, logger="raven.config.loader"): + _migrate_config(copy.deepcopy(raw), run_stamped=True) + _migrate_config(copy.deepcopy(raw), run_stamped=True) + + lines = [r.getMessage() for r in caplog.records if "contextWindowTokens" in r.getMessage()] + assert len(lines) == 1, lines