diff --git a/README.md b/README.md
index 94b78e6..08cacc7 100644
--- a/README.md
+++ b/README.md
@@ -111,6 +111,7 @@ account. The current catalog commonly includes:
- OpenAI-compatible `/v1/responses` (HTTP + WebSocket)
- Ollama-compatible endpoints
- Reasoning effort exposed as separate models (optional)
+- Codex usage and account status on `GET /v1/status` (see below)
@@ -158,6 +159,38 @@ All flags go after `chatmock serve`. These can also be set as environment variab
+## Usage and account status
+
+`GET /v1/status` reports the signed-in account and the plan usage Codex reported
+on the most recent proxied request — the same snapshot the `chatmock info`
+command prints. Like the Codex CLI, it is not a live probe: the backend only
+reports usage on replies it sends, so the numbers update after each request
+through the proxy and cost nothing to read.
+
+```json
+{
+ "account": {
+ "name": "Ada Lovelace",
+ "email": "ada@example.com",
+ "plan_type": "plus",
+ "account_id": "1f6f92a2-..."
+ },
+ "rate_limits": {
+ "captured_at": "2026-08-26T08:03:58+00:00",
+ "primary": { "used_percent": 12.5, "window_minutes": 10080, "resets_in_seconds": 345600 },
+ "secondary": { "used_percent": 3.0, "window_minutes": 300, "resets_in_seconds": 1799 }
+ }
+}
+```
+
+The account values are display metadata derived from the local token claims —
+no token is ever included. A field whose claim is missing is omitted; `account`
+is `null` when signed out, and `rate_limits` is `null` until a first request
+has been proxied. Do not assume `primary` is the shorter window; classify by
+`window_minutes`.
+
+
+
## Important notice
Use responsibly and at your own risk. This project is not affiliated with OpenAI.
diff --git a/chatmock/codex_status.py b/chatmock/codex_status.py
new file mode 100644
index 0000000..f5cd40b
--- /dev/null
+++ b/chatmock/codex_status.py
@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import os
+from typing import Any, Dict
+
+from .limits import StoredRateLimitSnapshot, load_rate_limit_snapshot
+from .utils import auth_file_candidates, load_chatgpt_tokens, parse_jwt_claims
+
+_OPENAI_AUTH_CLAIM = "https://api.openai.com/auth"
+
+# The account claims only change when the auth file does, so the derived
+# info is cached against its mtime/size: a request pays one os.stat per
+# candidate path, not a read + JSON parse + JWT decode.
+_CACHE_UNSET: Any = object()
+_account_cache_key: Any = _CACHE_UNSET
+_account_cache: Dict[str, str] = {}
+
+
+def _auth_files_state() -> tuple:
+ state = []
+ for path in auth_file_candidates():
+ try:
+ st = os.stat(path)
+ except OSError:
+ continue
+ state.append((path, st.st_mtime_ns, st.st_size))
+ return tuple(state)
+
+
+def reset_account_info_cache() -> None:
+ global _account_cache_key, _account_cache
+ _account_cache_key = _CACHE_UNSET
+ _account_cache = {}
+
+
+def account_info() -> Dict[str, str]:
+ """Display metadata about the signed-in account, derived from token claims.
+
+ Never carries a token, and never a claim set wholesale: only the display
+ values, each omitted when its claim is unavailable.
+ """
+ global _account_cache_key, _account_cache
+ key = _auth_files_state()
+ if key == _account_cache_key:
+ return dict(_account_cache)
+ info = _compute_account_info()
+ _account_cache_key = key
+ _account_cache = info
+ return dict(info)
+
+
+def _display_str(value: Any) -> str | None:
+ if not isinstance(value, str):
+ return None
+ cleaned = value.strip()
+ return cleaned or None
+
+
+def _compute_account_info() -> Dict[str, str]:
+ try:
+ access_token, account_id, id_token = load_chatgpt_tokens(ensure_fresh=False)
+ except Exception:
+ return {}
+ id_claims = parse_jwt_claims(id_token) if isinstance(id_token, str) else None
+ access_claims = parse_jwt_claims(access_token) if isinstance(access_token, str) else None
+ id_claims = id_claims if isinstance(id_claims, dict) else {}
+ access_claims = access_claims if isinstance(access_claims, dict) else {}
+
+ info: Dict[str, str] = {}
+
+ email = _display_str(id_claims.get("email"))
+ name = None
+ for candidate in (id_claims.get("name"), id_claims.get("email"), id_claims.get("preferred_username"), account_id):
+ name = _display_str(candidate)
+ if name:
+ break
+ if name:
+ info["name"] = name
+ if email:
+ info["email"] = email
+
+ auth_claims = access_claims.get(_OPENAI_AUTH_CLAIM)
+ if isinstance(auth_claims, dict):
+ plan = _display_str(auth_claims.get("chatgpt_plan_type"))
+ if plan:
+ info["plan_type"] = plan
+
+ safe_account_id = _display_str(account_id)
+ if safe_account_id:
+ info["account_id"] = safe_account_id
+ return info
+
+
+def _window_payload(window: Any) -> Dict[str, Any] | None:
+ if window is None:
+ return None
+ return {
+ "used_percent": window.used_percent,
+ "window_minutes": window.window_minutes,
+ "resets_in_seconds": window.resets_in_seconds,
+ }
+
+
+def build_status_payload() -> Dict[str, Any]:
+ """The signed-in account and the last usage snapshot Codex reported.
+
+ The usage half mirrors the CLI's `info` command: it is the snapshot
+ recorded from the most recent proxied request, not a live probe — the
+ Codex backend only reports usage on replies it sends.
+ """
+ account = account_info() or None
+
+ stored: StoredRateLimitSnapshot | None = load_rate_limit_snapshot()
+ rate_limits: Dict[str, Any] | None = None
+ if stored is not None:
+ rate_limits = {
+ "captured_at": stored.captured_at.isoformat(),
+ "primary": _window_payload(stored.snapshot.primary),
+ "secondary": _window_payload(stored.snapshot.secondary),
+ }
+
+ return {"account": account, "rate_limits": rate_limits}
diff --git a/chatmock/reasoning.py b/chatmock/reasoning.py
index 37c276c..162692b 100644
--- a/chatmock/reasoning.py
+++ b/chatmock/reasoning.py
@@ -18,14 +18,20 @@ def build_reasoning_param(
valid_efforts = allowed_efforts or DEFAULT_REASONING_EFFORTS
valid_summaries = {"auto", "concise", "detailed", "none"}
+ # A caller's explicit effort is forwarded when upstream knows it at all;
+ # upstream validates per model. Filtering against the catalog here silently
+ # rewrote efforts the catalog omits but upstream honours (e.g. `none`).
+ explicit_effort: str | None = None
if isinstance(overrides, dict):
o_eff = str(overrides.get("effort", "")).strip().lower()
o_sum = str(overrides.get("summary", "")).strip().lower()
- if o_eff in valid_efforts and o_eff:
- effort = o_eff
+ if o_eff in DEFAULT_REASONING_EFFORTS:
+ explicit_effort = o_eff
if o_sum in valid_summaries and o_sum:
summary = o_sum
- if effort not in valid_efforts:
+ if explicit_effort is not None:
+ effort = explicit_effort
+ elif effort not in valid_efforts:
effort = "medium"
if summary not in valid_summaries:
summary = "auto"
diff --git a/chatmock/responses_api.py b/chatmock/responses_api.py
index ab66803..bbc28be 100644
--- a/chatmock/responses_api.py
+++ b/chatmock/responses_api.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import copy
import json
from dataclasses import dataclass
from typing import Any, Dict, Iterable, Iterator, List
@@ -171,6 +172,8 @@ def aggregate_response_from_sse(
) -> tuple[Dict[str, Any] | None, Dict[str, Any] | None]:
response_obj: Dict[str, Any] | None = None
error_obj: Dict[str, Any] | None = None
+ completed_output_items: Dict[int, Dict[str, Any]] = {}
+ unindexed_output_items = 0
try:
for evt in iter_sse_event_payloads(upstream):
if callable(on_event):
@@ -182,6 +185,17 @@ def aggregate_response_from_sse(
if isinstance(response, dict):
response_obj = response
kind = evt.get("type")
+ if kind == "response.output_item.done":
+ item = evt.get("item")
+ if isinstance(item, dict):
+ output_index = evt.get("output_index")
+ if not isinstance(output_index, int):
+ # Indexed items are the protocol norm. Keep malformed or
+ # older unindexed events deterministically after them,
+ # preserving their arrival order.
+ output_index = 1_000_000 + unindexed_output_items
+ unindexed_output_items += 1
+ completed_output_items[output_index] = copy.deepcopy(item)
if kind == "response.failed":
if isinstance(response, dict) and isinstance(response.get("error"), dict):
error_obj = {"error": response.get("error")}
@@ -189,6 +203,16 @@ def aggregate_response_from_sse(
error_obj = {"error": {"message": "response.failed"}}
break
if kind == "response.completed":
+ if (
+ isinstance(response_obj, dict)
+ and completed_output_items
+ and not response_obj.get("output")
+ ):
+ response_obj = dict(response_obj)
+ response_obj["output"] = [
+ completed_output_items[index]
+ for index in sorted(completed_output_items)
+ ]
break
finally:
upstream.close()
diff --git a/chatmock/routes_openai.py b/chatmock/routes_openai.py
index 673e22f..ec0bbe4 100644
--- a/chatmock/routes_openai.py
+++ b/chatmock/routes_openai.py
@@ -6,6 +6,7 @@
from flask import Blueprint, Response, current_app, jsonify, make_response, request
+from .codex_status import build_status_payload
from .fast_mode import resolve_service_tier
from .limits import record_rate_limits_from_response
from .http import build_cors_headers
@@ -717,6 +718,14 @@ def responses_create() -> Response:
return resp
+@openai_bp.route("/v1/status", methods=["GET"])
+def codex_status() -> Response:
+ resp = make_response(jsonify(build_status_payload()), 200)
+ for k, v in build_cors_headers().items():
+ resp.headers.setdefault(k, v)
+ return resp
+
+
@openai_bp.route("/v1/models", methods=["GET"])
def list_models() -> Response:
expose_variants = bool(current_app.config.get("EXPOSE_REASONING_MODELS"))
diff --git a/chatmock/utils.py b/chatmock/utils.py
index 96dd314..1fe899f 100644
--- a/chatmock/utils.py
+++ b/chatmock/utils.py
@@ -53,16 +53,19 @@ def resolve_installation_id() -> str:
return str(uuid.uuid4())
-def read_auth_file() -> Dict[str, Any] | None:
- for base in [
+def auth_file_candidates() -> List[str]:
+ """The auth.json paths considered, in read priority order."""
+ bases = (
os.getenv("CHATGPT_LOCAL_HOME"),
os.getenv("CODEX_HOME"),
os.path.expanduser("~/.chatgpt-local"),
os.path.expanduser("~/.codex"),
- ]:
- if not base:
- continue
- path = os.path.join(base, "auth.json")
+ )
+ return [os.path.join(base, "auth.json") for base in bases if base]
+
+
+def read_auth_file() -> Dict[str, Any] | None:
+ for path in auth_file_candidates():
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
diff --git a/tests/test_routes.py b/tests/test_routes.py
index a490670..05059bd 100644
--- a/tests/test_routes.py
+++ b/tests/test_routes.py
@@ -1,13 +1,16 @@
from __future__ import annotations
import json
+import os
import socket
+import tempfile
import threading
import time
import unittest
from unittest.mock import patch
from chatmock.app import create_app
+from chatmock.codex_status import reset_account_info_cache
from chatmock.session import reset_session_state
from websockets.sync.client import connect as ws_connect
@@ -260,6 +263,108 @@ def test_responses_route_returns_completed_response_object(self, mock_start) ->
self.assertEqual(outbound_payload["reasoning"]["effort"], "medium")
self.assertIsInstance(outbound_payload["prompt_cache_key"], str)
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_responses_route_reconstructs_non_stream_output_from_item_events(self, mock_start) -> None:
+ output = [
+ {
+ "type": "reasoning",
+ "id": "reasoning_1",
+ "summary": [{"type": "summary_text", "text": "Need the tool."}],
+ "encrypted_content": "encrypted",
+ },
+ {
+ "type": "function_call",
+ "id": "fc_1",
+ "call_id": "call_1",
+ "name": "get_time",
+ "arguments": '{"city":"Paris"}',
+ "status": "completed",
+ },
+ {
+ "type": "message",
+ "role": "assistant",
+ "id": "msg_1",
+ "status": "completed",
+ "content": [{"type": "output_text", "text": '{"city":"Paris"}'}],
+ },
+ ]
+ events = [
+ {
+ "type": "response.created",
+ "response": {"id": "resp_items", "object": "response", "status": "in_progress"},
+ },
+ *[
+ {"type": "response.output_item.done", "output_index": index, "item": output[index]}
+ # Completion order is not output order; the protocol supplies
+ # output_index so non-stream aggregation can reconstruct it.
+ for index in (2, 0, 1)
+ ],
+ {
+ "type": "response.completed",
+ "response": {
+ "id": "resp_items",
+ "object": "response",
+ "status": "completed",
+ "output": [],
+ },
+ },
+ ]
+ mock_start.return_value = (
+ FakeUpstream(events, headers={"Content-Type": "text/event-stream"}),
+ None,
+ )
+
+ response = self.client.post(
+ "/v1/responses",
+ json={"model": "gpt-5.6-luna", "input": "Return structured output."},
+ )
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.get_json()["output"], output)
+
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_responses_route_keeps_output_from_completed_response(self, mock_start) -> None:
+ authoritative = {
+ "type": "message",
+ "role": "assistant",
+ "id": "msg_final",
+ "content": [{"type": "output_text", "text": "final"}],
+ }
+ mock_start.return_value = (
+ FakeUpstream(
+ [
+ {
+ "type": "response.output_item.done",
+ "item": {
+ "type": "message",
+ "role": "assistant",
+ "id": "msg_event",
+ "content": [{"type": "output_text", "text": "event"}],
+ },
+ },
+ {
+ "type": "response.completed",
+ "response": {
+ "id": "resp_final",
+ "object": "response",
+ "status": "completed",
+ "output": [authoritative],
+ },
+ },
+ ],
+ headers={"Content-Type": "text/event-stream"},
+ ),
+ None,
+ )
+
+ response = self.client.post(
+ "/v1/responses",
+ json={"model": "gpt-5.6-luna", "input": "hello"},
+ )
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.get_json()["output"], [authoritative])
+
@patch("chatmock.routes_openai.start_upstream_raw_request")
def test_responses_route_honors_debug_model_override(self, mock_start) -> None:
app = create_app(debug_model="gpt-5.4", model_sync=False)
@@ -660,5 +765,235 @@ def close(self) -> None:
)
+def _unsigned_jwt(claims: dict[str, object]) -> str:
+ import base64
+
+ def _b64(obj: dict[str, object]) -> str:
+ raw = json.dumps(obj, separators=(",", ":")).encode("utf-8")
+ return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+ return f"{_b64({'alg': 'none', 'typ': 'JWT'})}.{_b64(claims)}.sig"
+
+
+ACCESS_TOKEN = "access-token-secret-do-not-leak"
+ID_TOKEN_CLAIMS = {
+ "email": "athlete@example.com",
+ "name": "Athlete Example",
+ "preferred_username": "athlete",
+ "https://api.openai.com/auth": {"chatgpt_account_id": "acct_0123456789"},
+}
+ACCESS_TOKEN_CLAIMS = {"https://api.openai.com/auth": {"chatgpt_plan_type": "pro"}}
+
+USAGE_HEADERS = {
+ "x-codex-primary-used-percent": "12.5",
+ "x-codex-primary-window-minutes": "10080",
+ "x-codex-primary-reset-after-seconds": "345600",
+ "x-codex-secondary-used-percent": "3",
+ "x-codex-secondary-window-minutes": "300",
+ "x-codex-secondary-reset-after-seconds": "1799",
+}
+
+COMPLETED_EVENTS = [
+ {"type": "response.created", "response": {"id": "resp_1", "object": "response", "status": "in_progress"}},
+ {
+ "type": "response.output_item.done",
+ "output_index": 0,
+ "item": {"id": "msg_1", "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "ok"}]},
+ },
+ {"type": "response.completed", "response": {"id": "resp_1", "object": "response", "status": "completed", "output": []}},
+]
+
+
+class CodexStatusTests(unittest.TestCase):
+ """GET /v1/status reports the signed-in account and the last usage snapshot."""
+
+ def setUp(self) -> None:
+ reset_session_state()
+ reset_account_info_cache()
+ home = tempfile.TemporaryDirectory()
+ self.addCleanup(home.cleanup)
+ env = patch.dict(os.environ, {"CHATGPT_LOCAL_HOME": home.name})
+ env.start()
+ self.addCleanup(env.stop)
+ self.app = create_app(model_sync=False, verbose=True)
+ self.client = self.app.test_client()
+
+ def _signed_in(self):
+ id_token = _unsigned_jwt(ID_TOKEN_CLAIMS)
+ access_token = ACCESS_TOKEN + "." + _unsigned_jwt(ACCESS_TOKEN_CLAIMS).split(".", 1)[1]
+ return patch(
+ "chatmock.codex_status.load_chatgpt_tokens",
+ return_value=(access_token, "acct_0123456789", id_token),
+ )
+
+ @patch("chatmock.codex_status.load_chatgpt_tokens", return_value=(None, None, None))
+ def test_status_before_any_request_reports_nothing(self, _tokens) -> None:
+ response = self.client.get("/v1/status")
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.get_json(), {"account": None, "rate_limits": None})
+ self.assertEqual(response.headers.get("Access-Control-Allow-Origin"), "*")
+
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_responses_call_feeds_the_status_snapshot(self, mock_start) -> None:
+ mock_start.return_value = (
+ FakeUpstream(COMPLETED_EVENTS, headers={"Content-Type": "text/event-stream", **USAGE_HEADERS}),
+ None,
+ )
+ with self._signed_in():
+ response = self.client.post("/v1/responses", json={"model": "gpt-5.4", "input": "hello"})
+ self.assertEqual(response.status_code, 200)
+ # Nothing rides on the reply itself; the data is served by /v1/status.
+ for name in USAGE_HEADERS:
+ self.assertNotIn(name, response.headers)
+ status = self.client.get("/v1/status").get_json()
+ self.assertEqual(
+ status["account"],
+ {
+ "name": "Athlete Example",
+ "email": "athlete@example.com",
+ "plan_type": "pro",
+ "account_id": "acct_0123456789",
+ },
+ )
+ self.assertEqual(
+ status["rate_limits"]["primary"],
+ {"used_percent": 12.5, "window_minutes": 10080, "resets_in_seconds": 345600},
+ )
+ self.assertEqual(
+ status["rate_limits"]["secondary"],
+ {"used_percent": 3.0, "window_minutes": 300, "resets_in_seconds": 1799},
+ )
+ self.assertTrue(status["rate_limits"]["captured_at"])
+
+ @patch("chatmock.codex_status.load_chatgpt_tokens", return_value=(None, None, None))
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_partial_usage_is_stored_without_invention(self, mock_start, _tokens) -> None:
+ partial = {
+ "x-codex-primary-used-percent": "0",
+ "x-codex-primary-window-minutes": "10080",
+ "x-codex-primary-reset-after-seconds": "600",
+ }
+ mock_start.return_value = (
+ FakeUpstream(COMPLETED_EVENTS, headers={"Content-Type": "text/event-stream", **partial}),
+ None,
+ )
+ self.client.post("/v1/responses", json={"model": "gpt-5.4", "input": "hello"})
+ status = self.client.get("/v1/status").get_json()
+ self.assertEqual(
+ status["rate_limits"]["primary"],
+ {"used_percent": 0.0, "window_minutes": 10080, "resets_in_seconds": 600},
+ )
+ self.assertIsNone(status["rate_limits"]["secondary"])
+
+ @patch("chatmock.codex_status.load_chatgpt_tokens", return_value=(None, None, None))
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_usage_is_recorded_even_on_upstream_errors(self, mock_start, _tokens) -> None:
+ mock_start.return_value = (
+ FakeUpstream(
+ status_code=429,
+ headers={"Content-Type": "application/json", **USAGE_HEADERS},
+ content=json.dumps({"error": {"message": "usage limit reached"}}).encode("utf-8"),
+ text="usage limit reached",
+ ),
+ None,
+ )
+ response = self.client.post("/v1/responses", json={"model": "gpt-5.4", "input": "hello"})
+ self.assertEqual(response.status_code, 429)
+ status = self.client.get("/v1/status").get_json()
+ self.assertEqual(status["rate_limits"]["primary"]["used_percent"], 12.5)
+
+ def test_account_falls_back_through_email_and_username(self) -> None:
+ no_name = {k: v for k, v in ID_TOKEN_CLAIMS.items() if k != "name"}
+ with patch(
+ "chatmock.codex_status.load_chatgpt_tokens",
+ return_value=("a.b.c", "acct_0123456789", _unsigned_jwt(no_name)),
+ ):
+ account = self.client.get("/v1/status").get_json()["account"]
+ self.assertEqual(account["name"], "athlete@example.com")
+ # No plan claim on that access token: the plan stays unknown rather than defaulting.
+ self.assertNotIn("plan_type", account)
+
+ reset_account_info_cache()
+ username_only = {"preferred_username": "athlete"}
+ with patch(
+ "chatmock.codex_status.load_chatgpt_tokens",
+ return_value=(None, None, _unsigned_jwt(username_only)),
+ ):
+ account = self.client.get("/v1/status").get_json()["account"]
+ self.assertEqual(account, {"name": "athlete"})
+
+ def test_account_info_is_derived_once_not_per_request(self) -> None:
+ with self._signed_in() as loader:
+ first = self.client.get("/v1/status").get_json()
+ second = self.client.get("/v1/status").get_json()
+ self.assertEqual(first["account"]["name"], "Athlete Example")
+ self.assertEqual(second, first)
+ # The auth file was parsed for the first request only; while it is
+ # unchanged on disk, later requests reuse the derived info.
+ self.assertEqual(loader.call_count, 1)
+
+ @patch("chatmock.codex_status.load_chatgpt_tokens", return_value=(None, None, None))
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_upstream_request_keeps_model_effort_stream_and_store(self, mock_start, _tokens) -> None:
+ mock_start.return_value = (
+ FakeUpstream(
+ headers={"Content-Type": "text/event-stream"},
+ content=b'data: {"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[]}}\n\n',
+ ),
+ None,
+ )
+ # The static catalog lists no `none` for gpt-5.6-luna; upstream accepts it (issue #116).
+ response = self.client.post(
+ "/v1/responses",
+ json={
+ "model": "gpt-5.6-luna",
+ "input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "hi"}]}],
+ "tools": [],
+ "tool_choice": "auto",
+ "parallel_tool_calls": False,
+ "store": False,
+ "stream": True,
+ "reasoning": {"effort": "none"},
+ },
+ )
+ self.assertEqual(response.status_code, 200)
+ sent = mock_start.call_args.args[0]
+ self.assertEqual(sent["model"], "gpt-5.6-luna")
+ self.assertEqual(sent["reasoning"]["effort"], "none")
+ self.assertIs(sent["stream"], True)
+ self.assertIs(sent["store"], False)
+ self.assertEqual(sent["tools"], [])
+ self.assertEqual(sent["tool_choice"], "auto")
+ self.assertIs(sent["parallel_tool_calls"], False)
+ self.assertEqual(
+ sent["input"],
+ [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "hi"}]}],
+ )
+
+ @patch("chatmock.routes_openai.start_upstream_raw_request")
+ def test_no_token_reaches_the_status_body_headers_or_logs(self, mock_start) -> None:
+ import contextlib
+ import io
+
+ mock_start.return_value = (
+ FakeUpstream(COMPLETED_EVENTS, headers={"Content-Type": "text/event-stream", **USAGE_HEADERS}),
+ None,
+ )
+ captured = io.StringIO()
+ with self._signed_in(), contextlib.redirect_stdout(captured):
+ self.client.post("/v1/responses", json={"model": "gpt-5.4", "input": "hello"})
+ response = self.client.get("/v1/status")
+ self.assertEqual(response.status_code, 200)
+ id_token = _unsigned_jwt(ID_TOKEN_CLAIMS)
+ secrets = (ACCESS_TOKEN, id_token, id_token.split(".")[1], "Bearer ")
+ body = response.get_data(as_text=True)
+ header_blob = "\n".join(f"{k}: {v}" for k, v in response.headers.items())
+ logs = captured.getvalue()
+ for secret in secrets:
+ self.assertNotIn(secret, body)
+ self.assertNotIn(secret, header_blob)
+ self.assertNotIn(secret, logs)
+
+
if __name__ == "__main__":
unittest.main()