Skip to content

Commit 800ebc9

Browse files
Wibiaslidge-jun
andauthored
fix(bridge): CodeRabbit follow-ups for phase omit and content_filter (#556)
* fix(bridge): preserve omitted phases and batch content_filter * test(bridge): assert content filter skips compaction * test(bridge): harden heartbeat stall and content_filter compaction coverage Prove adapter heartbeats reset the effective 1s stall deadline via a test clock seam, require multiple wire heartbeats, and assert content_filter turns suppress compaction items. --------- Co-authored-by: bitkyc08-arch <bitkyc08@gmail.com>
1 parent df57bd9 commit 800ebc9

2 files changed

Lines changed: 199 additions & 33 deletions

File tree

src/bridge.ts

Lines changed: 43 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -133,8 +133,18 @@ export function bridgeToResponsesSSE(
133133
* from this callback instead of re-parsing the bridged SSE.
134134
*/
135135
onUsage?: (usage: OcxUsage | undefined) => void;
136+
/**
137+
* Test seam for the wire/stall beat loop. Production omits this and uses the
138+
* global timers; injecting here must not change scheduling semantics.
139+
*/
140+
timers?: {
141+
setInterval: (handler: () => void, ms: number) => unknown;
142+
clearInterval: (id: unknown) => void;
143+
};
136144
},
137145
): ReadableStream<Uint8Array> {
146+
const setBeatInterval = options?.timers?.setInterval ?? ((handler: () => void, ms: number) => setInterval(handler, ms));
147+
const clearBeatInterval = options?.timers?.clearInterval ?? ((id: unknown) => clearInterval(id as ReturnType<typeof setInterval>));
138148
// Freeform/custom tools (apply_patch) carry their body in `input`; the model is given a
139149
// function with `{input:string}`, so unwrap it here when relaying back as a custom_tool_call.
140150
const freeformInput = (args: string): string => {
@@ -194,7 +204,7 @@ export function bridgeToResponsesSSE(
194204
// buffering + raw-byte progress). Upstream activity only resets the stall watchdog.
195205
let upstreamActivity = false;
196206
let wireActivity = false;
197-
let beat: ReturnType<typeof setInterval> | undefined;
207+
let beat: unknown;
198208
let controller: ReadableStreamDefaultController<Uint8Array>;
199209
let emittedFrames = 0;
200210
let gated = false;
@@ -515,7 +525,11 @@ export function bridgeToResponsesSSE(
515525
if (currentRawReasoning) closeCurrentRawReasoning();
516526
flushHiddenRawReasoning();
517527
if (currentToolCall) closeCurrentToolCall();
518-
if (currentMsg && currentMsg.phase !== event.phase) closeCurrentMessage("commentary");
528+
// Only flush on an explicit phase change. A later delta that omits `phase` must
529+
// keep appending to the current message rather than wiping the earlier phase.
530+
if (currentMsg && event.phase !== undefined && currentMsg.phase !== event.phase) {
531+
closeCurrentMessage("commentary");
532+
}
519533
if (!currentMsg) {
520534
const itemId = `msg_${uuid()}`;
521535
const item = {
@@ -809,7 +823,7 @@ export function bridgeToResponsesSSE(
809823
stepping = false;
810824
return;
811825
}
812-
if (beat) { clearInterval(beat); beat = undefined; }
826+
if (beat !== undefined) { clearBeatInterval(beat); beat = undefined; }
813827

814828
if (!terminated) {
815829
// The adapter generator ended without an explicit done/error event. Mark as incomplete
@@ -848,7 +862,7 @@ export function bridgeToResponsesSSE(
848862
// The default ReadableStream strategy has HWM=1. Once one event's frames fill that
849863
// queue, pull stepping pauses; no custom FIFO or queuing strategy is layered on top.
850864
gated = true;
851-
beat = setInterval(() => {
865+
beat = setBeatInterval(() => {
852866
if (closed || gated) return;
853867
if (upstreamActivity) {
854868
upstreamActivity = false;
@@ -871,7 +885,7 @@ export function bridgeToResponsesSSE(
871885
terminated = true;
872886
returnIterator();
873887
emitDone();
874-
if (beat) clearInterval(beat);
888+
if (beat !== undefined) clearBeatInterval(beat);
875889
beat = undefined;
876890
try { controller.close(); } catch { /* already closed */ }
877891
closed = true;
@@ -902,14 +916,14 @@ export function bridgeToResponsesSSE(
902916
cancel() {
903917
// Client (Codex) disconnected. Stop emitting and let the caller abort the upstream fetch so a
904918
// cancelled turn does not leak the upstream stream or keep draining tokens (RC2).
905-
clientCancelled = true;
906-
closed = true;
907-
if (beat) clearInterval(beat);
908-
onCancel?.();
909-
returnIterator();
910-
},
911-
});
912-
}
919+
clientCancelled = true;
920+
closed = true;
921+
if (beat !== undefined) clearBeatInterval(beat);
922+
onCancel?.();
923+
returnIterator();
924+
},
925+
});
926+
}
913927

914928
export function buildResponseJSON(
915929
events: AdapterEvent[],
@@ -1048,15 +1062,17 @@ export function buildResponseJSON(
10481062
flushToolCall();
10491063
break;
10501064
case "text_delta":
1051-
if (currentText && currentTextPhase !== e.phase) flushText("commentary");
1065+
// Only flush on an explicit phase change. A later delta that omits `phase` must keep
1066+
// appending under the previously established phase.
1067+
if (currentText && e.phase !== undefined && currentTextPhase !== e.phase) flushText("commentary");
10521068
if (currentSummaryReasoning) flushSummaryReasoning();
10531069
if (currentRawReasoning) flushRawReasoning();
10541070
if (currentToolCallId) flushToolCall();
10551071
// Compaction turns keep the summary out of normal message output (replay dedup — see
10561072
// bridgeToResponsesSSE); it ships only inside the synthetic compaction item below.
10571073
if (options?.compaction) compactionText += e.text;
10581074
else {
1059-
currentTextPhase = e.phase;
1075+
if (e.phase !== undefined) currentTextPhase = e.phase;
10601076
currentText += e.text;
10611077
}
10621078
break;
@@ -1131,7 +1147,8 @@ export function buildResponseJSON(
11311147
endTurn = e.endTurn;
11321148
cleanDone = e.stopReason === undefined;
11331149
if (e.providerState) options?.onProviderState?.(e.providerState);
1134-
if (e.stopReason === "max_tokens") stopReason = "max_tokens";
1150+
// Match streaming: max_tokens and content_filter both terminate as incomplete.
1151+
if (e.stopReason === "max_tokens" || e.stopReason === "content_filter") stopReason = e.stopReason;
11351152
break;
11361153
}
11371154
}
@@ -1141,14 +1158,20 @@ export function buildResponseJSON(
11411158
flushToolCall();
11421159
// A truncated turn must never be installed as replacement history: emit the
11431160
// compaction item only when the turn actually completed (#422).
1144-
if (options?.compaction && !errorEvent && !incompleteEvent && stopReason !== "max_tokens") {
1161+
if (
1162+
options?.compaction
1163+
&& !errorEvent
1164+
&& !incompleteEvent
1165+
&& stopReason !== "max_tokens"
1166+
&& stopReason !== "content_filter"
1167+
) {
11451168
output.push({ type: "compaction", id: `cmp_${uuid()}`, encrypted_content: encodeCompactionSummary(compactionText) });
11461169
}
11471170

11481171
const failure = errorEvent ? adapterFailureFromEvent(errorEvent) : undefined;
11491172
const status = errorEvent
11501173
? "failed"
1151-
: incompleteEvent || stopReason === "max_tokens"
1174+
: incompleteEvent || stopReason === "max_tokens" || stopReason === "content_filter"
11521175
? "incomplete"
11531176
: "completed";
11541177
options?.onUsage?.(incompleteEvent?.usage ?? usage);
@@ -1168,6 +1191,8 @@ export function buildResponseJSON(
11681191
},
11691192
} : stopReason === "max_tokens" ? {
11701193
incomplete_details: { reason: "max_output_tokens" },
1194+
} : stopReason === "content_filter" ? {
1195+
incomplete_details: { reason: "content_filter" },
11711196
} : {}),
11721197
usage: responsesUsage(incompleteEvent?.usage ?? usage),
11731198
};

tests/bridge.test.ts

Lines changed: 156 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -529,6 +529,34 @@ describe("Responses bridge reasoning and usage parity", () => {
529529
expect((explicit.output as Record<string, unknown>[])[0]).toMatchObject({ phase: "commentary" });
530530
});
531531

532+
test("later text_delta omitting phase keeps the prior explicit phase", async () => {
533+
const frames = await collectSse(bridgeToResponsesSSE(replay([
534+
{ type: "text_delta", text: "Hello ", phase: "final_answer" },
535+
{ type: "text_delta", text: "world." },
536+
{ type: "done" },
537+
]), "routed/chat-model"));
538+
const completed = frames.find(frame => frame.event === "response.completed")?.data.response as Record<string, unknown>;
539+
const output = completed.output as Record<string, unknown>[];
540+
expect(output).toHaveLength(1);
541+
expect(output[0]).toMatchObject({
542+
type: "message",
543+
phase: "final_answer",
544+
content: [{ type: "output_text", text: "Hello world." }],
545+
});
546+
547+
const batch = buildResponseJSON([
548+
{ type: "text_delta", text: "Hello ", phase: "final_answer" },
549+
{ type: "text_delta", text: "world." },
550+
{ type: "done" },
551+
], "routed/chat-model");
552+
expect((batch.output as Record<string, unknown>[])).toHaveLength(1);
553+
expect((batch.output as Record<string, unknown>[])[0]).toMatchObject({
554+
type: "message",
555+
phase: "final_answer",
556+
content: [{ type: "output_text", text: "Hello world." }],
557+
});
558+
});
559+
532560
test("structured adapter errors override message heuristics", () => {
533561
const json = buildResponseJSON([
534562
{
@@ -684,44 +712,146 @@ describe("Responses bridge reasoning and usage parity", () => {
684712
// calls, the adapter emits `heartbeat` events. They must keep the stall watchdog alive (no
685713
// upstream_stall_timeout). Adapter heartbeats themselves are not translated into Responses
686714
// protocol items; wire keepalives use a separate `response.heartbeat` frame (see next test).
715+
//
716+
// resolveStallTimeoutSec ceils to a minimum of 1s, so sub-second stallTimeoutSec values cannot
717+
// prove the reset. Drive the beat loop through a test clock seam and run adapter-only progress
718+
// past the effective deadline.
719+
const heartbeatMs = 50;
720+
const stallTimeoutSec = 1; // effective after resolveStallTimeoutSec
721+
const maxStallTicks = Math.ceil((stallTimeoutSec * 1000) / heartbeatMs);
722+
const cycles = maxStallTicks + 5; // wall-clock equivalent >> stall deadline
723+
724+
let beatTick: (() => void) | undefined;
725+
const timers = {
726+
setInterval(handler: () => void, _ms: number) {
727+
beatTick = handler;
728+
return 1;
729+
},
730+
clearInterval(_id: unknown) {
731+
beatTick = undefined;
732+
},
733+
};
734+
735+
let waitResolve: (() => void) | undefined;
736+
const waitDelay = () => new Promise<void>(resolve => { waitResolve = resolve; });
737+
const releaseDelay = () => {
738+
const resolve = waitResolve;
739+
waitResolve = undefined;
740+
resolve?.();
741+
};
742+
const flush = async () => {
743+
for (let i = 0; i < 20; i++) await Promise.resolve();
744+
};
745+
687746
async function* heartbeatsThenDone(): AsyncGenerator<AdapterEvent> {
688-
// More heartbeats than maxStallTicks would allow if they did NOT reset the counter.
689-
for (let i = 0; i < 6; i++) {
747+
for (let i = 0; i < cycles; i++) {
690748
yield { type: "heartbeat" };
691-
await new Promise(r => setTimeout(r, 12));
749+
await waitDelay();
692750
}
693751
yield { type: "text_delta", text: "ok" };
694752
yield { type: "done" };
695753
}
696-
// heartbeatMs=10ms, stallTimeoutSec=0.03s -> maxStallTicks=3. 6 spaced heartbeats only survive
697-
// if each one resets stallTicks.
698-
const frames = await collectSse(bridgeToResponsesSSE(
699-
heartbeatsThenDone(), "model", undefined, undefined, undefined, undefined, 10, { stallTimeoutSec: 0.03 },
754+
755+
const framesPromise = collectSse(bridgeToResponsesSSE(
756+
heartbeatsThenDone(),
757+
"model",
758+
undefined,
759+
undefined,
760+
undefined,
761+
undefined,
762+
heartbeatMs,
763+
{ stallTimeoutSec, timers },
700764
));
701-
expect(frames.some(f => (f.data.response as Record<string, unknown> | undefined)?.incomplete_details)).toBe(false);
765+
766+
// First heartbeat is pulled; step blocks on the delay gate with gated=false.
767+
await flush();
768+
for (let i = 0; i < cycles; i++) {
769+
// maxStallTicks silent ticks would trip the watchdog if the preceding heartbeat did not
770+
// count as upstream activity (first tick clears the flag; the rest must not reach the limit).
771+
// Heartbeats between cycles reset the counter, so the stream must survive the full run.
772+
for (let t = 0; t < maxStallTicks; t++) beatTick?.();
773+
releaseDelay();
774+
await flush();
775+
}
776+
777+
const frames = await framesPromise;
778+
expect(frames.some(f => {
779+
const response = f.data.response as Record<string, unknown> | undefined;
780+
const details = response?.incomplete_details as Record<string, unknown> | undefined;
781+
return details?.reason === "upstream_stall_timeout";
782+
})).toBe(false);
702783
expect(frames.some(f => f.event === "response.completed")).toBe(true);
703-
// Adapter heartbeats must not be mis-translated into a rich protocol event of their own.
704-
expect(frames.some(f => f.event === "response.heartbeat" && f.data.type === "heartbeat" && Object.keys(f.data).length > 2)).toBe(false);
784+
// Adapter heartbeats must not be mis-translated into a protocol event of their own.
785+
expect(frames.some(f => f.data.type === "heartbeat")).toBe(false);
705786
});
706787

707788
test("wire response.heartbeat keeps firing while only adapter heartbeats flow", async () => {
708789
// Issue #521: web-search buffers semantic events and yields invisible adapter heartbeats from
709790
// raw-byte progress. Those must not suppress wire keepalives, or Codex Desktop idle-timeouts
710791
// (~5 min) while OCX still considers the upstream alive.
792+
const heartbeatMs = 50;
793+
const stallTimeoutSec = 1;
794+
const cycles = 4;
795+
796+
let beatTick: (() => void) | undefined;
797+
const timers = {
798+
setInterval(handler: () => void, _ms: number) {
799+
beatTick = handler;
800+
return 1;
801+
},
802+
clearInterval(_id: unknown) {
803+
beatTick = undefined;
804+
},
805+
};
806+
807+
let waitResolve: (() => void) | undefined;
808+
const waitDelay = () => new Promise<void>(resolve => { waitResolve = resolve; });
809+
const releaseDelay = () => {
810+
const resolve = waitResolve;
811+
waitResolve = undefined;
812+
resolve?.();
813+
};
814+
const flush = async () => {
815+
for (let i = 0; i < 20; i++) await Promise.resolve();
816+
};
817+
711818
async function* adapterHeartbeatsOnly(): AsyncGenerator<AdapterEvent> {
712-
for (let i = 0; i < 8; i++) {
819+
for (let i = 0; i < cycles; i++) {
713820
yield { type: "heartbeat" };
714-
await new Promise(r => setTimeout(r, 15));
821+
await waitDelay();
715822
}
716823
yield { type: "text_delta", text: "ok" };
717824
yield { type: "done" };
718825
}
719-
const frames = await collectSse(bridgeToResponsesSSE(
720-
adapterHeartbeatsOnly(), "model", undefined, undefined, undefined, undefined, 10, { stallTimeoutSec: 1 },
826+
827+
const framesPromise = collectSse(bridgeToResponsesSSE(
828+
adapterHeartbeatsOnly(),
829+
"model",
830+
undefined,
831+
undefined,
832+
undefined,
833+
undefined,
834+
heartbeatMs,
835+
{ stallTimeoutSec, timers },
721836
));
722-
expect(frames.some(f => f.event === "response.heartbeat" && f.data.type === "response.heartbeat")).toBe(true);
837+
838+
await flush();
839+
for (let i = 0; i < cycles; i++) {
840+
// Several silent beat ticks per adapter-only gap → multiple wire keepalives.
841+
for (let t = 0; t < 3; t++) beatTick?.();
842+
releaseDelay();
843+
await flush();
844+
}
845+
846+
const frames = await framesPromise;
847+
const wireHeartbeats = frames.filter(f =>
848+
f.event === "response.heartbeat" && f.data.type === "response.heartbeat"
849+
);
850+
expect(wireHeartbeats.length).toBeGreaterThan(1);
723851
expect(frames.some(f => f.event === "response.completed")).toBe(true);
724852
expect(frames.some(f => (f.data.response as Record<string, unknown> | undefined)?.incomplete_details)).toBe(false);
853+
// Reject every adapter-shaped heartbeat payload, regardless of event name or field count.
854+
expect(frames.some(f => f.data.type === "heartbeat")).toBe(false);
725855
});
726856
});
727857

@@ -872,6 +1002,17 @@ describe("Responses bridge stopReason threading (issue #246)", () => {
8721002
expect(json.incomplete_details).toEqual({ reason: "max_output_tokens" });
8731003
});
8741004

1005+
test("batch buildResponseJSON with stopReason content_filter returns incomplete status", () => {
1006+
const json = buildResponseJSON([
1007+
{ type: "text_delta", text: "partial" },
1008+
{ type: "done", stopReason: "content_filter" },
1009+
], "routed/model", { compaction: true });
1010+
expect(json.status).toBe("incomplete");
1011+
expect(json.incomplete_details).toEqual({ reason: "content_filter" });
1012+
// Truncated turns must not install a compaction replacement (#422).
1013+
expect((json.output as Record<string, unknown>[]).some(item => item.type === "compaction")).toBe(false);
1014+
});
1015+
8751016
test("batch buildResponseJSON without stopReason returns completed status", () => {
8761017
const json = buildResponseJSON([
8771018
{ type: "text_delta", text: "hello" },

0 commit comments

Comments
 (0)