Skip to content

fix(proxy): enforce structured stream fallback boundaries #1373

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ding113 merged 2 commits into from
Aug 1, 2026
Merged

fix(proxy): enforce structured stream fallback boundaries #1373

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8e19f91
fix(proxy): enforce structured stream fallback boundaries
ding113 Aug 1, 2026
947036d
fix(proxy): address stream review edge cases
ding113 Aug 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions messages/en/dashboard.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@
"local_manual": "Local manual pricing",
"cloud_exact": "Exact cloud provider pricing",
"cloud_model_fallback": "Fallback model provider pricing",
"cloud_official": "Official cloud pricing",
"priority_fallback": "Priority provider fallback pricing",
"single_provider_top_level": "Top-level single-provider pricing",
"official_fallback": "Official pricing"
Expand Down Expand Up @@ -594,6 +595,7 @@
"local_manual": "Local manual pricing",
"cloud_exact": "Exact cloud provider pricing",
"cloud_model_fallback": "Fallback model provider pricing",
"cloud_official": "Official cloud pricing",
"priority_fallback": "Priority provider fallback pricing",
"single_provider_top_level": "Top-level single-provider pricing",
"official_fallback": "Official pricing"
Expand Down
2 changes: 2 additions & 0 deletions messages/ja/dashboard.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@
"local_manual": "ローカル手動価格",
"cloud_exact": "クラウドの正確なプロバイダー価格",
"cloud_model_fallback": "フォールバックモデルのプロバイダー価格",
"cloud_official": "クラウド公式価格",
"priority_fallback": "優先フォールバック価格",
"single_provider_top_level": "トップレベル単一プロバイダー価格",
"official_fallback": "公式価格"
Expand Down Expand Up @@ -594,6 +595,7 @@
"local_manual": "ローカル手動価格",
"cloud_exact": "クラウドの正確なプロバイダー価格",
"cloud_model_fallback": "フォールバックモデルのプロバイダー価格",
"cloud_official": "クラウド公式価格",
"priority_fallback": "優先フォールバック価格",
"single_provider_top_level": "トップレベル単一プロバイダー価格",
"official_fallback": "公式価格"
Expand Down
2 changes: 2 additions & 0 deletions messages/ru/dashboard.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@
"local_manual": "Локальная ручная цена",
"cloud_exact": "Точная облачная цена провайдера",
"cloud_model_fallback": "Цена провайдера из резервной модели",
"cloud_official": "Официальная облачная цена",
"priority_fallback": "Приоритетная резервная цена",
"single_provider_top_level": "Цена верхнего уровня для одного провайдера",
"official_fallback": "Официальная цена"
Expand Down Expand Up @@ -594,6 +595,7 @@
"local_manual": "Локальная ручная цена",
"cloud_exact": "Точная облачная цена провайдера",
"cloud_model_fallback": "Цена провайдера из резервной модели",
"cloud_official": "Официальная облачная цена",
"priority_fallback": "Приоритетная резервная цена",
"single_provider_top_level": "Цена верхнего уровня для одного провайдера",
"official_fallback": "Официальная цена"
Expand Down
2 changes: 2 additions & 0 deletions messages/zh-CN/dashboard.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@
"local_manual": "本地手动价格",
"cloud_exact": "云端精确提供商价格",
"cloud_model_fallback": "回退模型的提供商价格",
"cloud_official": "云端官方价格",
"priority_fallback": "优先级回退价格",
"single_provider_top_level": "顶层单提供商价格",
"official_fallback": "官方价格"
Expand Down Expand Up @@ -594,6 +595,7 @@
"local_manual": "本地手动价格",
"cloud_exact": "云端精确提供商价格",
"cloud_model_fallback": "回退模型的提供商价格",
"cloud_official": "云端官方价格",
"priority_fallback": "优先级回退价格",
"single_provider_top_level": "顶层单提供商价格",
"official_fallback": "官方价格"
Expand Down
2 changes: 2 additions & 0 deletions messages/zh-TW/dashboard.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,7 @@
"local_manual": "本地手動價格",
"cloud_exact": "雲端精確供應商價格",
"cloud_model_fallback": "回退模型的供應商價格",
"cloud_official": "雲端官方價格",
"priority_fallback": "優先級回退價格",
"single_provider_top_level": "頂層單供應商價格",
"official_fallback": "官方價格"
Expand Down Expand Up @@ -594,6 +595,7 @@
"local_manual": "本地手動價格",
"cloud_exact": "雲端精確供應商價格",
"cloud_model_fallback": "回退模型的供應商價格",
"cloud_official": "雲端官方價格",
"priority_fallback": "優先級回退價格",
"single_provider_top_level": "頂層單供應商價格",
"official_fallback": "官方價格"
Expand Down
213 changes: 72 additions & 141 deletions src/app/v1/_lib/proxy/discovery-validity.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,9 @@
import {
classifyFrame,
type FrameVerdict,
type ProtocolFamily,
} from "./stream-gate/frame-classifier";

