diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index db46f1ed6..e6f414d7b 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -5,14 +5,14 @@ }, "metadata": { "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.", - "version": "0.21.7" + "version": "0.21.10" }, "plugins": [ { "name": "headroom", "source": "./plugins/headroom-agent-hooks", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", - "version": "0.21.7", + "version": "0.21.10", "author": { "name": "Headroom Contributors", "url": "https://github.com/chopratejas/headroom" diff --git a/.github/plugin/marketplace.json b/.github/plugin/marketplace.json index db46f1ed6..e6f414d7b 100644 --- a/.github/plugin/marketplace.json +++ b/.github/plugin/marketplace.json @@ -5,14 +5,14 @@ }, "metadata": { "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.", - "version": "0.21.7" + "version": "0.21.10" }, "plugins": [ { "name": "headroom", "source": "./plugins/headroom-agent-hooks", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", - "version": "0.21.7", + "version": "0.21.10", "author": { "name": "Headroom Contributors", "url": "https://github.com/chopratejas/headroom" diff --git a/headroom/paths.py b/headroom/paths.py index 7a3083325..dd24ab261 100644 --- a/headroom/paths.py +++ b/headroom/paths.py @@ -71,6 +71,7 @@ _BRIDGE_STATE_FILE = "bridge_state.json" _LOGS_DIR = "logs" _PROXY_LOG_FILE = "proxy.log" _DEBUG_400_DIR = "debug_400" +_CODEX_WIRE_DEBUG_DIR = "codex_wire" _BIN_DIR = "bin" _RTK_UNIX = "rtk" _RTK_WIN = "rtk.exe" @@ -254,6 +255,12 @@ def debug_400_dir() -> Path: return log_dir() / _DEBUG_400_DIR +def codex_wire_debug_dir() -> Path: + """Return the directory used for opt-in Codex wire debug captures.""" + + return log_dir() / _CODEX_WIRE_DEBUG_DIR + + def bin_dir() -> Path: """Return the directory where Headroom ships vendored binaries.""" @@ -339,6 +346,7 @@ __all__ = [ "log_dir", "proxy_log_path", "debug_400_dir", + "codex_wire_debug_dir", "bin_dir", "rtk_path", "deploy_root", diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index f0036feaf..3e55cc313 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -51,6 +51,39 @@ RESPONSES_CONTEXT_SEARCH_TIMEOUT_SECONDS = 2.0 WS_FIRST_FRAME_TIMEOUT_SECONDS = 60.0 +def _extract_responses_usage(event: dict[str, Any]) -> tuple[int, int, int, int]: + """Return input/output/cache usage from a Responses event. + + Codex WebSocket streams include usage on ``response.completed`` events. + The shape mirrors HTTP Responses usage: + ``response.usage.input_tokens`` plus + ``response.usage.input_tokens_details.cached_tokens``. + """ + + if event.get("type") != "response.completed": + return 0, 0, 0, 0 + + response = event.get("response") + if not isinstance(response, dict): + response = {} + usage = response.get("usage") or event.get("usage") + if not isinstance(usage, dict): + return 0, 0, 0, 0 + + def _int(value: Any) -> int: + try: + return max(int(value), 0) + except (TypeError, ValueError): + return 0 + + input_tokens = _int(usage.get("input_tokens")) + output_tokens = _int(usage.get("output_tokens")) + details = usage.get("input_tokens_details") + cached_tokens = _int(details.get("cached_tokens")) if isinstance(details, dict) else 0 + uncached_tokens = max(input_tokens - cached_tokens, 0) + return input_tokens, output_tokens, cached_tokens, uncached_tokens + + def _decode_openai_bearer_payload(headers: dict[str, str]) -> dict[str, Any] | None: """Best-effort decode of an OpenAI OAuth bearer token payload. @@ -1260,6 +1293,20 @@ class OpenAIHandlerMixin: model = body.get("model", "unknown") stream = body.get("stream", False) + from headroom.proxy.helpers import capture_codex_wire_debug + + capture_codex_wire_debug( + "http_inbound_request", + request_id=request_id, + transport="http", + direction="client_to_headroom", + method=request.method, + url=str(request.url), + headers=dict(request.headers.items()), + body=body, + metadata={"path": request.url.path, "stream": stream}, + ) + # PR-C5: Python no longer compresses /v1/responses — Rust handles # item-aware compression natively (see crates/headroom-proxy/src/ # handlers/responses.rs). We synthesise a minimal `messages` list @@ -1573,6 +1620,25 @@ class OpenAIHandlerMixin: f"forwarding original body: {type(_e).__name__}: {_e}" ) + capture_codex_wire_debug( + "http_upstream_request", + request_id=request_id, + transport="http", + direction="headroom_to_upstream", + method="POST", + url=url, + headers=headers, + body=body, + metadata={ + "path": request.url.path, + "stream": stream, + "auth_mode": auth_mode.value, + "is_chatgpt_auth": is_chatgpt_auth, + "tokens_saved": tokens_saved, + "transforms_applied": transforms_applied, + }, + ) + try: if stream: # Streaming for Responses API uses semantic events @@ -1594,6 +1660,28 @@ class OpenAIHandlerMixin: else: headers = await apply_copilot_api_auth(headers, url=url) response = await self._retry_request("POST", url, headers, body) + _response_body_for_debug: Any = None + _response_raw_for_debug: str | None = None + try: + _response_body_for_debug = response.json() + except Exception: + try: + _response_raw_for_debug = response.text[:200_000] + except Exception: + _response_raw_for_debug = None + capture_codex_wire_debug( + "http_upstream_response", + request_id=request_id, + transport="http", + direction="upstream_to_headroom", + method="POST", + url=url, + headers=dict(response.headers), + body=_response_body_for_debug, + raw_text=_response_raw_for_debug, + status_code=response.status_code, + metadata={"stream": stream, "auth_mode": auth_mode.value}, + ) total_latency = (time.time() - start_time) * 1000 total_input_tokens = original_tokens # fallback @@ -1807,6 +1895,21 @@ class OpenAIHandlerMixin: # Forward client headers to upstream, adding required OpenAI-Beta header ws_headers = dict(websocket.headers) + _ws_url_obj = getattr(websocket, "url", None) + _ws_url = str(_ws_url_obj) if _ws_url_obj is not None else "" + _ws_path = getattr(_ws_url_obj, "path", "") if _ws_url_obj is not None else "" + from headroom.proxy.helpers import capture_codex_wire_debug + + capture_codex_wire_debug( + "ws_inbound_handshake", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="client_to_headroom", + url=_ws_url, + headers=ws_headers, + metadata={"path": _ws_path}, + ) # Extract per-request tags from headers up front so the # session-end RequestLog can attach them. `_extract_tags` is # the same helper the HTTP handlers use; on a WebSocket the @@ -1914,6 +2017,20 @@ class OpenAIHandlerMixin: ws_base = base.replace("https://", "wss://").replace("http://", "ws://") upstream_url = build_copilot_upstream_url(ws_base, "/v1/responses") + capture_codex_wire_debug( + "ws_upstream_handshake", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="headroom_to_upstream", + url=upstream_url, + headers=upstream_headers, + metadata={ + "is_chatgpt_auth": is_chatgpt_auth, + "subprotocols": client_subprotocols, + }, + ) + # Unit 3: attach the resolved upstream URL to the session handle. if session_handle is not None: session_handle.upstream_url = upstream_url @@ -1988,6 +2105,20 @@ class OpenAIHandlerMixin: request_id=request_id, ) + capture_codex_wire_debug( + "ws_upstream_handshake_final", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="headroom_to_upstream", + url=upstream_url, + headers=upstream_headers, + metadata={ + "is_chatgpt_auth": is_chatgpt_auth, + "subprotocols": client_subprotocols, + }, + ) + logger.debug( f"[{request_id}] WS upstream headers: " f"{[k for k in upstream_headers if k.lower() != 'authorization']}, " @@ -2041,6 +2172,23 @@ class OpenAIHandlerMixin: except json.JSONDecodeError: # Not JSON — pass through as-is pass + ws_input_tokens_total = 0 + ws_output_tokens_total = 0 + ws_cache_read_tokens_total = 0 + ws_uncached_input_tokens_total = 0 + ws_response_create_frames = 1 + + capture_codex_wire_debug( + "ws_inbound_first_frame", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="client_to_headroom", + url=_ws_url, + body=body if body else None, + raw_text=None if body else first_msg_raw, + metadata={"frame": 1}, + ) # --- Memory: inject context, tools, and instructions --- memory_user_id: str | None = None @@ -2267,6 +2415,27 @@ class OpenAIHandlerMixin: f"forwarding original frame: {type(_ce).__name__}: {_ce}" ) + _first_upstream_body: Any = None + try: + _first_upstream_body = json.loads(first_msg_raw) + except json.JSONDecodeError: + _first_upstream_body = None + capture_codex_wire_debug( + "ws_upstream_first_frame", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="headroom_to_upstream", + url=upstream_url, + body=_first_upstream_body, + raw_text=None if _first_upstream_body is not None else first_msg_raw, + metadata={ + "frame": 1, + "tokens_saved": tokens_saved, + "transforms_applied": transforms_applied, + }, + ) + # --- Connect to upstream OpenAI WebSocket --- logger.info(f"[{request_id}] WS /v1/responses connecting to {upstream_url}") @@ -2417,11 +2586,54 @@ class OpenAIHandlerMixin: return rewritten async def _client_to_upstream() -> None: - nonlocal client_relay_error + nonlocal client_relay_error, ws_response_create_frames + client_frame_index = 1 try: while True: msg = await websocket.receive_text() + client_frame_index += 1 + _inbound_frame_body: Any = None + try: + _inbound_frame_body = json.loads(msg) + except json.JSONDecodeError: + _inbound_frame_body = None + capture_codex_wire_debug( + "ws_inbound_client_frame", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="client_to_headroom", + url=_ws_url, + body=_inbound_frame_body, + raw_text=None if _inbound_frame_body is not None else msg, + metadata={"frame": client_frame_index}, + ) + if ( + isinstance(_inbound_frame_body, dict) + and _inbound_frame_body.get("type") == "response.create" + ): + ws_response_create_frames += 1 msg = await _maybe_compress_response_create_frame(msg) + _outbound_frame_body: Any = None + try: + _outbound_frame_body = json.loads(msg) + except json.JSONDecodeError: + _outbound_frame_body = None + capture_codex_wire_debug( + "ws_upstream_client_frame", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="headroom_to_upstream", + url=upstream_url, + body=_outbound_frame_body, + raw_text=None if _outbound_frame_body is not None else msg, + metadata={ + "frame": client_frame_index, + "tokens_saved_total": tokens_saved, + "transforms_applied": transforms_applied, + }, + ) await upstream.send(msg) except asyncio.CancelledError: # Explicit cancel from the outer @@ -2463,6 +2675,8 @@ class OpenAIHandlerMixin: # over ``upstream_disconnect``. nonlocal response_completed_seen nonlocal upstream_relay_error + nonlocal ws_input_tokens_total, ws_output_tokens_total + nonlocal ws_cache_read_tokens_total, ws_uncached_input_tokens_total memory_enabled = bool(self.memory_handler and memory_user_id) @@ -2487,7 +2701,9 @@ class OpenAIHandlerMixin: _first_event_started_at = _upstream_first_event_started # noqa: B023 try: + upstream_frame_index = 0 async for msg in upstream: + upstream_frame_index += 1 if ( _first_event_started_at is not None and "upstream_first_event" not in stage_timer @@ -2498,13 +2714,39 @@ class OpenAIHandlerMixin: * 1000.0, ) if isinstance(msg, bytes): + capture_codex_wire_debug( + "ws_upstream_binary_frame", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="upstream_to_headroom", + url=upstream_url, + metadata={ + "frame": upstream_frame_index, + "byte_count": len(msg), + }, + ) await websocket.send_bytes(msg) continue msg_str = msg if isinstance(msg, str) else str(msg) - - if not memory_enabled: - await websocket.send_text(msg_str) - continue + _upstream_frame_body: Any = None + try: + _upstream_frame_body = json.loads(msg_str) + except json.JSONDecodeError: + _upstream_frame_body = None + capture_codex_wire_debug( + "ws_upstream_text_frame", + request_id=request_id, + session_id=session_id, + transport="websocket", + direction="upstream_to_headroom", + url=upstream_url, + body=_upstream_frame_body, + raw_text=None + if _upstream_frame_body is not None + else msg_str, + metadata={"frame": upstream_frame_index}, + ) # Parse event try: @@ -2514,6 +2756,21 @@ class OpenAIHandlerMixin: continue event_type = event.get("type", "") + ( + usage_input_tokens, + usage_output_tokens, + usage_cache_read_tokens, + usage_uncached_tokens, + ) = _extract_responses_usage(event) + if usage_input_tokens or usage_output_tokens: + ws_input_tokens_total += usage_input_tokens + ws_output_tokens_total += usage_output_tokens + ws_cache_read_tokens_total += usage_cache_read_tokens + ws_uncached_input_tokens_total += usage_uncached_tokens + + if not memory_enabled: + await websocket.send_text(msg_str) + continue # --- Phase 1: Buffer until first output item --- if not decided: @@ -2828,15 +3085,24 @@ class OpenAIHandlerMixin: **(ws_tags or {}), "auth_mode": _final_auth_mode.value, "endpoint": "responses_ws", + "compression_scope": "live_zone", + "cache_policy": "prefix_safe", + "transport": "websocket", + "route": "chatgpt_subscription" if is_chatgpt_auth else "openai_api", + "ws_response_create_frames": str(ws_response_create_frames), "ws_frames_compressed": str(ws_frames_compressed), + "cache_read_tokens": str(ws_cache_read_tokens_total), + "uncached_input_tokens": str(ws_uncached_input_tokens_total), } await self.metrics.record_request( provider="openai", model=model_name, - input_tokens=0, - output_tokens=0, + input_tokens=ws_input_tokens_total, + output_tokens=ws_output_tokens_total, tokens_saved=tokens_saved, latency_ms=ws_session_duration_ms, + cache_read_tokens=ws_cache_read_tokens_total, + uncached_input_tokens=ws_uncached_input_tokens_total, ) if getattr(self, "logger", None) is not None: from headroom.proxy.helpers import compute_turn_id @@ -2857,11 +3123,15 @@ class OpenAIHandlerMixin: timestamp=datetime.now().isoformat(), provider="openai", model=model_name, - input_tokens_original=0, - input_tokens_optimized=0, - output_tokens=0, + input_tokens_original=ws_input_tokens_total + tokens_saved, + input_tokens_optimized=ws_input_tokens_total, + output_tokens=ws_output_tokens_total, tokens_saved=tokens_saved, - savings_percent=0.0, + savings_percent=( + tokens_saved / (ws_input_tokens_total + tokens_saved) * 100 + ) + if ws_input_tokens_total + tokens_saved > 0 + else 0.0, optimization_latency_ms=0.0, total_latency_ms=ws_session_duration_ms, tags=ws_session_tags, diff --git a/headroom/proxy/handlers/streaming.py b/headroom/proxy/handlers/streaming.py index 9232d19f1..be15b7c33 100644 --- a/headroom/proxy/handlers/streaming.py +++ b/headroom/proxy/handlers/streaming.py @@ -696,6 +696,8 @@ class StreamingMixin: # transform mutated the body we re-serialize canonically; otherwise # we forward the original client bytes verbatim. from headroom.proxy.helpers import ( + capture_codex_wire_debug, + codex_wire_debug_enabled, log_outbound_request, prepare_outbound_body_bytes, ) @@ -716,6 +718,27 @@ class StreamingMixin: request_id=request_id, source=outbound_source, ) + _codex_wire_debug = ( + codex_wire_debug_enabled() and provider == "openai" and "/responses" in url + ) + if _codex_wire_debug: + capture_codex_wire_debug( + "http_stream_upstream_request", + request_id=request_id, + transport="http_sse", + direction="headroom_to_upstream", + method="POST", + url=url, + headers=outbound_headers, + body=body, + metadata={ + "body_bytes": len(outbound_bytes), + "body_mutated": body_mutated, + "mutation_reasons": list(mutation_reasons or []), + "tokens_saved": tokens_saved, + "transforms_applied": transforms_applied, + }, + ) # Mutable state for the generator to update stream_state: dict[str, Any] = { @@ -756,6 +779,17 @@ class StreamingMixin: "POST", url, content=outbound_bytes, headers=outbound_headers ) upstream_response = await self.http_client.send(_upstream_req, stream=True) + if _codex_wire_debug: + capture_codex_wire_debug( + "http_stream_upstream_response_headers", + request_id=request_id, + transport="http_sse", + direction="upstream_to_headroom", + method="POST", + url=url, + headers=dict(upstream_response.headers), + status_code=upstream_response.status_code, + ) break except (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) as e: last_connect_error = e @@ -827,6 +861,29 @@ class StreamingMixin: finally: await upstream_response.aclose() + if _codex_wire_debug: + _error_text: str | None = None + _error_body: Any = None + try: + _error_text = error_content.decode("utf-8") + _error_body = json.loads(_error_text) + _error_text = None + except Exception: + with contextlib.suppress(Exception): + _error_text = error_content.decode("utf-8", errors="replace") + capture_codex_wire_debug( + "http_stream_upstream_error_response", + request_id=request_id, + transport="http_sse", + direction="upstream_to_headroom", + method="POST", + url=url, + headers=response_headers, + body=_error_body, + raw_text=_error_text, + status_code=upstream_response.status_code, + ) + stream_state["total_bytes"] = len(error_content) await self._finalize_stream_response( body=body, @@ -871,7 +928,9 @@ class StreamingMixin: try: async with contextlib.aclosing(upstream_response) as response: + sse_chunk_index = 0 async for chunk in response.aiter_bytes(): + sse_chunk_index += 1 # Record TTFB on first chunk if stream_state["ttfb_ms"] is None: stream_state["ttfb_ms"] = (time.time() - start_time) * 1000 @@ -900,9 +959,26 @@ class StreamingMixin: # real-time clients (LangGraph, LangChain, etc.) yield chunk + if _codex_wire_debug: + capture_codex_wire_debug( + "http_stream_upstream_chunk", + request_id=request_id, + transport="http_sse", + direction="upstream_to_headroom", + method="POST", + url=url, + raw_text=chunk.decode("utf-8", errors="replace"), + metadata={ + "chunk": sse_chunk_index, + "byte_count": len(chunk), + }, + ) + # Buffer SSE data for memory processing and/or prefix tracker - _track_sse = memory_enabled or ( - prefix_tracker is not None and provider == "anthropic" + _track_sse = ( + _codex_wire_debug + or memory_enabled + or (prefix_tracker is not None and provider == "anthropic") ) if _track_sse: if memory_enabled: @@ -995,6 +1071,27 @@ class StreamingMixin: ) if ccr_parsed: self._record_ccr_feedback_from_response(ccr_parsed, provider, request_id) + if _codex_wire_debug: + _debug_parsed_response = ( + parsed_response + if parsed_response + else self._parse_sse_to_response(full_sse_data, provider) + if full_sse_data + else None + ) + capture_codex_wire_debug( + "http_stream_upstream_complete", + request_id=request_id, + transport="http_sse", + direction="upstream_to_headroom", + method="POST", + url=url, + headers=dict(upstream_response.headers), + body=_debug_parsed_response, + raw_text=full_sse_data, + status_code=upstream_response.status_code, + metadata={"total_bytes": stream_state["total_bytes"]}, + ) except (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout) as e: logger.error(f"[{request_id}] Connection error to upstream API: {e}") diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py index 8cfdf9e8f..2cb757c7e 100644 --- a/headroom/proxy/helpers.py +++ b/headroom/proxy/helpers.py @@ -26,6 +26,145 @@ if TYPE_CHECKING: logger = logging.getLogger("headroom.proxy") +_CODEX_WIRE_DEBUG_ENV = "HEADROOM_CODEX_WIRE_DEBUG" +_CODEX_WIRE_DEBUG_DIR_ENV = "HEADROOM_CODEX_WIRE_DEBUG_DIR" +_CODEX_WIRE_REDACTED = "[REDACTED]" +_CODEX_WIRE_SECRET_KEYS = ( + "authorization", + "cookie", + "set-cookie", + "api-key", + "x-api-key", + "openai-api-key", + "anthropic-api-key", + "access_token", + "refresh_token", + "id_token", + "bearer", + "password", + "secret", + "token", + "credential", +) + + +def codex_wire_debug_enabled() -> bool: + """Return whether opt-in Codex wire capture is enabled.""" + + return os.environ.get(_CODEX_WIRE_DEBUG_ENV, "").strip().lower() in ( + "1", + "true", + "yes", + "on", + ) + + +def _codex_wire_debug_dir() -> Path: + explicit = os.environ.get(_CODEX_WIRE_DEBUG_DIR_ENV, "").strip() + if explicit: + return Path(explicit).expanduser() + return _paths.codex_wire_debug_dir() + + +def _should_redact_key(key: str) -> bool: + normalized = key.lower().replace("-", "_") + if normalized in {marker.replace("-", "_") for marker in _CODEX_WIRE_SECRET_KEYS}: + return True + return ( + normalized.endswith("_api_key") + or normalized.endswith("_secret") + or normalized.endswith("_password") + or normalized.endswith("_access_token") + or normalized.endswith("_refresh_token") + ) + + +def _redact_value(value: Any) -> Any: + if isinstance(value, dict): + return { + k: (_CODEX_WIRE_REDACTED if _should_redact_key(str(k)) else _redact_value(v)) + for k, v in value.items() + } + if isinstance(value, list): + return [_redact_value(item) for item in value] + return value + + +def redact_for_wire_debug(value: Any) -> Any: + """Redact obvious secrets while preserving request/response shape.""" + + return _redact_value(value) + + +def _safe_event_name(event: str) -> str: + return "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in event)[:80] + + +def capture_codex_wire_debug( + event: str, + *, + request_id: str | None = None, + session_id: str | None = None, + transport: str, + direction: str, + method: str | None = None, + url: str | None = None, + headers: dict[str, Any] | None = None, + body: Any = None, + raw_text: str | None = None, + status_code: int | None = None, + metadata: dict[str, Any] | None = None, +) -> Path | None: + """Write an opt-in redacted Codex wire snapshot to disk. + + This is intentionally file-based rather than log-based: real Codex + requests can be large, and operators need the exact envelope shape without + mixing it into normal proxy logs. Header/body secret-looking keys are + redacted, but request content is otherwise preserved because this mode is + explicitly for local debugging. + """ + + if not codex_wire_debug_enabled(): + return None + + try: + out_dir = _codex_wire_debug_dir() + out_dir.mkdir(parents=True, exist_ok=True) + ts_ns = time.time_ns() + req = request_id or "no_request" + safe_req = _safe_event_name(req) + safe_event = _safe_event_name(event) + path = out_dir / f"{ts_ns}_{safe_req}_{safe_event}.json" + payload = { + "event": event, + "timestamp_ns": ts_ns, + "request_id": request_id, + "session_id": session_id, + "transport": transport, + "direction": direction, + "method": method, + "url": url, + "status_code": status_code, + "headers": redact_for_wire_debug(headers or {}), + "body": redact_for_wire_debug(body), + "raw_text": raw_text, + "metadata": redact_for_wire_debug(metadata or {}), + } + path.write_text( + json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8" + ) + logger.info( + "event=codex_wire_debug_capture path=%s request_id=%s wire_event=%s", + path, + request_id or "", + event, + ) + return path + except Exception as exc: # pragma: no cover - debug path must never break traffic + logger.warning("event=codex_wire_debug_capture_failed error=%s", exc) + return None + + # Memory injection mode (P0-1 fix in PR-A2). # # Values: diff --git a/plugins/headroom-agent-hooks/.claude-plugin/plugin.json b/plugins/headroom-agent-hooks/.claude-plugin/plugin.json index b8418c954..054de909f 100644 --- a/plugins/headroom-agent-hooks/.claude-plugin/plugin.json +++ b/plugins/headroom-agent-hooks/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "headroom", - "version": "0.21.7", + "version": "0.21.10", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", "author": { "name": "Headroom Contributors", diff --git a/plugins/headroom-agent-hooks/.github/plugin/plugin.json b/plugins/headroom-agent-hooks/.github/plugin/plugin.json index 60ee77725..0ca70de0e 100644 --- a/plugins/headroom-agent-hooks/.github/plugin/plugin.json +++ b/plugins/headroom-agent-hooks/.github/plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "headroom", - "version": "0.21.7", + "version": "0.21.10", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", "author": { "name": "Headroom Contributors", diff --git a/tests/test_openai_codex_ws_lifecycle.py b/tests/test_openai_codex_ws_lifecycle.py index ad3bac35f..e13f650e6 100644 --- a/tests/test_openai_codex_ws_lifecycle.py +++ b/tests/test_openai_codex_ws_lifecycle.py @@ -31,8 +31,10 @@ class _DummyMetrics: self.ws_session_durations: list[float] = [] self.stage_timings: list[tuple[str, dict[str, float]]] = [] self.termination_causes: list[str] = [] + self.recorded_requests: list[dict] = [] async def record_request(self, **kwargs): # pragma: no cover + self.recorded_requests.append(dict(kwargs)) return None async def record_stage_timings(self, path: str, timings: dict[str, float]) -> None: @@ -252,6 +254,43 @@ async def test_happy_path_registry_empty_after_response_completed(): } +@pytest.mark.asyncio +async def test_ws_session_metrics_include_response_completed_usage(): + """Codex WS sessions should report real upstream usage, not zero-token sessions.""" + + upstream_events = [ + json.dumps({"type": "response.created", "response": {"id": "r_1"}}), + json.dumps( + { + "type": "response.completed", + "response": { + "id": "r_1", + "usage": { + "input_tokens": 100, + "input_tokens_details": {"cached_tokens": 75}, + "output_tokens": 12, + }, + }, + } + ), + ] + upstream = _FakeUpstream(upstream_events) + fake_ws_mod = _make_fake_websockets_module(upstream) + + client_ws = _FakeWebSocket(frames=[_first_frame()]) + handler = _DummyOpenAIHandler() + + with patch.dict(sys.modules, {"websockets": fake_ws_mod}): + await handler.handle_openai_responses_ws(client_ws) + + assert handler.metrics.recorded_requests + recorded = handler.metrics.recorded_requests[-1] + assert recorded["input_tokens"] == 100 + assert recorded["output_tokens"] == 12 + assert recorded["cache_read_tokens"] == 75 + assert recorded["uncached_input_tokens"] == 25 + + @pytest.mark.asyncio async def test_client_disconnect_cancels_upstream_relay_within_100ms(): """**Failing-test-first** scenario from the plan.