mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix: reconstruct non-stream response output
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
Codex
+1
-1
@@ -43,7 +43,7 @@ Model name parsing supports configured aliases plus `-low`, `-medium`, `-high`,
|
||||
|
||||
Uses imported Codex accounts to call `POST /codex/responses` on the configured Codex backend. The upstream request is sent with Codex Desktop headers, account bearer token, optional account id, request id, and encrypted account-scoped Cookie replay.
|
||||
|
||||
When the client omits `stream` or sets `"stream": false`, the Rust route collects upstream HTTP SSE until `response.completed` and returns the completed OpenAI-compatible response JSON.
|
||||
When the client omits `stream` or sets `"stream": false`, the Rust route collects upstream HTTP SSE until `response.completed` and returns the completed OpenAI-compatible response JSON. If upstream sends assistant output through `response.output_item.done` or `response.output_text.delta` events while the completed payload has an empty `output`, the route reconstructs a standard assistant `output` message and synchronizes `output_text`.
|
||||
|
||||
When the client sets `"stream": true`, the Rust route returns `text/event-stream` and passes through upstream SSE frames while collecting usage for the account after the stream finishes. Both modes capture upstream `Set-Cookie`, replay encrypted account-scoped Cookies, record account usage when usage appears in SSE, and write a `v1.response` event log with `requestId`, `accountId`, `route`, `model`, `statusCode`, `latencyMs`, and non-secret metadata.
|
||||
|
||||
|
||||
@@ -16,11 +16,11 @@ Update this file before each feature commit.
|
||||
| Account pool and refresh | Partial | `0f97c5c` + `2f25db8` + `4dc0c9a` + `c2cf7a6` + `636fa79` + `fdb9374` | `cargo test account_pool_skips_expired_disabled_banned_and_quota_exhausted_accounts`; `cargo test --test account_pool_scheduling_test`; `cargo test --test account_refresh_test`; `cargo test --test oauth_refresh_test`; `cargo test --test v1_upstream_route_test`; `cargo test --test account_pool_startup_test`; full `fmt/test/clippy` before commit | Implements base account status filtering, refresh scheduler semantics, real OpenAI OAuth refresh-token exchange, 401-triggered refresh and single retry for `/v1/responses`, max concurrent slots, stale slot cleanup, least-used/round-robin/sticky strategies, tier priority, quota-cache filtering, Cloudflare cooldown skip, request staggering metadata, preferred account selection, exclude IDs, capacity summary, settings-backed pool construction, same-process import-to-pool sync, and startup pool snapshots from SQLite with usage-based last-used metadata. Still missing durable release-time slot state, broader upstream retry/fallback wiring, and admin account CRUD. |
|
||||
| Account repository | Completed | `14fe706` | `cargo test --test account_repository_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds encrypted SQLite account token storage, decrypting reads, cursor pagination, status/label updates, release-time usage counter persistence, and refresh-token-preserving token updates. |
|
||||
| Translation | Completed | `eff182e` | `cargo test chat_completion_translates_to_codex_response_request`; full `fmt/test/clippy` before commit | Adds OpenAI Chat request types, Codex Responses request type, translation, and OpenAI-compatible error body helper. |
|
||||
| HTTP routes | Partial | `ab0ad8f` + `3830b21` + `6179d44` + `cfec8ba` + `6341336` + `228febd` + `c2cf7a6` + `e9fdea0` + `ee0ed90` + `fdb9374` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; `cargo test --test admin_settings_route_test`; `cargo test --test model_catalog_test`; `cargo test --test admin_accounts_route_test`; `cargo test --test admin_api_keys_route_test`; `cargo test --test admin_usage_stats_route_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/debug/models`, `/admin/accounts`, `/admin/accounts/import`, `/admin/api-keys`, `/admin/usage-stats`, `/admin/usage-stats/summary`, `/v1/responses` HTTP SSE collection and `stream:true` passthrough through imported Codex accounts, `/v1` unauthorized OpenAI error body, global request-id response header, binary listener, `gpt-5.5` as the default/test model baseline, and read-only `/admin/settings` with admin-session auth and lower camelCase envelope. `/admin/accounts` reads metadata only and does not decrypt tokens for list views. Still missing OAuth/device/CLI account login flows, admin account mutation/health/quota/Cookie routes, full admin API-key mutation/export/import routes, settings mutation/persistence, diagnostics, backend model refresh, model cache persistence, and WebSocket Responses routes. |
|
||||
| HTTP routes | Partial | `ab0ad8f` + `3830b21` + `6179d44` + `cfec8ba` + `6341336` + `228febd` + `c2cf7a6` + `e9fdea0` + `ee0ed90` + `fdb9374` + `pending` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; `cargo test --test admin_settings_route_test`; `cargo test --test model_catalog_test`; `cargo test --test admin_accounts_route_test`; `cargo test --test admin_api_keys_route_test`; `cargo test --test admin_usage_stats_route_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/debug/models`, `/admin/accounts`, `/admin/accounts/import`, `/admin/api-keys`, `/admin/usage-stats`, `/admin/usage-stats/summary`, `/v1/responses` HTTP SSE collection with output item/delta reconstruction plus `output_text` sync for non-streaming clients and `stream:true` passthrough through imported Codex accounts, `/v1` unauthorized OpenAI error body, global request-id response header, binary listener, `gpt-5.5` as the default/test model baseline, and read-only `/admin/settings` with admin-session auth and lower camelCase envelope. `/admin/accounts` reads metadata only and does not decrypt tokens for list views. Still missing real `/v1/chat/completions` route wiring, OAuth/device/CLI account login flows, admin account mutation/health/quota/Cookie routes, full admin API-key mutation/export/import routes, settings mutation/persistence, diagnostics, backend model refresh, model cache persistence, and WebSocket Responses routes. |
|
||||
| Admin/frontend response contract | Completed | `1c34ea7` + `b0c10f2` | `cargo test admin_envelope_serializes_lower_camel_case_request_id`; `cargo test admin_logs_reject_missing_admin_session_cookie`; `cargo test v1_response_has_request_id_header_without_admin_body_field`; full `fmt/test/clippy` before commit | Adds reusable admin envelopes with body codes, lower camelCase JSON, paginated list shape, requestId bodies, admin session cookie enforcement, and keeps `/v1/*` OpenAI-compatible. |
|
||||
| HTTP auth boundary | Completed | `cec4607` + `e9fdea0` | `cargo test --test http_auth_test`; `cargo test --test routes_responses_test`; `cargo test --test admin_api_keys_route_test`; `cargo test --test admin_logs_route_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds a real `src/http/auth.rs` boundary for client API key and admin session cookie parsing. `/v1/*` accepts only enabled local `Bearer cpr_...` client API keys verified from SQLite, admin session parsing ignores client API keys, and route handlers no longer own ad hoc credential parsing. |
|
||||
| Model catalog | Partial | `6341336` | `cargo test --test model_catalog_test`; `cargo test model_catalog_route_returns_codex_metadata_without_alias_entries`; full `fmt/test/clippy` before commit | Adds a native static Codex model catalog with `gpt-5.5` default, configured aliases, reasoning/service-tier suffix parsing, OpenAI-compatible `/v1/models`, extended catalog/detail/info routes, and debug summary. Still missing backend Codex model fetching per plan, cache persistence, `/admin/refresh-models`, plan allowlist integration from fetched models, and full model metadata parity. |
|
||||
| Upstream lifecycle | Partial | `88cad87` + `225fbff` + `c2cf7a6` + `636fa79` + `ee0ed90` + `fdb9374` + `5bf012c` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; `cargo test --test codex_client_test`; `cargo test --test oauth_refresh_test`; `cargo test --test v1_upstream_route_test`; `cargo test --test account_pool_startup_test`; full `fmt/test/clippy` before commit | `/v1` enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, includes reusable SSE/usage helpers, sends Codex Desktop HTTP SSE requests through imported accounts, replays/captures encrypted Cookies, collects `response.completed` for non-streaming clients, passes through client `stream:true` SSE, refreshes and retries once after upstream 401 when a refresh token is stored, records token usage, writes v1 lifecycle event logs, records refresh failure classification for upstream 401, and restores the account pool from SQLite on startup. Still missing rate-limit header capture, retry/fallback across accounts beyond refresh retry, 429 quota cache, 402/403 classification, and previous-response/session affinity. |
|
||||
| Upstream lifecycle | Partial | `88cad87` + `225fbff` + `c2cf7a6` + `636fa79` + `ee0ed90` + `fdb9374` + `5bf012c` + `pending` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; `cargo test --test codex_client_test`; `cargo test --test oauth_refresh_test`; `cargo test --test v1_upstream_route_test`; `cargo test --test account_pool_startup_test`; full `fmt/test/clippy` before commit | `/v1` enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, includes reusable SSE/usage helpers, sends Codex Desktop HTTP SSE requests through imported accounts, replays/captures encrypted Cookies, collects `response.completed`, reconstructs missing assistant output from `response.output_item.done` or text deltas, synchronizes `output_text` for non-streaming clients, passes through client `stream:true` SSE, refreshes and retries once after upstream 401 when a refresh token is stored, records token usage, writes v1 lifecycle event logs, records refresh failure classification for upstream 401, and restores the account pool from SQLite on startup. Still missing Codex request context fields, SSE error/failed mapping, rate-limit header capture, retry/fallback across accounts beyond refresh retry, 429 quota cache, 402/403 classification, and previous-response/session affinity. |
|
||||
| Codex WebSocket boundary | Completed | `cc0516d` | `cargo test --test codex_websocket_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds `src/codex/websocket.rs` with tested transport classification for HTTP SSE versus WebSocket-only requests. `previous_response_id` and explicit WebSocket requests now fail through a typed boundary instead of being silently downgraded; real WebSocket transport remains an upstream lifecycle follow-up. |
|
||||
| Fingerprint updates | Partial | `dd26d31` + `e32220a` | `cargo test update_manifest_updates_app_version_and_build_number`; `cargo test fingerprint_updater_fetches_manifest_and_persists_history`; full `fmt/test/clippy` before commit | Parses a JSON update manifest and persists app version/build number history to SQLite. Still missing original appcast XML polling, update-state persistence, runtime header fingerprint selection from the latest stored fingerprint, and optional local Codex.app Chromium extraction. |
|
||||
| Runtime docs and packaging | Completed | `33b63f3` | `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test`; `cargo run`; `curl -s -i http://127.0.0.1:8080/health` | README scope, `.env.example`, Dockerfile, compose file, and health smoke test are complete. |
|
||||
|
||||
@@ -2969,6 +2969,7 @@ git commit -m "docs: document codex-proxy-rs scope"
|
||||
- [x] Add tested `/v1/responses` streaming passthrough and lifecycle event logs. Client `stream:true` returns `text/event-stream`, keeps the account slot until stream completion or disconnect, parses usage after SSE completion, writes account usage, and records non-secret `v1.response` logs with request/account/route/model/status/latency metadata. Non-streaming calls now write the same lifecycle log after `response.completed`.
|
||||
- [x] Add tested admin usage-stat read APIs. `/admin/usage-stats` returns cursor-paginated account usage counters, `/admin/usage-stats/summary` returns global totals, and both require the admin session cookie while exposing only non-secret account metadata.
|
||||
- [x] Add tested upstream-401 refresh failure event logs. When refresh fails, the service records a non-secret `account.refresh` event with request/account/model context, failure classification, and resulting account status.
|
||||
- [x] Add tested non-streaming output reconstruction. When Codex SSE carries assistant output in `response.output_item.done` or `response.output_text.delta` events but `response.completed.output` is empty, `/v1/responses` now returns a standard assistant `output` message and synchronizes `output_text` for non-streaming clients.
|
||||
- [x] Add real OpenAI OAuth refresh-token exchange and `/v1/responses` 401 retry wiring. The Rust service refreshes access tokens through the OAuth token endpoint, preserves the existing refresh token when the server omits a replacement, persists the rotated access token, updates the in-process account pool, and retries the failed non-streaming or streaming request once.
|
||||
|
||||
## 2026-06-11 Original Project Gap Audit
|
||||
@@ -2987,10 +2988,30 @@ The Rust project has no legacy compatibility requirement, but the in-scope behav
|
||||
- [ ] **Logs and usage stats:** add `/admin/logs/state`, clear/detail, error-log equivalents if retained, and richer bounded history queries if needed. Keep log file rotation and SQLite event pagination. Initial `/admin/usage-stats` pagination and `/admin/usage-stats/summary` are complete.
|
||||
- [ ] **Model catalog:** replace the hard-coded `/v1/models` list with a model store that supports static defaults, configured aliases, `-low/-medium/-high/-xhigh/-fast/-flex` suffix parsing, backend model fetching per plan, cache persistence, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/admin/refresh-models`, and `/debug/models`. Static defaults, aliases, suffix parsing, read-only model routes, and debug summary are complete. Backend fetch/cache, plan snapshots, and admin refresh remain.
|
||||
- [ ] **Account scheduling parity:** extend account acquisition beyond `active + least last_used`. Completed in the in-memory pool: max concurrent slots per account, stale slot cleanup, least-used/round-robin/sticky strategies, quota-cache skip, tier priority, model-plan filtering, exclude IDs, preferred account from session affinity, Cloudflare cooldown skip, request-staggering metadata, capacity summary, settings-backed construction, import-to-pool sync, and startup pool snapshots from SQLite. Remaining: durable release-time slot state and upstream retry/fallback wiring.
|
||||
- [ ] **Upstream Codex lifecycle:** implement account acquire -> Codex request -> SSE/collect-or-stream -> rate-limit capture -> usage release -> retry/fallback. Initial HTTP SSE collection, streaming passthrough, and 401-triggered refresh retry are complete: `/v1/responses` acquires imported/restored accounts, sends Codex Desktop headers, replays/captures Cookies, collects `response.completed` for non-streaming clients, passes through `stream:true` SSE, refreshes and retries once after upstream 401, records refresh failure classification when refresh cannot recover the account, records usage, writes lifecycle logs, and returns the correct OpenAI-compatible JSON or SSE shape. Remaining: rate-limit capture, retry/fallback across alternate accounts, 429 quota cache, 402 quota exhausted, 403 ban/Cloudflare distinction, path-block Cookie clearing, and richer OpenAI-compatible error mapping.
|
||||
- [ ] **Upstream Codex lifecycle:** implement account acquire -> Codex request -> SSE/collect-or-stream -> rate-limit capture -> usage release -> retry/fallback. Initial HTTP SSE collection, streaming passthrough, and 401-triggered refresh retry are complete: `/v1/responses` acquires imported/restored accounts, sends Codex Desktop headers, replays/captures Cookies, collects `response.completed` for non-streaming clients, reconstructs missing assistant output from done items or text deltas, passes through `stream:true` SSE, refreshes and retries once after upstream 401, records refresh failure classification when refresh cannot recover the account, records usage, writes lifecycle logs, and returns the correct OpenAI-compatible JSON or SSE shape. Remaining: full Codex request context fields, SSE `error`/`response.failed` mapping, rate-limit capture, retry/fallback across alternate accounts, 429 quota cache, 402 quota exhausted, 403 ban/Cloudflare distinction, path-block Cookie clearing, and richer OpenAI-compatible error mapping.
|
||||
- [ ] **Session affinity and previous response:** persist an in-memory response-id to account mapping with TTL, prompt-cache/conversation identity, turn-state replay, and preferred account selection. `src/codex/websocket.rs` must either implement the WebSocket path with verified TLS/header behavior or the API must reject WebSocket-only cases explicitly with documented limitations until fingerprint parity is proven.
|
||||
- [ ] **Fingerprint auto-update parity:** poll the real Codex Desktop appcast/update source, parse app version/build number, persist update state, select the latest stored fingerprint for runtime headers, and optionally extract Chromium version from a local Codex.app path when configured.
|
||||
|
||||
## 2026-06-12 Detailed Original Coverage Audit
|
||||
|
||||
This pass re-read the TypeScript routes and core modules under `/home/zyy/Codes/codex-proxy` and compared them with the Rust implementation. Deleted provider/proxy/Ollama/Electron features stay deleted; the items below are in-scope Codex/OpenAI-compatible behavior that is not yet fully rebuilt.
|
||||
|
||||
- [ ] **OpenAI Chat Completions route wiring:** Rust mounts `/v1/chat/completions` to the Responses handler, but that handler parses Responses input and ignores Chat `messages`. `src/translation/openai_to_codex.rs` exists and is unit-tested, but the route is not connected and no ChatCompletion response/SSE translation is wired.
|
||||
- [ ] **Responses request semantics:** Rust accepts only a minimal Responses body. Missing retained fields from the original include `service_tier` normalization (`fast` -> `priority` upstream), `prompt_cache_key`, `client_metadata`, `include`, `text.format` / tuple-schema reconversion, `tool_choice`, `parallel_tool_calls`, `Version`, `x-codex-turn-metadata`, `x-codex-beta-features`, `x-responsesapi-include-timing-metrics`, `x-codex-parent-thread-id`, `x-codex-window-id`, `session_id`, `x-codex-installation-id`, and sanitized reasoning input behavior.
|
||||
- [ ] **Responses stream/error parity:** Rust now reconstructs non-stream output from `response.output_item.done` and `response.output_text.delta`, but still ignores SSE `error` and `response.failed` in the non-stream collector, lacks upstream premature-close classification, empty-response retry, stream close `response.failed` injection, tuple reconversion, reasoning replay metadata, function-call metadata, image-generation usage, and detailed OpenAI-compatible error mapping.
|
||||
- [ ] **Account login flows:** Rust has admin login and account JSON import only. Still missing account-login flows for OAuth PKCE login-start/code-relay/callback, device-code login/poll, Codex CLI auth import, validated manual access-token import, account logout, and `/auth/status`-equivalent admin status. These must be rebuilt as admin/session-gated account-login endpoints, not by treating local client API keys as passwords.
|
||||
- [ ] **Admin account management:** Rust has paginated sanitized list and import. Missing export, manual add/validation, batch delete, batch status, per-account refresh/probe, label mutation, delete, reset usage, quota fetch/update, Cookie get/set/delete, quota warnings, persistence-health reporting, and JWT metadata/expires extraction during import. SQLite already has `quota_json`, but the repository/pool do not yet restore or update quota windows.
|
||||
- [ ] **Client API key management:** Rust creates/lists local `cpr_` keys and verifies them for `/v1/*`. Missing delete, batch-delete, label mutation, enable/disable, export, and import. Do not port the TypeScript provider/model binding or provider model-fetch APIs.
|
||||
- [ ] **Settings:** Rust exposes read-only `/admin/settings` for retained fields. Missing admin-session-gated mutation/writeback with validation for default model, default reasoning effort, service tier, aliases, refresh settings, max concurrent accounts, request interval, rotation strategy, tier priority, quota refresh/threshold/skip settings, log state/capacity/body capture, and usage retention. Do not re-add proxy URL, provider, Ollama, proxy API key, or Electron/self-update settings.
|
||||
- [ ] **Logs and error diagnostics:** Rust has cursor-paginated SQLite event logs and rotating files. Missing `/admin/logs/state`, pause/enable mutation, clear, detail-by-id, direction/search filters, bounded in-memory request/response body capture, and the original local error-log grouping/raw/count/seen/report/clear APIs if retained for troubleshooting.
|
||||
- [ ] **Usage and quota background jobs:** Rust records cumulative account usage from `/v1/*`. Missing usage history snapshots (`raw`, `five_min`, `hourly`, `daily`), retention cleanup, passive quota-window recovery, active quota refresher for dirty/limit-reached accounts, quota warning state, and startup background scheduling.
|
||||
- [ ] **Model catalog:** Rust has a static default catalog, aliases, suffix parsing, read routes, and debug summary. Missing `models.yaml`/cache loading, backend Codex model fetch/probe, per-plan snapshots, model-plan index feeding account scheduling, richer metadata fields, `/admin/refresh-models`, and runtime cache persistence.
|
||||
- [ ] **Account scheduling and retry:** Rust has in-memory slots, stale cleanup, strategies, tier priority, preferred/exclude IDs, and SQLite startup restore. Missing request-interval staggering enforcement, durable quota/cache restore, 429 `Retry-After`/rate-limit header application, fallback across alternate accounts, Cloudflare path-block tracker, cooldown clearing, WebSocket pool eviction on status/token mutation, and richer capacity/health endpoints.
|
||||
- [ ] **Refresh scheduler parity:** Rust has typed refresh policy and 401-triggered refresh retry. Missing long-running refresh timers at `exp - margin`, crash recovery for `refreshing`/`expired` accounts, exponential backoff, recovery scheduling, permanent-failure thresholding, per-account in-flight suppression, cross-process refresh locks, and disk refresh-token sync before consuming one-time RTs.
|
||||
- [ ] **Codex Desktop fingerprint and transport parity:** Rust uses pinned `reqwest`/`rustls` and core headers, but exact parity is not proven. Missing header-order/default-header parity (`sec-ch-ua`, accept language/encoding/fetch headers), `x-codex-installation-id`, runtime selection of the newest stored fingerprint, Sparkle appcast XML polling, update-state/version-state persistence, optional local Codex.app Chromium extraction, and real Desktop TLS/header capture comparison. `x-openai-internal-codex-residency` currently needs real-client verification because the builder inserts `global` and the reqwest client overwrites it with `us`.
|
||||
- [ ] **Health and diagnostics:** Rust `/health` only returns `status: ok`. Missing authenticated state, pool summary, capacity summary, transport/fingerprint status, paths, runtime metadata, local-only `/debug/fingerprint`, local-only `/debug/diagnostics`, and upstream test-connection/probe routes.
|
||||
- [ ] **Background service lifecycle:** Rust startup restores accounts and initializes tracing, but does not start/stop scheduler tasks for account refresh, model refresh, quota refresh, active quota refresh, session cleanup, fingerprint polling, or graceful shutdown drains. Ollama, proxy health timers, self-update, Electron, and provider router startup remain intentionally deleted.
|
||||
|
||||
### Keep Removed
|
||||
|
||||
- [ ] Do not rebuild per-account proxy assignment, proxy pool, proxy health checks, or proxy import/export.
|
||||
|
||||
+96
-6
@@ -792,18 +792,108 @@ fn ensure_stream_metadata(metadata: &mut Value, stream_value: bool) {
|
||||
|
||||
fn completed_response_json(body: &str) -> Result<Option<Value>, crate::codex::sse::SseError> {
|
||||
let events = parse_sse_events(body)?;
|
||||
let mut output_text = String::new();
|
||||
let mut output_items = Vec::new();
|
||||
let mut completed_response = None;
|
||||
for event in events {
|
||||
if event.event.as_deref() != Some("response.completed") {
|
||||
continue;
|
||||
}
|
||||
let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
|
||||
continue;
|
||||
};
|
||||
if let Some(response) = value.get("response") {
|
||||
return Ok(Some(response.clone()));
|
||||
match event.event.as_deref() {
|
||||
Some("response.output_text.delta") => {
|
||||
if let Some(delta) = value.get("delta").and_then(Value::as_str) {
|
||||
output_text.push_str(delta);
|
||||
}
|
||||
}
|
||||
Some("response.output_item.done") => {
|
||||
if let Some(item) = value.get("item") {
|
||||
output_items.push(item.clone());
|
||||
}
|
||||
}
|
||||
Some("response.completed") => {
|
||||
if let Some(response) = value.get("response") {
|
||||
completed_response = Some(response.clone());
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
let Some(mut response) = completed_response else {
|
||||
return Ok(None);
|
||||
};
|
||||
ensure_completed_response_output(&mut response, &output_items, &output_text);
|
||||
sync_output_text_from_output(&mut response);
|
||||
Ok(Some(response))
|
||||
}
|
||||
|
||||
fn ensure_completed_response_output(
|
||||
response: &mut Value,
|
||||
output_items: &[Value],
|
||||
output_text: &str,
|
||||
) {
|
||||
let output_is_empty = response
|
||||
.get("output")
|
||||
.and_then(Value::as_array)
|
||||
.is_none_or(Vec::is_empty);
|
||||
if !output_is_empty {
|
||||
return;
|
||||
}
|
||||
|
||||
if !output_items.is_empty() {
|
||||
response["output"] = Value::Array(output_items.to_vec());
|
||||
return;
|
||||
}
|
||||
if output_text.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
// 原版 passthrough 会用 done item 或文本 delta 回填 completed.output，避免非流式客户端拿到空正文。
|
||||
response["output"] = json!([{
|
||||
"type": "message",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{
|
||||
"type": "output_text",
|
||||
"text": output_text,
|
||||
"annotations": []
|
||||
}]
|
||||
}]);
|
||||
}
|
||||
|
||||
fn sync_output_text_from_output(response: &mut Value) {
|
||||
let output_text_is_empty = response
|
||||
.get("output_text")
|
||||
.and_then(Value::as_str)
|
||||
.is_none_or(str::is_empty);
|
||||
if !output_text_is_empty {
|
||||
return;
|
||||
}
|
||||
let Some(items) = response.get("output").and_then(Value::as_array) else {
|
||||
return;
|
||||
};
|
||||
let texts = items
|
||||
.iter()
|
||||
.filter_map(output_text_from_item)
|
||||
.collect::<Vec<_>>();
|
||||
if texts.is_empty() {
|
||||
return;
|
||||
}
|
||||
response["output_text"] = Value::String(texts.join("\n\n"));
|
||||
}
|
||||
|
||||
fn output_text_from_item(item: &Value) -> Option<String> {
|
||||
let content = item.get("content")?.as_array()?;
|
||||
let text = content
|
||||
.iter()
|
||||
.filter_map(|part| {
|
||||
let part_type = part.get("type")?.as_str()?;
|
||||
if part_type != "output_text" && part_type != "text" {
|
||||
return None;
|
||||
}
|
||||
part.get("text")?.as_str()
|
||||
})
|
||||
.collect::<String>();
|
||||
(!text.is_empty()).then_some(text)
|
||||
}
|
||||
|
||||
fn codex_client_error_response(error: CodexClientError) -> (StatusCode, Json<Value>) {
|
||||
|
||||
@@ -332,6 +332,105 @@ async fn v1_responses_should_use_imported_account_and_record_usage() {
|
||||
assert_eq!(event.4["usage"]["inputTokens"], 7);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn v1_responses_should_reconstruct_non_stream_output_text_from_sse_deltas() {
|
||||
let server = MockServer::start().await;
|
||||
let sse_body = concat!(
|
||||
"event: response.output_text.delta\n",
|
||||
"data: {\"delta\":\"你好，我是\"}\n",
|
||||
"\n",
|
||||
"event: response.output_text.delta\n",
|
||||
"data: {\"delta\":\"一个中文助手。\"}\n",
|
||||
"\n",
|
||||
"event: response.completed\n",
|
||||
"data: {\"response\":{\"id\":\"resp_text\",\"object\":\"response\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":6,\"output_tokens\":4}}}\n",
|
||||
"\n",
|
||||
);
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/codex/responses"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.insert_header("content-type", "text/event-stream")
|
||||
.set_body_string(sse_body),
|
||||
)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let imported = build_imported_app(server.uri()).await;
|
||||
|
||||
let response = imported
|
||||
.app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("POST")
|
||||
.uri("/v1/responses")
|
||||
.header(
|
||||
"authorization",
|
||||
format!("Bearer {}", imported.client_api_key),
|
||||
)
|
||||
.header("content-type", "application/json")
|
||||
.body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#))
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body = response_json(response).await;
|
||||
assert_eq!(body["id"], "resp_text");
|
||||
assert_eq!(
|
||||
body["output"][0]["content"][0]["text"],
|
||||
"你好，我是一个中文助手。"
|
||||
);
|
||||
assert_eq!(body["output_text"], "你好，我是一个中文助手。");
|
||||
assert_eq!(body["output"][0]["role"], "assistant");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn v1_responses_should_use_done_output_items_when_completed_output_is_empty() {
|
||||
let server = MockServer::start().await;
|
||||
let sse_body = concat!(
|
||||
"event: response.output_item.done\n",
|
||||
"data: {\"item\":{\"type\":\"message\",\"status\":\"completed\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"来自 done item\"}]}}\n",
|
||||
"\n",
|
||||
"event: response.completed\n",
|
||||
"data: {\"response\":{\"id\":\"resp_item\",\"object\":\"response\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":5,\"output_tokens\":3}}}\n",
|
||||
"\n",
|
||||
);
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/codex/responses"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.insert_header("content-type", "text/event-stream")
|
||||
.set_body_string(sse_body),
|
||||
)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let imported = build_imported_app(server.uri()).await;
|
||||
|
||||
let response = imported
|
||||
.app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("POST")
|
||||
.uri("/v1/responses")
|
||||
.header(
|
||||
"authorization",
|
||||
format!("Bearer {}", imported.client_api_key),
|
||||
)
|
||||
.header("content-type", "application/json")
|
||||
.body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#))
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let body = response_json(response).await;
|
||||
assert_eq!(body["id"], "resp_item");
|
||||
assert_eq!(body["output"][0]["content"][0]["text"], "来自 done item");
|
||||
assert_eq!(body["output_text"], "来自 done item");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn v1_responses_should_passthrough_stream_and_record_usage_and_log() {
|
||||
let server = MockServer::start().await;
|
||||
|
||||
Reference in New Issue
Block a user