mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-21 22:52:19 +03:00
fix(responses): replace synthetic reasoning keepalive (#10806)
Merged — locally validated (61/61 focused tests: early-stream-keepalive, chat-body-admission, responses-parse-once-4041, responses-route-early-keepalive-wiring; file-size/changelog gates clean, merges conflict-free against the current release tip). Good catch replacing the synthetic reasoning placeholder with a real response.in_progress bookkeeping event — keeps event-level watchdogs (Codex etc.) happy without any replayable fake reasoning content. Thanks!
This commit is contained in:
Xiangzhe
@@ -2929,11 +2929,6 @@
|
||||
"count": 83
|
||||
}
|
||||
},
|
||||
"tests/unit/responses-parse-once-4041.test.ts": {
|
||||
"@typescript-eslint/no-explicit-any": {
|
||||
"count": 6
|
||||
}
|
||||
},
|
||||
"tests/unit/responses-translation-fixes.test.ts": {
|
||||
"@typescript-eslint/no-explicit-any": {
|
||||
"count": 35
|
||||
|
||||
@@ -31,7 +31,6 @@
|
||||
* to 200, so the HTTP status can no longer change).
|
||||
*/
|
||||
|
||||
import { ResponsesOutputIndexStack } from "./responsesOutputIndexStack.ts";
|
||||
import { recordEarlyKeepaliveBytes } from "./earlyKeepaliveByteBuffer.ts";
|
||||
|
||||
const ENCODER = new TextEncoder();
|
||||
@@ -52,91 +51,6 @@ export const OPENAI_STARTUP_FRAME = OPENAI_KEEPALIVE_FRAME;
|
||||
// token the comment frame lets the client abort and retry the stream. Anthropic's own
|
||||
// API emits `event: ping` for exactly this reason; the /v1/messages route mirrors it.
|
||||
export const ANTHROPIC_PING_FRAME = ENCODER.encode('event: ping\ndata: {"type":"ping"}\n\n');
|
||||
// Responses API keepalive: a self-contained, self-closed synthetic reasoning
|
||||
// item (added -> summary_part.added -> text.delta -> summary_part.done ->
|
||||
// output_item.done). Unlike open-sse/utils/stream.ts's own
|
||||
// emitSyntheticResponsesReasoningSummary — which only supplements a REAL
|
||||
// upstream item that the real provider stream will close on its own — this
|
||||
// placeholder item has no real counterpart: the upstream response, once it
|
||||
// arrives, starts its own independent response.created lifecycle from
|
||||
// scratch and will never close this one. It must therefore send its own
|
||||
// response.output_item.done here, not just reasoning_summary_part.done
|
||||
// (that only closes the nested summary part, not the output item itself).
|
||||
// Without it, a strict client tracking open items by output_index (as the
|
||||
// Responses API spec requires) sees this item still open at index 0 and
|
||||
// throws a collision the moment the real response's own output_item.added
|
||||
// reuses that same index — reproduced live 2026-08-13, OpenClaw issue
|
||||
// https://github.com/openclaw/openclaw/issues/123342.
|
||||
//
|
||||
// The output_index is allocated from ResponsesOutputIndexStack instead of a
|
||||
// hardcoded literal so this stays structurally correct: forgetting the
|
||||
// close() call throws at module load (assertAllClosed() below), not
|
||||
// silently at some future real request.
|
||||
const RESPONSES_STARTUP_ITEM_ID = "rs_keepalive";
|
||||
// Brand-neutral placeholder — clients persist this as visible reasoning.
|
||||
const STARTUP_THINKING_TEXT = "✨";
|
||||
const startupIndexStack = new ResponsesOutputIndexStack();
|
||||
const RESPONSES_STARTUP_OUTPUT_INDEX = startupIndexStack.open();
|
||||
const startupEvents = [
|
||||
{
|
||||
event: "response.output_item.added",
|
||||
data: {
|
||||
type: "response.output_item.added",
|
||||
output_index: RESPONSES_STARTUP_OUTPUT_INDEX,
|
||||
item: { id: RESPONSES_STARTUP_ITEM_ID, type: "reasoning", summary: [] },
|
||||
},
|
||||
},
|
||||
{
|
||||
event: "response.reasoning_summary_part.added",
|
||||
data: {
|
||||
type: "response.reasoning_summary_part.added",
|
||||
item_id: RESPONSES_STARTUP_ITEM_ID,
|
||||
output_index: RESPONSES_STARTUP_OUTPUT_INDEX,
|
||||
summary_index: 0,
|
||||
part: { type: "summary_text", text: "" },
|
||||
},
|
||||
},
|
||||
{
|
||||
event: "response.reasoning_summary_text.delta",
|
||||
data: {
|
||||
type: "response.reasoning_summary_text.delta",
|
||||
item_id: RESPONSES_STARTUP_ITEM_ID,
|
||||
output_index: RESPONSES_STARTUP_OUTPUT_INDEX,
|
||||
summary_index: 0,
|
||||
delta: STARTUP_THINKING_TEXT,
|
||||
},
|
||||
},
|
||||
{
|
||||
event: "response.reasoning_summary_part.done",
|
||||
data: {
|
||||
type: "response.reasoning_summary_part.done",
|
||||
item_id: RESPONSES_STARTUP_ITEM_ID,
|
||||
output_index: RESPONSES_STARTUP_OUTPUT_INDEX,
|
||||
summary_index: 0,
|
||||
part: { type: "summary_text", text: STARTUP_THINKING_TEXT },
|
||||
},
|
||||
},
|
||||
];
|
||||
// close() runs before the output_item.done event is built (not just before
|
||||
// it's appended) so assertAllClosed() below is a real check, not scaffolding
|
||||
// that always trivially passes.
|
||||
startupIndexStack.close(RESPONSES_STARTUP_OUTPUT_INDEX);
|
||||
startupEvents.push({
|
||||
event: "response.output_item.done",
|
||||
data: {
|
||||
type: "response.output_item.done",
|
||||
output_index: RESPONSES_STARTUP_OUTPUT_INDEX,
|
||||
item: {
|
||||
id: RESPONSES_STARTUP_ITEM_ID,
|
||||
type: "reasoning",
|
||||
summary: [{ type: "summary_text", text: STARTUP_THINKING_TEXT }],
|
||||
},
|
||||
},
|
||||
});
|
||||
startupIndexStack.assertAllClosed();
|
||||
export const RESPONSES_STARTUP_THINKING_FRAME = ENCODER.encode(
|
||||
startupEvents.map((e) => `event: ${e.event}\ndata: ${JSON.stringify(e.data)}\n\n`).join("")
|
||||
);
|
||||
// Anthropic Messages API default — Anthropic's own spec really does use a named
|
||||
// `event: error` SSE frame, so this is correct there. It is WRONG for the OpenAI-
|
||||
// format routes below: Chat Completions and Responses streaming never use the SSE
|
||||
@@ -192,11 +106,15 @@ export type EarlyStreamKeepaliveOptions = {
|
||||
/**
|
||||
* Frame emitted ONCE, immediately, as the very first byte of the slow path —
|
||||
* before the recurring `keepaliveFrame` ticks start. Defaults to
|
||||
* `keepaliveFrame` when omitted (today's behavior, unchanged). Pass a
|
||||
* content-bearing frame (e.g. `OPENAI_STARTUP_THINKING_FRAME`) so the client
|
||||
* sees visible progress instead of an empty/no-op keepalive on the first byte.
|
||||
* `keepaliveFrame` when omitted (today's behavior, unchanged).
|
||||
*/
|
||||
startupFrame?: Uint8Array;
|
||||
/**
|
||||
* Optional parser-visible frame emitted at a slower cadence than the transport
|
||||
* heartbeat. A due application frame replaces that interval's keepalive frame,
|
||||
* so both cadences share one timer and never burst after an event-loop stall.
|
||||
*/
|
||||
applicationKeepalive?: { frame: Uint8Array; intervalMs: number };
|
||||
/** Extra headers to include in the keepalive response (e.g. X-Correlation-Id). */
|
||||
extraHeaders?: Record<string, string>;
|
||||
/**
|
||||
@@ -241,6 +159,13 @@ export async function withEarlyStreamKeepalive(
|
||||
const signal = options.signal ?? null;
|
||||
const keepaliveFrame = options.keepaliveFrame ?? KEEPALIVE_FRAME;
|
||||
const startupFrame = options.startupFrame ?? keepaliveFrame;
|
||||
const applicationKeepalive =
|
||||
options.applicationKeepalive && options.applicationKeepalive.intervalMs > 0
|
||||
? {
|
||||
frame: options.applicationKeepalive.frame,
|
||||
intervalMs: Math.max(intervalMs, options.applicationKeepalive.intervalMs),
|
||||
}
|
||||
: null;
|
||||
const extraHeaders = options.extraHeaders ?? {};
|
||||
const errorFrame = options.errorFrame ?? ERROR_FRAME;
|
||||
// Single source of truth for whether THIS route's error framing uses a named SSE
|
||||
@@ -291,22 +216,30 @@ export async function withEarlyStreamKeepalive(
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
async start(controller) {
|
||||
let stopped = false;
|
||||
let nextApplicationKeepaliveAt = applicationKeepalive
|
||||
? performance.now() + applicationKeepalive.intervalMs
|
||||
: Number.POSITIVE_INFINITY;
|
||||
const interval = setInterval(() => {
|
||||
if (stopped) return;
|
||||
try {
|
||||
controller.enqueue(keepaliveFrame);
|
||||
recordClientBytes(keepaliveFrame);
|
||||
const now = performance.now();
|
||||
let frame = keepaliveFrame;
|
||||
if (applicationKeepalive && now >= nextApplicationKeepaliveAt) {
|
||||
frame = applicationKeepalive.frame;
|
||||
nextApplicationKeepaliveAt = now + applicationKeepalive.intervalMs;
|
||||
}
|
||||
controller.enqueue(frame);
|
||||
recordClientBytes(frame);
|
||||
} catch {
|
||||
stopped = true;
|
||||
clearInterval(interval);
|
||||
}
|
||||
}, intervalMs);
|
||||
if (interval && typeof interval === "object" && "unref" in interval) {
|
||||
if (typeof interval === "object" && interval !== null && "unref" in interval) {
|
||||
interval.unref?.();
|
||||
}
|
||||
// First frame immediately on commit so the client sees a byte right away.
|
||||
// Use `startupFrame` (e.g. OPENAI_STARTUP_THINKING_FRAME / ANTHROPIC_PING_FRAME)
|
||||
// — an SSE comment here would be ignored by Anthropic clients' watchdog on a
|
||||
// An SSE comment here would be ignored by Anthropic clients' watchdog on a
|
||||
// sub-interval gap, defeating the keepalive for exactly the case it targets.
|
||||
try {
|
||||
controller.enqueue(startupFrame);
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
/**
|
||||
* @file responsesOutputIndexStack.ts
|
||||
* @description Structural guard against the Responses-API output_index
|
||||
* collision bug class (OpenClaw issue #123342): a hand-tracked output_index
|
||||
* that an emitter forgets to close before the same number gets reused.
|
||||
*
|
||||
* Responses-API output items open and close one at a time within any single
|
||||
* emitter — there is never a real need to hold two indices open
|
||||
* simultaneously from one emitter's own bookkeeping. Modeling allocation as
|
||||
* a stack makes "forgot to close" a structural impossibility instead of a
|
||||
* silent bug: open() always returns the next sequential index, close()
|
||||
* requires the caller to name the index being closed and throws if it does
|
||||
* not match the top of the stack, and assertAllClosed() — called once the
|
||||
* caller has finished building its frame/events — throws if anything is
|
||||
* still open. For a module-level constant frame (like the early keepalive
|
||||
* placeholder), that last check runs at import time: a regression here fails
|
||||
* the build/boot instead of shipping a malformed stream to production.
|
||||
*/
|
||||
|
||||
export class ResponsesOutputIndexStack {
|
||||
private readonly openIndices: number[] = [];
|
||||
private nextIndex = 0;
|
||||
|
||||
open(): number {
|
||||
const index = this.nextIndex;
|
||||
this.nextIndex += 1;
|
||||
this.openIndices.push(index);
|
||||
return index;
|
||||
}
|
||||
|
||||
close(index: number): void {
|
||||
const top = this.openIndices.at(-1);
|
||||
if (top !== index) {
|
||||
throw new Error(
|
||||
`ResponsesOutputIndexStack: closing output_index ${index} but the open top was ${String(top)}`
|
||||
);
|
||||
}
|
||||
this.openIndices.pop();
|
||||
}
|
||||
|
||||
assertAllClosed(): void {
|
||||
if (this.openIndices.length > 0) {
|
||||
throw new Error(
|
||||
`ResponsesOutputIndexStack: output_index(es) still open with no close(): ${this.openIndices.join(", ")}`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5,8 +5,16 @@
|
||||
* @changes
|
||||
* - [2026-07-28] [Cursor Grok 4.5] - Brand-neutral default OpenAI keepalive id/model
|
||||
*/
|
||||
const HEARTBEAT_ENCODER = new TextEncoder();
|
||||
const OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD = 'data: {"type":"response.in_progress"}\n\n';
|
||||
|
||||
export const DEFAULT_SSE_HEARTBEAT_INTERVAL_MS = 15_000;
|
||||
|
||||
/** Shared Responses API heartbeat frame for early and mid-stream keepalives. */
|
||||
export const OPENAI_RESPONSES_IN_PROGRESS_FRAME = HEARTBEAT_ENCODER.encode(
|
||||
OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD
|
||||
);
|
||||
|
||||
export const HEARTBEAT_SHAPES = {
|
||||
COMMENT: "comment",
|
||||
ANTHROPIC_PING: "anthropic-ping",
|
||||
@@ -41,7 +49,7 @@ function buildHeartbeatPayload(
|
||||
case HEARTBEAT_SHAPES.ANTHROPIC_PING:
|
||||
return 'event: ping\ndata: {"type":"ping"}\n\n';
|
||||
case HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS:
|
||||
return 'data: {"type":"response.in_progress"}\n\n';
|
||||
return OPENAI_RESPONSES_IN_PROGRESS_PAYLOAD;
|
||||
case HEARTBEAT_SHAPES.OPENAI_CHUNK: {
|
||||
const payload = {
|
||||
id: opts.chunkId ?? "chatcmpl-keepalive",
|
||||
@@ -66,8 +74,6 @@ type SseHeartbeatTransformOptions = {
|
||||
chunkModel?: string;
|
||||
};
|
||||
|
||||
const HEARTBEAT_ENCODER = new TextEncoder();
|
||||
|
||||
/**
|
||||
* Whether OmniRoute may emit SSE `:` comment lines (e.g. the `: keepalive` heartbeat).
|
||||
* Some strict OpenAI-compatible clients parse every SSE line as JSON and crash on `:` comments.
|
||||
|
||||
@@ -1,15 +1,26 @@
|
||||
import { handleChat } from "@/sse/handlers/chat";
|
||||
import {
|
||||
withEarlyStreamKeepalive,
|
||||
RESPONSES_STARTUP_THINKING_FRAME,
|
||||
OPENAI_RESPONSES_ERROR_FRAME,
|
||||
} from "@omniroute/open-sse/utils/earlyStreamKeepalive";
|
||||
import { withInjectionGuard } from "@/middleware/promptInjectionGuard";
|
||||
import { CORS_HEADERS } from "@/shared/utils/cors";
|
||||
import { createInjectionGuard } from "@/middleware/promptInjectionGuard";
|
||||
import { resolveResponsesApiModel } from "@/app/api/internal/codex-responses-ws/modelResolution";
|
||||
import { getModelInfo, getComboForModel } from "@/sse/services/model";
|
||||
import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
|
||||
import { resolveStreamFlag } from "@omniroute/open-sse/utils/aiSdkCompat";
|
||||
import { generateRequestId } from "@/shared/utils/requestId";
|
||||
import {
|
||||
admitChatRequest,
|
||||
admitChatStructure,
|
||||
CHAT_ADMISSION_QUEUE_MAX_MS,
|
||||
releaseChatAdmissionAfterHandler,
|
||||
releaseChatAdmissionWhenDone,
|
||||
resolveSessionId,
|
||||
} from "@/shared/middleware/chatBodyAdmission";
|
||||
import { SSE_HEARTBEAT_INTERVAL_MS } from "@omniroute/open-sse/config/constants";
|
||||
import { resolveStreamFlag } from "@omniroute/open-sse/utils/aiSdkCompat";
|
||||
import { errorResponse } from "@omniroute/open-sse/utils/error";
|
||||
import {
|
||||
withEarlyStreamKeepalive,
|
||||
OPENAI_RESPONSES_ERROR_FRAME,
|
||||
} from "@omniroute/open-sse/utils/earlyStreamKeepalive";
|
||||
import { resolveKeepaliveThreshold } from "@omniroute/open-sse/utils/keepaliveThreshold";
|
||||
import { OPENAI_RESPONSES_IN_PROGRESS_FRAME } from "@omniroute/open-sse/utils/sseHeartbeat";
|
||||
|
||||
// NOTE: We do NOT call initTranslators() here — the translator registry is
|
||||
// bootstrapped at module level inside open-sse/translator/index.ts when it
|
||||
@@ -20,6 +31,8 @@ import { generateRequestId } from "@/shared/utils/requestId";
|
||||
// The translators are always initialized via the open-sse side (chatCore),
|
||||
// so /v1/responses just delegates to handleChat which handles everything.
|
||||
|
||||
const injectionGuard = createInjectionGuard();
|
||||
|
||||
export async function OPTIONS() {
|
||||
return new Response(null, {
|
||||
headers: {
|
||||
@@ -35,8 +48,8 @@ export async function OPTIONS() {
|
||||
* the CLI sends bare "gpt-5.5" over HTTP after WS closes (1008 Policy), and
|
||||
* without this rewrite OmniRoute routes it to openrouter instead of codex.
|
||||
*
|
||||
* Accepts an optional `preParsedBody` (threaded from withInjectionGuard via #4041)
|
||||
* to avoid re-cloning the request when the body was already parsed upstream.
|
||||
* Accepts an optional `preParsedBody` so the route-level admission and injection
|
||||
* checks can parse once and avoid re-cloning the request on the hot path.
|
||||
*
|
||||
* Safe: only rewrites when codex/model is genuinely registered; all other models
|
||||
* pass through unchanged. Errors are caught and the original request + body are returned.
|
||||
@@ -80,42 +93,115 @@ export async function withCodexPreferredModel(
|
||||
/**
|
||||
* POST /v1/responses - OpenAI Responses API format
|
||||
* Handled by the unified chat handler (openai-responses format auto-detected).
|
||||
*
|
||||
* `preParsedBody` is threaded from withInjectionGuard (#4041) so the body is
|
||||
* parsed at most once per request instead of 3-4x on the hot codex path.
|
||||
*/
|
||||
async function postHandler(request: any, context: any, preParsedBody: any = null) {
|
||||
// Codex CLI (wire_api="responses") consumes this endpoint over SSE and its reqwest
|
||||
// client drops the connection if no bytes arrive within ~5s. Keep the connection
|
||||
// warm with early keepalives while the upstream produces its first token (#2544).
|
||||
// Non-streaming callers (JSON) keep the original verbatim path untouched.
|
||||
const { request: resolved, body: resolvedBody } = await withCodexPreferredModel(
|
||||
request,
|
||||
preParsedBody
|
||||
);
|
||||
const accept = String(request.headers?.get?.("accept") || "");
|
||||
const wantsStreaming = resolveStreamFlag(resolvedBody?.stream, accept, "openai-responses");
|
||||
if (wantsStreaming) {
|
||||
// Adaptive threshold: web-session and anonymous-fallback providers are slower
|
||||
// to produce the first byte, so use a longer keepalive threshold (15s vs 2s).
|
||||
// Reuse resolvedBody.model — no extra clone/parse needed (#4041).
|
||||
const model = resolvedBody?.model;
|
||||
const thresholdMs = resolveKeepaliveThreshold(model);
|
||||
// Generated here (rather than left to handleChatImplementation's own
|
||||
// fallback) so withEarlyStreamKeepalive can tag its own direct-to-client
|
||||
// writes with the same id chatCore.ts ends up persisting the call log
|
||||
// under — see earlyKeepaliveByteBuffer.ts for why this is the only way
|
||||
// the two sides of that boundary can agree on "which request."
|
||||
const correlationId = generateRequestId();
|
||||
return await withEarlyStreamKeepalive(handleChat(resolved, null, resolvedBody, correlationId), {
|
||||
async function postHandler(request: any) {
|
||||
const sessionId = resolveSessionId(request);
|
||||
const admissionResult = await admitChatRequest(request, {
|
||||
sessionId,
|
||||
queueMs: CHAT_ADMISSION_QUEUE_MAX_MS,
|
||||
});
|
||||
if (admissionResult.admit === false) return admissionResult.response;
|
||||
|
||||
const admission = admissionResult;
|
||||
request = admission.request;
|
||||
const finishAdmission = (response: Response) =>
|
||||
releaseChatAdmissionWhenDone(response, admission.lease);
|
||||
|
||||
try {
|
||||
let parsedBody;
|
||||
try {
|
||||
parsedBody = await request.json();
|
||||
} catch {
|
||||
return finishAdmission(errorResponse(400, "Invalid JSON body"));
|
||||
}
|
||||
if (!parsedBody || typeof parsedBody !== "object" || Array.isArray(parsedBody)) {
|
||||
return finishAdmission(errorResponse(400, "Request body must be a JSON object"));
|
||||
}
|
||||
|
||||
const structuralAdmission = await admitChatStructure(parsedBody, admission.lease, {
|
||||
sessionId,
|
||||
queueMs: CHAT_ADMISSION_QUEUE_MAX_MS,
|
||||
signal: request.signal,
|
||||
thresholdMs,
|
||||
startupFrame: RESPONSES_STARTUP_THINKING_FRAME,
|
||||
errorFrame: OPENAI_RESPONSES_ERROR_FRAME,
|
||||
correlationId,
|
||||
});
|
||||
if (structuralAdmission.admit === false) {
|
||||
admission.lease?.release();
|
||||
return finishAdmission(structuralAdmission.response);
|
||||
}
|
||||
admission.lease = structuralAdmission.lease;
|
||||
|
||||
let guardResult;
|
||||
try {
|
||||
guardResult = injectionGuard(parsedBody);
|
||||
} catch (error) {
|
||||
console.error("[SECURITY] Injection guard error:", error);
|
||||
return finishAdmission(
|
||||
new Response(JSON.stringify({ error: "Security check failed" }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
const { blocked, result } = guardResult;
|
||||
if (blocked) {
|
||||
return finishAdmission(
|
||||
new Response(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
message: "Request blocked: potential prompt injection detected",
|
||||
type: "injection_detected",
|
||||
code: "SECURITY_001",
|
||||
detections: result.detections.length,
|
||||
},
|
||||
}),
|
||||
{ status: 400, headers: { ...CORS_HEADERS, "Content-Type": "application/json" } }
|
||||
)
|
||||
);
|
||||
}
|
||||
if (result.flagged) {
|
||||
try {
|
||||
request.headers.set("X-Injection-Flagged", "true");
|
||||
request.headers.set("X-Injection-Detections", String(result.detections.length));
|
||||
} catch {
|
||||
// Detection already ran; metadata propagation is best-effort.
|
||||
}
|
||||
}
|
||||
|
||||
// Codex CLI (wire_api="responses") consumes this endpoint over SSE and its reqwest
|
||||
// client drops the connection if no bytes arrive within ~5s. Keep the connection
|
||||
// warm with transport comments plus sparse parser-visible events while the upstream
|
||||
// produces its first token (#2544).
|
||||
const { request: resolved, body: resolvedBody } = await withCodexPreferredModel(
|
||||
request,
|
||||
parsedBody
|
||||
);
|
||||
const accept = String(request.headers?.get?.("accept") || "");
|
||||
const wantsStreaming = resolveStreamFlag(resolvedBody?.stream, accept, "openai-responses");
|
||||
if (wantsStreaming) {
|
||||
const thresholdMs = resolveKeepaliveThreshold(resolvedBody?.model);
|
||||
const correlationId = generateRequestId();
|
||||
const handlerResponse = releaseChatAdmissionAfterHandler(
|
||||
handleChat(resolved, null, resolvedBody, correlationId),
|
||||
admission.lease
|
||||
);
|
||||
return await withEarlyStreamKeepalive(handlerResponse, {
|
||||
signal: request.signal,
|
||||
thresholdMs,
|
||||
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
applicationKeepalive: {
|
||||
frame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
intervalMs: SSE_HEARTBEAT_INTERVAL_MS,
|
||||
},
|
||||
errorFrame: OPENAI_RESPONSES_ERROR_FRAME,
|
||||
correlationId,
|
||||
});
|
||||
}
|
||||
|
||||
return finishAdmission(await handleChat(resolved, null, resolvedBody));
|
||||
} catch (error) {
|
||||
admission.lease?.release();
|
||||
throw error;
|
||||
}
|
||||
return await handleChat(resolved, null, resolvedBody);
|
||||
}
|
||||
|
||||
export const POST = withInjectionGuard(postHandler);
|
||||
export const POST = postHandler;
|
||||
|
||||
@@ -655,9 +655,8 @@ export async function admitChatStructure(
|
||||
} = {}
|
||||
): Promise<ChatStructureAdmission> {
|
||||
if (!body || typeof body !== "object" || Array.isArray(body)) return { admit: true, lease };
|
||||
|
||||
const record = body as Record<string, unknown>;
|
||||
const messages = Array.isArray(record.messages) ? record.messages : [];
|
||||
const messages = [record.messages, record.input].flat().filter((item) => item != null);
|
||||
const tools = Array.isArray(record.tools) ? record.tools : [];
|
||||
const maxMessages = options.maxMessages ?? CHAT_HARD_MAX_MESSAGES;
|
||||
// Opt-in only: `0`/unset means no history cap, so oversized conversations reach the
|
||||
|
||||
@@ -1,55 +0,0 @@
|
||||
/**
|
||||
* Validates the Responses-API output_index lifecycle invariant that real
|
||||
* clients (e.g. OpenClaw's outputSlots tracker) enforce: an output_index
|
||||
* claimed by response.output_item.added must be closed by a matching
|
||||
* response.output_item.done before any later item reuses that same index.
|
||||
*
|
||||
* Existing coverage (responses-reasoning-close-before-message-466.test.ts)
|
||||
* asserts this invariant by hand for one specific emitter path (the real
|
||||
* translator/transformer). This helper generalizes that check so any SSE
|
||||
* event sequence — including hand-rolled synthetic frames like the early
|
||||
* keepalive placeholder — can be verified against the same contract a real
|
||||
* downstream client applies, without duplicating the tracking logic per test.
|
||||
*
|
||||
* Mirrors OpenClaw's createResponsesOutputSlotTracker() closely enough to
|
||||
* reproduce the exact failure mode: "Responses stream reused active output
|
||||
* index N" (see OpenClaw issue #123342 / the RESPONSES_STARTUP_THINKING_FRAME
|
||||
* missing-output_item.done incident this helper was added for).
|
||||
*/
|
||||
|
||||
export type ResponsesLifecycleEvent = { event?: string; data: Record<string, unknown> };
|
||||
|
||||
export function assertResponsesOutputIndexLifecycle(
|
||||
events: ResponsesLifecycleEvent[],
|
||||
options: { requireAllClosed?: boolean } = {}
|
||||
): void {
|
||||
const open = new Map<number, unknown>();
|
||||
|
||||
for (const { data } of events) {
|
||||
const type = data?.type;
|
||||
if (type !== "response.output_item.added" && type !== "response.output_item.done") continue;
|
||||
|
||||
const outputIndex = data.output_index;
|
||||
if (typeof outputIndex !== "number") continue;
|
||||
|
||||
if (type === "response.output_item.added") {
|
||||
if (open.has(outputIndex)) {
|
||||
const item = data.item as { id?: unknown; type?: unknown } | undefined;
|
||||
throw new Error(
|
||||
`Responses stream reused active output index ${outputIndex} ` +
|
||||
`(item id=${String(item?.id)} type=${String(item?.type)} was still open)`
|
||||
);
|
||||
}
|
||||
open.set(outputIndex, data.item);
|
||||
} else {
|
||||
open.delete(outputIndex);
|
||||
}
|
||||
}
|
||||
|
||||
if (options.requireAllClosed !== false && open.size > 0) {
|
||||
const stillOpen = [...open.keys()].join(", ");
|
||||
throw new Error(
|
||||
`Responses stream left output index(es) open with no output_item.done: ${stillOpen}`
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -80,6 +80,24 @@ test("a byte-light request above the message threshold acquires heavyweight capa
|
||||
assert.equal(controller.activeHeavy, 0);
|
||||
});
|
||||
|
||||
test("Responses input items count toward heavyweight admission", async () => {
|
||||
const controller = new ChatAdmissionController(1);
|
||||
const result = await admitChatStructure(
|
||||
{
|
||||
input: [
|
||||
{ role: "user", content: "one" },
|
||||
{ role: "user", content: "two" },
|
||||
],
|
||||
},
|
||||
null,
|
||||
{ controller, maxMessages: 10, heavyMessages: 2, heavyTools: 10, heavyTokens: 10_000 }
|
||||
);
|
||||
|
||||
assert.equal(result.admit, true);
|
||||
assert.equal(controller.activeHeavy, 1);
|
||||
if (result.admit) result.lease?.release();
|
||||
});
|
||||
|
||||
test("a byte-light request above the tool threshold is rejected when heavy capacity is busy AND the heap is genuinely under pressure (#10183/#10268)", async () => {
|
||||
const controller = new ChatAdmissionController(1);
|
||||
const occupied = controller.tryAcquireHeavy();
|
||||
@@ -201,6 +219,21 @@ test("a conservative token estimate classifies string messages and tool schemas
|
||||
if (result.admit) result.lease?.release();
|
||||
});
|
||||
|
||||
test("Responses string input contributes to the conservative token estimate", async () => {
|
||||
const controller = new ChatAdmissionController(1);
|
||||
const result = await admitChatStructure({ messages: [], input: "abcdefgh" }, null, {
|
||||
controller,
|
||||
maxMessages: 10,
|
||||
heavyMessages: 10,
|
||||
heavyTools: 10,
|
||||
heavyTokens: 2,
|
||||
});
|
||||
|
||||
assert.equal(result.admit, true);
|
||||
assert.equal(controller.activeHeavy, 1);
|
||||
if (result.admit) result.lease?.release();
|
||||
});
|
||||
|
||||
test("exhausting the bounded structural inspection is conservatively heavyweight", async () => {
|
||||
const controller = new ChatAdmissionController(1);
|
||||
const result = await admitChatStructure(
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
* @description Unit tests for withEarlyStreamKeepalive (fast/slow path, frames, abort).
|
||||
*
|
||||
* @changes
|
||||
* - [2026-07-28] [Cursor Grok 4.5] - Assert brand-neutral startup thinking text (✨)
|
||||
* - [2026-08-16] - Assert Responses startup and recurring keepalives are neutral JSON events
|
||||
*/
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
@@ -13,12 +13,11 @@ import {
|
||||
ANTHROPIC_PING_FRAME,
|
||||
OPENAI_KEEPALIVE_FRAME,
|
||||
OPENAI_STARTUP_FRAME,
|
||||
RESPONSES_STARTUP_THINKING_FRAME,
|
||||
OPENAI_CHAT_ERROR_FRAME,
|
||||
OPENAI_RESPONSES_ERROR_FRAME,
|
||||
} from "../../open-sse/utils/earlyStreamKeepalive.ts";
|
||||
import { assertResponsesOutputIndexLifecycle } from "../helpers/assertResponsesOutputIndexLifecycle.ts";
|
||||
import { takeEarlyKeepaliveBytes } from "../../open-sse/utils/earlyKeepaliveByteBuffer.ts";
|
||||
import { OPENAI_RESPONSES_IN_PROGRESS_FRAME } from "../../open-sse/utils/sseHeartbeat.ts";
|
||||
|
||||
async function readAll(response: Response): Promise<string> {
|
||||
const reader = response.body!.getReader();
|
||||
@@ -175,131 +174,46 @@ test("startupFrame defaults to keepaliveFrame when omitted (no behavior change)"
|
||||
);
|
||||
});
|
||||
|
||||
// #7360 follow-up round 2: OpenClaw calls via /v1/responses (Responses API
|
||||
// format), which only had the generic bare-comment keepalive — a live
|
||||
// incident showed it disconnecting after ~56s waiting on a slow gemma-4
|
||||
// response. RESPONSES_STARTUP_THINKING_FRAME gives Responses-API clients the
|
||||
// same real-content keepalive OpenAI chat/completions already got, as a
|
||||
// self-contained (opened AND closed within this one frame) synthetic
|
||||
// reasoning item — it never claims a response_id, so it can't collide with
|
||||
// the real response's own independent response.created lifecycle that follows.
|
||||
test("RESPONSES_STARTUP_THINKING_FRAME is a self-closed synthetic reasoning item with the expected text", () => {
|
||||
const decoded = new TextDecoder().decode(RESPONSES_STARTUP_THINKING_FRAME);
|
||||
const events = decoded
|
||||
.split("\n\n")
|
||||
.filter(Boolean)
|
||||
.map((frame) => {
|
||||
const [eventLine, dataLine] = frame.split("\n");
|
||||
return {
|
||||
event: eventLine.replace(/^event: /, ""),
|
||||
data: JSON.parse(dataLine.replace(/^data: /, "")),
|
||||
};
|
||||
});
|
||||
|
||||
assert.deepEqual(
|
||||
events.map((e) => e.event),
|
||||
[
|
||||
"response.output_item.added",
|
||||
"response.reasoning_summary_part.added",
|
||||
"response.reasoning_summary_text.delta",
|
||||
"response.reasoning_summary_part.done",
|
||||
"response.output_item.done",
|
||||
]
|
||||
);
|
||||
|
||||
const [added, partAdded, delta, partDone, itemDone] = events;
|
||||
assert.equal(added.data.item.type, "reasoning");
|
||||
const itemId = added.data.item.id;
|
||||
assert.ok(itemId, "reasoning item must have an id");
|
||||
|
||||
assert.equal(partAdded.data.item_id, itemId);
|
||||
assert.equal(delta.data.item_id, itemId);
|
||||
assert.equal(delta.data.delta, "✨");
|
||||
assert.equal(partDone.data.item_id, itemId);
|
||||
assert.equal(partDone.data.part.text, "✨");
|
||||
|
||||
// Regression for the live 2026-08-13 incident (OpenClaw issue #123342):
|
||||
// reasoning_summary_part.done only closes the nested summary part, not the
|
||||
// output item itself. Without a matching response.output_item.done here,
|
||||
// a client tracking open items by output_index still sees this synthetic
|
||||
// item open at index 0 when the real upstream response later reuses that
|
||||
// same index for its own response.output_item.added, and throws a
|
||||
// collision ("Responses stream reused active output index 0").
|
||||
assert.equal(itemDone.data.output_index, added.data.output_index);
|
||||
assert.equal(itemDone.data.item.id, itemId);
|
||||
assert.equal(itemDone.data.item.type, "reasoning");
|
||||
|
||||
// General-purpose form of the same check: this frame alone must be a fully
|
||||
// self-closed lifecycle (no output_item left open at the end).
|
||||
assertResponsesOutputIndexLifecycle(events);
|
||||
});
|
||||
|
||||
test("RESPONSES_STARTUP_THINKING_FRAME does not collide when the real upstream response reuses output_index 0", () => {
|
||||
// Reproduces the actual live failure shape (OpenClaw issue #123342): the
|
||||
// keepalive placeholder fires, then the real upstream response starts its
|
||||
// own independent response.created lifecycle and reuses output_index 0 for
|
||||
// its own real reasoning item. Concatenating the two and replaying them
|
||||
// through the same output_index-lifecycle contract a real client enforces
|
||||
// is what actually would have caught the missing output_item.done — the
|
||||
// frame-shape-only test above could pass while this still failed.
|
||||
const decoded = new TextDecoder().decode(RESPONSES_STARTUP_THINKING_FRAME);
|
||||
const keepaliveEvents = decoded
|
||||
.split("\n\n")
|
||||
.filter(Boolean)
|
||||
.map((frame) => {
|
||||
const [eventLine, dataLine] = frame.split("\n");
|
||||
return {
|
||||
event: eventLine.replace(/^event: /, ""),
|
||||
data: JSON.parse(dataLine.replace(/^data: /, "")),
|
||||
};
|
||||
});
|
||||
|
||||
const realResponseEvents = [
|
||||
{ event: "response.created", data: { type: "response.created" } },
|
||||
{ event: "response.in_progress", data: { type: "response.in_progress" } },
|
||||
{
|
||||
event: "response.output_item.added",
|
||||
data: {
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: { id: "rs_real", type: "reasoning", summary: [] },
|
||||
},
|
||||
},
|
||||
{
|
||||
event: "response.output_item.done",
|
||||
data: {
|
||||
type: "response.output_item.done",
|
||||
output_index: 0,
|
||||
item: { id: "rs_real", type: "reasoning", summary: [] },
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
assert.doesNotThrow(() =>
|
||||
assertResponsesOutputIndexLifecycle([...keepaliveEvents, ...realResponseEvents])
|
||||
);
|
||||
});
|
||||
|
||||
test("slow handler emits the Responses API startup frame before the real body", async () => {
|
||||
// Responses clients need both frequent raw bytes and occasional parsed events while
|
||||
// upstream readiness is pending. Keep those cadences separate: comments cover the
|
||||
// short idle-read timeout, while sparse response.in_progress events reset parsers that
|
||||
// ignore comments without flooding the application event stream.
|
||||
test("slow Responses handler uses comments plus sparse in_progress events", async () => {
|
||||
const slow = new Promise<Response>((resolve) => {
|
||||
setTimeout(
|
||||
() => resolve(sseResponse("event: response.created\ndata: {}\n\ndata: [DONE]\n\n")),
|
||||
120
|
||||
);
|
||||
setTimeout(() => resolve(sseResponse('data: {"type":"response.completed"}\n\n')), 900);
|
||||
});
|
||||
|
||||
const result = await withEarlyStreamKeepalive(slow, {
|
||||
thresholdMs: 25,
|
||||
intervalMs: 20,
|
||||
startupFrame: RESPONSES_STARTUP_THINKING_FRAME,
|
||||
thresholdMs: 20,
|
||||
intervalMs: 250,
|
||||
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
applicationKeepalive: {
|
||||
frame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
intervalMs: 500,
|
||||
},
|
||||
});
|
||||
|
||||
const body = await readAll(result);
|
||||
assert.match(body, /event: response\.output_item\.added/);
|
||||
assert.match(body, /✨/);
|
||||
assert.match(body, /event: response\.reasoning_summary_part\.done/);
|
||||
assert.match(body, /event: response\.created/, "should forward the real upstream body");
|
||||
assert.match(body, /data: \[DONE\]/);
|
||||
const frames = body.split("\n\n").filter(Boolean);
|
||||
const earlyFrames = frames.slice(0, -1);
|
||||
assert.equal(earlyFrames[0], 'data: {"type":"response.in_progress"}');
|
||||
assert.ok(
|
||||
earlyFrames.some((frame) => frame === ": keepalive"),
|
||||
"transport ticks must remain lightweight SSE comments"
|
||||
);
|
||||
const applicationFrames = earlyFrames.filter((frame) => frame.startsWith("data: "));
|
||||
assert.ok(applicationFrames.length >= 2, "expected startup and sparse application keepalives");
|
||||
for (const frame of applicationFrames) {
|
||||
assert.deepEqual(JSON.parse(frame.slice("data: ".length)), {
|
||||
type: "response.in_progress",
|
||||
});
|
||||
assert.doesNotMatch(frame, /output_item|reasoning|✨/);
|
||||
}
|
||||
assert.ok(
|
||||
applicationFrames.length < earlyFrames.length,
|
||||
"application events must be sparser than transport heartbeats"
|
||||
);
|
||||
assert.match(body, /data: {"type":"response.completed"}/, "real upstream body forwarded");
|
||||
});
|
||||
|
||||
test("a correlationId records the startup frame and keepalive ticks, but not the forwarded body", async () => {
|
||||
@@ -307,20 +221,29 @@ test("a correlationId records the startup frame and keepalive ticks, but not the
|
||||
const slow = new Promise<Response>((resolve) => {
|
||||
setTimeout(
|
||||
() => resolve(sseResponse("event: response.created\ndata: {}\n\ndata: [DONE]\n\n")),
|
||||
65
|
||||
650
|
||||
);
|
||||
});
|
||||
|
||||
const result = await withEarlyStreamKeepalive(slow, {
|
||||
thresholdMs: 25,
|
||||
intervalMs: 20,
|
||||
startupFrame: RESPONSES_STARTUP_THINKING_FRAME,
|
||||
intervalMs: 250,
|
||||
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
applicationKeepalive: {
|
||||
frame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
intervalMs: 500,
|
||||
},
|
||||
correlationId,
|
||||
});
|
||||
await readAll(result);
|
||||
|
||||
const recorded = takeEarlyKeepaliveBytes(correlationId).join("");
|
||||
assert.match(recorded, /event: response\.output_item\.added/, "startup frame must be recorded");
|
||||
assert.match(
|
||||
recorded,
|
||||
/data: {"type":"response\.in_progress"}/,
|
||||
"startup frame must be recorded"
|
||||
);
|
||||
assert.match(recorded, /: keepalive/, "transport heartbeat must be recorded");
|
||||
assert.doesNotMatch(
|
||||
recorded,
|
||||
/event: response\.created/,
|
||||
@@ -337,7 +260,8 @@ test("omitting correlationId leaves the buffer untouched (today's behavior, unch
|
||||
const result = await withEarlyStreamKeepalive(slow, {
|
||||
thresholdMs: 25,
|
||||
intervalMs: 20,
|
||||
startupFrame: RESPONSES_STARTUP_THINKING_FRAME,
|
||||
keepaliveFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
startupFrame: OPENAI_RESPONSES_IN_PROGRESS_FRAME,
|
||||
});
|
||||
await readAll(result);
|
||||
|
||||
|
||||
@@ -1,46 +0,0 @@
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
import { ResponsesOutputIndexStack } from "../../open-sse/utils/responsesOutputIndexStack.ts";
|
||||
|
||||
test("open() allocates sequential indices starting at 0", () => {
|
||||
const stack = new ResponsesOutputIndexStack();
|
||||
assert.equal(stack.open(), 0);
|
||||
assert.equal(stack.open(), 1);
|
||||
});
|
||||
|
||||
test("close() on the current top does not throw", () => {
|
||||
const stack = new ResponsesOutputIndexStack();
|
||||
const index = stack.open();
|
||||
assert.doesNotThrow(() => stack.close(index));
|
||||
});
|
||||
|
||||
test("close() with a mismatched index throws (catches the exact keepalive bug shape)", () => {
|
||||
const stack = new ResponsesOutputIndexStack();
|
||||
const first = stack.open();
|
||||
stack.open();
|
||||
assert.throws(() => stack.close(first), /closing output_index 0 but the open top was 1/);
|
||||
});
|
||||
|
||||
test("assertAllClosed() passes when everything opened was closed", () => {
|
||||
const stack = new ResponsesOutputIndexStack();
|
||||
const index = stack.open();
|
||||
stack.close(index);
|
||||
assert.doesNotThrow(() => stack.assertAllClosed());
|
||||
});
|
||||
|
||||
test("assertAllClosed() throws when an index was never closed — the exact regression this stack prevents", () => {
|
||||
const stack = new ResponsesOutputIndexStack();
|
||||
stack.open();
|
||||
assert.throws(() => stack.assertAllClosed(), /still open with no close/);
|
||||
});
|
||||
|
||||
test("a later open() after a forgotten close() gets the next index, never a reused one", () => {
|
||||
// This is the structural guarantee replacing the old hand-tracked literal
|
||||
// output_index: 0 in RESPONSES_STARTUP_THINKING_FRAME: even if a caller
|
||||
// forgets to close(), the next open() can never collide with it.
|
||||
const stack = new ResponsesOutputIndexStack();
|
||||
const first = stack.open();
|
||||
const second = stack.open();
|
||||
assert.notEqual(first, second);
|
||||
});
|
||||
@@ -1,15 +1,16 @@
|
||||
import test from "node:test";
|
||||
import test, { after } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
// #4041: /v1/responses (Codex wire_api=responses hot path) parsed the JSON body 3-4x per
|
||||
// request — once in withInjectionGuard, once in withCodexPreferredModel, once for model
|
||||
// detection before SSE keepalive, and once more inside handleChat via resolveChatRequestBody.
|
||||
//
|
||||
// The fix threads the already-parsed body from withInjectionGuard into the wrapped handler
|
||||
// as a third argument (preParsedBody), mirroring the existing /v1/chat/completions pattern
|
||||
// (#4380). This test confirms: (a) withInjectionGuard passes the body it parsed to the inner
|
||||
// handler as a 3rd arg, and (b) withCodexPreferredModel reuses an already-parsed body
|
||||
// instead of re-cloning+re-parsing the request.
|
||||
const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-responses-parse-once-"));
|
||||
process.env.DATA_DIR = dataDir;
|
||||
after(() => fs.rmSync(dataDir, { recursive: true, force: true }));
|
||||
|
||||
// #4041: AI routes must parse each JSON body at most once and thread the parsed value
|
||||
// through model resolution and handleChat. /v1/responses now parses after raw-body admission;
|
||||
// withInjectionGuard retains the same preParsedBody contract for routes that still wrap it.
|
||||
|
||||
// ─── Part A: withInjectionGuard threads the parsed body ──────────────────────
|
||||
|
||||
@@ -18,7 +19,7 @@ const { withInjectionGuard } = await import("../../src/middleware/promptInjectio
|
||||
test("#4041 withInjectionGuard passes the parsed body as 3rd arg to the inner handler", async () => {
|
||||
let receivedPreParsed: unknown = undefined;
|
||||
|
||||
const innerHandler = async (_request: any, _context: any, preParsedBody: unknown) => {
|
||||
const innerHandler = async (_request: Request, _context: unknown, preParsedBody: unknown) => {
|
||||
receivedPreParsed = preParsedBody;
|
||||
return new Response("ok");
|
||||
};
|
||||
@@ -44,7 +45,7 @@ test("#4041 withInjectionGuard passes the parsed body as 3rd arg to the inner ha
|
||||
test("#4041 withInjectionGuard passes null as 3rd arg when body cannot be parsed", async () => {
|
||||
let receivedPreParsed: unknown = "sentinel";
|
||||
|
||||
const innerHandler = async (_request: any, _context: any, preParsedBody: unknown) => {
|
||||
const innerHandler = async (_request: Request, _context: unknown, preParsedBody: unknown) => {
|
||||
receivedPreParsed = preParsedBody;
|
||||
return new Response("ok");
|
||||
};
|
||||
@@ -70,83 +71,30 @@ test("#4041 withInjectionGuard passes null as 3rd arg when body cannot be parsed
|
||||
|
||||
// ─── Part B: withCodexPreferredModel reuses pre-parsed body ──────────────────
|
||||
|
||||
// Import the internal helper directly. It is not exported as a named export from
|
||||
// the route file by default, but we can import the module and access it.
|
||||
// We spy on Request.prototype behaviour by counting .clone() calls instead.
|
||||
|
||||
test("#4041 withCodexPreferredModel accepts a pre-parsed body and avoids re-cloning the request", async () => {
|
||||
// Stub out resolveResponsesApiModel and its dependencies so we can test the
|
||||
// parse-counting in isolation without hitting the database.
|
||||
const originalFetch = globalThis.fetch;
|
||||
|
||||
const { withCodexPreferredModel } = await import("../../src/app/api/v1/responses/route.ts");
|
||||
const body = { model: "openai/gpt-4o", input: "hello" };
|
||||
const request = new Request("http://localhost/v1/responses", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
let cloneCount = 0;
|
||||
let jsonCount = 0;
|
||||
|
||||
const fakeBody = { model: "gpt-4o", messages: [] };
|
||||
|
||||
// Build a minimal fake request whose .clone() / .json() we can count
|
||||
function makeCountingRequest(body: object): Request {
|
||||
const req = new Request("http://localhost/v1/responses", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
|
||||
// Wrap clone so we count calls
|
||||
const origClone = req.clone.bind(req);
|
||||
Object.defineProperty(req, "clone", {
|
||||
value: () => {
|
||||
cloneCount++;
|
||||
return origClone();
|
||||
},
|
||||
writable: true,
|
||||
});
|
||||
|
||||
return req;
|
||||
}
|
||||
|
||||
// We import the route module to call withCodexPreferredModel.
|
||||
// Because the module-level DB / getModelInfo calls are side-effecting, we only
|
||||
// check that the function, when given a preParsedBody, returns early without cloning.
|
||||
// We test this by ensuring cloneCount === 0 after a call where the model is unknown
|
||||
// (resolveResponsesApiModel returns changed=false → early return).
|
||||
//
|
||||
// Simplest approach: inline-test the contract via the re-exported helper.
|
||||
|
||||
const req = makeCountingRequest(fakeBody);
|
||||
|
||||
// Simulate the behavior we want: if preParsedBody is supplied and the model field is
|
||||
// already resolved, clone() must not be called on the original request.
|
||||
// This is a white-box contract test — if the impl calls clone() when preParsedBody is
|
||||
// provided, cloneCount will be > 0 and the assertion fails.
|
||||
|
||||
// Before fix: withCodexPreferredModel always does `const clone = request.clone()`
|
||||
// After fix: it should use the pre-parsed body directly.
|
||||
|
||||
// We can test this without importing the whole route by checking that `resolveChatRequestBody`
|
||||
// (the terminal consumer) also does not re-parse when given a pre-parsed body — verifying
|
||||
// that the end-to-end threading avoids the extra parse.
|
||||
|
||||
const { resolveChatRequestBody } = await import("../../src/sse/handlers/requestBody.ts");
|
||||
|
||||
let innerJsonCalls = 0;
|
||||
const countingReq = {
|
||||
json: async () => {
|
||||
innerJsonCalls++;
|
||||
return fakeBody;
|
||||
const originalClone = request.clone.bind(request);
|
||||
Object.defineProperty(request, "clone", {
|
||||
value: () => {
|
||||
cloneCount += 1;
|
||||
return originalClone();
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const result = await resolveChatRequestBody(countingReq, fakeBody);
|
||||
assert.deepEqual(result, fakeBody);
|
||||
assert.equal(
|
||||
innerJsonCalls,
|
||||
0,
|
||||
"resolveChatRequestBody must not call request.json() when preParsedBody is provided"
|
||||
);
|
||||
const result = await withCodexPreferredModel(request, body);
|
||||
|
||||
assert.equal(cloneCount, 0);
|
||||
assert.equal(result.body, body);
|
||||
});
|
||||
|
||||
// ─── Part C: full integration — count .json() calls through withInjectionGuard ──
|
||||
// ─── Part C: wrapped routes parse once before invoking their handler ─────────
|
||||
|
||||
test("#4041 the body is parsed AT MOST ONCE through withInjectionGuard + inner handler", async () => {
|
||||
let jsonParseCount = 0;
|
||||
@@ -185,7 +133,7 @@ test("#4041 the body is parsed AT MOST ONCE through withInjectionGuard + inner h
|
||||
const spyRequest = wrapWithJsonSpy(origRequest);
|
||||
|
||||
let preParsedBodyReceived: unknown = undefined;
|
||||
const innerHandler = async (_req: any, _ctx: any, preParsedBody: unknown) => {
|
||||
const innerHandler = async (_req: Request, _ctx: unknown, preParsedBody: unknown) => {
|
||||
preParsedBodyReceived = preParsedBody;
|
||||
return new Response("ok");
|
||||
};
|
||||
|
||||
64
tests/unit/responses-route-early-keepalive-wiring.test.ts
Normal file
64
tests/unit/responses-route-early-keepalive-wiring.test.ts
Normal file
@@ -0,0 +1,64 @@
|
||||
import test, { after } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
const routeSource = fs.readFileSync("src/app/api/v1/responses/route.ts", "utf8");
|
||||
const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-responses-route-test-"));
|
||||
process.env.DATA_DIR = dataDir;
|
||||
process.env.REQUIRE_API_KEY = "false";
|
||||
after(() => fs.rmSync(dataDir, { recursive: true, force: true }));
|
||||
|
||||
test("Responses route wires dual-cadence neutral keepalives", () => {
|
||||
assert.match(
|
||||
routeSource,
|
||||
/startupFrame:\s*OPENAI_RESPONSES_IN_PROGRESS_FRAME/,
|
||||
"the immediate frame must not create a synthetic reasoning item"
|
||||
);
|
||||
assert.match(
|
||||
routeSource,
|
||||
/applicationKeepalive:\s*\{[\s\S]*frame:\s*OPENAI_RESPONSES_IN_PROGRESS_FRAME,[\s\S]*intervalMs:\s*SSE_HEARTBEAT_INTERVAL_MS/,
|
||||
"parser-visible events must use the configured mid-stream heartbeat cadence"
|
||||
);
|
||||
assert.doesNotMatch(
|
||||
routeSource,
|
||||
/keepaliveFrame:\s*OPENAI_RESPONSES_IN_PROGRESS_FRAME/,
|
||||
"frequent transport ticks must not all become application events"
|
||||
);
|
||||
assert.doesNotMatch(routeSource, /RESPONSES_STARTUP_THINKING_FRAME/);
|
||||
});
|
||||
|
||||
test("Responses route applies heavyweight admission through the SSE lifecycle", () => {
|
||||
assert.match(routeSource, /admitChatRequest\(request,/);
|
||||
assert.match(routeSource, /admitChatStructure\(parsedBody, admission\.lease,/);
|
||||
assert.match(routeSource, /releaseChatAdmissionAfterHandler\(/);
|
||||
assert.match(routeSource, /releaseChatAdmissionWhenDone\(/);
|
||||
});
|
||||
|
||||
test("Responses route rejects malformed JSON and releases raw-body admission", async () => {
|
||||
const [{ POST }, admission] = await Promise.all([
|
||||
import("../../src/app/api/v1/responses/route.ts"),
|
||||
import("../../src/shared/middleware/chatBodyAdmission.ts"),
|
||||
]);
|
||||
const malformed = `{"input":"${"x".repeat(admission.CHAT_LARGE_BODY_BYTES)}`;
|
||||
assert.equal(admission.perConnectionAdmissionController.activeHeavy, 0);
|
||||
|
||||
const response = await POST(
|
||||
new Request("http://localhost/v1/responses", {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
"content-length": String(Buffer.byteLength(malformed)),
|
||||
},
|
||||
body: malformed,
|
||||
})
|
||||
);
|
||||
|
||||
assert.equal(response.status, 400);
|
||||
assert.match(response.headers.get("content-type") || "", /application\/json/);
|
||||
const body = await response.text();
|
||||
assert.match(body, /invalid_request_error/);
|
||||
assert.doesNotMatch(body, /Body is unusable|stack|node:internal/);
|
||||
assert.equal(admission.perConnectionAdmissionController.activeHeavy, 0);
|
||||
});
|
||||
Reference in New Issue
Block a user