pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-26 09:52:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

wip(sse): extract responsesCommentaryFilter from stream.ts (rescue; behavior já shipou via #6232)

Browse Source
This commit is contained in:
Diego Rodrigues de Sa e Souza
2026-07-05 18:20:59 -03:00
parent 1ef375e810
commit d0ff56613c
2 changed files with 72 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
open-sse/utils/responsesCommentaryFilter.ts Normal file
View File

@@ -0,0 +1,59 @@
/**
* Responses API passthrough — internal commentary-phase suppression (#6199).
*
* The Responses streaming passthrough must not forward internal commentary-phase
* output to clients. A commentary item is announced by `response.output_item.added`
* (which carries the `phase`), but its follow-up `response.output_text.delta` /
* `response.output_text.done` / `response.output_item.done` events only carry
* `item_id` / `output_index`. This helper tracks the announced commentary item id +
* output index in two caller-owned Sets and reports whether the current event should
* be dropped, so the hot per-chunk loop in stream.ts stays a single call.
*/
import { isResponsesCommentaryMessageItem } from "../handlers/responseSanitizer.ts";
type JsonRecord = Record<string, unknown>;
/**
* Decide whether a single Responses SSE event is internal commentary that must be
* dropped before forwarding. Mutates the two tracking Sets: an `output_item.added`
* for a commentary item is recorded (and dropped); its trailing events are dropped
* while tracked; the item's `output_item.done` clears the tracking.
*
* @returns true when the event should be dropped (the caller should `continue`).
*/
export function shouldDropResponsesCommentaryEvent(
parsed: JsonRecord,
commentaryItemIds: Set<string>,
commentaryIndexes: Set<number>
): boolean {
const eventType = parsed.type as string;
const eventOutputIndex = typeof parsed.output_index === "number" ? parsed.output_index : null;
const eventItem =
parsed.item && typeof parsed.item === "object" && !Array.isArray(parsed.item)
? (parsed.item as JsonRecord)
: null;
const eventItemId =
typeof parsed.item_id === "string"
? parsed.item_id
: eventItem && typeof eventItem.id === "string"
? eventItem.id
: null;
if (eventType === "response.output_item.added" && isResponsesCommentaryMessageItem(parsed.item)) {
if (eventItemId) commentaryItemIds.add(eventItemId);
if (eventOutputIndex !== null) commentaryIndexes.add(eventOutputIndex);
return true;
}
const belongsToCommentary =
(eventItemId !== null && commentaryItemIds.has(eventItemId)) ||
(eventOutputIndex !== null && commentaryIndexes.has(eventOutputIndex));
if (!belongsToCommentary) return false;
if (eventType === "response.output_item.done") {
if (eventItemId) commentaryItemIds.delete(eventItemId);
if (eventOutputIndex !== null) commentaryIndexes.delete(eventOutputIndex);
}
return true;
}

56
open-sse/utils/stream.ts
View File

@@ -30,8 +30,8 @@ import { STREAM_IDLE_TIMEOUT_MS, FETCH_BODY_TIMEOUT_MS, HTTP_STATUS } from "../c
import {
OMIT_STREAMING_CHUNK_MARKER,
sanitizeStreamingChunk,
isResponsesCommentaryMessageItem,
} from "../handlers/responseSanitizer.ts";
import { shouldDropResponsesCommentaryEvent } from "./responsesCommentaryFilter.ts";
import { isFeatureFlagEnabled } from "@/shared/utils/featureFlags";
import { buildErrorBody } from "./error.ts";
import { parseTextualToolCallCandidate, isValidToolCallHeaderPrefix } from "./textualToolCall.ts";
@@ -1308,48 +1308,18 @@ export function createSSEStream(options: StreamOptions = {}) {
parsed.type === "error");
if (isResponsesSSE) {
// #6199 — statefully drop internal commentary-phase output. The
// `response.output_item.added` announces the phase; the follow-up
// delta/done events only carry `item_id`/`output_index`, so we key
// off those. Happy-path (non-commentary) events are untouched.
if (shouldDropResponsesCommentary) {
const responsesEventType = parsed.type as string;
const eventOutputIndex =
typeof parsed.output_index === "number" ? parsed.output_index : null;
const eventItem =
parsed.item && typeof parsed.item === "object" && !Array.isArray(parsed.item)
? (parsed.item as JsonRecord)
: null;
const eventItemId =
typeof parsed.item_id === "string"
? parsed.item_id
: eventItem && typeof eventItem.id === "string"
? eventItem.id
: null;
if (
responsesEventType === "response.output_item.added" &&
isResponsesCommentaryMessageItem(parsed.item)
) {
if (eventItemId) passthroughResponsesCommentaryItemIds.add(eventItemId);
if (eventOutputIndex !== null)
passthroughResponsesCommentaryIndexes.add(eventOutputIndex);
continue;
}
const belongsToCommentary =
(eventItemId !== null &&
passthroughResponsesCommentaryItemIds.has(eventItemId)) ||
(eventOutputIndex !== null &&
passthroughResponsesCommentaryIndexes.has(eventOutputIndex));
if (belongsToCommentary) {
if (responsesEventType === "response.output_item.done") {
if (eventItemId) passthroughResponsesCommentaryItemIds.delete(eventItemId);
if (eventOutputIndex !== null)
passthroughResponsesCommentaryIndexes.delete(eventOutputIndex);
}
continue;
}
// #6199 — statefully drop internal commentary-phase output
// (logic extracted to responsesCommentaryFilter.ts). Happy-path
// (non-commentary) events are untouched.
if (
shouldDropResponsesCommentary &&
shouldDropResponsesCommentaryEvent(
parsed as JsonRecord,
passthroughResponsesCommentaryItemIds,
passthroughResponsesCommentaryIndexes
)
) {
continue;
}
const responsesIdsNormalized = normalizeResponsesSseIds(parsed as JsonRecord);