export type DiscoveryProtocol =
| "anthropic"
| "openai-chat"
Expand All @@ -16,76 +22,12 @@ export const DISCOVERY_PREFIX_MAX_BYTES = 1024 * 1024;
export const DISCOVERY_EVENT_MAX_COUNT = 1024;
const DISCOVERY_TEXT_ENCODER = new TextEncoder();

function hasContent(value: unknown): boolean {
if (typeof value === "string") return value.trim().length > 0;
if (!value || typeof value !== "object") return false;
if (Array.isArray(value)) return value.some(hasContent);
const object = value as Record<string, unknown>;
return [
"text",
"content",
"delta",
"output_text",
"thinking",
"tool_use",
"tool_calls",
"functionCall",
"function_call",
"function",
"arguments",
"partial_json",
"id",
"name",
"input",
"parts",
].some((key) => hasContent(object[key]));
}

function hasAnthropicContentBlock(value: unknown): boolean {
if (!value || typeof value !== "object" || Array.isArray(value)) return false;
const block = value as Record<string, unknown>;
if (typeof block.type !== "string" || block.type.length === 0) return false;
// Text blocks need non-empty text; tool_use/thinking/image blocks are
// deliverable as soon as their typed block starts, even with empty input.
return block.type === "text" ? hasContent(block.text) : true;
}

function hasOpenAIResponsesOutputItem(value: unknown): boolean {
if (!value || typeof value !== "object" || Array.isArray(value)) return false;
const item = value as Record<string, unknown>;
if (typeof item.type !== "string") return false;

switch (item.type) {
case "message":
return hasContent(item.content);
case "reasoning":
return hasContent(item.summary) || hasContent(item.content);
case "function_call":
case "mcp_call":
return hasContent(item.name) || hasContent(item.arguments);
case "custom_tool_call":
return hasContent(item.name) || hasContent(item.input);
case "computer_call":
case "web_search_call":
case "file_search_call":
case "code_interpreter_call":
case "local_shell_call":
case "shell_call":
case "apply_patch_call":
return [
item.action,
item.arguments,
item.input,
item.queries,
item.query,
item.code,
item.command,
item.operation,
].some(hasContent);
default:
return false;
}
}
const DISCOVERY_PROTOCOL_FAMILIES: Partial<Record<DiscoveryProtocol, ProtocolFamily>> = {
anthropic: "anthropic",
"openai-chat": "openai-chat",
"openai-responses": "openai-responses",
gemini: "gemini",
};

