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..3795a4fe 100644 --- a/raven/config/loader.py +++ b/raven/config/loader.py @@ -53,6 +53,39 @@ # 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) + + +# 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 @@ -193,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, ) @@ -360,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 = _logging.getLogger(__name__) + _log = _migration_log 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..406953c8 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -481,3 +481,108 @@ 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" + + +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 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 == {}