diff --git a/Dockerfile b/Dockerfile index b2c2bcc908..d24444e2f4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -44,7 +44,7 @@ WORKDIR /app RUN apt-get update \ && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends --only-upgrade \ bsdutils libblkid1 libc-bin libc6 libcap2 libmount1 libsmartcols1 libssl3t64 \ - libsystemd0 libudev1 libuuid1 openssl sed util-linux \ + libsystemd0 libudev1 libuuid1 openssl perl-base sed util-linux \ && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ openssl-provider-legacy \ && rm -rf /var/lib/apt/lists/* diff --git a/app/core/clients/codex_continuation.py b/app/core/clients/codex_continuation.py index 307f817f3d..437bd1aca6 100644 --- a/app/core/clients/codex_continuation.py +++ b/app/core/clients/codex_continuation.py @@ -731,6 +731,10 @@ def _reconstruct_terminal( ) -> dict[str, Any]: terminal_response = _response_payload(terminal or {}) response = dict(base_response or terminal_response) + if "service_tier" in terminal_response: + response["service_tier"] = terminal_response["service_tier"] + else: + response.pop("service_tier", None) response["output"] = output_items response["usage"] = usage response["status"] = terminal_response.get("status", "completed") @@ -754,6 +758,7 @@ def _synthetic_incomplete( billed_usage: dict[str, Any] | None, ) -> dict[str, Any]: response = dict(base_response or {}) + response.pop("service_tier", None) response["output"] = output_items response["usage"] = usage response["status"] = "incomplete" diff --git a/app/core/request_locality.py b/app/core/request_locality.py index a1e1f53d69..e21197543e 100644 --- a/app/core/request_locality.py +++ b/app/core/request_locality.py @@ -36,6 +36,7 @@ def _combined_chain_header(headers: Mapping[str, str], header_name: str) -> str return ",".join(values) if values else None return headers.get(header_name) + def _has_non_empty_header_value( headers: Mapping[str, str], header_name: str, diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index c3a834fb59..6785299014 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -837,10 +837,11 @@ async def _process_http_bridge_upstream_text( matched_request_state.latency_response_created_ms = int( max(0.0, now - matched_request_state.started_at) * 1000 ) - actual_service_tier = _service_tier_from_event_payload(payload) - if actual_service_tier is not None: - matched_request_state.actual_service_tier = actual_service_tier - matched_request_state.service_tier = actual_service_tier + # Preserve any-event tier observations for billing. The shared + # terminal finalizer alone assigns actual-tier evidence. + observed_service_tier = _service_tier_from_event_payload(payload) + if observed_service_tier is not None: + matched_request_state.service_tier = observed_service_tier _record_http_bridge_tool_call_lifecycle( matched_request_state, event_type=event_type, diff --git a/app/modules/proxy/_service/observability.py b/app/modules/proxy/_service/observability.py index 46e32d2971..e0de89776c 100644 --- a/app/modules/proxy/_service/observability.py +++ b/app/modules/proxy/_service/observability.py @@ -152,6 +152,7 @@ def _maybe_log_proxy_service_tier_trace( kind, requested_service_tier=requested_service_tier, actual_service_tier=actual_service_tier, + status=status, ) settings = _service_get_settings() if not settings.log_proxy_service_tier_trace: @@ -176,7 +177,10 @@ def _record_service_tier_mismatch( *, requested_service_tier: str | None, actual_service_tier: str | None, + status: str | None, ) -> None: + if status != "success": + return requested_tier = _service_tier_metric_label(requested_service_tier) actual_tier = _service_tier_metric_label(actual_service_tier) if requested_tier == "none" or actual_tier == "none" or requested_tier == actual_tier: diff --git a/app/modules/proxy/_service/streaming/mixin.py b/app/modules/proxy/_service/streaming/mixin.py index c8a498ec85..3a297228bb 100644 --- a/app/modules/proxy/_service/streaming/mixin.py +++ b/app/modules/proxy/_service/streaming/mixin.py @@ -598,8 +598,9 @@ async def _record_or_defer_owner_recovery_health(error: UpstreamError, code: str ) event_service_tier = _facade()._service_tier_from_event_payload(first_payload) if event_service_tier is not None: - actual_service_tier = event_service_tier service_tier = event_service_tier + if event_type == "response.completed": + actual_service_tier = event_service_tier if event and event.response and event.response.id: response_id = event.response.id settlement.response_id = response_id @@ -765,8 +766,9 @@ async def _record_or_defer_owner_recovery_health(error: UpstreamError, code: str ) event_service_tier = _facade()._service_tier_from_event_payload(event_payload) if event_service_tier is not None: - actual_service_tier = event_service_tier service_tier = event_service_tier + if event_type == "response.completed": + actual_service_tier = event_service_tier line, event_payload, event, event_type = rewrite_parallel_tool_call_sse_line(line, event_payload) if event_type in _facade()._TEXT_DELTA_EVENT_TYPES: saw_text_delta = True @@ -986,6 +988,8 @@ async def _record_or_defer_owner_recovery_health(error: UpstreamError, code: str output_tokens = usage_accounting.output_tokens cached_input_tokens = usage_accounting.cached_input_tokens reasoning_tokens = usage_accounting.reasoning_tokens + if status != "success": + actual_service_tier = None settlement.status = status settlement.model = model settlement.service_tier = service_tier diff --git a/app/modules/proxy/_service/websocket/mixin.py b/app/modules/proxy/_service/websocket/mixin.py index 31a8bcb501..2589536424 100644 --- a/app/modules/proxy/_service/websocket/mixin.py +++ b/app/modules/proxy/_service/websocket/mixin.py @@ -3784,10 +3784,11 @@ async def _process_upstream_websocket_text( ) if event_type not in {"response.completed", "response.failed", "response.incomplete", "error"}: _record_response_event(request_state, event_type) - actual_service_tier = _facade()._service_tier_from_event_payload(payload) - if actual_service_tier is not None: - request_state.actual_service_tier = actual_service_tier - request_state.service_tier = actual_service_tier + # Any upstream event may refine billing, but only the final + # response.completed event supplies actual-tier evidence. + observed_service_tier = _facade()._service_tier_from_event_payload(payload) + if observed_service_tier is not None: + request_state.service_tier = observed_service_tier completed_tool_call = _facade()._response_output_item_done_tool_call(payload) if completed_tool_call is not None: completed_call_id, completed_call_type = completed_tool_call @@ -4007,6 +4008,8 @@ async def _process_upstream_websocket_text( # to this pending request (and is then folded/suppressed). fold_request_state.awaiting_response_created = True fold_request_state.response_id = None + # A swallowed round cannot provide final actual-tier evidence. + fold_request_state.actual_service_tier = None upstream_control.continuation_resend_body = fold_outcome.continuation_request upstream_control.suppress_downstream_event = True return text @@ -4580,6 +4583,8 @@ async def _finalize_websocket_request_state( error_payload: UpstreamError | None = None response_id = request_state.response_id or request_state.request_id response_service_tier = request_state.service_tier + # Clear earlier attempts and in-flight events before every terminal path. + request_state.actual_service_tier = None if request_state.draining_until_terminal: await _release_websocket_response_create_gate(request_state, response_create_gate) @@ -4622,10 +4627,11 @@ async def _finalize_websocket_request_state( billed_usage_payload = _proxy_billed_usage_from_event_payload(payload) usage_accounting = _stream_usage_accounting(usage, billed_usage_payload) - actual_service_tier = _facade()._service_tier_from_event_payload(payload) - if actual_service_tier is not None: - request_state.actual_service_tier = actual_service_tier - response_service_tier = actual_service_tier + observed_service_tier = _facade()._service_tier_from_event_payload(payload) + if observed_service_tier is not None: + response_service_tier = observed_service_tier + if event_type == "response.completed": + request_state.actual_service_tier = observed_service_tier settlement = _StreamSettlement( status=status, @@ -4765,6 +4771,7 @@ async def _write_websocket_connect_failure( ) -> None: proxy = cast(_WebSocketServiceProtocol, self) _ = proxy + request_state.actual_service_tier = None if request_state.skip_request_log: return await proxy._write_request_log( @@ -4950,6 +4957,7 @@ async def _fail_pending_websocket_requests( last_index = len(remaining) - 1 for index, request_state in enumerate(remaining): + request_state.actual_service_tier = None proxy._cancel_request_state_api_key_reservation_heartbeat(request_state) request_error_code = request_state.error_code_override or error_code request_error_message = request_state.error_message_override or error_message diff --git a/openspec/changes/fix-terminal-service-tier-provenance/.openspec.yaml b/openspec/changes/fix-terminal-service-tier-provenance/.openspec.yaml new file mode 100644 index 0000000000..a40cb63c10 --- /dev/null +++ b/openspec/changes/fix-terminal-service-tier-provenance/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-09-14 diff --git a/openspec/changes/fix-terminal-service-tier-provenance/design.md b/openspec/changes/fix-terminal-service-tier-provenance/design.md new file mode 100644 index 0000000000..298e4e4b3d --- /dev/null +++ b/openspec/changes/fix-terminal-service-tier-provenance/design.md @@ -0,0 +1,19 @@ +## Context + +The initial created response is useful for stable IDs and output reconstruction, but its service tier is not a final processing report. Continuation currently copies that initial field into the terminal response. Direct collection and WebSocket finalization then treat the copied value as actual. Separately, any-event collection retains created-only tiers when completion omits the field, and a finally-block counter can count unsuccessful attempts. + +## Decisions + +1. The final upstream terminal owns its service-tier field. Reconstruction copies that field when present and removes any inherited field when absent. Synthetic incomplete output removes created tier claims. +2. Streaming actual-tier evidence requires an explicit nonblank tier on `response.completed`. Failed, incomplete, cancelled and interrupted attempts do not gain evidence from earlier events. Hidden-round completions do not supply the final logical request's actual tier. +3. Billable `service_tier` remains a separate fallback. Existing observed-event/request fallback is retained; explicit final terminal data takes precedence. This repair does not redesign multi-round pricing or historical accounting. +4. The existing mismatch counter is gated on successful completion and actual terminal evidence. Compact success is terminal by definition. Counter names, labels and transport coverage stay unchanged. +5. An explicit terminal `auto` is faithfully recorded but remains inconclusive about upstream scheduling. Neither a zero counter nor an absent tier proves Fast delivery. + +## Verification and rollout + +Use synthetic created, terminal, hidden-round, omitted-tier and interrupted-stream fixtures through folding and service/route settlement. Keep useful existing accounting assertions. Run all checks only after worker edits settle. Publish to ericjuta/codex-lb main through a reviewed PR, then use the existing direct deployment script with unchanged configuration. Verify image/source identity, readiness and safe error counts. Inspect grouped post-deployment metadata from ordinary traffic only, without prompts, response bodies, account identities or model benchmarks. + +## Risks and rollback + +New logs will have null actual tiers for created-only observations. Existing rows are not rewritten and must be separated from post-deployment observations by time. The deployment restarts the local proxy and interrupts in-flight connections. Keep the previous image available for rollback; do not change continuation or transport settings to make the evidence look better. diff --git a/openspec/changes/fix-terminal-service-tier-provenance/proposal.md b/openspec/changes/fix-terminal-service-tier-provenance/proposal.md new file mode 100644 index 0000000000..40331f9be6 --- /dev/null +++ b/openspec/changes/fix-terminal-service-tier-provenance/proposal.md @@ -0,0 +1,22 @@ +## Why + +Continuation reconstruction can replace a terminal response's explicit service tier with the tier from the first `response.created`. Request logs and mismatch counters can therefore report `auto` even when the terminal reports `priority` or `default`. Created-only tiers and failed attempts must not be presented as successful terminal evidence. + +## What Changes + +- Preserve the final upstream terminal's service-tier field through continuation reconstruction, including its absence. +- Record streaming `actual_service_tier` only from an explicit successful completion, across direct streaming, native WebSocket, and the HTTP bridge. +- Keep the existing accounting-tier fallback separate from terminal evidence. +- Count service-tier mismatches only for successful completions with an explicit reported tier. Keep existing counter coverage and bounded labels. +- Verify the repaired response and request-log behavior with synthetic upstream events, then deploy and inspect aggregate normal-traffic metadata. + +## Capabilities + +### Modified Capabilities + +- `responses-api-compat`: terminal tier preservation and actual-versus-accounting request-log semantics. +- `proxy-runtime-observability`: successful-terminal mismatch evidence and missing/auto interpretation. + +## Impact + +The change affects continuation reconstruction and proxy tier reporting. It adds no database columns or runtime settings, does not rewrite historical request logs, and does not alter request tiers, reasoning effort, transport, account selection, or continuation limits. Repairing reporting does not itself prove Fast delivery or improve model latency. diff --git a/openspec/changes/fix-terminal-service-tier-provenance/specs/proxy-runtime-observability/spec.md b/openspec/changes/fix-terminal-service-tier-provenance/specs/proxy-runtime-observability/spec.md new file mode 100644 index 0000000000..c95c9b9504 --- /dev/null +++ b/openspec/changes/fix-terminal-service-tier-provenance/specs/proxy-runtime-observability/spec.md @@ -0,0 +1,22 @@ +## ADDED Requirements + +### Requirement: Service-tier mismatches use successful terminal evidence +The service-tier mismatch counter MUST count only successful requests with an explicit nonblank terminal-reported tier that differs from the normalized requested tier. A created or in-progress tier, requested-tier fallback, missing terminal tier, or unsuccessful attempt MUST NOT increment the counter. The change MUST preserve existing counter coverage and bounded labels rather than adding identities or payload data. + +An explicitly returned `auto` MUST remain observable as the reported tier. Operator guidance MUST describe it as inconclusive for Fast delivery rather than proof of a downgrade. Counter equality or absence MUST NOT be described as evidence of upstream scheduling. + +#### Scenario: Explicit terminal default differs from requested priority +- **WHEN** a successful counted request asks for `priority` and completes with explicit `default` +- **THEN** one mismatch is counted with requested `priority` and actual `default` + +#### Scenario: Created auto followed by terminal priority is not a mismatch +- **WHEN** a successful counted request asks for `priority`, starts with created `auto`, and completes with explicit `priority` +- **THEN** the created tier does not increment the mismatch counter + +#### Scenario: Missing terminal tier does not count a created fallback +- **WHEN** a counted request starts with a tier different from the requested tier but completes without a tier +- **THEN** the request does not increment the mismatch counter + +#### Scenario: Failure does not count a tier mismatch +- **WHEN** a counted attempt fails, is cancelled, disconnects, or finishes incomplete +- **THEN** it does not increment the mismatch counter even if an earlier event included a different tier diff --git a/openspec/changes/fix-terminal-service-tier-provenance/specs/responses-api-compat/spec.md b/openspec/changes/fix-terminal-service-tier-provenance/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..b7a04975ea --- /dev/null +++ b/openspec/changes/fix-terminal-service-tier-provenance/specs/responses-api-compat/spec.md @@ -0,0 +1,38 @@ +## MODIFIED Requirements + +### Requirement: Request logs persist requested, actual, and billable service tiers separately +For Responses proxy traffic, the system MUST persist the operator-requested tier, the upstream-reported actual tier when available, and the effective billable tier used for pricing as separate request-log fields. + +The legacy `fast` alias MUST be normalized to `priority` before forwarding and before it is stored as the requested tier. For streaming requests, `actual_service_tier` MUST come only from a nonblank service tier explicitly present in `response.completed.response`. Created, in-progress, failed, incomplete, cancelled, and disconnected streams MUST NOT supply actual-tier evidence. A completed response that omits the tier MUST leave actual-tier evidence absent even if earlier events supplied one. Non-streaming successful compact responses MAY supply actual-tier evidence directly. + +The system MUST preserve the existing billable-tier fallback independently of actual-tier evidence. An explicit terminal tier MUST take precedence for billing; otherwise the existing observed-event/request fallback remains available without being copied into `actual_service_tier`. + +Continuation reconstruction MUST preserve the final upstream terminal response's `service_tier`, including its absence. It MUST NOT inherit a tier from the initial created response when the terminal omits the field. Proxy-synthesized incomplete responses MUST NOT claim an upstream terminal tier from the created response. + +#### Scenario: Upstream reports a downgraded actual tier +- **WHEN** a client sends a Responses request with `service_tier: "priority"` +- **AND** the upstream completion explicitly reports `service_tier: "default"` +- **THEN** the persisted request log records requested tier `priority`, actual tier `default`, and billable tier `default` + +#### Scenario: Fast alias is logged as a priority request +- **WHEN** a client sends `service_tier: "fast"` +- **AND** the upstream completion explicitly reports `service_tier: "default"` +- **THEN** the persisted request log records requested tier `priority`, actual tier `default`, and billable tier `default` + +#### Scenario: Upstream omits the actual tier +- **WHEN** a client requests `priority` and no upstream event reports a tier +- **THEN** actual tier is null and billable tier falls back to requested `priority` + +#### Scenario: Initial auto does not replace terminal priority +- **WHEN** a continued response starts with created tier `auto` and finishes with explicit terminal tier `priority` +- **THEN** the folded completion and persisted actual tier are `priority` +- **AND** every upstream continuation request retains the requested tier + +#### Scenario: Created-only tier remains accounting fallback +- **WHEN** a created response reports `default` and the completion omits `service_tier` +- **THEN** the completion does not inherit `service_tier` +- **AND** actual tier is null while the existing billable-tier fallback remains `default` + +#### Scenario: Failed or interrupted stream does not prove a terminal tier +- **WHEN** a stream reports a created tier but fails, is cancelled, disconnects, or finishes incomplete +- **THEN** actual-tier evidence remains absent diff --git a/openspec/changes/fix-terminal-service-tier-provenance/tasks.md b/openspec/changes/fix-terminal-service-tier-provenance/tasks.md new file mode 100644 index 0000000000..185b017444 --- /dev/null +++ b/openspec/changes/fix-terminal-service-tier-provenance/tasks.md @@ -0,0 +1,17 @@ +## 1. Implementation + +- [x] 1.1 Preserve terminal service-tier values and omission through continuation folding. +- [x] 1.2 Separate direct-stream terminal evidence from accounting fallback and gate mismatch counters on successful completion. +- [x] 1.3 Apply the same actual-tier contract to native WebSocket and HTTP bridge settlement. + +## 2. Verification + +- [x] 2.1 Prove conflicting, missing, and failed-terminal cases through response and request-log regressions. +- [x] 2.2 Run focused tests, required repository validation, and independent review on the settled tree. +- [ ] 2.3 Synchronize the normative specs and operational context, then validate and archive this change. + +## 3. Release + +- [ ] 3.1 Publish a PR targeting ericjuta/codex-lb main and satisfy current-head CI and review gates. +- [ ] 3.2 Merge and deploy the verified revision without changing runtime settings. +- [ ] 3.3 Verify deployment identity and readiness, then inspect aggregate normal-traffic terminal-tier evidence without model benchmarks or payload capture. diff --git a/openspec/specs/api-keys/spec.md b/openspec/specs/api-keys/spec.md index baf889461e..509470a23b 100644 --- a/openspec/specs/api-keys/spec.md +++ b/openspec/specs/api-keys/spec.md @@ -801,9 +801,7 @@ When API-key authentication and proxy-header trust are disabled, a loopback sock ### Requirement: GPT-5.6 usage cost pricing matches the current published rates -When computing API-key usage, request-log, reservation, or aggregate cost for -the canonical GPT-5.6 models, the system MUST use these USD-per-1M-token rates -for input, cached input, and output: +When computing API-key usage, request-log, reservation, or aggregate cost for the canonical GPT-5.6 models, the system MUST use these USD-per-1M-token rates for input, cached input, and output: | Model | Standard | Fast/priority | Flex | Standard long context | | --- | --- | --- | --- | --- | diff --git a/openspec/specs/proxy-runtime-observability/context.md b/openspec/specs/proxy-runtime-observability/context.md index 932e5af864..5135406b2d 100644 --- a/openspec/specs/proxy-runtime-observability/context.md +++ b/openspec/specs/proxy-runtime-observability/context.md @@ -12,7 +12,7 @@ See `openspec/specs/proxy-runtime-observability/spec.md` for normative requireme - **Request tracing is opt-in:** outbound request summary and payload tracing remain configurable because payload logs can be noisy or sensitive. - **Error logs must be correlated:** request id, endpoint, status, code, and message are the minimum useful fields for debugging 4xx/5xx failures. - **Metrics stay low-cardinality:** Prometheus labels may include bounded values such as route shape, status, transport, error class, bridge outcome, SQLite retry outcome, and service tier. They must not include prompts, bearer tokens, raw request ids, response ids, or auth headers. -- **Service-tier verification is observational:** codex-lb preserves the requested tier and records the upstream actual tier when upstream reports one. A mismatch is visible through request-log snapshots and the service-tier mismatch counter; it does not require changing the operator's tier preference. +- **Service-tier verification is observational:** codex-lb records requested intent, successful explicit terminal-tier evidence, and billable fallback separately. These labels do not independently prove upstream scheduling or the exact egress payload. An explicit terminal `auto` is inconclusive, not proof of a downgrade or priority delivery. The repair does not change requested priority or transport. - **Health and performance are separate reads:** a green container, readiness probe, and clean local error window prove the proxy is locally serving. They do not prove every request is fast, so slow-tail analysis starts with request-log percentiles, output-size/request-shape buckets, upstream failure classes, and requested-versus-actual service tier before changing local runtime settings. ## Operational Notes @@ -21,11 +21,52 @@ See `openspec/specs/proxy-runtime-observability/spec.md` for normative requireme - Prefer summary tracing in normal debugging sessions; enable payload tracing only when the exact normalized outbound request matters. - For direct compact `5xx` failures, look for `proxy_compact_failure` alongside `upstream_request_complete`; together they show the compact failure phase, failure detail, exception type, retry metadata, and affinity source. - For the greenfield sustained baseline, enable `CODEX_LB_METRICS_ENABLED=true` and scrape the metrics port. The minimum alert bundle covers readiness, request success rate, p95 latency, upstream timeout or `stream_incomplete` spikes, SQLite lock pressure, bridge continuity errors, service-tier mismatches, and container restart/OOM signals. -- For service-tier verification, capture a snapshot with `scripts/codex_lb_live_snapshot.py` and inspect `request_logs.service_tier_counts` plus `request_logs.tier_mismatches`. The snapshot summarizes requested and actual tiers without printing prompts or auth headers. +- For service-tier verification, capture a snapshot with `scripts/codex_lb_live_snapshot.py` and inspect `request_logs.service_tier_counts` plus `request_logs.tier_mismatches`. For postrepair records, actual tier means explicit successful terminal evidence, not a billable fallback. The snapshot prints neither prompts nor auth headers. - Treat health and latency as separate signals. A container can be running, restart-free, and answering `/health/ready` plus `/backend-api/codex/health` quickly while request logs still show slow-tail latency, upstream websocket open timeouts, or `stream_incomplete` errors. In that case the proxy is healthy enough to route traffic, but operators should investigate upstream service tier, prompt size, output size, websocket stability, and account/upstream behavior before calling the deployment fully healthy. - `upstream_websocket_open_timeout` storms that hit all accounts uniformly with a diurnal shape (quiet ~03-05 UTC) indicate chatgpt.com edge stalls, not a local fault; verify with direct socket probes against chatgpt.com vs other Cloudflare-fronted hosts. Tuning levers are `CODEX_LB_UPSTREAM_CONNECT_TIMEOUT_SECONDS` (keep above the observed 3-8s SYN-stall band so healthy-but-slow handshakes still land) and `CODEX_LB_PROXY_UPSTREAM_WEBSOCKET_CONNECT_LIMIT` (too low serializes handshakes behind the admission gate during storms). After lowering the connect timeout, watch `continuity_fail_closed` counts; a rise means the timeout is trimming recoverable handshakes. -- When snapshots show requested `ultrafast` with actual upstream `default`, keep reporting it as a tier mismatch observation rather than a local proxy failure by itself. Correlate it with p95/max latency and recent error classes; do not hide it by changing the configured Codex CLI tier. -- For a quick live health/performance read, pair `scripts/codex_lb_live_snapshot.py` with the configured Codex CLI smoke (`codex exec --ephemeral ... "Reply with OK only."`). The snapshot shows local health, latency percentiles, upstream failure classes, and tier mismatches; the CLI smoke proves the configured provider chain without exposing credentials. +- When postrepair snapshots show requested `ultrafast` with an explicit successful terminal `default`, report it as a tier metadata mismatch rather than a local proxy failure or independent proof of upstream scheduling. Correlate it with p95/max latency and recent error classes; do not hide it by changing the configured Codex CLI tier. +- For a quick live health/performance read, pair `scripts/codex_lb_live_snapshot.py` with the configured Codex CLI smoke (`codex exec --ephemeral ... "Reply with OK only."`). The snapshot shows local health, latency percentiles, upstream failure classes, and tier metadata; the CLI smoke proves the configured provider chain without exposing credentials. Neither proves how upstream scheduled the request. + +## Service-tier evidence boundaries + +For streaming traffic, `actual_service_tier` comes only from a nonblank +`response.completed.response.service_tier` on a successful request. A completed +response that omits the tier leaves actual tier null even when a created or +in-progress event supplied one. Failed, incomplete, cancelled, and disconnected +streams provide no actual-tier evidence. + +Billable tier is separate. An explicit successful terminal tier takes precedence; +otherwise the existing observed-event/request fallback remains available for +accounting without being copied into `actual_service_tier`. A successful compact +response may report actual tier directly. Continuation folding preserves the +final terminal tier or its absence, and a synthesized incomplete response cannot +inherit a created tier as terminal evidence. + +The mismatch counter covers direct streaming and compact requests only. It does +not cover native WebSocket or persistent HTTP bridge traffic, even though those +paths receive the request-log terminal-tier repair. Count only successful +requests with an explicit nonblank actual tier different from the normalized +requested tier. Missing actual evidence and unsuccessful outcomes never count. + +Treat records produced by the repaired logic separately from historical records. +Old logs and counter values are not backfilled or reinterpreted as terminal +proof. In particular, an old `auto` may have come from an earlier event or from +continuation folding. A time range that crosses the repair boundary cannot be +read as a uniform set of successful terminal observations. + +Synthetic examples, not runtime observations, for requests asking for `priority`: + +| Created tier | Successful final response tier | Actual tier after repair | Billable tier | Counter result | +| --- | --- | --- | --- | --- | +| `auto` | `priority` | `priority` | `priority` | No mismatch. The final tier wins. | +| `default` | Missing | null | `default` | No mismatch. The created tier is accounting fallback only. | +| Missing | Missing | null | `priority` | No mismatch. Requested fallback is not actual evidence. | +| `priority` | `auto` | `auto` | `auto` | Mismatch on direct streaming or compact only; `auto` does not establish scheduling. | + +If the second sequence fails, finishes incomplete, is cancelled, or disconnects, +actual tier remains null and the counter does not increment. The same labels in +an old record do not prove that its terminal sequence matched any row above. +These reporting rules leave request priority and transport unchanged. ## Prompt-cache economics triage diff --git a/openspec/specs/proxy-runtime-observability/spec.md b/openspec/specs/proxy-runtime-observability/spec.md index 7f2bceeb29..b1061804df 100644 --- a/openspec/specs/proxy-runtime-observability/spec.md +++ b/openspec/specs/proxy-runtime-observability/spec.md @@ -220,6 +220,27 @@ When the greenfield runtime baseline is documented or enabled, the project MUST - **AND** upstream reports actual response.service_tier as default or auto - **THEN** metrics or documented request-log queries allow operators to count the mismatch without inspecting raw prompt data or secret-bearing logs +### Requirement: Service-tier mismatches use successful terminal evidence +The service-tier mismatch counter MUST count only successful requests with an explicit nonblank terminal-reported tier that differs from the normalized requested tier. A created or in-progress tier, requested-tier fallback, missing terminal tier, or unsuccessful attempt MUST NOT increment the counter. The change MUST preserve existing counter coverage and bounded labels rather than adding identities or payload data. + +An explicitly returned `auto` MUST remain observable as the reported tier. Operator guidance MUST describe it as inconclusive for Fast delivery rather than proof of a downgrade. Counter equality or absence MUST NOT be described as evidence of upstream scheduling. + +#### Scenario: Explicit terminal default differs from requested priority +- **WHEN** a successful counted request asks for `priority` and completes with explicit `default` +- **THEN** one mismatch is counted with requested `priority` and actual `default` + +#### Scenario: Created auto followed by terminal priority is not a mismatch +- **WHEN** a successful counted request asks for `priority`, starts with created `auto`, and completes with explicit `priority` +- **THEN** the created tier does not increment the mismatch counter + +#### Scenario: Missing terminal tier does not count a created fallback +- **WHEN** a counted request starts with a tier different from the requested tier but completes without a tier +- **THEN** the request does not increment the mismatch counter + +#### Scenario: Failure does not count a tier mismatch +- **WHEN** a counted attempt fails, is cancelled, disconnects, or finishes incomplete +- **THEN** it does not increment the mismatch counter even if an earlier event included a different tier + ### Requirement: Performance triage separates local health from upstream latency Operator-facing health or performance guidance MUST distinguish local proxy health from request-tail latency and upstream behavior. The guidance MUST combine process/container health, readiness checks, recent local error-log patterns, request latency percentiles, recent upstream failure classes, requested service tier, and actual upstream service tier before recommending a local code change, restart, worker-count change, or upstream/tier investigation. diff --git a/openspec/specs/responses-api-compat/context.md b/openspec/specs/responses-api-compat/context.md index d0b1a53ec5..46335ac698 100644 --- a/openspec/specs/responses-api-compat/context.md +++ b/openspec/specs/responses-api-compat/context.md @@ -39,23 +39,34 @@ codex-lb accepts the OpenAI/Codex `service_tier` field on Responses and Chat Completions compatible routes. The legacy `fast` spelling is accepted as an alias and is forwarded upstream as the canonical `priority` tier. -Fast Mode is request-level intent, not a local speed guarantee. The upstream -Codex backend decides the actual tier for each completed response. codex-lb -therefore records three separate values in request logs: - -- `requestedServiceTier`: what the client or API key asked for, after alias - normalization. -- `actualServiceTier`: what upstream reported in the completed response, when - upstream included it. -- `serviceTier`: the effective billable tier. This uses `actualServiceTier` - when present and falls back to `requestedServiceTier` only when upstream omits - the actual tier. - -If a request is sent with `service_tier: "fast"` or `service_tier: "priority"` -and the completed row shows `requestedServiceTier: "priority"` but -`actualServiceTier: "default"`, codex-lb forwarded the priority request and -upstream chose the default tier. That can happen even when websocket transport -is active. +Fast Mode is request-level intent, not a local speed guarantee. The terminal-tier +provenance repair separates three request-log values: + +- `requestedServiceTier`: the client or API key's requested tier after policy + and alias normalization, not an independent capture of upstream egress. +- `actualServiceTier`: for streams, only a nonblank tier explicitly present in + a successful `response.completed.response`. Successful non-streaming compact + responses may supply actual-tier evidence directly. +- `serviceTier`: the effective billable tier. An explicit terminal tier takes + precedence; otherwise the existing observed-event/request fallback remains + available independently of actual-tier evidence. + +A completion without a tier leaves `actualServiceTier` null even if a created +or in-progress event supplied one. Missing successful terminal evidence, failed, +incomplete, cancelled, and disconnected streams provide no actual-tier evidence. +An accounting fallback must not be copied into `actualServiceTier`. + +Continuation reconstruction preserves the final upstream terminal response's +`service_tier`, including its absence. It does not inherit the initial created +tier when the final terminal omits it. Proxy-synthesized incomplete responses +cannot claim a created tier as upstream terminal evidence. + +These fields describe request intent, reported response metadata, and accounting. +They do not independently prove upstream scheduling or the exact egress payload. +An explicit successful terminal `auto` remains observable but is inconclusive: +it proves neither a downgrade nor priority delivery. A reported `default` is a +metadata mismatch with requested `priority`, not independent scheduling proof. +The repair leaves requested priority and transport behavior unchanged. For OpenCode or Codex-compatible clients, enable Fast Mode by sending a Responses request with: @@ -73,17 +84,34 @@ API keys can also force the tier for traffic that uses that key. Set the key's enforced service tier to `priority` or `fast`; both values are stored and returned as `priority`. -To verify a completed Fast Mode request: - -1. `Transport` should be `WS` if you are verifying the websocket Codex path. -2. `requestedServiceTier` should be `priority` when the client requested Fast - Mode or the API key enforced it. -3. `actualServiceTier` is the upstream result. `default` means upstream did not - grant priority for that response. - -This distinction matters for quota and cost accounting: codex-lb prices the -request from the effective billable `serviceTier`, not from the requested tier -when upstream reports a different actual tier. +Interpret records produced by the repaired logic separately from older records. +Old request logs and mismatch counters are not backfilled, corrected, or +reinterpreted as successful terminal proof. An old `auto` value may reflect an +earlier event or continuation folding; it cannot establish what the terminal +reported. Do not assume an aggregate mixes only postrepair records. + +The mismatch counter covers direct streaming and compact only, not all Responses +traffic. Native WebSocket and persistent HTTP bridge request logs receive the +actual-tier repair, but those paths remain outside the counter. A zero counter, +equal tiers, or absent actual tier does not prove priority delivery. + +Synthetic examples, not runtime observations, for requests asking for `priority`: + +| Created tier | Successful final response tier | Actual tier after repair | Billable tier | Interpretation | +| --- | --- | --- | --- | --- | +| `auto` | `priority` | `priority` | `priority` | Continuation preserves the final reported tier, not the created tier. | +| `default` | Missing | null | `default` | Earlier tier remains an accounting fallback only; no mismatch is counted. | +| Missing | Missing | null | `priority` | Requested-tier billing fallback is not actual-tier evidence. | +| `priority` | `auto` | `auto` | `auto` | Reported metadata differs from the request; scheduling remains unknown. | + +If the second example instead fails, is incomplete, is cancelled, or disconnects, +actual tier is still null and no mismatch is counted. In a pre-repair record, +a created `auto` might have obscured a terminal `priority`; the new rules do not +retroactively establish that any old record followed this synthetic sequence. + +codex-lb prices requests using the effective billable `serviceTier`. Keep this +accounting choice separate from terminal evidence and from latency analysis; +these examples do not call for changing requested priority or transport. ## Include Allowlist (Reference) diff --git a/openspec/specs/responses-api-compat/spec.md b/openspec/specs/responses-api-compat/spec.md index 57292db265..e05b797f88 100644 --- a/openspec/specs/responses-api-compat/spec.md +++ b/openspec/specs/responses-api-compat/spec.md @@ -508,31 +508,39 @@ When a request depends on hard continuity ownership, the service MUST fail close ### Requirement: Request logs persist requested, actual, and billable service tiers separately For Responses proxy traffic, the system MUST persist the operator-requested tier, the upstream-reported actual tier when available, and the effective billable tier used for pricing as separate request-log fields. -The legacy `fast` alias MUST be normalized to the canonical upstream value -`priority` before forwarding and before it is stored as the requested tier. -The upstream-reported `response.service_tier`, when present, remains the -authoritative actual tier even when it differs from the requested tier. +The legacy `fast` alias MUST be normalized to `priority` before forwarding and before it is stored as the requested tier. For streaming requests, `actual_service_tier` MUST come only from a nonblank service tier explicitly present in `response.completed.response`. Created, in-progress, failed, incomplete, cancelled, and disconnected streams MUST NOT supply actual-tier evidence. A completed response that omits the tier MUST leave actual-tier evidence absent even if earlier events supplied one. Non-streaming successful compact responses MAY supply actual-tier evidence directly. + +The system MUST preserve the existing billable-tier fallback independently of actual-tier evidence. An explicit terminal tier MUST take precedence for billing; otherwise the existing observed-event/request fallback remains available without being copied into `actual_service_tier`. + +Continuation reconstruction MUST preserve the final upstream terminal response's `service_tier`, including its absence. It MUST NOT inherit a tier from the initial created response when the terminal omits the field. Proxy-synthesized incomplete responses MUST NOT claim an upstream terminal tier from the created response. #### Scenario: Upstream reports a downgraded actual tier - **WHEN** a client sends a Responses request with `service_tier: "priority"` -- **AND** the upstream response later reports `service_tier: "default"` -- **THEN** the persisted request log entry records `requested_service_tier = "priority"` -- **AND** the persisted request log entry records `actual_service_tier = "default"` -- **AND** the persisted request log entry records billable `service_tier = "default"` +- **AND** the upstream completion explicitly reports `service_tier: "default"` +- **THEN** the persisted request log records requested tier `priority`, actual tier `default`, and billable tier `default` #### Scenario: Fast alias is logged as a priority request -- **WHEN** a client sends a Responses request with `service_tier: "fast"` -- **AND** the upstream response later reports `service_tier: "default"` -- **THEN** the persisted request log entry records `requested_service_tier = "priority"` -- **AND** the persisted request log entry records `actual_service_tier = "default"` -- **AND** the persisted request log entry records billable `service_tier = "default"` +- **WHEN** a client sends `service_tier: "fast"` +- **AND** the upstream completion explicitly reports `service_tier: "default"` +- **THEN** the persisted request log records requested tier `priority`, actual tier `default`, and billable tier `default` #### Scenario: Upstream omits the actual tier -- **WHEN** a client sends a Responses request with `service_tier: "priority"` -- **AND** the upstream response omits `service_tier` -- **THEN** the persisted request log entry records `requested_service_tier = "priority"` -- **AND** the persisted request log entry records `actual_service_tier = null` -- **AND** the persisted request log entry records billable `service_tier = "priority"` +- **WHEN** a client requests `priority` and no upstream event reports a tier +- **THEN** actual tier is null and billable tier falls back to requested `priority` + +#### Scenario: Initial auto does not replace terminal priority +- **WHEN** a continued response starts with created tier `auto` and finishes with explicit terminal tier `priority` +- **THEN** the folded completion and persisted actual tier are `priority` +- **AND** every upstream continuation request retains the requested tier + +#### Scenario: Created-only tier remains accounting fallback +- **WHEN** a created response reports `default` and the completion omits `service_tier` +- **THEN** the completion does not inherit `service_tier` +- **AND** actual tier is null while the existing billable-tier fallback remains `default` + +#### Scenario: Failed or interrupted stream does not prove a terminal tier +- **WHEN** a stream reports a created tier but fails, is cancelled, disconnects, or finishes incomplete +- **THEN** actual-tier evidence remains absent ### Requirement: API key service tier enforcement applies to upstream Responses requests diff --git a/tests/conftest.py b/tests/conftest.py index 26fdeb94f3..a5da451bc6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -163,6 +163,12 @@ def _reset_global_state() -> None: get_api_key_cache().clear() except Exception: pass + try: + import app.core.cache.invalidation as cache_invalidation + + cache_invalidation._poller = None + except Exception: + pass try: from app.core.middleware.firewall_cache import get_firewall_ip_cache as get_firewall_cache diff --git a/tests/integration/test_api_keys_api.py b/tests/integration/test_api_keys_api.py index 4572a5576d..b2abd26685 100644 --- a/tests/integration/test_api_keys_api.py +++ b/tests/integration/test_api_keys_api.py @@ -1337,7 +1337,9 @@ async def fake_stream(_payload, _headers, _access_token, _account_id, base_url=N @pytest.mark.asyncio -async def test_stream_usage_logs_actual_service_tier_when_response_created_echoes_default(async_client, monkeypatch): +async def test_stream_usage_preserves_created_service_tier_for_billing_without_actual_evidence( + async_client, monkeypatch +): enable = await async_client.put( "/api/settings", json={ @@ -1416,8 +1418,9 @@ async def fake_stream(_payload, _headers, _access_token, _account_id, base_url=N latest_log = result.scalars().first() assert latest_log is not None assert latest_log.api_key_id == key_id + assert latest_log.status == "success" assert latest_log.requested_service_tier == "priority" - assert latest_log.actual_service_tier == "default" + assert latest_log.actual_service_tier is None assert latest_log.service_tier == "default" @@ -2721,6 +2724,7 @@ async def test_allowed_but_unsupported_model_is_not_exposed(async_client): # Reservation lifecycle regression tests (fix-api-key-reservation-leak) # --------------------------------------------------------------------------- + @pytest.mark.asyncio @pytest.mark.parametrize("surface", ["stream", "collect", "compact", "transcribe"]) async def test_rate_limit_header_failure_releases_reservation_once( diff --git a/tests/integration/test_health_and_errors.py b/tests/integration/test_health_and_errors.py index 515ef14707..39f41b78d5 100644 --- a/tests/integration/test_health_and_errors.py +++ b/tests/integration/test_health_and_errors.py @@ -73,6 +73,7 @@ async def test_api_not_found_returns_dashboard_payload(async_client): assert payload["error"]["message"] == "Not Found" assert response.headers["X-App-Version"] == __version__ + @pytest.mark.asyncio @pytest.mark.parametrize( "path", @@ -96,6 +97,8 @@ async def test_exact_openai_root_error_envelope_matches_equivalent_paths(async_c "code": "not_found", } } + + @pytest.mark.asyncio async def test_spa_route_path_returns_index_html(async_client, tmp_path): index = _STATIC_DIR / "index.html" diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index acbbd2860e..440852c3f6 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -559,7 +559,12 @@ async def send_text(self, text: str) -> None: text=json.dumps( { "type": "response.created", - "response": {"id": response_id, "object": "response", "status": "in_progress"}, + "response": { + "id": response_id, + "object": "response", + "status": "in_progress", + "service_tier": "auto", + }, }, separators=(",", ":"), ), @@ -577,7 +582,12 @@ async def send_text(self, text: str) -> None: { "type": "response.created", "sequence_number": 0, - "response": {"id": response_id, "object": "response", "status": "in_progress"}, + "response": { + "id": response_id, + "object": "response", + "status": "in_progress", + "service_tier": "auto", + }, }, { "type": "response.output_item.added", @@ -4129,9 +4139,23 @@ async def fake_open_upstream_websocket_with_budget(self, account, headers, *, ti @pytest.mark.asyncio +@pytest.mark.parametrize( + ("first_tier", "final_tier", "observed_tier", "expected_second_billable"), + [ + pytest.param("default", "priority", "auto", "priority", id="final-priority"), + pytest.param("priority", "default", "auto", "default", id="final-default"), + pytest.param("priority", None, "auto", "auto", id="missing-final-tier-observed"), + pytest.param("default", None, None, "priority", id="missing-final-tier-requested"), + ], +) async def test_v1_responses_http_bridge_reuses_upstream_websocket_and_preserves_previous_response_id( async_client, + app_instance, monkeypatch, + first_tier, + final_tier, + observed_tier, + expected_second_billable, ): _install_bridge_settings(monkeypatch, enabled=True) account_id = await _import_account(async_client, "acc_http_bridge_reuse", "http-bridge-reuse@example.com") @@ -4139,6 +4163,44 @@ async def test_v1_responses_http_bridge_reuses_upstream_websocket_and_preserves_ fake_upstream = _FakeBridgeUpstreamWebSocket() connect_calls: list[tuple[str | None, str | None]] = [] + async def fake_send_text(text: str) -> None: + fake_upstream.sent_text.append(text) + response_id = f"resp_bridge_{len(fake_upstream.sent_text)}" + response = {"id": response_id, "object": "response", "status": "in_progress"} + if observed_tier is not None: + response["service_tier"] = observed_tier + for event_type in ("response.created", "response.in_progress"): + await fake_upstream._messages.put( + _FakeUpstreamMessage("text", text=json.dumps({"type": event_type, "response": response})) + ) + completed = { + "id": response_id, + "object": "response", + "status": "completed", + "output": [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "OK"}], + } + ], + "usage": { + "input_tokens": 24, + "output_tokens": 2, + "total_tokens": 26, + "input_tokens_details": {"cached_tokens": 20}, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + } + tier = first_tier if response_id == "resp_bridge_1" else final_tier + if tier is not None: + completed["service_tier"] = tier + await fake_upstream._messages.put( + _FakeUpstreamMessage("text", text=json.dumps({"type": "response.completed", "response": completed})) + ) + + monkeypatch.setattr(fake_upstream, "send_text", fake_send_text) + async def fake_select_account_with_budget( self, deadline, @@ -4206,6 +4268,7 @@ async def fail_legacy_stream(*args, **kwargs): "instructions": "Return exactly OK.", "input": "hello", "prompt_cache_key": "http-bridge-thread-1", + "service_tier": "priority", "client_metadata": { "keep": "yes", "x-codex-installation-id": "client-spoofed-installation-id", @@ -4232,7 +4295,15 @@ async def fail_legacy_stream(*args, **kwargs): second_body = second.json() assert first_body["id"] == "resp_bridge_1" + assert first_body["service_tier"] == first_tier assert second_body["id"] == "resp_bridge_2" + if final_tier is None: + assert "service_tier" not in second_body + else: + assert second_body["service_tier"] == final_tier + assert first_body["status"] == second_body["status"] == "completed" + assert first_body["output"][0]["content"][0]["text"] == "OK" + assert second_body["output"][0]["content"][0]["text"] == "OK" assert connect_calls == [(account_id, account.chatgpt_account_id)] assert len(fake_upstream.sent_text) == 2 first_upstream_payload = json.loads(fake_upstream.sent_text[0]) @@ -4248,6 +4319,38 @@ async def fail_legacy_stream(*args, **kwargs): assert second_upstream_payload["client_metadata"]["x-openai-subagent"] == "collab_spawn" assert second_upstream_payload["client_metadata"]["x-codex-parent-thread-id"] == "parent-thread" assert second_upstream_payload["client_metadata"]["x-codex-window-id"] == "child-thread:0" + assert first_upstream_payload["service_tier"] == "priority" + assert second_upstream_payload["service_tier"] == "priority" + + service = get_proxy_service_for_app(app_instance) + rows: list[RequestLog] = [] + deadline = time.monotonic() + _TEST_SYNC_TIMEOUT_SECONDS + while time.monotonic() < deadline: + assert await service.drain_persistence_tasks(timeout_seconds=10) + async with SessionLocal() as session: + rows = list( + (await session.execute(select(RequestLog).where(RequestLog.account_id == account_id))).scalars() + ) + if len(rows) == 2: + break + await asyncio.sleep(0.05) + assert len(rows) == 2 + logs = {row.request_id: row for row in rows} + assert set(logs) == {first_body["id"], second_body["id"]} + first_log = logs[first_body["id"]] + assert first_log.actual_service_tier == first_tier + assert first_log.service_tier == first_tier + assert first_log.requested_service_tier == "priority" + second_log = logs[second_body["id"]] + assert second_log.actual_service_tier == final_tier + assert second_log.service_tier == expected_second_billable + assert second_log.requested_service_tier == "priority" + for row in rows: + assert row.status == "success" + assert row.transport == "http" + assert row.upstream_transport == "websocket" + assert row.input_tokens == 24 + assert row.output_tokens == 2 @pytest.mark.asyncio @@ -9220,13 +9323,22 @@ async def fake_create_http_bridge_session( @pytest.mark.asyncio @pytest.mark.parametrize( - ("upstream_type", "prime_reused_session", "expected_event_types", "expected_failure_sequence"), + ( + "upstream_type", + "prime_reused_session", + "expected_event_types", + "expected_failure_sequence", + "expected_error_code", + "emit_failed_terminal", + ), [ ( _CreatedThenCloseUpstreamWebSocket, False, ["response.created", "response.failed"], 0, + "stream_incomplete", + False, ), ( _ReasoningThenAbruptCloseUpstreamWebSocket, @@ -9239,6 +9351,8 @@ async def fake_create_http_bridge_session( "response.failed", ], 4, + "stream_incomplete", + False, ), ( _CompleteThenReasoningAbruptCloseUpstreamWebSocket, @@ -9251,17 +9365,35 @@ async def fake_create_http_bridge_session( "response.failed", ], 4, + "stream_incomplete", + False, + ), + ( + _FakeBridgeUpstreamWebSocket, + False, + ["response.created", "response.failed"], + 1, + "server_error", + True, ), ], - ids=["created-then-close", "reasoning-then-abrupt-close", "reused-reasoning-then-abrupt-close"], + ids=[ + "created-then-close", + "reasoning-then-abrupt-close", + "reused-reasoning-then-abrupt-close", + "created-then-response-failed", + ], ) async def test_v1_responses_http_bridge_stream_failure_remains_valid_sse( async_client, + app_instance, monkeypatch, upstream_type, prime_reused_session, expected_event_types, expected_failure_sequence, + expected_error_code, + emit_failed_terminal, ): _install_bridge_settings(monkeypatch, enabled=True) account_id = await _import_account( @@ -9272,6 +9404,54 @@ async def test_v1_responses_http_bridge_stream_failure_remains_valid_sse( account = await _get_account(account_id) upstream = upstream_type() + if emit_failed_terminal: + + async def fake_send_text(text: str) -> None: + upstream.sent_text.append(text) + response_id = "resp_created_then_failed" + await upstream._messages.put( + _FakeUpstreamMessage( + "text", + text=json.dumps( + { + "type": "response.created", + "sequence_number": 0, + "response": { + "id": response_id, + "object": "response", + "status": "in_progress", + "service_tier": "auto", + }, + }, + separators=(",", ":"), + ), + ) + ) + await upstream._messages.put( + _FakeUpstreamMessage( + "text", + text=json.dumps( + { + "type": "response.failed", + "sequence_number": 1, + "response": { + "id": response_id, + "object": "response", + "status": "failed", + "error": { + "type": "server_error", + "code": "server_error", + "message": "Synthetic upstream failure", + }, + }, + }, + separators=(",", ":"), + ), + ) + ) + + monkeypatch.setattr(upstream, "send_text", fake_send_text) + async def fake_select_account_with_budget( self, deadline, @@ -9325,9 +9505,13 @@ async def fake_connect_responses_websocket( del headers, access_token, account_id_header, base_url, session return upstream + async def fail_legacy_stream(*args, **kwargs): + raise AssertionError("legacy core_stream_responses path must not be used when HTTP bridge is enabled") + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + monkeypatch.setattr(proxy_module, "core_stream_responses", fail_legacy_stream) headers = {"x-codex-turn-state": "turn-sse-failure"} if prime_reused_session else {} if prime_reused_session: @@ -9339,6 +9523,7 @@ async def fake_connect_responses_websocket( "instructions": "Return exactly OK.", "input": "prime-sse-session", "prompt_cache_key": "sse-failure-key", + "service_tier": "priority", }, ) assert prime.status_code == 200 @@ -9352,6 +9537,7 @@ async def fake_connect_responses_websocket( "instructions": "Return exactly OK.", "input": "trigger-sse-failure", "prompt_cache_key": "sse-failure-key", + "service_tier": "priority", "stream": True, }, ) as response: @@ -9361,8 +9547,38 @@ async def fake_connect_responses_websocket( events = [json.loads(line[6:]) for line in lines if line[6:] != "[DONE]"] assert [event["type"] for event in events] == expected_event_types assert events[0]["response"]["id"] == events[-1]["response"]["id"] + assert events[0]["response"]["service_tier"] == "auto" assert events[-1]["sequence_number"] == expected_failure_sequence - assert events[-1]["response"]["error"]["code"] == "stream_incomplete" + assert events[-1]["response"]["error"]["code"] == expected_error_code + + service = get_proxy_service_for_app(app_instance) + rows: list[RequestLog] = [] + deadline = time.monotonic() + _TEST_SYNC_TIMEOUT_SECONDS + while time.monotonic() < deadline: + assert await service.drain_persistence_tasks(timeout_seconds=10) + async with SessionLocal() as session: + rows = list( + ( + await session.execute( + select(RequestLog).where( + RequestLog.account_id == account_id, + RequestLog.status == "error", + ) + ) + ).scalars() + ) + if len(rows) == 1: + break + await asyncio.sleep(0.05) + assert len(rows) == 1 + failure_log = rows[0] + assert failure_log.actual_service_tier is None + assert failure_log.service_tier == "auto" + assert failure_log.requested_service_tier == "priority" + assert failure_log.status == "error" + assert failure_log.error_code == expected_error_code + assert failure_log.transport == "http" + assert failure_log.upstream_transport == "websocket" @pytest.mark.asyncio diff --git a/tests/integration/test_oauth_flow.py b/tests/integration/test_oauth_flow.py index 5744a5b4b3..b375fe3dab 100644 --- a/tests/integration/test_oauth_flow.py +++ b/tests/integration/test_oauth_flow.py @@ -815,6 +815,7 @@ async def test_only_expired_pending_browser_flow_no_longer_keeps_callback_server assert oauth_module._OAUTH_STORE._flows == {} assert oauth_module._OAUTH_STORE.state.status == "idle" + @pytest.mark.asyncio async def test_callback_access_log_omits_code_and_state(caplog, unused_tcp_port): code_secret = "CALLBACK_CODE_SECRET" @@ -843,6 +844,8 @@ async def handler(request: web.Request) -> web.StreamResponse: assert body == "callback accepted" assert code_secret not in caplog.text assert state_secret not in caplog.text + + @pytest.mark.asyncio async def test_callback_server_remains_reserved_until_stop_completes(): await oauth_module._OAUTH_STORE.reset() diff --git a/tests/integration/test_proxy_websocket_responses.py b/tests/integration/test_proxy_websocket_responses.py index d047f209fc..f081b749c7 100644 --- a/tests/integration/test_proxy_websocket_responses.py +++ b/tests/integration/test_proxy_websocket_responses.py @@ -1671,22 +1671,48 @@ def _ws_msg(event: dict[str, Any]) -> _FakeUpstreamMessage: return _FakeUpstreamMessage("text", text=json.dumps(event, separators=(",", ":"))) -def test_backend_responses_websocket_folds_truncated_reasoning_continuation(app_instance, monkeypatch): +@pytest.mark.parametrize( + ("final_service_tier", "expected_actual_tier", "expected_billable_tier"), + [ + pytest.param("priority", "priority", "priority", id="terminal-priority"), + pytest.param("default", "default", "default", id="terminal-default"), + pytest.param(None, None, "auto", id="terminal-omitted"), + ], +) +def test_backend_responses_websocket_folds_truncated_reasoning_continuation( + app_instance, + monkeypatch, + final_service_tier: str | None, + expected_actual_tier: str | None, + expected_billable_tier: str, +): # A visible round that truncates on the 518*n-2 fingerprint is folded with a # hidden continuation round over the same upstream/account: the truncated # final output is suppressed, the final answer comes from the hidden round, - # and settlement bills the summed usage. + # and settlement bills the summed usage. Created and hidden-round tiers may + # inform billing, but only the final completion supplies the actual tier. + round_one_completion = _ws_completed("resp_ws_v", input_tokens=100, output_tokens=600, reasoning_tokens=516) + round_one_completion["response"]["service_tier"] = "default" + final_completion = _ws_completed("resp_ws_h", input_tokens=120, output_tokens=20, reasoning_tokens=10) + if final_service_tier is not None: + final_completion["response"]["service_tier"] = final_service_tier round_one = [ - {"type": "response.created", "response": {"id": "resp_ws_v", "status": "in_progress", "output": []}}, + { + "type": "response.created", + "response": {"id": "resp_ws_v", "status": "in_progress", "output": [], "service_tier": "auto"}, + }, *_ws_reasoning_events(output_index=0, item_id="rs_1", encrypted_content="enc1"), *_ws_message_events(output_index=1, item_id="msg_partial", text="partial answer"), - _ws_completed("resp_ws_v", input_tokens=100, output_tokens=600, reasoning_tokens=516), + round_one_completion, ] round_two = [ - {"type": "response.created", "response": {"id": "resp_ws_h", "status": "in_progress", "output": []}}, + { + "type": "response.created", + "response": {"id": "resp_ws_h", "status": "in_progress", "output": [], "service_tier": "auto"}, + }, *_ws_reasoning_events(output_index=0, item_id="rs_2", encrypted_content="enc2"), *_ws_message_events(output_index=1, item_id="msg_final", text="final answer"), - _ws_completed("resp_ws_h", input_tokens=120, output_tokens=20, reasoning_tokens=10), + final_completion, ] fake_upstream = _FakeUpstreamWebSocket([_ws_msg(e) for e in (*round_one, *round_two)]) log_calls: list[dict[str, object]] = [] @@ -1739,6 +1765,7 @@ async def fake_settle(self, api_key, reservation, settlement, response_id): "model": "gpt-5.5", "instructions": "", "reasoning": {"effort": "high"}, + "service_tier": "priority", # Upstream-unsupported field: must be stripped from the hidden round too. "max_output_tokens": 6000, "input": [{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}], @@ -1766,6 +1793,11 @@ async def fake_settle(self, api_key, reservation, settlement, response_id): types = [event["type"] for event in received] assert types.count("response.created") == 1 assert types.count("response.completed") == 1 + terminal_response = next(event["response"] for event in received if event["type"] == "response.completed") + if expected_actual_tier is None: + assert "service_tier" not in terminal_response + else: + assert terminal_response["service_tier"] == expected_actual_tier deltas = "".join( str(event.get("delta", "")) for event in received if event.get("type") == "response.output_text.delta" ) @@ -1778,6 +1810,7 @@ async def fake_settle(self, api_key, reservation, settlement, response_id): # Hidden round inherits the prepared/stripped request shape, not the raw # client payload, so upstream-unsupported fields must not reappear. assert "max_output_tokens" not in continuation + assert continuation["service_tier"] == "priority" replay_input = continuation["input"] assert {"id": "rs_1", "type": "reasoning", "encrypted_content": "enc1"} in replay_input assert any(isinstance(item, dict) and item.get("phase") == "commentary" for item in replay_input) @@ -1785,9 +1818,13 @@ async def fake_settle(self, api_key, reservation, settlement, response_id): assert len(log_calls) == 1 assert log_calls[0]["input_tokens"] == 220 assert log_calls[0]["output_tokens"] == 620 + assert log_calls[0]["requested_service_tier"] == "priority" + assert log_calls[0]["service_tier"] == expected_billable_tier + assert log_calls[0]["actual_service_tier"] == expected_actual_tier assert len(settlements) == 1 assert settlements[0].input_tokens == 220 assert settlements[0].output_tokens == 620 + assert settlements[0].service_tier == expected_billable_tier def test_backend_responses_websocket_non_truncated_reasoning_passes_through(app_instance, monkeypatch): @@ -9592,7 +9629,7 @@ def upstream_created_then_eof( ) -> _FakeUpstreamWebSocket: created_payload: dict[str, object] = { "type": "response.created", - "response": {"id": response_id, "status": "in_progress"}, + "response": {"id": response_id, "status": "in_progress", "service_tier": "auto"}, } if sequence_number is not None: created_payload["sequence_number"] = sequence_number @@ -9684,12 +9721,15 @@ async def fake_write_request_log(self, **kwargs): assert created_event["type"] == "response.created" assert failed_event["type"] == "response.failed" assert failed_event["response"]["id"] == "resp_ws_eof" + assert "service_tier" not in failed_event["response"] assert failed_event["response"]["error"]["code"] == "stream_incomplete" assert "close_code=1011" in failed_event["response"]["error"]["message"] assert len(log_calls) == 1 assert log_calls[0]["request_id"] == "resp_ws_eof_retry" assert log_calls[0]["status"] == "error" assert log_calls[0]["error_code"] == "stream_incomplete" + assert log_calls[0]["service_tier"] == "auto" + assert log_calls[0]["actual_service_tier"] is None def test_backend_responses_websocket_closes_before_replaying_exposed_sequence( diff --git a/tests/integration/test_request_logs_api.py b/tests/integration/test_request_logs_api.py index 4220704b14..67cd78d7a2 100644 --- a/tests/integration/test_request_logs_api.py +++ b/tests/integration/test_request_logs_api.py @@ -265,9 +265,7 @@ async def test_request_logs_api_excludes_client_ip_from_guest_search_and_preserv admin_response = await async_client.get("/api/request-logs?search=198.51.100.222") assert admin_response.status_code == 200 - assert [entry["requestId"] for entry in admin_response.json()["requests"]] == [ - "req_ip_search_target" - ] + assert [entry["requestId"] for entry in admin_response.json()["requests"]] == ["req_ip_search_target"] assert admin_response.json()["total"] == 1 diff --git a/tests/unit/test_bulkhead.py b/tests/unit/test_bulkhead.py index c08602dd67..f8475245b3 100644 --- a/tests/unit/test_bulkhead.py +++ b/tests/unit/test_bulkhead.py @@ -68,6 +68,7 @@ async def work(): assert overloaded.headers["retry-after"] == "5" assert first_response.status_code == 200 + @pytest.mark.parametrize( "path", [ @@ -132,6 +133,8 @@ async def send(message: dict[str, object]) -> None: assert payload["error"]["type"] == "rate_limit_error" assert payload["error"]["code"] == "proxy_overloaded" assert "proxy_http lane" in payload["error"]["message"] + + @pytest.mark.asyncio async def test_bulkhead_compact_lane_isolated_from_general_proxy_http(): app = FastAPI() diff --git a/tests/unit/test_codex_continuation.py b/tests/unit/test_codex_continuation.py index 458de9f4fb..6335a71255 100644 --- a/tests/unit/test_codex_continuation.py +++ b/tests/unit/test_codex_continuation.py @@ -1,8 +1,9 @@ from __future__ import annotations from collections.abc import AsyncGenerator, AsyncIterator +from copy import deepcopy from types import SimpleNamespace -from typing import cast +from typing import Any, cast import pytest @@ -164,30 +165,85 @@ async def _collect_events(chunks: AsyncIterator[str]) -> list[dict[str, JsonValu @pytest.mark.asyncio +@pytest.mark.parametrize( + ("created_tier", "terminal_tier"), + [ + pytest.param("auto", "priority", id="created-auto-terminal-priority"), + pytest.param("auto", "default", id="created-auto-terminal-default"), + pytest.param(None, "priority", id="created-absent-terminal-priority"), + pytest.param("auto", None, id="created-auto-terminal-omitted"), + ], +) async def test_fold_responses_stream_continues_truncated_round_and_reuses_payload_shape( decision_counter: _ObservedCounter, + monkeypatch: pytest.MonkeyPatch, + created_tier: str | None, + terminal_tier: str | None, ) -> None: base_payload: JsonObject = { "model": "gpt-5.5", "instructions": "solve", "input": [{"role": "user", "content": "question"}], "previous_response_id": "resp_previous", + "service_tier": "priority", "stream": True, } + base_payload_snapshot = deepcopy(base_payload) + first_created = _created("resp_visible") + created_response = cast(dict[str, JsonValue], first_created["response"]) + created_response["model"] = "gpt-5.5" + created_response["metadata"] = {"user_label": "visible"} + if created_tier is not None: + created_response["service_tier"] = created_tier + first_terminal = _completed("resp_visible", input_tokens=100, output_tokens=600, reasoning_tokens=516) + cast(dict[str, JsonValue], first_terminal["response"])["service_tier"] = "priority" + hidden_created = _created("resp_hidden") + cast(dict[str, JsonValue], hidden_created["response"])["service_tier"] = "flex" + hidden_terminal = _completed("resp_hidden", input_tokens=120, output_tokens=20, reasoning_tokens=10) + if terminal_tier is not None: + cast(dict[str, JsonValue], hidden_terminal["response"])["service_tier"] = terminal_tier round_events = [ [ - _created("resp_visible"), + first_created, *_reasoning_events(output_index=0, item_id="rs_1", encrypted_content="enc1"), *_message_events(output_index=1, item_id="msg_partial", text="partial answer"), - _completed("resp_visible", input_tokens=100, output_tokens=600, reasoning_tokens=516), + first_terminal, ], [ - _created("resp_hidden"), + hidden_created, *_reasoning_events(output_index=0, item_id="rs_2", encrypted_content="enc2"), *_message_events(output_index=1, item_id="msg_final", text="final answer"), - _completed("resp_hidden", input_tokens=120, output_tokens=20, reasoning_tokens=10), + hidden_terminal, ], ] + reconstruction_inputs: list[tuple[dict[str, Any] | None, dict[str, Any] | None]] = [] + reconstruction_snapshots: list[tuple[dict[str, Any] | None, dict[str, Any] | None]] = [] + real_reconstruct_terminal = codex_continuation_module._reconstruct_terminal + + def observe_reconstruct_terminal( + terminal: dict[str, Any] | None, + base_response: dict[str, Any] | None, + output_items: list[dict[str, Any]], + usage: dict[str, Any], + seq: int, + rounds: list[dict[str, Any]], + stopped_reason: str | None, + billed_usage: dict[str, Any] | None, + ) -> dict[str, Any]: + reconstruction_inputs.append((terminal, base_response)) + reconstruction_snapshots.append((deepcopy(terminal), deepcopy(base_response))) + return real_reconstruct_terminal( + terminal, + base_response, + output_items, + usage, + seq, + rounds, + stopped_reason, + billed_usage, + ) + + monkeypatch.setattr(codex_continuation_module, "_reconstruct_terminal", observe_reconstruct_terminal) opened_payloads: list[JsonObject] = [] async def open_round(payload: JsonObject) -> AsyncGenerator[str, None]: @@ -205,9 +261,31 @@ async def open_round(payload: JsonObject) -> AsyncGenerator[str, None]: ) assert len(opened_payloads) == 2 + assert opened_payloads[0]["service_tier"] == "priority" + assert opened_payloads[1]["service_tier"] == "priority" + assert base_payload == base_payload_snapshot assert opened_payloads[0]["previous_response_id"] == "resp_previous" assert opened_payloads[0]["include"] == ["reasoning.encrypted_content"] assert "previous_response_id" not in opened_payloads[1] + assert len(reconstruction_inputs) == 1 + captured_terminal, captured_created = reconstruction_inputs[0] + terminal_snapshot, created_snapshot = reconstruction_snapshots[0] + assert captured_terminal == terminal_snapshot + assert captured_created == created_snapshot + assert captured_terminal is not None + assert captured_created is not None + captured_terminal_response = cast(dict[str, JsonValue], captured_terminal["response"]) + captured_created_response = cast(dict[str, JsonValue], captured_created) + assert captured_terminal_response["id"] == "resp_hidden" + if terminal_tier is None: + assert "service_tier" not in captured_terminal_response + else: + assert captured_terminal_response["service_tier"] == terminal_tier + assert captured_created_response["id"] == "resp_visible" + if created_tier is None: + assert "service_tier" not in captured_created_response + else: + assert captured_created_response["service_tier"] == created_tier replay_input = cast(list[JsonValue], opened_payloads[1]["input"]) assert replay_input[0] == {"role": "user", "content": "question"} assert replay_input[1] == { @@ -227,6 +305,12 @@ async def open_round(payload: JsonObject) -> AsyncGenerator[str, None]: assert event_types.count("response.completed") == 1 assert event_types.count("response.output_item.done") == 3 assert [event["sequence_number"] for event in events] == list(range(len(events))) + assert events[0]["type"] == "response.created" + emitted_created = cast(dict[str, JsonValue], events[0]["response"]) + if created_tier is None: + assert "service_tier" not in emitted_created + else: + assert emitted_created["service_tier"] == created_tier deltas = "".join( str(event.get("delta", "")) for event in events if event.get("type") == "response.output_text.delta" @@ -237,7 +321,13 @@ async def open_round(payload: JsonObject) -> AsyncGenerator[str, None]: terminal = events[-1] response = cast(dict[str, JsonValue], terminal["response"]) assert response["id"] == "resp_visible" + assert response["model"] == "gpt-5.5" + if terminal_tier is None: + assert "service_tier" not in response + else: + assert response["service_tier"] == terminal_tier metadata = cast(dict[str, JsonValue], response["metadata"]) + assert metadata["user_label"] == "visible" assert metadata["proxy_rounds"] == [ {"round": 1, "reasoning_tokens": 516, "n": 1}, {"round": 2, "reasoning_tokens": 10, "n": None}, @@ -271,6 +361,126 @@ async def open_round(payload: JsonObject) -> AsyncGenerator[str, None]: ] +@pytest.mark.asyncio +async def test_fold_responses_stream_hidden_eof_synthetic_terminal_omits_service_tier( + monkeypatch: pytest.MonkeyPatch, +) -> None: + base_payload: JsonObject = { + "model": "gpt-5.5", + "input": [{"role": "user", "content": "question"}], + "service_tier": "priority", + "stream": True, + } + synthetic_inputs: list[dict[str, Any] | None] = [] + synthetic_snapshots: list[dict[str, Any] | None] = [] + real_synthetic_incomplete = codex_continuation_module._synthetic_incomplete + + def observe_synthetic_incomplete( + base_response: dict[str, Any] | None, + output_items: list[dict[str, Any]], + usage: dict[str, Any], + seq: int, + reason: str, + rounds: list[dict[str, Any]], + billed_usage: dict[str, Any] | None, + ) -> dict[str, Any]: + synthetic_inputs.append(base_response) + synthetic_snapshots.append(deepcopy(base_response)) + return real_synthetic_incomplete( + base_response, + output_items, + usage, + seq, + reason, + rounds, + billed_usage, + ) + + monkeypatch.setattr(codex_continuation_module, "_synthetic_incomplete", observe_synthetic_incomplete) + opened_payloads: list[JsonObject] = [] + + async def open_round(payload: JsonObject) -> AsyncGenerator[str, None]: + opened_payloads.append(payload) + if len(opened_payloads) != 1: + return + + created = _created("resp_visible") + created_response = cast(dict[str, JsonValue], created["response"]) + created_response["model"] = "gpt-5.5" + created_response["metadata"] = {"user_label": "visible"} + created_response["service_tier"] = "auto" + yield _event(created) + for event in _reasoning_events(output_index=0, item_id="rs_1", encrypted_content="enc1"): + yield _event(event) + completed = _completed("resp_visible", input_tokens=100, output_tokens=600, reasoning_tokens=516) + cast(dict[str, JsonValue], completed["response"])["service_tier"] = "default" + yield _event(completed) + + events = await _collect_events( + fold_responses_stream_with_codex_continuation( + base_payload=base_payload, + open_round=open_round, + config=CodexContinuationConfig(max_continue=1, rechunk_size=64), + ) + ) + + assert len(opened_payloads) == 2 + assert len(synthetic_inputs) == 1 + captured_created = synthetic_inputs[0] + assert captured_created == synthetic_snapshots[0] + assert captured_created is not None + assert captured_created["service_tier"] == "auto" + assert opened_payloads[0]["service_tier"] == "priority" + assert opened_payloads[1]["service_tier"] == "priority" + assert "previous_response_id" not in opened_payloads[1] + assert [event["type"] for event in events] == [ + "response.created", + "response.output_item.added", + "response.output_item.done", + "response.incomplete", + ] + assert [event["sequence_number"] for event in events] == list(range(len(events))) + assert cast(dict[str, JsonValue], events[0]["response"])["service_tier"] == "auto" + + terminal = events[-1] + assert terminal["type"] == "response.incomplete" + response = cast(dict[str, JsonValue], terminal["response"]) + assert response["id"] == "resp_visible" + assert response["model"] == "gpt-5.5" + assert "service_tier" not in response + assert response["status"] == "incomplete" + assert response["incomplete_details"] == {"reason": "upstream_eof"} + assert response["output"] == [ + { + "id": "rs_1", + "type": "reasoning", + "encrypted_content": "enc1", + } + ] + assert events[2]["item"] == cast(list[JsonValue], response["output"])[0] + assert response["usage"] == { + "input_tokens": 100, + "output_tokens": 516, + "total_tokens": 616, + "input_tokens_details": {"cached_tokens": 25}, + "output_tokens_details": {"reasoning_tokens": 516}, + } + metadata = cast(dict[str, JsonValue], response["metadata"]) + assert metadata["user_label"] == "visible" + assert metadata["proxy_stopped_reason"] == "upstream_eof" + assert metadata["proxy_rounds"] == [ + {"round": 1, "reasoning_tokens": 516, "n": 1}, + {"round": 2, "reasoning_tokens": None, "n": None}, + ] + assert metadata["proxy_billed_usage"] == { + "input_tokens": 100, + "output_tokens": 600, + "total_tokens": 700, + "input_tokens_details": {"cached_tokens": 25}, + "output_tokens_details": {"reasoning_tokens": 516}, + } + + @pytest.mark.asyncio async def test_fold_responses_stream_stops_and_delivers_truncated_round_tool_calls( decision_counter: _ObservedCounter, diff --git a/tests/unit/test_db_sqlite_maintenance.py b/tests/unit/test_db_sqlite_maintenance.py index cdb75126c2..525963b278 100644 --- a/tests/unit/test_db_sqlite_maintenance.py +++ b/tests/unit/test_db_sqlite_maintenance.py @@ -59,9 +59,7 @@ def tracked(self: Path, *args: object, **kwargs: object) -> object: assert connections, f"expected tracked sqlite connections before Path.{method_name}" states = tuple(connection.closed for connection in connections) snapshots.append((self, states)) - assert all(states), ( - f"sqlite connection still open at Path.{method_name} for {self}; closed={states}" - ) + assert all(states), f"sqlite connection still open at Path.{method_name} for {self}; closed={states}" return original(self, *args, **kwargs) monkeypatch.setattr(Path, method_name, tracked) diff --git a/tests/unit/test_multipart_openapi.py b/tests/unit/test_multipart_openapi.py index 3128ec0d6a..2523c9cd11 100644 --- a/tests/unit/test_multipart_openapi.py +++ b/tests/unit/test_multipart_openapi.py @@ -140,10 +140,11 @@ def test_multipart_request_body_schemas_remain_semantically_identical() -> None: } -def test_non_multipart_codex_image_and_file_surfaces_are_unchanged() -> None: +def test_codex_image_and_file_surfaces_do_not_advertise_multipart() -> None: openapi = create_app().openapi() - assert "/backend-api/codex/images/edits" not in openapi["paths"] + image_edit = openapi["paths"]["/backend-api/codex/images/edits"]["post"] + assert "multipart/form-data" not in image_edit.get("requestBody", {}).get("content", {}) files_body = openapi["paths"]["/backend-api/files"]["post"]["requestBody"] assert files_body == { "required": True, diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index e5ccb3789e..2f356c2223 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -5125,6 +5125,23 @@ def post( return self._response +class _QueuedSseSession: + def __init__(self, responses: list[_SsePostResponse]) -> None: + self._responses = responses + self.calls: list[dict[str, object]] = [] + + def post( + self, + url: str, + *, + json=None, + headers: dict[str, str] | None = None, + timeout=None, + ): + self.calls.append({"url": url, "json": json, "headers": headers, "timeout": timeout}) + return self._responses.pop(0) + + class _SessionGenerationHarness: def __init__(self, failed_session: object, replacement_session: object) -> None: self.current_session = failed_session @@ -5820,57 +5837,6 @@ class Settings: assert "proxy_service_tier_trace" not in caplog.text -def test_service_tier_mismatch_metric_records_without_trace_logging(monkeypatch, caplog): - counter = _ObservedCounter() - - class Settings: - log_proxy_request_payload = False - log_proxy_request_shape = False - log_proxy_request_shape_raw_cache_key = False - log_proxy_service_tier_trace = False - - monkeypatch.setattr(proxy_service, "get_settings", lambda: Settings()) - monkeypatch.setattr(proxy_service, "PROMETHEUS_AVAILABLE", True) - monkeypatch.setattr(proxy_service, "service_tier_mismatch_total", counter) - - caplog.set_level(logging.WARNING) - proxy_service._maybe_log_proxy_service_tier_trace( - "stream", - requested_service_tier="ultrafast", - actual_service_tier="default", - ) - - assert "proxy_service_tier_trace" not in caplog.text - assert counter.samples == [ - { - "labels": {"kind": "stream", "requested_tier": "ultrafast", "actual_tier": "default"}, - "value": 1.0, - } - ] - - -def test_service_tier_mismatch_metric_skips_matching_tiers(monkeypatch): - counter = _ObservedCounter() - - class Settings: - log_proxy_request_payload = False - log_proxy_request_shape = False - log_proxy_request_shape_raw_cache_key = False - log_proxy_service_tier_trace = False - - monkeypatch.setattr(proxy_service, "get_settings", lambda: Settings()) - monkeypatch.setattr(proxy_service, "PROMETHEUS_AVAILABLE", True) - monkeypatch.setattr(proxy_service, "service_tier_mismatch_total", counter) - - proxy_service._maybe_log_proxy_service_tier_trace( - "compact", - requested_service_tier="priority", - actual_service_tier="priority", - ) - - assert counter.samples == [] - - def test_log_upstream_request_trace(monkeypatch, caplog): class Settings: log_upstream_request_summary = True @@ -6034,22 +6000,6 @@ class Settings: codex_continuation_rechunk_size = 64 codex_continuation_max_total_output_tokens = 0 - class QueueSseSession: - def __init__(self, responses: list[_SsePostResponse]) -> None: - self._responses = responses - self.calls: list[dict[str, object]] = [] - - def post( - self, - url: str, - *, - json=None, - headers: dict[str, str] | None = None, - timeout=None, - ): - self.calls.append({"url": url, "json": json, "headers": headers, "timeout": timeout}) - return self._responses.pop(0) - def sse(payload: Mapping[str, JsonValue]) -> bytes: return proxy_module.format_sse_event(payload).encode("utf-8") @@ -6180,7 +6130,7 @@ def sse(payload: Mapping[str, JsonValue]) -> bytes: "stream": True, } ) - session = QueueSseSession([round_one, round_two]) + session = _QueuedSseSession([round_one, round_two]) events = [ parse_sse_data_json(event) @@ -10008,14 +9958,197 @@ def test_logged_error_json_response_emits_proxy_error_log(caplog): @pytest.mark.asyncio -async def test_stream_responses_logs_actual_service_tier_and_requested_tier_trace(monkeypatch, caplog): - settings = _make_proxy_settings(log_proxy_service_tier_trace=True) +@pytest.mark.parametrize( + "terminal_case", + [ + "conflicting-completed-tier", + "completed-without-tier", + "absent-completion", + "failed-completion", + ], +) +async def test_stream_responses_logs_actual_service_tier_and_requested_tier_trace( + monkeypatch, + caplog, + terminal_case: Literal[ + "conflicting-completed-tier", + "completed-without-tier", + "absent-completion", + "failed-completion", + ], +): + trace_enabled = terminal_case == "conflicting-completed-tier" + settings = _make_proxy_settings(log_proxy_service_tier_trace=trace_enabled) + settings.upstream_base_url = "https://chatgpt.com/backend-api" + settings.upstream_connect_timeout_seconds = 8.0 + settings.stream_idle_timeout_seconds = 600.0 + settings.http_responses_stream_request_budget_seconds = 7200.0 + settings.image_inline_fetch_enabled = False + settings.log_upstream_request_payload = False + settings.log_upstream_request_summary = False + settings.upstream_stream_transport = "http" + settings.codex_continuation_enabled = True + settings.codex_continuation_max_continue = 1 request_logs = _RequestLogsRecorder() + counter = _ObservedCounter() service = proxy_service.ProxyService(_repo_factory(request_logs)) account = _make_account("acc_trace_stream") + def sse(event: Mapping[str, JsonValue]) -> bytes: + return proxy_module.format_sse_event(event).encode("utf-8") + + created = sse( + { + "type": "response.created", + "response": {"id": "resp_trace_stream", "status": "in_progress", "service_tier": "auto"}, + } + ) + if terminal_case == "conflicting-completed-tier": + upstream_responses = [ + _SsePostResponse( + [ + created, + sse( + { + "type": "response.output_item.added", + "output_index": 0, + "item": {"id": "rs_trace_stream", "type": "reasoning"}, + } + ), + sse( + { + "type": "response.output_item.done", + "output_index": 0, + "item": { + "id": "rs_trace_stream", + "type": "reasoning", + "encrypted_content": "enc-trace-stream", + }, + } + ), + sse( + { + "type": "response.completed", + "response": { + "id": "resp_trace_stream", + "status": "completed", + "service_tier": "auto", + "usage": { + "input_tokens": 100, + "output_tokens": 518, + "total_tokens": 618, + "output_tokens_details": {"reasoning_tokens": 516}, + }, + }, + } + ), + ] + ), + _SsePostResponse( + [ + sse( + { + "type": "response.completed", + "response": { + "id": "resp_final_trace_stream", + "status": "completed", + "service_tier": "default", + "usage": { + "input_tokens": 120, + "output_tokens": 2, + "total_tokens": 122, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + }, + } + ) + ] + ), + ] + elif terminal_case == "completed-without-tier": + upstream_responses = [ + _SsePostResponse( + [ + created, + sse( + { + "type": "response.completed", + "response": { + "id": "resp_trace_stream", + "status": "completed", + "usage": { + "input_tokens": 1, + "output_tokens": 1, + "total_tokens": 2, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + }, + } + ), + ] + ) + ] + elif terminal_case == "absent-completion": + upstream_responses = [_SsePostResponse([created])] + else: + upstream_responses = [ + _SsePostResponse( + [ + sse( + { + "type": "response.in_progress", + "response": { + "id": "resp_trace_stream", + "status": "in_progress", + "service_tier": "auto", + }, + } + ), + sse( + { + "type": "response.failed", + "response": { + "id": "resp_trace_stream", + "status": "failed", + "service_tier": "default", + "error": {"code": "upstream_error", "message": "provider failed"}, + }, + } + ), + ] + ) + ] + + session = _QueuedSseSession(upstream_responses) + + async def folded_stream( + payload, + headers, + access_token, + account_id, + base_url=None, + raise_for_status=False, + ): + async for event in proxy_module.stream_responses( + payload, + headers, + access_token, + account_id, + base_url=base_url, + raise_for_status=raise_for_status, + session=cast(proxy_module.aiohttp.ClientSession, session), + ): + yield event + monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings)) monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(proxy_module, "get_settings", lambda: settings) + monkeypatch.setattr(proxy_module, "_maybe_log_upstream_request_start", lambda **kwargs: None) + monkeypatch.setattr(proxy_module, "_maybe_log_upstream_request_complete", lambda **kwargs: None) + monkeypatch.setattr(proxy_module, "archive_json", lambda **kwargs: None) + monkeypatch.setattr(proxy_module, "archive_text", lambda **kwargs: None) + monkeypatch.setattr(proxy_service, "PROMETHEUS_AVAILABLE", True) + monkeypatch.setattr(proxy_service, "service_tier_mismatch_total", counter) monkeypatch.setattr( service._load_balancer, "select_account", @@ -10023,11 +10156,7 @@ async def test_stream_responses_logs_actual_service_tier_and_requested_tier_trac ) monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(return_value=account)) monkeypatch.setattr(service, "_settle_stream_api_key_usage", AsyncMock(return_value=True)) - - async def fake_stream(payload, headers, access_token, account_id, base_url=None, raise_for_status=False): - yield 'data: {"type":"response.completed","response":{"id":"resp_trace_stream","service_tier":"default"}}\n\n' - - monkeypatch.setattr(proxy_service, "core_stream_responses", fake_stream) + monkeypatch.setattr(proxy_service, "core_stream_responses", folded_stream) payload = ResponsesRequest.model_validate( { @@ -10047,19 +10176,68 @@ async def fake_stream(payload, headers, access_token, account_id, base_url=None, finally: reset_request_id(token) - assert chunks + downstream = [event for chunk in chunks if (event := parse_sse_data_json(chunk)) is not None] + terminal = downstream[-1] + response = cast(dict[str, JsonValue], terminal["response"]) + expected_terminal_type = { + "conflicting-completed-tier": "response.completed", + "completed-without-tier": "response.completed", + "absent-completion": "response.failed", + "failed-completion": "response.failed", + }[terminal_case] + expected_status = ( + "success" if terminal_case in {"conflicting-completed-tier", "completed-without-tier"} else "error" + ) + expected_service_tier = ( + "default" if terminal_case in {"conflicting-completed-tier", "failed-completion"} else "auto" + ) + expected_actual_tier = "default" if terminal_case == "conflicting-completed-tier" else None + + assert terminal["type"] == expected_terminal_type + assert response["id"] == "resp_trace_stream" + if terminal_case == "conflicting-completed-tier": + assert response["service_tier"] == "default" + metadata = cast(dict[str, JsonValue], response["metadata"]) + assert metadata["proxy_rounds"] == [ + {"round": 1, "reasoning_tokens": 516, "n": 1}, + {"round": 2, "reasoning_tokens": 0, "n": None}, + ] + elif terminal_case in {"completed-without-tier", "absent-completion"}: + assert "service_tier" not in response + else: + assert response["service_tier"] == "default" + error = cast(dict[str, JsonValue], response["error"]) + assert error["code"] == "upstream_error" + assert error["message"] == "provider failed" assert request_id - assert request_logs.calls[0]["service_tier"] == "default" - assert request_logs.calls[0]["requested_service_tier"] == "priority" - assert request_logs.calls[0]["actual_service_tier"] == "default" - assert f"request_id={request_id}" in caplog.text - assert "response_id=resp_trace_stream" in caplog.text - assert "kind=stream" in caplog.text - assert "model=gpt-5.1" in caplog.text - assert "transport=http" in caplog.text - assert "status=success" in caplog.text - assert "requested_service_tier=priority" in caplog.text - assert "actual_service_tier=default" in caplog.text + assert len(request_logs.calls) == 1 + request_log = request_logs.calls[0] + assert request_log["request_id"] == "resp_trace_stream" + assert request_log["status"] == expected_status + assert request_log["service_tier"] == expected_service_tier + assert request_log["requested_service_tier"] == "priority" + assert request_log["actual_service_tier"] == expected_actual_tier + assert counter.samples == ( + [ + { + "labels": {"kind": "stream", "requested_tier": "priority", "actual_tier": "default"}, + "value": 1.0, + } + ] + if terminal_case == "conflicting-completed-tier" + else [] + ) + if trace_enabled: + assert f"request_id={request_id}" in caplog.text + assert "response_id=resp_trace_stream" in caplog.text + assert "kind=stream" in caplog.text + assert "model=gpt-5.1" in caplog.text + assert "transport=http" in caplog.text + assert "status=success" in caplog.text + assert "requested_service_tier=priority" in caplog.text + assert "actual_service_tier=default" in caplog.text + else: + assert "proxy_service_tier_trace" not in caplog.text @pytest.mark.asyncio @@ -11031,14 +11209,24 @@ async def fake_stream(payload, headers, access_token, account_id, base_url=None, @pytest.mark.asyncio -async def test_compact_responses_logs_service_tier_trace_and_generates_request_id(monkeypatch, caplog): - settings = _make_proxy_settings(log_proxy_service_tier_trace=True) +@pytest.mark.parametrize( + ("actual_tier", "trace_enabled"), + [("default", False), ("priority", True)], + ids=["mismatch-with-trace-disabled", "matching-with-trace-enabled"], +) +async def test_compact_responses_logs_service_tier_trace_and_generates_request_id( + monkeypatch, caplog, actual_tier: str, trace_enabled: bool +): + settings = _make_proxy_settings(log_proxy_service_tier_trace=trace_enabled) request_logs = _RequestLogsRecorder() + counter = _ObservedCounter() service = proxy_service.ProxyService(_repo_factory(request_logs)) account = _make_account("acc_trace_compact") monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings)) monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(proxy_service, "PROMETHEUS_AVAILABLE", True) + monkeypatch.setattr(proxy_service, "service_tier_mismatch_total", counter) monkeypatch.setattr( service._load_balancer, "select_account", @@ -11047,8 +11235,15 @@ async def test_compact_responses_logs_service_tier_trace_and_generates_request_i monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(return_value=account)) monkeypatch.setattr(service, "_settle_compact_api_key_usage", AsyncMock(return_value=(True, True))) - async def fake_compact(payload, headers, access_token, account_id): - return OpenAIResponsePayload.model_validate({"output": [], "service_tier": "default"}) + async def fake_compact(payload, headers, access_token, account_id) -> CompactResponsePayload: + return CompactResponsePayload.model_validate( + { + "object": "response.compaction", + "id": "resp_trace_compact", + "output": [], + "service_tier": actual_tier, + } + ) monkeypatch.setattr(proxy_service, "core_compact_responses", fake_compact) @@ -11069,16 +11264,38 @@ async def fake_compact(payload, headers, access_token, account_id): finally: reset_request_id(token) - assert proxy_service._service_tier_from_response(response) == "default" - assert request_logs.calls[0]["service_tier"] == "default" + assert response.model_dump(exclude_none=True) == { + "object": "response.compaction", + "id": "resp_trace_compact", + "output": [], + "service_tier": actual_tier, + } + assert len(request_logs.calls) == 1 + assert request_logs.calls[0]["status"] == "success" + assert request_logs.calls[0]["service_tier"] == actual_tier assert request_logs.calls[0]["requested_service_tier"] == "priority" - assert request_logs.calls[0]["actual_service_tier"] == "default" + assert request_logs.calls[0]["actual_service_tier"] == actual_tier assert request_id - assert f"request_id={request_id}" in caplog.text - assert "kind=compact" in caplog.text - assert "requested_service_tier=priority" in caplog.text - assert "actual_service_tier=default" in caplog.text + assert request_logs.calls[0]["request_id"] == request_id + if trace_enabled: + assert f"request_id={request_id}" in caplog.text + assert "kind=compact" in caplog.text + assert "status=success" in caplog.text + assert "requested_service_tier=priority" in caplog.text + assert f"actual_service_tier={actual_tier}" in caplog.text + else: + assert "proxy_service_tier_trace" not in caplog.text assert request_logs.calls[0]["transport"] == "http" + assert counter.samples == ( + [ + { + "labels": {"kind": "compact", "requested_tier": "priority", "actual_tier": "default"}, + "value": 1.0, + } + ] + if actual_tier == "default" + else [] + ) @pytest.mark.asyncio @@ -18330,8 +18547,12 @@ class Settings: fresh_payload = json.loads(prepared.request_state.fresh_upstream_request_text) assert "previous_response_id" not in fresh_payload assert fresh_payload["input"] == [*replay_safe_historical_input, new_input] - assert inbound_input[1]["id"] == "tsc_injected_replay" - assert inbound_input[2]["id"] == "tso_injected_replay" + tool_search_call = inbound_input[1] + tool_search_output = inbound_input[2] + assert isinstance(tool_search_call, dict) + assert isinstance(tool_search_output, dict) + assert tool_search_call["id"] == "tsc_injected_replay" + assert tool_search_output["id"] == "tso_injected_replay" assert prepared.request_state.previous_response_id == "resp_completed_tool_search_anchor" @@ -18809,6 +19030,8 @@ class Settings: fresh_payload = json.loads(prepared.request_state.fresh_upstream_request_text) assert "previous_response_id" not in fresh_payload assert fresh_payload["input"] == replay_safe_full_resend_input + assert isinstance(tool_search_call, dict) + assert isinstance(tool_search_output, dict) assert tool_search_call["id"] == "tsc_replayed" assert tool_search_output["id"] == "tso_replayed" @@ -19226,8 +19449,12 @@ class Settings: assert upstream_payload["input"] == full_resend_input assert prepared.request_state.fresh_upstream_request_is_retry_safe is False assert prepared.request_state.fresh_upstream_request_text is None - assert full_resend_input[1]["id"] == "tsc_server" - assert full_resend_input[2]["id"] == "tso_server" + tool_search_call = full_resend_input[1] + tool_search_output = full_resend_input[2] + assert isinstance(tool_search_call, dict) + assert isinstance(tool_search_output, dict) + assert tool_search_call["id"] == "tsc_server" + assert tool_search_output["id"] == "tso_server" @pytest.mark.asyncio @@ -19298,7 +19525,9 @@ class Settings: assert upstream_payload["input"] == full_resend_input assert prepared.request_state.fresh_upstream_request_is_retry_safe is False assert prepared.request_state.fresh_upstream_request_text is None - assert full_resend_input[1]["id"] == "tso_orphan" + tool_search_output = full_resend_input[1] + assert isinstance(tool_search_output, dict) + assert tool_search_output["id"] == "tso_orphan" @pytest.mark.asyncio @@ -19376,7 +19605,9 @@ class Settings: assert upstream_payload["input"] == full_resend_input assert prepared.request_state.fresh_upstream_request_is_retry_safe is False assert prepared.request_state.fresh_upstream_request_text is None - assert full_resend_input[1]["id"] == "tsc_incomplete" + tool_search_call = full_resend_input[1] + assert isinstance(tool_search_call, dict) + assert tool_search_call["id"] == "tsc_incomplete" @pytest.mark.asyncio @@ -26735,7 +26966,8 @@ async def core_stream(*_args: object, **_kwargs: object): ) monkeypatch.setattr(service, "_ensure_fresh_with_budget", AsyncMock(return_value=account)) monkeypatch.setattr(service, "_settle_stream_api_key_usage", settle_usage) - monkeypatch.setattr(service, "_handle_stream_error", AsyncMock()) + handle_stream_error = AsyncMock() + monkeypatch.setattr(service, "_handle_stream_error", handle_stream_error) monkeypatch.setattr(service._load_balancer, "record_success", AsyncMock()) payload = ResponsesRequest.model_validate( @@ -26757,7 +26989,7 @@ async def core_stream(*_args: object, **_kwargs: object): assert any('"code":"previous_response_owner_unavailable"' in chunk for chunk in chunks) assert order == ["settle"] - service._handle_stream_error.assert_not_awaited() + handle_stream_error.assert_not_awaited() service._load_balancer.record_success.assert_not_awaited() @@ -27063,7 +27295,8 @@ async def select_account(*_args: object, **kwargs: object) -> AccountSelection: monkeypatch.setattr(service, "_ensure_fresh_with_budget", AsyncMock(side_effect=lambda account, **kwargs: account)) monkeypatch.setattr(service, "_stream_once", fake_stream_once) monkeypatch.setattr(service, "_settle_stream_api_key_usage", settle_usage) - monkeypatch.setattr(service, "_handle_stream_error", AsyncMock(side_effect=handle_stream_error)) + handle_stream_error_mock = AsyncMock(side_effect=handle_stream_error) + monkeypatch.setattr(service, "_handle_stream_error", handle_stream_error_mock) release_unsettled = AsyncMock(return_value=True) monkeypatch.setattr(service, "_release_unsettled_stream_api_key_usage", release_unsettled) monkeypatch.setattr(asyncio, "sleep", AsyncMock()) @@ -27088,7 +27321,7 @@ async def select_account(*_args: object, **kwargs: object) -> AccountSelection: assert "response.completed" in chunks[-1] assert stream_calls == [account_a.id, account_a.id, account_b.id] assert order == ["health:invalid_api_key", "settle"] - service._handle_stream_error.assert_awaited_once() + handle_stream_error_mock.assert_awaited_once() release_unsettled.assert_awaited_once() @@ -32779,7 +33012,9 @@ async def test_compact_successful_upstream_unconfirmed_settlement_raises_usage_s settle_result: tuple[bool, bool], ): settings = _make_proxy_settings() - service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder())) + request_logs = _RequestLogsRecorder() + counter = _ObservedCounter() + service = proxy_service.ProxyService(_repo_factory(request_logs)) account = _make_account("acc_compact_success_unconfirmed") api_key = _make_api_key_data("key_compact_success_unconfirmed") reservation = proxy_service.ApiKeyUsageReservationData( @@ -32792,6 +33027,8 @@ async def test_compact_successful_upstream_unconfirmed_settlement_raises_usage_s monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings)) monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(proxy_service, "PROMETHEUS_AVAILABLE", True) + monkeypatch.setattr(proxy_service, "service_tier_mismatch_total", counter) monkeypatch.setattr( service._load_balancer, "select_account", @@ -32804,10 +33041,16 @@ async def test_compact_successful_upstream_unconfirmed_settlement_raises_usage_s monkeypatch.setattr( proxy_service, "core_compact_responses", - AsyncMock(return_value=CompactResponsePayload.model_validate({"object": "response.compaction", "output": []})), + AsyncMock( + return_value=CompactResponsePayload.model_validate( + {"object": "response.compaction", "output": [], "service_tier": "default"} + ) + ), ) - payload = ResponsesCompactRequest.model_validate({"model": "gpt-5.1", "instructions": "hi", "input": []}) + payload = ResponsesCompactRequest.model_validate( + {"model": "gpt-5.1", "instructions": "hi", "input": [], "service_tier": "priority"} + ) with pytest.raises(proxy_module.ProxyResponseError) as exc_info: await service.compact_responses( payload, @@ -32820,6 +33063,13 @@ async def test_compact_successful_upstream_unconfirmed_settlement_raises_usage_s assert _proxy_error_code(exc_info.value) == "usage_settlement_failed" record_error.assert_not_awaited() handle_stream_error.assert_not_awaited() + assert len(request_logs.calls) == 1 + assert request_logs.calls[0]["status"] == "error" + assert request_logs.calls[0]["error_code"] == "usage_settlement_failed" + assert request_logs.calls[0]["service_tier"] == "default" + assert request_logs.calls[0]["requested_service_tier"] == "priority" + assert request_logs.calls[0]["actual_service_tier"] == "default" + assert counter.samples == [] @pytest.mark.asyncio diff --git a/tests/unit/test_request_decompression_middleware.py b/tests/unit/test_request_decompression_middleware.py index dc10b261d1..bdfc57d1fb 100644 --- a/tests/unit/test_request_decompression_middleware.py +++ b/tests/unit/test_request_decompression_middleware.py @@ -280,6 +280,7 @@ async def test_request_decompression_rejects_one_byte_over_decoded_boundary( assert response.status_code == 413 assert response.json()["error"]["code"] == "payload_too_large" + @pytest.mark.asyncio async def test_zstd_streams_without_one_shot_allocation_attempt(monkeypatch): payload = {"input": "x" * 512} @@ -315,6 +316,8 @@ def stream_reader(self, source): assert response.status_code == 200 assert response.json()["data"] == payload + + @pytest.mark.asyncio async def test_request_decompression_uses_openai_envelope_for_unsupported_encoding(monkeypatch): monkeypatch.setenv("CODEX_LB_MAX_DECOMPRESSED_BODY_BYTES", "2048") diff --git a/tests/unit/test_select_with_stickiness.py b/tests/unit/test_select_with_stickiness.py index b45eeaff75..f0c082b2ce 100644 --- a/tests/unit/test_select_with_stickiness.py +++ b/tests/unit/test_select_with_stickiness.py @@ -1082,6 +1082,8 @@ async def test_sticky_thread_below_threshold_does_not_reallocate(): async def test_fresh_sticky_mapping_uses_normal_budget_gate(): acc_a = _active("a", used_percent=10.0, secondary_used_percent=99.0) acc_b = _active("b", used_percent=20.0, secondary_used_percent=10.0) + acc_a.last_selected_at = 1.0 + acc_b.last_selected_at = 2.0 repo = _make_sticky_repo(existing_account_id=None) result = await _invoke_stickiness( diff --git a/tests/unit/test_settings_multi_replica.py b/tests/unit/test_settings_multi_replica.py index a86fd579b0..6c099bb8ab 100644 --- a/tests/unit/test_settings_multi_replica.py +++ b/tests/unit/test_settings_multi_replica.py @@ -78,11 +78,14 @@ def test_settings_metrics_port_from_env(monkeypatch, port: int): settings = Settings() assert settings.metrics_port == port + @pytest.mark.parametrize("port", [0, -1, 65536, 70000]) def test_settings_rejects_out_of_range_metrics_port(monkeypatch, port: int): monkeypatch.setenv("CODEX_LB_METRICS_PORT", str(port)) with pytest.raises(ValidationError, match="metrics_port"): Settings() + + def test_settings_rejects_metrics_port_2455(monkeypatch): monkeypatch.setenv("CODEX_LB_METRICS_PORT", "2455") with pytest.raises(ValidationError) as exc_info: @@ -96,10 +99,13 @@ def test_settings_log_format_from_env(monkeypatch, log_format: str): settings = Settings() assert settings.log_format == log_format + def test_settings_rejects_unknown_log_format(monkeypatch): monkeypatch.setenv("CODEX_LB_LOG_FORMAT", "jsoon") with pytest.raises(ValidationError, match="log_format"): Settings() + + def test_settings_conversation_archive_queue_max_bytes_from_env(monkeypatch): monkeypatch.setenv("CODEX_LB_CONVERSATION_ARCHIVE_QUEUE_MAX_BYTES", "16777216") settings = Settings() diff --git a/tests/unit/test_websocket_continuation_fold.py b/tests/unit/test_websocket_continuation_fold.py index e419bfe545..8197586490 100644 --- a/tests/unit/test_websocket_continuation_fold.py +++ b/tests/unit/test_websocket_continuation_fold.py @@ -447,6 +447,87 @@ def test_ws_fold_continues_truncated_round_then_reconstructs_final(decision_coun ] +@pytest.mark.parametrize("final_tier", ["priority", "default", None], ids=["priority", "default", "missing"]) +def test_ws_fold_completed_tier_comes_only_from_final_terminal(final_tier: str | None) -> None: + fold = _WebSocketContinuationFold( + CodexContinuationConfig(max_continue=2, rechunk_size=64), + { + "model": "gpt-5.5", + "input": [{"role": "user", "content": "question"}], + "service_tier": "priority", + "stream": True, + }, + ) + downstream: list[dict[str, Any]] = [] + # Both truncated terminals are suppressed, including one from a hidden round. + for round_number, response_id, input_tokens in [(1, "resp_visible", 100), (2, "resp_hidden", 120)]: + completed = _completed(response_id, input_tokens=input_tokens, output_tokens=600, reasoning_tokens=516) + completed["response"]["service_tier"] = "priority" + events = [ + { + "type": "response.created", + "response": {"id": response_id, "status": "in_progress", "output": [], "service_tier": "auto"}, + }, + *_reasoning_events(output_index=0, item_id=f"rs_{round_number}", encrypted_content=f"enc{round_number}"), + *_message_events(output_index=1, item_id=f"msg_partial_{round_number}", text="partial answer"), + completed, + ] + emitted, continuation, terminal = _drive(fold, events) + assert continuation is not None + assert terminal is None + assert all(event["type"] != "response.completed" for event in emitted) + downstream.extend(emitted) + + completed = _completed("resp_final", input_tokens=140, output_tokens=20, reasoning_tokens=10) + if final_tier is not None: + completed["response"]["service_tier"] = final_tier + emitted, continuation, terminal = _drive( + fold, + [ + {"type": "response.created", "response": {"id": "resp_final", "status": "in_progress", "output": []}}, + *_reasoning_events(output_index=0, item_id="rs_3", encrypted_content="enc3"), + *_message_events(output_index=1, item_id="msg_final", text="final answer"), + completed, + ], + ) + downstream.extend(emitted) + assert continuation is None + downstream_completed = [event for event in downstream if event["type"] == "response.completed"] + assert downstream_completed == [terminal] + response = downstream_completed[0]["response"] + if final_tier is None: + assert "service_tier" not in response + else: + assert response["service_tier"] == final_tier + assert response["id"] == "resp_visible" + assert response["status"] == "completed" + assert response["output"] == [ + {"id": "rs_1", "type": "reasoning", "encrypted_content": "enc1"}, + {"id": "rs_2", "type": "reasoning", "encrypted_content": "enc2"}, + {"id": "rs_3", "type": "reasoning", "encrypted_content": "enc3"}, + { + "id": "msg_final", + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "final answer"}], + }, + ] + assert response["usage"] == { + "input_tokens": 100, + "output_tokens": 1052, + "total_tokens": 1152, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 1042}, + } + assert response["metadata"]["proxy_billed_usage"] == { + "input_tokens": 360, + "output_tokens": 1220, + "total_tokens": 1580, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 1042}, + } + + def test_folded_terminal_function_call_ids_prune_only_delivered_calls() -> None: # Defense-in-depth invariant: the relay prunes pending-call tracking to # calls present in the folded terminal's delivered output. No fold mode diff --git a/tests/unit/test_websocket_upstream_transport_observability.py b/tests/unit/test_websocket_upstream_transport_observability.py index 89fd489395..d846c8e404 100644 --- a/tests/unit/test_websocket_upstream_transport_observability.py +++ b/tests/unit/test_websocket_upstream_transport_observability.py @@ -395,6 +395,7 @@ def record_metric(**labels: object) -> None: } ] + @pytest.mark.asyncio async def test_fail_pending_websocket_requests_attributes_request_state_api_key(monkeypatch): """Bridge failure fan-out passes api_key=None at the session level; each