From ea7c4d0736afaa570e46448e2e58975fc328b90c Mon Sep 17 00:00:00 2001 From: jasper-yl Date: Mon, 2 Nov 2026 03:35:53 -0800 Subject: [PATCH 1/3] fix(gateway): forward Responses requests intact and replay streams after release Codex could not run through the Responses ingress: - Every input item with a role was rebuilt from role and content, so Codex's additional_tools item lost its tool definitions and developer messages became system messages without their type. - Unmodeled top-level fields (reasoning, include, prompt_cache_key, parallel_tool_calls) were silently dropped although Core admits them. - Streaming requests were forwarded as streaming and withheld by the send check, which only supports openai-chat streaming. Forward every item that is not a plain user message verbatim, pass the unmodeled fields through for Core's request allowlist to decide (dropping only the client-local client_metadata), always request a non-streaming response upstream, and replay a released response to streaming clients as Responses SSE events. Nothing is released before the release check. Refs #2398 Co-Authored-By: Claude Opus 5.5 --- gateway/README.md | 1 + gateway/crates/pc-gateway-app/src/ingress.rs | 255 ++++++++++++++++-- .../tests/executor_semantics.rs | 1 + .../pc-gateway-protocol/src/canonical.rs | 4 + .../crates/pc-gateway-protocol/src/render.rs | 16 +- .../tests/openai_renderer.rs | 6 + .../tests/openai_responses_renderer.rs | 9 + 7 files changed, 275 insertions(+), 17 deletions(-) diff --git a/gateway/README.md b/gateway/README.md index 13c2c3fcda..cccbc262f8 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -25,6 +25,7 @@ An optional Rust executor for PowerContext, defined by the gateway design in #21 - **Two deployment forms**: *combined* (Rust owns the public TCP port; Core listens only on a private Unix socket; all non-`/gateway` routes are reverse-proxied to Core over UDS) or *standalone* (TCP to a remote Core). - **UDS client**: hand-rolled HTTP/1.1 over the Unix socket with a pooled, keep-alive connection pool (bounded idle set, connect/read timeouts, stale-connection single retry, content-length truncation and chunked-trailer consumption — covered by a real-socket regression test). - **OpenAI-compatible ingress** at `/gateway/v1/chat/completions` and `/gateway/v1/responses` (same format in and out, no cross-format translation), authenticated by a gateway key that replaces provider keys. +- **Responses fidelity**: on `/gateway/v1/responses`, every input item other than a plain user message (developer messages, tool calls and outputs, reasoning, client-specific items such as Codex's `additional_tools`) is forwarded unchanged, and request fields the ingress does not model (for example `reasoning`, `include`, `prompt_cache_key`) are passed through for Core's request allowlist to decide; only the client-local `client_metadata` is dropped. The upstream request is always non-streaming; a streaming client receives the released response replayed as Responses SSE events (`response.created`, `response.output_item.added`/`done`, `response.completed`). - **Middleware chain**: composable `GatewayMiddleware` trait; ships with resource limits (request-body precheck + concurrency semaphore, health probe exempt) and request-id propagation (`x-powercontext-request-id`, client-supplied or generated, echoed on responses). - **Body-free evidence**: usage completeness (known/estimated/unknown), cache tokens, TTFT vs first-release latency, attempt counters — never request/response bodies. diff --git a/gateway/crates/pc-gateway-app/src/ingress.rs b/gateway/crates/pc-gateway-app/src/ingress.rs index 4149feaf52..f8111d8a30 100644 --- a/gateway/crates/pc-gateway-app/src/ingress.rs +++ b/gateway/crates/pc-gateway-app/src/ingress.rs @@ -85,6 +85,10 @@ struct OpenAiResponsesRequest { conversation: Option, #[serde(default)] stream: Option, + /// 入口未建模的其余顶层字段（Codex 发送 reasoning、include、prompt_cache_key、 + /// parallel_tool_calls 等）。原样透传，由 Core 的请求字段白名单裁决。 + #[serde(flatten)] + extra: serde_json::Map, } pub async fn chat_completions( @@ -150,6 +154,7 @@ async fn ingress_common( .into_response(); } }; + let mut client_stream = false; let canonical = match format { Format::Chat => { let chat: OpenAiChatRequest = match serde_json::from_slice(&body) { @@ -184,6 +189,7 @@ async fn ingress_common( .into_response() } }; + client_stream = r.stream.unwrap_or(false); match to_canonical_responses(r) { Ok(c) => c, Err(msg) => { @@ -241,9 +247,74 @@ async fn ingress_common( .executor .run(call, &renderer, state.evidence.as_ref()) .await; + if client_stream { + return respond_responses_sse(outcome); + } respond(outcome) } +/// 流式 Responses 客户端：把放行后的完整响应重放为 Responses SSE 事件。 +/// 非 200 或无法解析的结果按普通 JSON 返回，错误状态码不被包装成流。 +fn respond_responses_sse(outcome: ManagedOutcome) -> Response { + let events = match &outcome { + ManagedOutcome::Released { + status: 200, body, .. + } => serde_json::from_slice::(body) + .ok() + .map(|resp| responses_sse(&resp)), + _ => None, + }; + match events { + Some(text) => ( + StatusCode::OK, + [(header::CONTENT_TYPE, "text/event-stream")], + text, + ) + .into_response(), + None => respond(outcome), + } +} + +/// 事件序列：response.created → 每个 output item 的 output_item.added/done → response.completed。 +fn responses_sse(resp: &serde_json::Value) -> String { + fn push(out: &mut String, event: &str, data: serde_json::Value) { + out.push_str("event: "); + out.push_str(event); + out.push_str("\ndata: "); + out.push_str(&data.to_string()); + out.push_str("\n\n"); + } + let mut out = String::new(); + let mut created = resp.clone(); + created["status"] = serde_json::json!("in_progress"); + created["output"] = serde_json::json!([]); + push( + &mut out, + "response.created", + serde_json::json!({"type": "response.created", "response": created}), + ); + if let Some(items) = resp.get("output").and_then(|v| v.as_array()) { + for (index, item) in items.iter().enumerate() { + push( + &mut out, + "response.output_item.added", + serde_json::json!({"type": "response.output_item.added", "output_index": index, "item": item}), + ); + push( + &mut out, + "response.output_item.done", + serde_json::json!({"type": "response.output_item.done", "output_index": index, "item": item}), + ); + } + } + push( + &mut out, + "response.completed", + serde_json::json!({"type": "response.completed", "response": resp}), + ); + out +} + fn bearer_headers(key: &str) -> Vec<(String, String)> { if key.is_empty() { Vec::new() @@ -291,6 +362,7 @@ fn to_canonical_chat(chat: OpenAiChatRequest) -> Result Result Result { let mut messages = Vec::with_capacity(r.input.len()); for item in r.input { - let role_str = item.get("role").and_then(|v| v.as_str()).unwrap_or(""); - let role = match role_str { - "system" | "developer" => Role::System, - "user" => Role::User, - "assistant" => Role::Assistant, - "tool" => Role::Tool, - // 无 role 的顶层 item（function_call/function_call_output 等）： - // 原样保留，role 占位为 tool。 - _ => Role::Tool, - }; - if role_str.is_empty() { + let role = item.get("role").and_then(|v| v.as_str()).unwrap_or(""); + let kind = item + .get("type") + .and_then(|v| v.as_str()) + .unwrap_or("message"); + if role != "user" || kind != "message" { messages.push(CanonicalMessage { role: Role::Tool, content: vec![item], @@ -334,13 +404,16 @@ fn to_canonical_responses(r: OpenAiResponsesRequest) -> Result vec![], }; messages.push(CanonicalMessage { - role, + role: Role::User, content, content_null: false, tool_calls: None, tool_call_id: None, }); } + let mut extra = r.extra; + // client_metadata 是客户端本地的回合元数据（Codex 的 x-codex-turn-metadata），不是发给模型的字段。 + extra.remove("client_metadata"); Ok(CanonicalRequest { model: r.model, messages, @@ -350,7 +423,10 @@ fn to_canonical_responses(r: OpenAiResponsesRequest) -> Result Result serde_json::Value { + serde_json::json!({ + "model": "gpt-5.6-sol", + "input": [ + {"type": "additional_tools", "role": "developer", + "tools": [{"type": "custom", "name": "exec", "description": "Run code"}]}, + {"type": "message", "role": "developer", + "content": [{"type": "input_text", "text": "base instructions"}]}, + {"role": "developer", "content": "shorthand developer message"}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "Fix the bug"}]}, + {"type": "reasoning", "summary": [], "encrypted_content": "gAAAA"}, + {"type": "custom_tool_call", "call_id": "c1", "name": "exec", "input": "ls"}, + {"type": "custom_tool_call_output", "call_id": "c1", "output": "util.py"} + ], + "tool_choice": "auto", + "parallel_tool_calls": false, + "reasoning": {"effort": "low", "context": "all_turns"}, + "store": false, + "stream": stream, + "include": ["reasoning.encrypted_content"], + "prompt_cache_key": "session-1", + "client_metadata": {"x-codex-turn-metadata": "{}"} + }) + } + + fn render_responses(client: &serde_json::Value) -> serde_json::Value { + use pc_gateway_protocol::WireRenderer; + let r: OpenAiResponsesRequest = serde_json::from_value(client.clone()).unwrap(); + let canonical = to_canonical_responses(r).unwrap(); + let payload = pc_gateway_protocol::OpenAiResponsesRenderer + .render(&canonical, None) + .unwrap(); + serde_json::from_slice(&payload.bytes).unwrap() + } + + #[test] + fn responses_ingress_forwards_non_user_items_verbatim() { + let client = codex_request(false); + let upstream = render_responses(&client); + let sent = client["input"].as_array().unwrap(); + let got = upstream["input"].as_array().unwrap(); + assert_eq!(got.len(), sent.len()); + for (i, (s, g)) in sent.iter().zip(got).enumerate() { + if s["role"] == "user" { + // 用户消息走规范形态（供注入使用），内容部件不变。 + assert_eq!(g["role"], "user"); + assert_eq!(g["content"], s["content"]); + } else { + assert_eq!(g, s, "input[{i}] must reach the upstream unchanged"); + } + } + } + + #[test] + fn responses_ingress_passes_through_unmodeled_fields() { + let upstream = render_responses(&codex_request(false)); + for field in [ + "reasoning", + "include", + "prompt_cache_key", + "parallel_tool_calls", + ] { + assert_eq!( + upstream[field], + codex_request(false)[field], + "{field} must be forwarded" + ); + } + assert!( + upstream.get("client_metadata").is_none(), + "client-local metadata is not sent" + ); + assert_eq!(upstream["tool_choice"], "auto"); + assert_eq!(upstream["store"], false); + } + + #[test] + fn passthrough_fields_never_override_modeled_ones() { + let r: OpenAiResponsesRequest = serde_json::from_value(serde_json::json!({ + "model": "m", "input": [], "store": false, "metadata": {"k": "v"} + })) + .unwrap(); + let canonical = to_canonical_responses(r).unwrap(); + let extra = canonical.passthrough_fields.clone().unwrap(); + assert!(extra.contains_key("metadata")); + assert!( + !extra.contains_key("store"), + "modeled fields are not collected twice" + ); + } + + #[test] + fn streaming_responses_clients_get_a_non_streaming_upstream_request() { + let upstream = render_responses(&codex_request(true)); + assert_eq!(upstream["stream"], false); + } + + #[test] + fn released_responses_are_replayed_as_sse_events() { + let resp = serde_json::json!({ + "id": "resp_1", "object": "response", "status": "completed", "model": "m", + "output": [ + {"type": "reasoning", "summary": []}, + {"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "ok"}]} + ], + "usage": {"input_tokens": 3, "output_tokens": 1} + }); + let text = responses_sse(&resp); + let events: Vec = text + .split("\n\n") + .filter(|e| !e.is_empty()) + .map(|e| serde_json::from_str(e.split("\ndata: ").nth(1).unwrap()).unwrap()) + .collect(); + let types: Vec<&str> = events.iter().map(|e| e["type"].as_str().unwrap()).collect(); + assert_eq!( + types, + [ + "response.created", + "response.output_item.added", + "response.output_item.done", + "response.output_item.added", + "response.output_item.done", + "response.completed" + ] + ); + assert_eq!(events[0]["response"]["status"], "in_progress"); + assert_eq!(events[4]["item"], resp["output"][1]); + assert_eq!(events[5]["response"], resp); + } + + #[test] + fn upstream_errors_are_not_wrapped_in_a_stream() { + let outcome = ManagedOutcome::Released { + binding: None, + status: 400, + content_type: "application/json".into(), + body: bytes::Bytes::from_static(br#"{"error":{"message":"bad"}}"#), + }; + let response = respond_responses_sse(outcome); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.headers()[header::CONTENT_TYPE], "application/json"); + } + #[test] fn chat_ingress_null_content_without_tool_calls_also_preserved() { let body = diff --git a/gateway/crates/pc-gateway-core/tests/executor_semantics.rs b/gateway/crates/pc-gateway-core/tests/executor_semantics.rs index 9d0d8148fb..dbaffd4158 100644 --- a/gateway/crates/pc-gateway-core/tests/executor_semantics.rs +++ b/gateway/crates/pc-gateway-core/tests/executor_semantics.rs @@ -214,6 +214,7 @@ fn call(degrade: DegradePolicy) -> ManagedCall { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }, target_url: "http://upstream.invalid/v1/chat/completions".into(), diff --git a/gateway/crates/pc-gateway-protocol/src/canonical.rs b/gateway/crates/pc-gateway-protocol/src/canonical.rs index 69622b8074..53cfcbe763 100644 --- a/gateway/crates/pc-gateway-protocol/src/canonical.rs +++ b/gateway/crates/pc-gateway-protocol/src/canonical.rs @@ -52,6 +52,10 @@ pub struct CanonicalRequest { /// Responses 的会话标识（conversation 参数），原样透传。 #[serde(default, skip_serializing_if = "Option::is_none")] pub conversation: Option, + /// 入口未建模的其余顶层请求字段（如 reasoning、include、prompt_cache_key），原样透传。 + /// 是否允许发送由 Core 的请求字段白名单裁决：被拒是显式 withhold，而不是静默丢弃。 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub passthrough_fields: Option>, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/gateway/crates/pc-gateway-protocol/src/render.rs b/gateway/crates/pc-gateway-protocol/src/render.rs index 4c76e2ddcb..69ab79760c 100644 --- a/gateway/crates/pc-gateway-protocol/src/render.rs +++ b/gateway/crates/pc-gateway-protocol/src/render.rs @@ -229,7 +229,12 @@ impl WireRenderer for OpenAiResponsesRenderer { let mut input = Vec::with_capacity(request.messages.len()); for msg in &request.messages { // 单元素 content 且是 Responses 专有 item → 原样作为顶层 item。 - if msg.content.len() == 1 && is_responses_item(&msg.content[0]) { + // 入口把非用户消息的整个 item 以 Tool 角色原样封装（含 developer 消息、 + // Codex 的 additional_tools 等不在白名单里的类型），同样原样透出。 + let verbatim_item = msg.role == crate::Role::Tool + && msg.content.len() == 1 + && msg.content[0].is_object(); + if msg.content.len() == 1 && (verbatim_item || is_responses_item(&msg.content[0])) { input.push(msg.content[0].clone()); continue; } @@ -284,6 +289,15 @@ impl WireRenderer for OpenAiResponsesRenderer { if let Some(conv) = &request.conversation { body["conversation"] = serde_json::json!(conv); } + if let Some(fields) = &request.passthrough_fields { + for (k, v) in fields { + // 已建模的字段以规范值为准，透传字段不得覆盖。 + body.as_object_mut() + .expect("body is an object") + .entry(k.clone()) + .or_insert_with(|| v.clone()); + } + } if let Some(fields) = request.native_cache_fields.as_ref() { merge_top_level(&mut body, fields).map_err(|_| RenderError::InvalidRenderPlan { renderer: self.name(), diff --git a/gateway/crates/pc-gateway-protocol/tests/openai_renderer.rs b/gateway/crates/pc-gateway-protocol/tests/openai_renderer.rs index 488a619b59..8b30463a83 100644 --- a/gateway/crates/pc-gateway-protocol/tests/openai_renderer.rs +++ b/gateway/crates/pc-gateway-protocol/tests/openai_renderer.rs @@ -38,6 +38,7 @@ fn request() -> CanonicalRequest { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, } } @@ -128,6 +129,7 @@ fn tool_call_metadata_round_trip_preserved() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiChatRenderer.render(&req, None).unwrap(); @@ -175,6 +177,7 @@ fn omitted_token_limit_not_emitted() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiChatRenderer.render(&req, None).unwrap(); @@ -208,6 +211,7 @@ fn explicit_token_limit_still_emitted() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiChatRenderer.render(&req, None).unwrap(); @@ -240,6 +244,7 @@ fn max_completion_tokens_field_name_preserved() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: Some("max_completion_tokens".into()), }; let p = OpenAiChatRenderer.render(&req, None).unwrap(); @@ -278,6 +283,7 @@ fn max_tokens_field_name_preserved() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: Some("max_tokens".into()), }; let p = OpenAiChatRenderer.render(&req, None).unwrap(); diff --git a/gateway/crates/pc-gateway-protocol/tests/openai_responses_renderer.rs b/gateway/crates/pc-gateway-protocol/tests/openai_responses_renderer.rs index e1075bc99e..9f6409c377 100644 --- a/gateway/crates/pc-gateway-protocol/tests/openai_responses_renderer.rs +++ b/gateway/crates/pc-gateway-protocol/tests/openai_responses_renderer.rs @@ -38,6 +38,7 @@ fn request() -> CanonicalRequest { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, } } @@ -95,6 +96,7 @@ fn responses_items_pass_through_verbatim() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -141,6 +143,7 @@ fn typed_content_parts_pass_through_as_array() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -182,6 +185,7 @@ fn structured_output_via_text_field_reaches_wire() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -224,6 +228,7 @@ fn store_false_reaches_responses_wire_body() { store: Some(false), previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -258,6 +263,7 @@ fn store_absent_omitted_from_wire_body() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -292,6 +298,7 @@ fn previous_response_id_reaches_wire_body() { store: Some(false), previous_response_id: Some("resp_prior_turn".into()), conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -325,6 +332,7 @@ fn responses_omitted_token_limit_not_emitted() { store: None, previous_response_id: None, conversation: None, + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); @@ -357,6 +365,7 @@ fn conversation_field_reaches_wire_body() { store: None, previous_response_id: None, conversation: Some("conv_prior".into()), + passthrough_fields: None, token_limit_field: None, }; let p = OpenAiResponsesRenderer.render(&req, None).unwrap(); From bbe65484cb2dd7160084ddcfd5d7ce2e6210293a Mon Sep 17 00:00:00 2001 From: jasper-yl Date: Mon, 2 Nov 2026 04:26:54 -0800 Subject: [PATCH 2/3] fix(gateway): replay Responses text events, failure status and degraded streams The SSE replay for streaming Responses clients emitted only item snapshots, always ended with response.completed, and covered only released responses: - Replay each message content part as content_part.added/done around one buffered output_text (or refusal) delta/done, so delta-loop consumers see the answer; message items are announced empty as upstream streams do. - Choose the terminal event from the response status: failed/cancelled or a non-null error yield response.failed, incomplete yields response.incomplete. - Replay a permitted degraded passthrough 200 response as well, keeping the x-pc-degraded header. - Number events with sequence_number. Refs #2398 Co-Authored-By: Claude Opus 5.5 --- gateway/README.md | 2 +- gateway/crates/pc-gateway-app/src/ingress.rs | 276 +++++++++++++++---- 2 files changed, 228 insertions(+), 50 deletions(-) diff --git a/gateway/README.md b/gateway/README.md index cccbc262f8..5d0b9e8ecf 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -25,7 +25,7 @@ An optional Rust executor for PowerContext, defined by the gateway design in #21 - **Two deployment forms**: *combined* (Rust owns the public TCP port; Core listens only on a private Unix socket; all non-`/gateway` routes are reverse-proxied to Core over UDS) or *standalone* (TCP to a remote Core). - **UDS client**: hand-rolled HTTP/1.1 over the Unix socket with a pooled, keep-alive connection pool (bounded idle set, connect/read timeouts, stale-connection single retry, content-length truncation and chunked-trailer consumption — covered by a real-socket regression test). - **OpenAI-compatible ingress** at `/gateway/v1/chat/completions` and `/gateway/v1/responses` (same format in and out, no cross-format translation), authenticated by a gateway key that replaces provider keys. -- **Responses fidelity**: on `/gateway/v1/responses`, every input item other than a plain user message (developer messages, tool calls and outputs, reasoning, client-specific items such as Codex's `additional_tools`) is forwarded unchanged, and request fields the ingress does not model (for example `reasoning`, `include`, `prompt_cache_key`) are passed through for Core's request allowlist to decide; only the client-local `client_metadata` is dropped. The upstream request is always non-streaming; a streaming client receives the released response replayed as Responses SSE events (`response.created`, `response.output_item.added`/`done`, `response.completed`). +- **Responses fidelity**: on `/gateway/v1/responses`, every input item other than a plain user message (developer messages, tool calls and outputs, reasoning, client-specific items such as Codex's `additional_tools`) is forwarded unchanged, and request fields the ingress does not model (for example `reasoning`, `include`, `prompt_cache_key`) are passed through for Core's request allowlist to decide; only the client-local `client_metadata` is dropped. The upstream request is always non-streaming; a streaming client receives the released (or permitted degraded-passthrough, still marked `x-pc-degraded`) 200 response replayed as Responses SSE events: `response.created`, `response.output_item.added`/`done` per item with each message content part as `response.content_part.added`/`done` around one buffered `response.output_text.delta`/`done` (or `response.refusal.delta`/`done`), and a terminal event chosen from the response status (`response.completed`, `response.incomplete`, or `response.failed` when the status is `failed`/`cancelled` or an `error` is present). - **Middleware chain**: composable `GatewayMiddleware` trait; ships with resource limits (request-body precheck + concurrency semaphore, health probe exempt) and request-id propagation (`x-powercontext-request-id`, client-supplied or generated, echoed on responses). - **Body-free evidence**: usage completeness (known/estimated/unknown), cache tokens, TTFT vs first-release latency, attempt counters — never request/response bodies. diff --git a/gateway/crates/pc-gateway-app/src/ingress.rs b/gateway/crates/pc-gateway-app/src/ingress.rs index f8111d8a30..803135d51b 100644 --- a/gateway/crates/pc-gateway-app/src/ingress.rs +++ b/gateway/crates/pc-gateway-app/src/ingress.rs @@ -253,65 +253,149 @@ async fn ingress_common( respond(outcome) } -/// 流式 Responses 客户端：把放行后的完整响应重放为 Responses SSE 事件。 -/// 非 200 或无法解析的结果按普通 JSON 返回，错误状态码不被包装成流。 +/// 流式 Responses 客户端：把放行（或允许降级直通）后的完整响应重放为 Responses SSE 事件。 +/// 非 200 或无法解析的结果按普通 JSON 返回，错误状态码不被包装成流； +/// 降级直通的流保留 `x-pc-degraded` 头。 fn respond_responses_sse(outcome: ManagedOutcome) -> Response { - let events = match &outcome { + let (body, degraded) = match &outcome { ManagedOutcome::Released { status: 200, body, .. - } => serde_json::from_slice::(body) - .ok() - .map(|resp| responses_sse(&resp)), - _ => None, + } => (body, false), + ManagedOutcome::Degraded { + status: 200, body, .. + } => (body, true), + _ => return respond(outcome), }; - match events { - Some(text) => ( - StatusCode::OK, - [(header::CONTENT_TYPE, "text/event-stream")], - text, - ) - .into_response(), - None => respond(outcome), + let Ok(resp) = serde_json::from_slice::(body) else { + return respond(outcome); + }; + let mut response = ( + StatusCode::OK, + [(header::CONTENT_TYPE, "text/event-stream")], + responses_sse(&resp), + ) + .into_response(); + if degraded { + response.headers_mut().insert( + "x-pc-degraded", + header::HeaderValue::from_static("passthrough_without_memory"), + ); } + response } -/// 事件序列：response.created → 每个 output item 的 output_item.added/done → response.completed。 +/// 事件序列：response.created → 每个 output item 的 output_item.added/done +/// （message 的每个内容部件在其间重放 content_part.added → 一次性 delta → done → content_part.done） +/// → 按响应状态选择的终止事件（completed / incomplete / failed）。 fn responses_sse(resp: &serde_json::Value) -> String { - fn push(out: &mut String, event: &str, data: serde_json::Value) { + use serde_json::{json, Value}; + let mut out = String::new(); + let mut seq: u64 = 0; + let mut push = |event: &str, mut data: Value| { + data["type"] = json!(event); + data["sequence_number"] = json!(seq); + seq += 1; out.push_str("event: "); out.push_str(event); out.push_str("\ndata: "); out.push_str(&data.to_string()); out.push_str("\n\n"); - } - let mut out = String::new(); + }; let mut created = resp.clone(); - created["status"] = serde_json::json!("in_progress"); - created["output"] = serde_json::json!([]); - push( - &mut out, - "response.created", - serde_json::json!({"type": "response.created", "response": created}), - ); - if let Some(items) = resp.get("output").and_then(|v| v.as_array()) { - for (index, item) in items.iter().enumerate() { + created["status"] = json!("in_progress"); + created["output"] = json!([]); + if created.get("error").is_some() { + created["error"] = Value::Null; + } + push("response.created", json!({"response": created})); + let items = resp.get("output").and_then(Value::as_array); + for (output_index, item) in items.into_iter().flatten().enumerate() { + let parts = match ( + item.get("type").and_then(Value::as_str), + item.get("content"), + ) { + (Some("message"), Some(Value::Array(parts))) => Some(parts), + _ => None, + }; + let Some(parts) = parts else { push( - &mut out, "response.output_item.added", - serde_json::json!({"type": "response.output_item.added", "output_index": index, "item": item}), + json!({"output_index": output_index, "item": item}), ); push( - &mut out, "response.output_item.done", - serde_json::json!({"type": "response.output_item.done", "output_index": index, "item": item}), + json!({"output_index": output_index, "item": item}), + ); + continue; + }; + // 与上游流一致：added 时 message 尚无内容，内容由随后的部件事件补齐。 + let mut added = item.clone(); + added["content"] = json!([]); + added["status"] = json!("in_progress"); + push( + "response.output_item.added", + json!({"output_index": output_index, "item": added}), + ); + let item_id = item.get("id").cloned().unwrap_or(Value::Null); + for (content_index, part) in parts.iter().enumerate() { + let at = json!({"item_id": item_id, "output_index": output_index, "content_index": content_index}); + let with = |mut base: Value, extra: Value| { + if let (Some(b), Some(e)) = (base.as_object_mut(), extra.as_object()) { + b.extend(e.clone()); + } + base + }; + let mut empty = part.clone(); + let text_kind = match part.get("type").and_then(Value::as_str) { + Some("output_text") => { + empty["text"] = json!(""); + if empty.get("annotations").is_some() { + empty["annotations"] = json!([]); + } + Some(("text", "response.output_text")) + } + Some("refusal") => { + empty["refusal"] = json!(""); + Some(("refusal", "response.refusal")) + } + _ => None, + }; + push( + "response.content_part.added", + with(at.clone(), json!({"part": empty})), + ); + if let Some((field, prefix)) = text_kind { + let text = part.get(field).cloned().unwrap_or(json!("")); + let logprobs = json!({"logprobs": []}); + let is_text = field == "text"; + let mut delta = with(at.clone(), json!({"delta": text})); + let mut done = with(at.clone(), json!({ field: text })); + if is_text { + delta = with(delta, logprobs.clone()); + done = with(done, logprobs); + } + push(&format!("{prefix}.delta"), delta); + push(&format!("{prefix}.done"), done); + } + push( + "response.content_part.done", + with(at, json!({"part": part})), ); } + push( + "response.output_item.done", + json!({"output_index": output_index, "item": item}), + ); } - push( - &mut out, - "response.completed", - serde_json::json!({"type": "response.completed", "response": resp}), - ); + // 终止事件取自已获准的响应状态：失败或不完整的响应不能被报告为完成。 + let has_error = resp.get("error").is_some_and(|e| !e.is_null()); + let terminal = match resp.get("status").and_then(Value::as_str) { + _ if has_error => "response.failed", + Some("failed") | Some("cancelled") => "response.failed", + Some("incomplete") => "response.incomplete", + _ => "response.completed", + }; + push(terminal, json!({"response": resp})); out } @@ -637,38 +721,132 @@ mod tests { assert_eq!(upstream["stream"], false); } + fn sse_events(text: &str) -> Vec { + text.split("\n\n") + .filter(|e| !e.is_empty()) + .map(|e| { + let (head, data) = e.split_once("\ndata: ").unwrap(); + let event: serde_json::Value = serde_json::from_str(data).unwrap(); + assert_eq!(head.strip_prefix("event: "), event["type"].as_str()); + event + }) + .collect() + } + + fn event_types(events: &[serde_json::Value]) -> Vec<&str> { + events.iter().map(|e| e["type"].as_str().unwrap()).collect() + } + #[test] fn released_responses_are_replayed_as_sse_events() { let resp = serde_json::json!({ "id": "resp_1", "object": "response", "status": "completed", "model": "m", "output": [ {"type": "reasoning", "summary": []}, - {"type": "message", "role": "assistant", - "content": [{"type": "output_text", "text": "ok"}]} + {"id": "msg_1", "type": "message", "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": "fixture-ok", "annotations": []}]} ], "usage": {"input_tokens": 3, "output_tokens": 1} }); - let text = responses_sse(&resp); - let events: Vec = text - .split("\n\n") - .filter(|e| !e.is_empty()) - .map(|e| serde_json::from_str(e.split("\ndata: ").nth(1).unwrap()).unwrap()) - .collect(); - let types: Vec<&str> = events.iter().map(|e| e["type"].as_str().unwrap()).collect(); + let events = sse_events(&responses_sse(&resp)); assert_eq!( - types, + event_types(&events), [ "response.created", "response.output_item.added", "response.output_item.done", "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", "response.output_item.done", "response.completed" ] ); assert_eq!(events[0]["response"]["status"], "in_progress"); - assert_eq!(events[4]["item"], resp["output"][1]); - assert_eq!(events[5]["response"], resp); + assert_eq!(events[3]["item"]["content"], serde_json::json!([])); + assert_eq!(events[8]["item"], resp["output"][1]); + assert_eq!(events[9]["response"], resp); + for (i, e) in events.iter().enumerate() { + assert_eq!(e["sequence_number"], i); + } + } + + #[test] + fn replayed_text_reaches_delta_stream_consumers() { + let resp = serde_json::json!({ + "id": "resp_1", "object": "response", "status": "completed", + "output": [ + {"id": "msg_1", "type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "fixture-", "annotations": []}, + {"type": "refusal", "refusal": "no"}, + {"type": "output_text", "text": "ok", "annotations": []}]} + ] + }); + let events = sse_events(&responses_sse(&resp)); + // 只读 output_text.delta 的标准消费循环必须拿到完整答案。 + let text: String = events + .iter() + .filter(|e| e["type"] == "response.output_text.delta") + .map(|e| e["delta"].as_str().unwrap()) + .collect(); + assert_eq!(text, "fixture-ok"); + let refusal = events + .iter() + .find(|e| e["type"] == "response.refusal.delta") + .unwrap(); + assert_eq!(refusal["delta"], "no"); + assert_eq!(refusal["content_index"], 1); + let done = events + .iter() + .find(|e| e["type"] == "response.output_text.done" && e["content_index"] == 2) + .unwrap(); + assert_eq!(done["text"], "ok"); + assert_eq!(done["item_id"], "msg_1"); + } + + #[test] + fn replay_terminal_event_follows_response_status() { + let failed = serde_json::json!({ + "id": "resp_1", "object": "response", "status": "failed", "output": [], + "error": {"code": "server_error", "message": "upstream failed"} + }); + let events = sse_events(&responses_sse(&failed)); + assert_eq!( + event_types(&events), + ["response.created", "response.failed"] + ); + assert_eq!(events[1]["response"]["error"]["code"], "server_error"); + assert!(events[0]["response"]["error"].is_null()); + + let incomplete = serde_json::json!({ + "id": "resp_2", "object": "response", "status": "incomplete", "output": [], + "incomplete_details": {"reason": "max_output_tokens"} + }); + let events = sse_events(&responses_sse(&incomplete)); + assert_eq!(event_types(&events).last(), Some(&"response.incomplete")); + } + + #[test] + fn degraded_responses_are_replayed_as_a_marked_stream() { + let outcome = ManagedOutcome::Degraded { + status: 200, + content_type: "application/json".into(), + body: bytes::Bytes::from_static( + br#"{"id":"resp_1","object":"response","status":"completed","output":[]}"#, + ), + }; + let response = respond_responses_sse(outcome); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers()[header::CONTENT_TYPE], + "text/event-stream" + ); + assert_eq!( + response.headers()["x-pc-degraded"], + "passthrough_without_memory" + ); } #[test] From 83a15578d79bf873a8746e5a3817a05103b0a032 Mon Sep 17 00:00:00 2001 From: hidb4ai Date: Mon, 2 Nov 2026 20:51:26 +0800 Subject: [PATCH 3/3] fix(gateway): preserve logprobs in Responses replay --- gateway/crates/pc-gateway-app/src/ingress.rs | 72 +++++++++++++++++++- 1 file changed, 71 insertions(+), 1 deletion(-) diff --git a/gateway/crates/pc-gateway-app/src/ingress.rs b/gateway/crates/pc-gateway-app/src/ingress.rs index 803135d51b..17f531c1ba 100644 --- a/gateway/crates/pc-gateway-app/src/ingress.rs +++ b/gateway/crates/pc-gateway-app/src/ingress.rs @@ -352,6 +352,9 @@ fn responses_sse(resp: &serde_json::Value) -> String { if empty.get("annotations").is_some() { empty["annotations"] = json!([]); } + if empty.get("logprobs").is_some() { + empty["logprobs"] = json!([]); + } Some(("text", "response.output_text")) } Some("refusal") => { @@ -366,11 +369,13 @@ fn responses_sse(resp: &serde_json::Value) -> String { ); if let Some((field, prefix)) = text_kind { let text = part.get(field).cloned().unwrap_or(json!("")); - let logprobs = json!({"logprobs": []}); let is_text = field == "text"; let mut delta = with(at.clone(), json!({"delta": text})); let mut done = with(at.clone(), json!({ field: text })); if is_text { + let logprobs = json!({ + "logprobs": part.get("logprobs").filter(|value| !value.is_null()).cloned().unwrap_or(json!([])) + }); delta = with(delta, logprobs.clone()); done = with(done, logprobs); } @@ -828,6 +833,71 @@ mod tests { assert_eq!(event_types(&events).last(), Some(&"response.incomplete")); } + #[tokio::test] + async fn replayed_text_events_preserve_logprobs() { + let logprobs = serde_json::json!([{ + "token": "hello", "logprob": -0.125, "bytes": [104, 101, 108, 108, 111], + "top_logprobs": [{"token": "hi", "logprob": -2.5, "bytes": [104, 105]}] + }]); + let resp = serde_json::json!({ + "id": "resp_1", "object": "response", "status": "completed", + "output": [{ + "id": "msg_1", "type": "message", "role": "assistant", "status": "completed", + "content": [ + {"type": "output_text", "text": "hello", "annotations": [], "logprobs": logprobs}, + {"type": "output_text", "text": " world", "annotations": [], "logprobs": null}, + {"type": "output_text", "text": "!", "annotations": []} + ] + }] + }); + for degraded in [false, true] { + let body = bytes::Bytes::from(serde_json::to_vec(&resp).unwrap()); + let outcome = if degraded { + ManagedOutcome::Degraded { + status: 200, + content_type: "application/json".into(), + body, + } + } else { + ManagedOutcome::Released { + binding: None, + status: 200, + content_type: "application/json".into(), + body, + } + }; + let response = respond_responses_sse(outcome); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let events = sse_events(std::str::from_utf8(&body).unwrap()); + let added = events + .iter() + .find(|e| e["type"] == "response.content_part.added" && e["content_index"] == 0) + .unwrap(); + assert_eq!(added["part"]["logprobs"], serde_json::json!([])); + for (content_index, expected) in [ + (0, logprobs.clone()), + (1, serde_json::json!([])), + (2, serde_json::json!([])), + ] { + for kind in ["response.output_text.delta", "response.output_text.done"] { + let event = events + .iter() + .find(|e| e["type"] == kind && e["content_index"] == content_index) + .unwrap(); + assert_eq!(event["logprobs"], expected); + } + } + let done = events + .iter() + .find(|e| e["type"] == "response.content_part.done" && e["content_index"] == 0) + .unwrap(); + assert_eq!(done["part"], resp["output"][0]["content"][0]); + assert_eq!(events.last().unwrap()["response"], resp); + } + } + #[test] fn degraded_responses_are_replayed_as_a_marked_stream() { let outcome = ManagedOutcome::Degraded {