/**
* Protocol-level error signals that must remain terminal even if a provider
Expand Down Expand Up @@ -114,69 +56,57 @@ export function isDiscoveryProtocolErrorPayload(value: unknown): boolean {
);
}

function classifyJson(value: unknown, protocol: DiscoveryProtocol): DiscoveryValidity {
if (!value || typeof value !== "object") return { ready: false, terminal: false, error: true };
const object = value as Record<string, unknown>;
if (isDiscoveryProtocolErrorPayload(value)) {
return { ready: false, terminal: true, error: true };
}
if (protocol === "openai-chat") {
const choices = Array.isArray(object.choices) ? object.choices : [];
const ready = choices.some((choice) => {
if (!choice || typeof choice !== "object") return false;
const choiceObject = choice as Record<string, unknown>;
const delta = choiceObject.delta;
return hasContent(delta) || hasContent(choiceObject.message);
});
return { ready, terminal: false, error: false };
}
if (protocol === "openai-responses") {
if (object.type === "response.completed" || object.type === "response.done") {
return { ready: false, terminal: true, error: false };
function validityFromVerdict(verdict: FrameVerdict): DiscoveryValidity {
return {
ready: verdict === "content",
terminal: verdict === "terminal" || verdict === "error" || verdict === "malformed",
error: verdict === "error" || verdict === "malformed",
};
}

function classifyProtocolFrame(
data: string,
protocol: DiscoveryProtocol,
eventName: string | null
): DiscoveryValidity {
const family = DISCOVERY_PROTOCOL_FAMILIES[protocol];
if (!family) return { ready: false, terminal: false, error: false };

let parsed: unknown;
try {
parsed = JSON.parse(data) as unknown;
// 同帧的通用失败标志优先于 content；fake-200 失败响应不能赢得 Discovery。
if (isDiscoveryProtocolErrorPayload(parsed)) {
return { ready: false, terminal: true, error: true };
}
return {
ready:
(object.type === "response.output_text.delta" && hasContent(object.delta)) ||
(object.type === "response.function_call_arguments.delta" && hasContent(object.delta)) ||
(object.type === "response.reasoning_summary_text.delta" && hasContent(object.delta)) ||
(object.type === "response.output_item.added" && hasOpenAIResponsesOutputItem(object.item)),
terminal: false,
error: false,
};
}
if (protocol === "gemini") {
const response =
object.response && typeof object.response === "object" && !Array.isArray(object.response)
? (object.response as Record<string, unknown>)
: null;
const candidatesValue = response?.candidates ?? object.candidates;
const candidates = Array.isArray(candidatesValue) ? candidatesValue : [];
return {
ready: candidates.some((candidate) => hasContent(candidate)),
terminal: false,
error: false,
};
} catch {
parsed = undefined;
}
// Anthropic SSE data events: message_start/message_delta are metadata; a
// content_block_delta or tool use is the first deliverable event.

let verdict = classifyFrame(family, eventName, data);
if (verdict !== "neutral") return validityFromVerdict(verdict);

// Gemini SDK wrappers may expose the native candidate chunk under response.
if (
object.type === "message_start" ||
object.type === "message_delta" ||
object.type === "ping"
family === "gemini" &&
parsed &&
typeof parsed === "object" &&
!Array.isArray(parsed) &&
(parsed as Record<string, unknown>).response &&
typeof (parsed as Record<string, unknown>).response === "object"
) {
return { ready: false, terminal: false, error: false };
}
if (object.type === "message_stop") {
return { ready: false, terminal: true, error: false };
try {
verdict = classifyFrame(
family,
eventName,
JSON.stringify((parsed as Record<string, unknown>).response)
);
} catch {
return validityFromVerdict("malformed");
Comment on lines +75 to +105

@coderabbitai coderabbitai Bot Aug 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

确认通用包装事件名(如 "message")是否仍导致分类误判。

历史评审在旧提交的行 233 指出一个问题:当 SSE 帧使用通用 event: message 包装、而 data.type 携带真实类型(如 content_block_delta)时,classifyFrame 会把该帧误判为 neutral。原因是分类器只在 eventName 为空时才回退检查 data.type。这会导致 Discovery 阶段在有效流上耗尽 precommit 窗口,触发不必要的 provider fallback。

本次重构后的 classifyProtocolFrame 在行 86 和行 99-103 仍然把原始 eventName 直接传给 classifyFrame,没有对 "message" 这类通用包装事件名做归一化处理。由于 classifyFrameInner(位于 frame-classifier.ts)的内部实现未包含在本次审查内容中,无法确认该问题是否已在分类器内部修复。

如果该问题尚未修复,请在调用 classifyFrame 前对 eventName 做归一化,以恢复对 data.type 的回退检查。

🐛 建议修复(如问题仍然存在) 
-  let verdict = classifyFrame(family, eventName, data);
+  const effectiveEventName = eventName?.trim() === "message" ? null : eventName;
+  let verdict = classifyFrame(family, effectiveEventName, data);
   if (verdict !== "neutral") return validityFromVerdict(verdict);

Gemini response 展开分支(行 99-103)需要使用同一个 effectiveEventName:

       verdict = classifyFrame(
         family,
-        eventName,
+        effectiveEventName,
         JSON.stringify((parsed as Record<string, unknown>).response)
       );

请运行以下脚本,确认 classifyFrameInner 是否已处理该场景:

#!/bin/bash
# 描述:检查 frame-classifier.ts 中对通用包装事件名(如 "message")的处理逻辑。
ast-grep outline src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts --items all

echo "---"
rg -n -B5 -A30 'function classifyFrameInner' src/app/v1/_lib/proxy/stream-gate/frame-classifier.ts

请补充一条针对 event: message 包装 + data.type 携带真实类型帧的回归测试,验证该帧不会被误判为 neutral。

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/app/v1/_lib/proxy/discovery-validity.ts` around lines 75 - 105, 检查
classifyFrameInner 是否已支持通用 eventName（如 "message"）回退读取 data.type；若未支持，在
classifyProtocolFrame 中生成归一化的 effectiveEventName，并在原始调用及 Gemini response
展开分支中统一传给 classifyFrame。补充回归测试覆盖 event: message 且 data.type 携带真实协议类型的帧，确认不会返回
neutral。

}
}
return {
ready:
(object.type === "content_block_delta" && hasContent(object.delta)) ||
(object.type === "content_block_start" && hasAnthropicContentBlock(object.content_block)) ||
hasContent(object.content),
terminal: false,
error: false,
};

return validityFromVerdict(verdict);
}

export function classifyDiscoveryChunk(
Expand All @@ -189,6 +119,7 @@ export function classifyDiscoveryChunk(
export class DiscoveryValidityParser {
private buffered = "";
private dataLines: string[] = [];
private eventName: string | null = null;
private readonly decoder = new TextDecoder();
private _ready = false;
private _terminal = false;
Expand All @@ -211,6 +142,7 @@ export class DiscoveryValidityParser {
this._limitExceeded = true;
this.buffered = "";
this.dataLines = [];
this.eventName = null;
return this.result;
}
}
Expand All @@ -229,6 +161,7 @@ export class DiscoveryValidityParser {
if (this._error) {
this.buffered = "";
this.dataLines = [];
this.eventName = null;
return this.result;
}
}
Expand Down Expand Up @@ -263,6 +196,10 @@ export class DiscoveryValidityParser {

const colonIndex = line.indexOf(":");
const field = colonIndex === -1 ? line : line.slice(0, colonIndex);
if (field === "event") {
this.eventName = (colonIndex === -1 ? "" : line.slice(colonIndex + 1)).trim();
return;
}
if (field === "data") {
let value = colonIndex === -1 ? "" : line.slice(colonIndex + 1);
if (value.startsWith(" ")) value = value.slice(1);
Expand All @@ -273,7 +210,7 @@ export class DiscoveryValidityParser {
// event/id/retry and unknown SSE fields carry framing metadata only. A
// bare JSON line is supported for providers returning non-SSE JSON, but
// never while an SSE data event is pending.
if (field === "event" || field === "id" || field === "retry" || this.dataLines.length > 0) {
if (field === "id" || field === "retry" || this.dataLines.length > 0) {
return;
}
const candidate = line.trim();
Expand All @@ -287,24 +224,18 @@ export class DiscoveryValidityParser {
}

private flushSseEvent(): void {
const eventName = this.eventName;
this.eventName = null;
if (this.dataLines.length === 0) return;
const candidate = this.dataLines.join("\n");
this.dataLines = [];
if (!this.beginEvent()) return;
if (candidate.trim() === "[DONE]") {
this._terminal = true;
return;
}
try {
this.consumeValue(JSON.parse(candidate) as unknown);
} catch {
// A complete but non-JSON SSE event cannot establish protocol validity.
}
this.consumeFrame(candidate, eventName);

@github-actions github-actions Bot Aug 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

**[HIGH] [LOGIC-BUG] Generic event: message headers now hide the payload event type

Why this is a problem: This new call passes the SSE event field into classifyFrame. The classifier only falls back to data.type when eventName is empty, so a valid relay frame such as event: message with data: {"type":"content_block_delta","delta":{"text":"hi"}} is classified as neutral (the same happens for Responses delta frames). Before this refactor, discovery parsed the JSON without the event name and recognized data.type; now Discovery can exhaust its precommit window and fail over on a valid stream. The completion code already treats message as a supported wrapper event, so this format is intended to be accepted.

Suggested fix:

const effectiveEventName = eventName?.trim() === "message" ? null : eventName;
this.consumeFrame(candidate, effectiveEventName);

Alternatively, make classifyFrame prefer the embedded type for generic wrapper event names, and add a regression test for event: message plus a typed content frame.

}

private consumeEventValue(value: unknown): void {
if (!this.beginEvent()) return;
this.consumeValue(value);
this.consumeFrame(JSON.stringify(value), null);
}

private beginEvent(): boolean {
Expand All @@ -317,8 +248,8 @@ export class DiscoveryValidityParser {
return true;
}

private consumeValue(value: unknown): void {
const result = classifyJson(value, this.protocol);
private consumeFrame(data: string, eventName: string | null): void {
const result = classifyProtocolFrame(data, this.protocol, eventName);
this._ready ||= result.ready;
this._terminal ||= result.terminal;
this._error ||= result.error;
Expand Down
4 changes: 3 additions & 1 deletion src/app/v1/_lib/proxy/forwarder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6449,7 +6449,9 @@ export class ProxyForwarder {
kind: effectiveKind,
finalRescue: effectiveFinalRescue,
controller,
parser: new DiscoveryValidityParser(protocol),
parser: new DiscoveryValidityParser(
mapProviderTypeToFamily(provider.providerType) ?? protocol
),
chunks: [],
pending: true,
ready: false,
Expand Down
Loading
Loading