shujakuin/codex-retry-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add SSE passthrough stream action

Browse Source
This commit is contained in:
yunyaozhou
2026-07-20 00:40:14 +08:00
parent ca8935d73e
commit 31e673f777
4 changed files with 282 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2
View File
@@ -24,6 +24,7 @@
- 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`、`stream disconnected before completion: Concurrency limit exceeded for account, please retry later`，以及 Responses 错误码 `server_is_overloaded` / `slow_down`），也会自动重试；重试耗尽后转成本地 `502`
- 除了 `429/503` JSON 错误响应，也会识别 `200` 但返回体本质是错误、以及流式失败事件里携带同样文案的情况
- 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502`
- 可选 `passthrough` 直通模式会对 `stream: true` 请求跳过 reasoning retry；若上游实际返回 `text/event-stream`，则把原始 SSE chunk 直接转发给客户端，不解析、缓存、规范化或拦截
- 流式成功响应在严格检查模式下仍会先缓存完成；成功后会按真实且规范化的 Responses 生命周期与输出顺序逐块回放给 Codex，每个 SSE 块之间至少间隔 5ms，并关闭 TCP 小包聚合，避免大量 delta 在同一事件循环突发到达；不会伪造缺少 response ID 的生命周期事件
- 默认同时拦截 root 路径和 `/v1` 路径：
- `/responses`
@@ -388,6 +389,7 @@ macOS / Linux: ~/.codex-retry-gateway/config/config.json
- 默认 `strict_502`
- `strict_502`：先缓存整个流，命中 `516` 时统一返回 `502`
- `disconnect`：兼容旧行为；若命中发生在已透传 chunk 之后，则直接断开连接
- `passthrough`：对 `stream: true` 请求关闭 reasoning retry；只有上游实际返回 `text/event-stream` 时原样转发 SSE chunk，非 SSE 响应继续按单次既有规则处理
- `log_match`
- 是否记录命中日志
gateway.mjs
+166 -4
View File
@@ -80,6 +80,7 @@ const DEFAULT_CONFIG = {
log_match: true,
health_path: "/__codex_retry_gateway/health",
};
const STREAM_ACTIONS = ["strict_502", "disconnect", "passthrough"];
const REASONING_POINTERS = [
"/usage/output_tokens_details/reasoning_tokens",
@@ -935,6 +936,15 @@ function normalizeReasoningMatchMode(value) {
return DEFAULT_CONFIG.reasoning_match_mode;
}
function normalizeStreamAction(value) {
const action = `${value || DEFAULT_CONFIG.stream_action}`.trim().toLowerCase();
return STREAM_ACTIONS.includes(action) ? action : DEFAULT_CONFIG.stream_action;
}
function streamPassthroughEnabled(config) {
return normalizeStreamAction(config?.stream_action) === "passthrough";
}
function normalizeReasoningEquals(values, fallback = DEFAULT_CONFIG.reasoning_equals) {
const normalized = normalizeIntegerList(values, fallback);
return normalized.length > 0 ? normalized : [...fallback];
@@ -1646,6 +1656,7 @@ function buildProfileFormModel(env) {
env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS || `${DEFAULT_CONFIG.upstream_fetch_retry_attempts}`,
upstream_fetch_retry_backoff_ms:
env.CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS || `${DEFAULT_CONFIG.upstream_fetch_retry_backoff_ms}`,
stream_action: normalizeStreamAction(env.CODEX_RETRY_GATEWAY_STREAM_ACTION),
endpoints: normalizeStringList(env.CODEX_RETRY_GATEWAY_ENDPOINTS || DEFAULT_CONFIG.endpoints, DEFAULT_CONFIG.endpoints),
};
}
@@ -1758,7 +1769,7 @@ function buildConfigFromProfileEnv(profileName, env, imageConfig = {}) {
non_stream_status_code: env.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE
? Number.parseInt(`${env.CODEX_RETRY_GATEWAY_NON_STREAM_STATUS_CODE}`, 10)
: DEFAULT_CONFIG.non_stream_status_code,
stream_action: env.CODEX_RETRY_GATEWAY_STREAM_ACTION || DEFAULT_CONFIG.stream_action,
stream_action: normalizeStreamAction(env.CODEX_RETRY_GATEWAY_STREAM_ACTION),
log_match: env.CODEX_RETRY_GATEWAY_LOG_MATCH === undefined
? DEFAULT_CONFIG.log_match
: ["1", "true", "yes", "on"].includes(`${env.CODEX_RETRY_GATEWAY_LOG_MATCH}`.trim().toLowerCase()),
@@ -1895,6 +1906,7 @@ async function buildProfileEnvText(payload) {
if (upstreamFetchRetryBackoffMs < 0) {
throw new Error("upstream_fetch_retry_backoff_ms 不能为负数");
}
const streamAction = normalizeStreamAction(payload.stream_action);
const envPairs = [
["CODEX_RETRY_GATEWAY_LISTEN_HOST", `${payload.listen_host || DEFAULT_CONFIG.listen_host}`.trim()],
@@ -1907,6 +1919,7 @@ async function buildProfileEnvText(payload) {
["CODEX_RETRY_GATEWAY_RETRYABLE_ERROR_MESSAGES", retryableErrorMessages.join("\n")],
["CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_ATTEMPTS", `${upstreamFetchRetryAttempts}`],
["CODEX_RETRY_GATEWAY_UPSTREAM_FETCH_RETRY_BACKOFF_MS", `${upstreamFetchRetryBackoffMs}`],
["CODEX_RETRY_GATEWAY_STREAM_ACTION", streamAction],
["CODEX_RETRY_GATEWAY_REQUEST_HISTORY_LIMIT", `${requestHistoryLimit}`],
["CODEX_RETRY_GATEWAY_ENDPOINTS", endpoints.join(",")],
];
@@ -2132,6 +2145,7 @@ async function buildTextProfileExportBundle(runtime, payload) {
retryable_error_messages: config.retryable_error_messages,
upstream_fetch_retry_attempts: config.upstream_fetch_retry_attempts,
upstream_fetch_retry_backoff_ms: config.upstream_fetch_retry_backoff_ms,
stream_action: config.stream_action,
endpoints: config.endpoints,
},
};
@@ -2242,6 +2256,7 @@ async function importTextProfileBundle(runtime, payload) {
retryable_error_messages: imported.profile.retryable_error_messages,
upstream_fetch_retry_attempts: imported.profile.upstream_fetch_retry_attempts,
upstream_fetch_retry_backoff_ms: imported.profile.upstream_fetch_retry_backoff_ms,
stream_action: imported.profile.stream_action,
endpoints: imported.profile.endpoints,
});
return {
@@ -2544,6 +2559,7 @@ async function loadConfig(configPath) {
config.retryable_error_messages,
DEFAULT_CONFIG.retryable_error_messages,
);
config.stream_action = normalizeStreamAction(config.stream_action);
config.management_access_key = normalizeManagementAccessKey(config.management_access_key);
config.upstream_fetch_retry_attempts = normalizePositiveInteger(
config.upstream_fetch_retry_attempts,
@@ -2941,6 +2957,7 @@ async function listProfiles(runtime) {
auth_source: summarizeProfileAuthSource(env),
reasoning_match_mode: form.reasoning_match_mode,
reasoning_equals: form.reasoning_equals,
stream_action: form.stream_action,
},
form,
});
@@ -3703,6 +3720,9 @@ function buildEditableConfig(currentConfig, payload) {
payload.non_stream_status_code === undefined
? currentConfig.non_stream_status_code
: Number.parseInt(`${payload.non_stream_status_code}`, 10);
const nextStreamAction = normalizeStreamAction(
payload.stream_action === undefined ? currentConfig.stream_action : payload.stream_action,
);
if (nextRetryableStatusCodes.length === 0) {
throw new Error("retryable_status_codes 不能为空");
@@ -3737,6 +3757,7 @@ function buildEditableConfig(currentConfig, payload) {
upstream_fetch_retry_attempts: nextUpstreamFetchRetryAttempts,
upstream_fetch_retry_backoff_ms: nextUpstreamFetchRetryBackoffMs,
non_stream_status_code: nextStatusCode,
stream_action: nextStreamAction,
log_match: payload.log_match === undefined ? currentConfig.log_match : Boolean(payload.log_match),
};
}
@@ -4322,7 +4343,7 @@ async function handleManagementRequest(runtime, req, res, requestUrl) {
await writeConfig(runtime.configPath, nextConfig);
runtime.config = nextConfig;
runtime.logger(
`[config] updated reasoning_match_mode=${nextConfig.reasoning_match_mode} reasoning_equals=${nextConfig.reasoning_equals.join(",")} retryable_status_codes=${nextConfig.retryable_status_codes.join(",")} endpoints=${nextConfig.endpoints.join(",")}`,
`[config] updated reasoning_match_mode=${nextConfig.reasoning_match_mode} reasoning_equals=${nextConfig.reasoning_equals.join(",")} retryable_status_codes=${nextConfig.retryable_status_codes.join(",")} stream_action=${nextConfig.stream_action} endpoints=${nextConfig.endpoints.join(",")}`,
);
const state = await readRuntimeState(runtime);
jsonResponse(req, res, 200, {
@@ -4538,6 +4559,34 @@ async function writeBufferedStreamChunks(res, chunks) {
}
}
async function writeResponseChunk(res, chunk) {
if (!chunk || chunk.length === 0 || res.destroyed || res.writableEnded) {
return false;
}
if (res.write(chunk)) {
return true;
}
await new Promise((resolve) => {
let settled = false;
const finish = () => {
if (settled) {
return;
}
settled = true;
res.off("drain", onDrain);
res.off("close", onClose);
res.off("error", onClose);
resolve();
};
const onDrain = () => finish();
const onClose = () => finish();
res.once("drain", onDrain);
res.once("close", onClose);
res.once("error", onClose);
});
return !res.destroyed && !res.writableEnded;
}
async function writeCapturedResponse(res, delivery) {
if (!delivery) {
throw new Error("missing captured response delivery");
@@ -5281,7 +5330,7 @@ async function handleStreaming({
captureOnly = false,
persistEntry = true,
}) {
const strict502Mode = captureOnly || config.stream_action !== "disconnect";
const strict502Mode = captureOnly || normalizeStreamAction(config.stream_action) !== "disconnect";
const reader = upstreamResponse.body.getReader();
const sseState = {
decoder: new TextDecoder("utf8"),
@@ -5659,6 +5708,97 @@ async function handleStreaming({
}
}
async function handlePassthroughStreaming({
runtime,
upstreamResponse,
res,
requestEntry,
requestAbortSignal = null,
persistEntry = true,
}) {
const reader = upstreamResponse.body?.getReader();
if (!reader) {
throw new Error("upstream SSE response is missing a readable body");
}
requestEntry.inspected = false;
copyHeadersToClient(upstreamResponse.headers, res);
res.writeHead(upstreamResponse.status);
res.socket?.setNoDelay(true);
res.flushHeaders?.();
while (true) {
let readResult;
try {
readResult = await reader.read();
} catch (error) {
if (requestAbortSignal?.aborted) {
throw requestAbortSignal.reason || error;
}
if (persistEntry) {
persistStreamingProgress(runtime, requestEntry, { force: true }, new Date());
}
if (!res.destroyed && !res.writableEnded) {
res.destroy(error);
}
return {
inspected: false,
matched: false,
status_code: upstreamResponse.status,
upstream_status_code: upstreamResponse.status,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
error: `upstream SSE stream terminated: ${error?.message || error}`,
};
}
if (readResult.done) {
if (persistEntry) {
persistStreamingProgress(runtime, requestEntry, { force: true }, new Date());
}
if (!res.destroyed && !res.writableEnded) {
res.end();
}
return {
inspected: false,
matched: false,
status_code: upstreamResponse.status,
upstream_status_code: upstreamResponse.status,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
const chunk = Buffer.from(readResult.value);
const now = new Date();
markRequestEntryFirstResponse(runtime, requestEntry, { persistEntry, at: now });
updateStreamingProgress(requestEntry, { chunkBytes: chunk.length, at: now });
if (persistEntry) {
persistStreamingProgress(
runtime,
requestEntry,
{ force: requestEntry.stream_chunk_count === 1 },
now,
);
}
await writeResponseChunk(res, chunk);
if (res.destroyed || res.writableEnded) {
return {
cancelled: true,
cancel_reason: REASONING_RETRY_ABORT_CLIENT,
response_id: requestEntry.response_id,
thread_id: requestEntry.thread_id,
response_bytes_received: requestEntry.response_bytes_received,
stream_chunk_count: requestEntry.stream_chunk_count,
};
}
}
}
function buildAttemptRequestEntry(baseEntry) {
return {
...baseEntry,
@@ -5778,6 +5918,22 @@ async function executeGatewayQuery({
);
}
if (isSseContentType(responseContentType) && !captureOnly && streamPassthroughEnabled(config)) {
const result = await handlePassthroughStreaming({
runtime,
upstreamResponse,
res,
requestEntry,
requestAbortSignal: queryAbortLink.controller.signal,
persistEntry,
});
return {
response_stream: true,
...result,
total_upstream_attempts: totalUpstreamAttempts,
};
}
if (!shouldInspect) {
markRequestEntryFirstResponse(runtime, requestEntry, { persistEntry });
const body = Buffer.from(await upstreamResponse.arrayBuffer());
@@ -6303,6 +6459,12 @@ async function proxyRequest(runtime, req, res) {
requestEntry.reasoning_summary = extractRequestReasoningSummary(requestJson);
requestEntry.request_stream = requestIsStream;
applyThreadReasoningState(runtime, requestEntry, pathname);
const streamPassthroughRequest = streamPassthroughEnabled(config) && requestIsStream;
if (streamPassthroughRequest && isResponsesReasoningRetryPath(pathname)) {
requestEntry.reasoning_retry_enabled = false;
requestEntry.reasoning_retry_thread_mode = "stream_passthrough";
requestEntry.reasoning_retry_stop_reason = "stream_passthrough";
}
upsertRequestEntry(runtime, requestEntry);
const upstreamRoute = selectUpstreamRoute(config, pathname);
@@ -6323,7 +6485,7 @@ async function proxyRequest(runtime, req, res) {
}
const clientAbortContext = createClientAbortContext(req, res);
try {
const result = isResponsesReasoningRetryEligible(pathname, requestEntry)
const result = !streamPassthroughRequest && isResponsesReasoningRetryEligible(pathname, requestEntry)
? await runResponsesReasoningRetry({
runtime,
config,
scripts/test-gateway-e2e.mjs
+63
View File
@@ -822,6 +822,7 @@ async function run() {
retryable_error_messages: ["capacity test"],
upstream_fetch_retry_attempts: 3,
upstream_fetch_retry_backoff_ms: 50,
stream_action: "passthrough",
endpoints: ["/responses", "/v1/responses"],
}),
},
@@ -829,6 +830,15 @@ async function run() {
assert(exportableTextProfileResponse.status === 200, `可导出文本 profile 保存失败: ${exportableTextProfileResponse.status}`);
const exportableTextProfilePayload = await exportableTextProfileResponse.json();
assert(!JSON.stringify(exportableTextProfilePayload).includes("text-export-secret"), "普通 profile API 不应返回导出 key");
const exportableTextProfile = (exportableTextProfilePayload.profiles || []).find(
(profile) => profile?.name === "export-source",
);
assert(exportableTextProfile?.form?.stream_action === "passthrough", "文本 profile 未保留 SSE 直通模式");
assert(
(await readFile(path.join(profilesDir, "export-source.env"), "utf8"))
.includes("CODEX_RETRY_GATEWAY_STREAM_ACTION=passthrough"),
"文本 profile env 未写入 SSE 直通模式",
);
const passthroughExportResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/profiles/export`,
@@ -1916,6 +1926,59 @@ async function run() {
}
}
const enablePassthroughResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/config`, {
method: "POST",
headers: { ...adminHeaders, "content-type": "application/json" },
body: JSON.stringify({ stream_action: "passthrough" }),
});
const enablePassthroughPayload = await enablePassthroughResponse.json();
assert(enablePassthroughResponse.status === 200, `启用 SSE 直通模式失败: ${enablePassthroughResponse.status}`);
assert(enablePassthroughPayload?.config?.stream_action === "passthrough", "当前配置未切换到 SSE 直通模式");
const passthroughThreadId = "thread_sse_passthrough";
const passthroughRetryKey = reasoningRetryKeyForRequest("/responses", {
stream: true,
thread_id: passthroughThreadId,
test_reasoning_retry_key: "sse-passthrough",
});
const passthroughStream = await readSseUntilClose(
`http://127.0.0.1:${gatewayPort}/responses`,
{
stream: true,
thread_id: passthroughThreadId,
test_reasoning_before_success_times: 1,
test_reasoning_retry_key: "sse-passthrough",
test_stream_include_lifecycle: true,
test_stream_lifecycle_marker: "passthrough-raw-marker",
test_stream_delta_chunks: 2,
test_stream_chunk_delay_ms: 2,
},
);
assert(passthroughStream.status === 200, `SSE 直通状态异常: ${passthroughStream.status}`);
assert(passthroughStream.headers.get("x-upstream-reasoning-attempt") === "1", "SSE 直通不应发起 reasoning 重试");
assert(passthroughStream.text.includes("passthrough-raw-marker"), "SSE 直通不应改写上游生命周期 payload");
assert(passthroughStream.text.includes('"reasoning_tokens":516'), "SSE 直通不应拦截匹配的 reasoning 响应");
const passthroughRetryStats = upstream.getReasoningRetryStat(passthroughRetryKey);
assert(passthroughRetryStats.totalRequests === 1, "SSE 直通不应启动多轮 reasoning 请求");
const passthroughRequestsResponse = await fetch(
`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/requests?query=${encodeURIComponent(passthroughThreadId)}`,
{ headers: adminHeaders },
);
const passthroughRequestsPayload = await passthroughRequestsResponse.json();
const passthroughEntry = (passthroughRequestsPayload?.entries || []).find(
(entry) => entry.thread_id === passthroughThreadId,
);
assert(passthroughEntry?.inspected === false, "SSE 直通请求不应标记为已检查");
assert(passthroughEntry?.reasoning_retry_stop_reason === "stream_passthrough", "SSE 直通请求未记录跳过重试原因");
assert((passthroughEntry?.stream_chunk_count || 0) >= 1, "SSE 直通请求未累计流式 chunk");
const restoreStrictStreamResponse = await fetch(`http://127.0.0.1:${gatewayPort}/__codex_retry_gateway/api/config`, {
method: "POST",
headers: { ...adminHeaders, "content-type": "application/json" },
body: JSON.stringify({ stream_action: "strict_502" }),
});
assert(restoreStrictStreamResponse.status === 200, "测试后恢复严格 SSE 模式失败");
const capturedLifecycleStream = await readSseUntilClose(
`http://127.0.0.1:${gatewayPort}/responses`,
{
ui-src/src/App.tsx
+51
View File
@@ -4,6 +4,7 @@ type PageKey = "overview" | "requests" | "profiles" | "rules" | "logs";
type Tone = "" | "success" | "error";
type ReasoningMatchMode = "formula_518n_minus_2" | "manual";
type AuthMode = "passthrough" | "fixed_bearer" | "manual_bearer" | "auth_json";
type StreamAction = "strict_502" | "disconnect" | "passthrough";
type GatewayConfig = {
profile_name?: string;
@@ -30,6 +31,7 @@ type GatewayConfig = {
upstream_fetch_retry_attempts?: number;
upstream_fetch_retry_backoff_ms?: number;
non_stream_status_code?: number;
stream_action?: StreamAction;
log_match?: boolean;
};
@@ -190,6 +192,7 @@ type ProfileFormModel = {
retryable_error_messages?: string[];
upstream_fetch_retry_attempts?: string;
upstream_fetch_retry_backoff_ms?: string;
stream_action?: StreamAction;
endpoints?: string[];
};
@@ -207,6 +210,7 @@ type Profile = {
model_remap?: string;
reasoning_match_mode?: ReasoningMatchMode;
reasoning_equals?: string;
stream_action?: StreamAction;
};
form?: ProfileFormModel;
};
@@ -321,6 +325,7 @@ type ProfileFormState = {
retryable_error_messages: string;
upstream_fetch_retry_attempts: string;
upstream_fetch_retry_backoff_ms: string;
stream_action: StreamAction;
endpoints: string;
};
@@ -346,6 +351,7 @@ type RuleFormState = {
upstream_fetch_retry_backoff_ms: string;
endpoints: string;
non_stream_status_code: string;
stream_action: StreamAction;
log_match: boolean;
};
@@ -454,6 +460,7 @@ const defaultProfileForm: ProfileFormState = {
retryable_error_messages: "Selected model is at capacity. Please try a different model.\nstream disconnected before completion: Concurrency limit exceeded for account, please retry later",
upstream_fetch_retry_attempts: "5",
upstream_fetch_retry_backoff_ms: "350",
stream_action: "strict_502",
endpoints: "/responses\n/chat/completions\n/v1/responses\n/v1/chat/completions",
};
@@ -694,6 +701,24 @@ function normalizeReasoningMode(value: unknown): ReasoningMatchMode {
return value === "manual" ? "manual" : "formula_518n_minus_2";
}
function normalizeStreamAction(value: unknown): StreamAction {
if (value === "disconnect" || value === "passthrough") {
return value;
}
return "strict_502";
}
function formatStreamAction(value: unknown) {
const action = normalizeStreamAction(value);
if (action === "passthrough") {
return "直通";
}
if (action === "disconnect") {
return "透传后断开";
}
return "严格检查";
}
function formatReasoningMode(mode: ReasoningMatchMode) {
return mode === "manual" ? "manual" : "518n-2";
}
@@ -782,6 +807,7 @@ function profileFormFromStatus(status: StatusPayload | null): ProfileFormState {
upstream_fetch_retry_backoff_ms: String(
config.upstream_fetch_retry_backoff_ms ?? defaultProfileForm.upstream_fetch_retry_backoff_ms,
),
stream_action: normalizeStreamAction(config.stream_action),
endpoints: Array.isArray(config.endpoints) ? config.endpoints.join("\n") : defaultProfileForm.endpoints,
};
}
@@ -814,6 +840,7 @@ function profileFormFromProfile(profile: Profile): ProfileFormState {
form.upstream_fetch_retry_attempts || defaultProfileForm.upstream_fetch_retry_attempts,
upstream_fetch_retry_backoff_ms:
form.upstream_fetch_retry_backoff_ms || defaultProfileForm.upstream_fetch_retry_backoff_ms,
stream_action: normalizeStreamAction(form.stream_action),
endpoints: Array.isArray(form.endpoints) ? form.endpoints.join("\n") : "",
};
}
@@ -866,6 +893,7 @@ function ruleFormFromStatus(status: StatusPayload | null): RuleFormState {
),
endpoints: Array.isArray(config.endpoints) ? config.endpoints.join("\n") : "",
non_stream_status_code: String(config.non_stream_status_code || 502),
stream_action: normalizeStreamAction(config.stream_action),
log_match: Boolean(config.log_match),
};
}
@@ -1168,6 +1196,7 @@ export default function App() {
upstream_fetch_retry_backoff_ms: Number.parseInt(ruleForm.upstream_fetch_retry_backoff_ms, 10),
endpoints: splitLines(ruleForm.endpoints),
non_stream_status_code: Number.parseInt(ruleForm.non_stream_status_code, 10),
stream_action: ruleForm.stream_action,
log_match: ruleForm.log_match,
}),
});
@@ -1276,6 +1305,7 @@ export default function App() {
retryable_error_messages: splitLines(profileForm.retryable_error_messages),
upstream_fetch_retry_attempts: Number.parseInt(profileForm.upstream_fetch_retry_attempts, 10),
upstream_fetch_retry_backoff_ms: Number.parseInt(profileForm.upstream_fetch_retry_backoff_ms, 10),
stream_action: profileForm.stream_action,
endpoints: splitLines(profileForm.endpoints),
}),
});
@@ -2202,6 +2232,7 @@ export default function App() {
label="Rule Mode"
value={formatReasoningMode(normalizeReasoningMode(profile.summary?.reasoning_match_mode))}
/>
<MiniStat label="SSE" value={formatStreamAction(profile.summary?.stream_action)} />
</div>
</article>
))
@@ -2388,6 +2419,16 @@ export default function App() {
<input type="number" min={0} step={50} value={profileForm.upstream_fetch_retry_backoff_ms} onChange={(event) => setProfileForm({ ...profileForm, upstream_fetch_retry_backoff_ms: event.target.value })} />
</Field>
</div>
<Field label="stream_action" hint="passthrough 对 stream:true 请求关闭 reasoning retry；若上游返回 text/event-stream，会把原始 chunk 直接转发，不解析、缓存或规范化。">
<select
value={profileForm.stream_action}
onChange={(event) => setProfileForm({ ...profileForm, stream_action: normalizeStreamAction(event.target.value) })}
>
<option value="strict_502">strict_502：完整检查后回放</option>
<option value="disconnect">disconnect：边转发边检查，命中后断开</option>
<option value="passthrough">passthrough：SSE 原样直通</option>
</select>
</Field>
<Field label="endpoints">
<textarea value={profileForm.endpoints} onChange={(event) => setProfileForm({ ...profileForm, endpoints: event.target.value })} />
</Field>
@@ -2717,6 +2758,16 @@ export default function App() {
命中时写日志
</label>
</div>
<Field label="stream_action" hint="passthrough 会立即使当前实例的 SSE 原样直通；若要让重启或切换 profile 后仍保持，请在 Profiles 页保存该字段。">
<select
value={ruleForm.stream_action}
onChange={(event) => setRuleForm({ ...ruleForm, stream_action: normalizeStreamAction(event.target.value) })}
>
<option value="strict_502">strict_502：完整检查后回放</option>
<option value="disconnect">disconnect：边转发边检查，命中后断开</option>
<option value="passthrough">passthrough：SSE 原样直通</option>
</select>
</Field>
<div className="toolbar">
<button className="primary" type="submit">
保存并立即生效