pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-03 22:02:08 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve streamed tool call arguments (#3762)

Browse Source 
Integrated into release/v3.8.24 — tool-call argument dedup made snapshot-only to prevent silent truncation (fix-in-place + tests).
This commit is contained in:
Artem
2026-06-13 14:34:23 +03:00
committed by GitHub
parent d1088f0dc6
commit 06e85a90c0
16 changed files with 748 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
docs/fixes/TOOL_CALL_INTEGRITY.md Normal file
View File

@@ -0,0 +1,49 @@
# Tool Call Integrity Fix
## Problem
In OmniRoute v3.8.21, during streaming tool calls, function arguments could become corrupted on the client side due to duplication or re-insertion of fragments. For example, `find` would turn into `fifnd`, and `grep` into `grreep`. The symptom appeared only on machine JSON fields of tool calls (`function.arguments` / `partial_json`) and was independent of the provider, because the corruption occurred within the shared OmniRoute SSE/translation pipeline after the upstream response.
## Root Cause
Tool-call argument chunks were being processed as regular human-readable text across several shared layers:
- `src/lib/sseTextTransform.ts` recursively passed string fields like `arguments` and `partial_json` to the text processor.
- `src/lib/streamingPiiTransform.ts` buffered these fields through a rolling-window PII sanitizer. This is unacceptable for machine JSON deltas: a chunk could be a delta, a snapshot, or an overlap-fragment, and the sanitizer does not understand the semantics of tool-call JSON.
- `open-sse/transformer/responsesTransformer.ts`, `open-sse/translator/response/openai-to-claude.ts`, `open-sse/translator/response/openai-responses.ts`, and `open-sse/handlers/sseParser.ts` accumulated arguments using a simple `+=`. As a result, a repeated snapshot or overlapping delta was added a second time.
Regular chat was not broken because text `content` deltas tolerate sanitization and buffering. Tool calls were broken because `arguments` is a machine JSON contract that must pass byte-preserving to the client.
## Fix
Explicit protection for tool-call JSON was added to the core source:
1. `src/lib/sseTextTransform.ts` skips `toolArgs` and `partialJson` without applying the text processor.
2. `src/lib/streamingPiiTransform.ts` returns `toolArgs` and `partialJson` as-is, bypassing rolling-window buffering.
3. Shared stream assemblers now use `appendToolCallArgumentDelta()` instead of blindly using `+=`, ensuring that repeated snapshots and overlapping chunks are added exactly once.
4. Responses/OpenAI/Claude translation paths emit only the new suffix of tool arguments to the client, rather than repeating the snapshot.
## How to Prevent Regression
- `tool_calls.function.name`, `tool_calls.function.arguments`, Responses `function_call.arguments`, and Claude `input_json_delta.partial_json` must never pass through text/PII/compression/dedup transforms.
- Any transform for SSE must distinguish between human text (`content`, `reasoning`) and machine JSON (`arguments`, `partial_json`).
- Regression tests are located in:
- `tests/unit/sseTextTransform.test.ts`
- `tests/unit/streamingPiiTransform.test.ts`
- `tests/unit/sse-parser.test.ts`
- `tests/unit/responses-transformer.test.ts`
- `tests/unit/translator-resp-openai-responses.test.ts`
- E2E smoke script: `tests/e2e-tool-calls.sh`.
## Configuration
For coding sessions, you can optionally disable risky text transforms:
```env
PII_RESPONSE_SANITIZATION=false
COMPRESSION_LEVEL=off
RTK_ENABLED=false
CAVEMAN_ENABLED=false
```
The core fix does not depend on these env variables: machine tool-call JSON is protected in the core pipeline and should not be modified even if PII response sanitization is enabled.

15
open-sse/handlers/sseParser.ts
View File

@@ -1,3 +1,4 @@
import { appendToolCallArgumentDelta } from "../utils/toolCallArguments.ts";
/**
* Convert OpenAI-style SSE chunks into a single non-streaming JSON response.
* Used as a fallback when upstream returns text/event-stream for stream=false.
@@ -181,7 +182,10 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
existing.function.name = tc.function.name;
}
existing.function = existing.function || {};
existing.function.arguments = `${existing.function.arguments || ""}${deltaArgs}`;
existing.function.arguments = appendToolCallArgumentDelta(
existing.function.arguments,
deltaArgs
);
accumulatedToolCalls.set(key, existing);
}
}
@@ -505,7 +509,10 @@ function ensureResponsesReasoningItem(outputItems, outputIndex, itemId) {
const next = {
...(existing && typeof existing === "object" ? existing : {}),
id: itemId || (existing?.id != null ? String(existing.id) : null) || `rs_${Date.now()}_${outputIndex}`,
id:
itemId ||
(existing?.id != null ? String(existing.id) : null) ||
`rs_${Date.now()}_${outputIndex}`,
type: "reasoning",
summary: Array.isArray(existing?.summary)
? existing.summary.map((summaryPart) => ({ ...toRecord(summaryPart) }))
@@ -539,9 +546,7 @@ function ensureResponsesFunctionCallItem(outputItems, outputIndex, itemId, callI
const next = {
...(existing && typeof existing === "object" ? existing : {}),
id:
normalizedItemId ||
existingId ||
`fc_${normalizedCallId || `${Date.now()}_${outputIndex}`}`,
normalizedItemId || existingId || `fc_${normalizedCallId || `${Date.now()}_${outputIndex}`}`,
type: "function_call",
call_id: normalizedCallId || existingCallId || "",
name: name || existing?.name || "",

11
open-sse/transformer/responsesTransformer.ts
View File

@@ -1,3 +1,4 @@
import { appendToolCallArgumentDelta } from "../utils/toolCallArguments.ts";
import * as fs from "fs";
import * as path from "path";
/**
@@ -527,15 +528,19 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
.replace(/"[a-zA-Z0-9_]+":\s*\[\s*\],?/g, "");
}
if (refCallId) {
const existingArgs = state.funcArgsBuf[tcIdx] || "";
const nextArgs = appendToolCallArgumentDelta(existingArgs, deltaStr);
const emittedDelta = nextArgs.slice(existingArgs.length);
state.funcArgsBuf[tcIdx] = nextArgs;
if (refCallId && emittedDelta) {
emit(controller, "response.function_call_arguments.delta", {
type: "response.function_call_arguments.delta",
item_id: `fc_${refCallId}`,
output_index: tcIdx,
delta: deltaStr,
delta: emittedDelta,
});
}
state.funcArgsBuf[tcIdx] += deltaStr;
}
}
}

11
open-sse/translator/response/openai-responses.ts
View File

@@ -4,6 +4,7 @@
*/
import { register } from "../registry.ts";
import { FORMATS } from "../formats.ts";
import { appendToolCallArgumentDelta } from "../../utils/toolCallArguments.ts";
function normalizeToolName(value) {
return typeof value === "string" ? value.trim() : "";
@@ -348,15 +349,19 @@ function emitToolCall(state, emit, tc) {
if (tc.function?.arguments) {
const refCallId = state.funcCallIds[tcIdx] || newCallId;
if (refCallId) {
const existingArgs = state.funcArgsBuf[tcIdx] || "";
const nextArgs = appendToolCallArgumentDelta(existingArgs, tc.function.arguments);
const emittedDelta = nextArgs.slice(existingArgs.length);
state.funcArgsBuf[tcIdx] = nextArgs;
if (refCallId && emittedDelta) {
emit("response.function_call_arguments.delta", {
type: "response.function_call_arguments.delta",
item_id: `fc_${refCallId}`,
output_index: tcIdx,
delta: tc.function.arguments,
delta: emittedDelta,
});
}
state.funcArgsBuf[tcIdx] += tc.function.arguments;
}
}

12
open-sse/translator/response/openai-to-claude.ts
View File

@@ -2,6 +2,7 @@ import { register } from "../registry.ts";
import { FORMATS } from "../formats.ts";
import { CLAUDE_OAUTH_TOOL_PREFIX } from "../request/openai-to-claude.ts";
import { hasToolCallShim, applyToolCallShimToBuffer } from "../helpers/toolCallShim.ts";
import { appendToolCallArgumentDelta } from "../../utils/toolCallArguments.ts";
// Helper: stop thinking block if started
function stopThinkingBlock(state, results) {
@@ -192,15 +193,16 @@ export function openaiToClaudeResponse(chunk, state) {
if (toolInfo) {
// Always buffer the raw stream so shimmed tools can re-emit a
// corrected JSON at stop time.
toolInfo.argBuffer = (toolInfo.argBuffer || "") + tc.function.arguments;
const existingArgs = toolInfo.argBuffer || "";
const nextArgs = appendToolCallArgumentDelta(existingArgs, tc.function.arguments);
let deltaStr = nextArgs.slice(existingArgs.length);
toolInfo.argBuffer = nextArgs;
if (toolInfo.shimmed) {
// Suppress passthrough; we emit one corrective delta at finish.
if (toolInfo.shimmed || !deltaStr) {
// Suppress passthrough for shimmed tools; emit one corrective delta at finish.
continue;
}
let deltaStr = tc.function.arguments;
// Fix #1852: Strip empty string and array placeholders from streaming tool arguments
if (deltaStr.includes('""') || deltaStr.includes("[]") || deltaStr.includes("[ ]")) {
deltaStr = deltaStr

32
open-sse/utils/toolCallArguments.ts Normal file
View File

@@ -0,0 +1,32 @@
/**
* Accumulate streamed tool-call `arguments` fragments without corrupting them.
*
* Providers stream tool-call arguments in one of two shapes:
* - Incremental deltas: each chunk carries only the NEW fragment. These must
* be concatenated verbatim — even when a fragment's leading bytes repeat the
* tail of what we already have (e.g. the doubled `l` in `ls -ll`).
* - Full snapshots: each chunk re-sends the ENTIRE accumulated arguments so
* far. Concatenating those would duplicate the payload (issue #3701).
*
* We only dedup the snapshot case when it is UNAMBIGUOUS: an identical repeat,
* or a growing superset that still starts with everything seen so far. Every
* other fragment is treated as an incremental delta and appended as-is.
*
* A fuzzy suffix/prefix-overlap heuristic must NOT be used here: it silently
* drops bytes from legitimate incremental deltas (turning `ll` into `l`, `xx`
* into `x`), which trades a visible duplication bug for a silent truncation bug.
*/
export function appendToolCallArgumentDelta(current: unknown, incoming: unknown): string {
const existing = typeof current === "string" ? current : "";
const next = typeof incoming === "string" ? incoming : "";
if (!existing) return next;
if (!next) return existing;
// Unambiguous snapshot repeat / growth — replace instead of concatenating.
if (next === existing) return existing;
if (next.startsWith(existing)) return next;
// Incremental delta fragment — append verbatim (preserves repeated chars).
return existing + next;
}

2
package-lock.json generated
View File

@@ -94,7 +94,7 @@
"@testing-library/react": "^16.3.2",
"@types/bcryptjs": "^3.0.0",
"@types/better-sqlite3": "^7.6.13",
"@types/bun": "latest",
"@types/bun": "*",
"@types/keytar": "^4.4.2",
"@types/node": "^25.9.1",
"@types/react": "^19.2.15",

91
src/lib/sseTextTransform.ts
View File

@@ -5,7 +5,7 @@ const CATEGORY_MAP: Record<string, FieldCategory> = {
thinking: "reasoning",
reasoning_content: "reasoning",
arguments: "toolArgs",
partial_json: "partialJson"
partial_json: "partialJson",
};
export function getFieldCategory(key: string): FieldCategory {
@@ -13,13 +13,22 @@ export function getFieldCategory(key: string): FieldCategory {
}
const STOP_EVENT_TYPES = new Set([
"response.done", "response.completed", "response.cancelled", "response.failed"
"response.done",
"response.completed",
"response.cancelled",
"response.failed",
]);
export function checkIfStopSignal(json: any): boolean {
if (!json || typeof json !== "object") return false;
if (json.choices && Array.isArray(json.choices) && json.choices.some((c: any) => c.finish_reason)) return true;
if (json.candidates && Array.isArray(json.candidates) && json.candidates.some((c: any) => c.finishReason)) return true;
if (json.choices && Array.isArray(json.choices) && json.choices.some((c: any) => c.finish_reason))
return true;
if (
json.candidates &&
Array.isArray(json.candidates) &&
json.candidates.some((c: any) => c.finishReason)
)
return true;
if (json.type === "content_block_stop") return true;
if (json.type === "message_stop") return true;
if (json.type === "message_delta" && json.delta?.stop_reason) return true;
@@ -39,9 +48,15 @@ export function checkIfSnapshot(json: any): boolean {
const fallbackDecoder = new TextDecoder();
export function createSseTextTransform(
processor: (text: string, field: FieldCategory, isStopSignal?: boolean, index?: string | number, isSnapshot?: boolean) => string,
processor: (
text: string,
field: FieldCategory,
isStopSignal?: boolean,
index?: string | number,
isSnapshot?: boolean
) => string,
onFlush?: (lastJson: any, isJsonStream?: boolean, lastContentJson?: any) => any,
onCancel?: () => void,
onCancel?: () => void
): TransformStream {
const encoder = new TextEncoder();
const decoder = new TextDecoder("utf-8");
@@ -80,7 +95,8 @@ export function createSseTextTransform(
const flushedValue = onFlush(lastJson, isJsonStream, lastContentJson);
if (flushedValue) {
const prefix = lastPrefix || "data: ";
const payload = typeof flushedValue === "string" ? flushedValue : JSON.stringify(flushedValue);
const payload =
typeof flushedValue === "string" ? flushedValue : JSON.stringify(flushedValue);
if (lastEventLine) {
controller.enqueue(encoder.encode(lastEventLine + "\n"));
}
@@ -101,18 +117,36 @@ export function createSseTextTransform(
try {
const json = JSON.parse(trimmedSegment);
isJsonStream = true;
let matched = false;
const isStopSignal = checkIfStopSignal(json);
const isSnapshot = checkIfSnapshot(json);
const METADATA_KEYS = [
"id", "model", "object", "created", "finish_reason", "finishReason",
"role", "type", "index", "stop_reason", "stop_sequence",
"system_fingerprint", "service_tier", "usage", "prompt_tokens",
"completion_tokens", "total_tokens", "input_tokens", "output_tokens",
"logprobs", "refusal", "name", "event"
"id",
"model",
"object",
"created",
"finish_reason",
"finishReason",
"role",
"type",
"index",
"stop_reason",
"stop_sequence",
"system_fingerprint",
"service_tier",
"usage",
"prompt_tokens",
"completion_tokens",
"total_tokens",
"input_tokens",
"output_tokens",
"logprobs",
"refusal",
"name",
"event",
];
// Recursively sanitize all string properties (except system metadata)
@@ -141,6 +175,11 @@ export function createSseTextTransform(
if (typeof obj[key] === "string") {
const val = obj[key];
const field: FieldCategory = getFieldCategory(key);
if (field === "toolArgs" || field === "partialJson") {
obj[key] = val;
matched = true;
continue;
}
obj[key] = processor(val, field, isStopSignal, compositeKey, isSnapshot);
matched = true;
} else if (typeof obj[key] === "object") {
@@ -152,16 +191,24 @@ export function createSseTextTransform(
sanitizeObject(json, 0, 0);
if (!matched) {
console.warn("[SSE-TRANSFORM] No string fields sanitized in SSE JSON chunk. Keys:", Object.keys(json).slice(0, 5).join(", "));
console.warn(
"[SSE-TRANSFORM] No string fields sanitized in SSE JSON chunk. Keys:",
Object.keys(json).slice(0, 5).join(", ")
);
} else {
lastContentJson = json;
}
if (isStopSignal && onFlush && !flushed) {
const flushedValue = onFlush(lastJson || json, isJsonStream, lastContentJson || lastJson || json); // Use json as fallback just in case
const flushedValue = onFlush(
lastJson || json,
isJsonStream,
lastContentJson || lastJson || json
); // Use json as fallback just in case
if (flushedValue) {
const prefix = lastPrefix || "data: ";
const payload = typeof flushedValue === "string" ? flushedValue : JSON.stringify(flushedValue);
const payload =
typeof flushedValue === "string" ? flushedValue : JSON.stringify(flushedValue);
// Only enqueue if the flushed value actually has content (onFlush usually returns null if buffer is empty now)
if (lastEventLine) {
controller.enqueue(encoder.encode(lastEventLine + "\n"));
@@ -188,7 +235,10 @@ export function createSseTextTransform(
if (err instanceof SyntaxError) {
// JSON parsing failed. Check if it looks like JSON that failed to parse.
if (trimmedSegment.startsWith("{") || trimmedSegment.startsWith("[")) {
console.warn("[SSE-TRANSFORM] Dropping malformed JSON chunk to prevent syntax injection:", trimmedSegment.slice(0, 100));
console.warn(
"[SSE-TRANSFORM] Dropping malformed JSON chunk to prevent syntax injection:",
trimmedSegment.slice(0, 100)
);
pendingEventLine = "";
} else {
if (pendingEventLine) {
@@ -274,7 +324,8 @@ export function createSseTextTransform(
const flushedValue = onFlush(lastJson, isJsonStream, lastContentJson);
if (flushedValue) {
const prefix = lastPrefix || "data: ";
const payload = typeof flushedValue === "string" ? flushedValue : JSON.stringify(flushedValue);
const payload =
typeof flushedValue === "string" ? flushedValue : JSON.stringify(flushedValue);
if (lastEventLine) {
controller.enqueue(encoder.encode(lastEventLine + "\n"));
}
@@ -290,6 +341,6 @@ export function createSseTextTransform(
if (onCancel) {
onCancel();
}
}
},
} as any);
}

99
src/lib/streamingPiiTransform.ts
View File

@@ -16,14 +16,17 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
content: "",
reasoning: "",
toolArgs: "",
partialJson: ""
partialJson: "",
};
choiceBuffers.set(index, buf);
}
return buf;
};
let windowSize = Math.max(200, options?.windowSize ?? (parseInt(process.env.PII_WINDOW_SIZE || "", 10) || 200));
let windowSize = Math.max(
200,
options?.windowSize ?? (parseInt(process.env.PII_WINDOW_SIZE || "", 10) || 200)
);
if (options?.windowSize !== undefined && process.env.PII_TEST_BYPASS_MIN_WINDOW === "true") {
windowSize = options.windowSize;
}
@@ -36,6 +39,9 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
index: string | number = "0_0",
isSnapshot = false
): string => {
if (field === "toolArgs" || field === "partialJson") {
return text;
}
if (isSnapshot) {
return sanitizePII(text).text;
}
@@ -43,12 +49,12 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
buffers[field] += text;
const { text: sanitized, endMatchIndex } = sanitizePII(buffers[field], !isStopSignal);
let emitLength = isStopSignal ? sanitized.length : Math.max(0, sanitized.length - W);
// Cap emitLength at the start of any PII that touched the end of the buffer
if (!isStopSignal && endMatchIndex !== undefined && emitLength > endMatchIndex) {
emitLength = endMatchIndex;
}
// Prevent slicing in the middle of a UTF-16 surrogate pair (e.g. emojis)
if (emitLength > 0 && emitLength < sanitized.length) {
const charCode = sanitized.charCodeAt(emitLength - 1);
@@ -57,7 +63,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
emitLength -= 1;
}
}
const toEmit = sanitized.slice(0, emitLength);
buffers[field] = sanitized.slice(emitLength);
return toEmit;
@@ -91,17 +97,17 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
if (buffers.content) {
const remaining = buffers.content;
buffers.content = "";
if (isJsonStream) {
// Wrap in a safe default OpenAI format to prevent client-side SDK crashes
return {
choices: [
{
delta: {
content: remaining
}
}
]
content: remaining,
},
},
],
};
} else {
return remaining;
@@ -112,15 +118,36 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
// Explicitly target formats to prevent metadata corruption and leakage
const METADATA_KEYS = [
"id", "model", "object", "created", "finish_reason", "finishReason",
"role", "type", "index", "stop_reason", "stop_sequence",
"system_fingerprint", "service_tier", "usage", "prompt_tokens",
"completion_tokens", "total_tokens", "input_tokens", "output_tokens",
"logprobs", "refusal", "name", "event"
"id",
"model",
"object",
"created",
"finish_reason",
"finishReason",
"role",
"type",
"index",
"stop_reason",
"stop_sequence",
"system_fingerprint",
"service_tier",
"usage",
"prompt_tokens",
"completion_tokens",
"total_tokens",
"input_tokens",
"output_tokens",
"logprobs",
"refusal",
"name",
"event",
];
// 1. Claude format
if (typeof lastJson.type === "string" && (lastJson.type.startsWith("message") || lastJson.type.startsWith("content_block"))) {
if (
typeof lastJson.type === "string" &&
(lastJson.type.startsWith("message") || lastJson.type.startsWith("content_block"))
) {
const buffers = getBuffers("0_0");
const delta: any = { type: "text_delta" };
let hasDelta = false;
@@ -143,7 +170,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
return {
type: "content_block_delta",
index: typeof lastJson.index === "number" ? lastJson.index : 0,
delta
delta,
};
}
return null;
@@ -152,10 +179,15 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
// 2. OpenAI Chat Completions
if (lastJson.choices && Array.isArray(lastJson.choices)) {
const finalJson = JSON.parse(JSON.stringify(lastJson));
const presentIndexes = new Set(finalJson.choices.map((c: any) => c.index).filter((idx: any) => typeof idx === "number"));
const presentIndexes = new Set(
finalJson.choices.map((c: any) => c.index).filter((idx: any) => typeof idx === "number")
);
for (const [compositeKey, choiceBuf] of choiceBuffers.entries()) {
const choiceIdx = parseInt(compositeKey.split("_")[0] || "0", 10);
if (!presentIndexes.has(choiceIdx) && (choiceBuf.content || choiceBuf.reasoning || choiceBuf.toolArgs)) {
if (
!presentIndexes.has(choiceIdx) &&
(choiceBuf.content || choiceBuf.reasoning || choiceBuf.toolArgs)
) {
finalJson.choices.push({ index: choiceIdx, delta: {} });
presentIndexes.add(choiceIdx);
}
@@ -163,15 +195,16 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
for (const choice of finalJson.choices) {
const choiceIdx = typeof choice.index === "number" ? choice.index : 0;
// Find if we have tool buffers for this choice
const toolEntries = Array.from(choiceBuffers.entries())
.filter(([key]) => key.startsWith(`${choiceIdx}_`) && key !== `${choiceIdx}_0`);
const toolEntries = Array.from(choiceBuffers.entries()).filter(
([key]) => key.startsWith(`${choiceIdx}_`) && key !== `${choiceIdx}_0`
);
const choiceBuf = getBuffers(`${choiceIdx}_0`);
if (!choice.delta) choice.delta = {};
const delta = choice.delta;
if (choiceBuf.content) {
delta.content = choiceBuf.content;
choiceBuf.content = "";
@@ -186,11 +219,11 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
}
if (choiceBuf.toolArgs || toolEntries.length > 0) {
if (!choice.delta.tool_calls) choice.delta.tool_calls = [];
if (choiceBuf.toolArgs) {
choice.delta.tool_calls.push({
index: 0,
function: { arguments: choiceBuf.toolArgs }
function: { arguments: choiceBuf.toolArgs },
});
choiceBuf.toolArgs = "";
}
@@ -200,7 +233,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
const toolIdx = parseInt(key.split("_")[1] || "0", 10);
choice.delta.tool_calls.push({
index: toolIdx,
function: { arguments: buf.toolArgs }
function: { arguments: buf.toolArgs },
});
buf.toolArgs = "";
}
@@ -223,7 +256,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
}
if (buffers.toolArgs) {
finalJson.item = {
arguments: buffers.toolArgs
arguments: buffers.toolArgs,
};
buffers.toolArgs = "";
}
@@ -238,7 +271,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
const buffers = getBuffers(`${idx}_0`);
if (!cand.content) cand.content = {};
cand.content.parts = [];
if (buffers.content) {
cand.content.parts.push({ text: buffers.content });
buffers.content = "";
@@ -267,10 +300,10 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
const populateRemaining = (obj: any, currentChoiceIdx = 0, currentToolIdx = 0) => {
if (!obj || typeof obj !== "object") return;
let choiceIdx = currentChoiceIdx;
let toolIdx = currentToolIdx;
if (typeof obj.index === "number") {
if (obj.delta || obj.message || obj.finish_reason) {
choiceIdx = obj.index;
@@ -280,7 +313,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
choiceIdx = obj.index;
}
}
const compositeKey = `${choiceIdx}_${toolIdx}`;
for (const key of Object.keys(obj)) {
@@ -301,7 +334,7 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
};
populateRemaining(finalJson, 0, 0);
// Clear all buffers
for (const buffers of choiceBuffers.values()) {
buffers.content = "";

142
tests/e2e-tool-calls.sh Executable file
View File

@@ -0,0 +1,142 @@
#!/usr/bin/env bash
set -Eeuo pipefail
OMNIROUTE_URL="${OMNIROUTE_URL:-http://localhost:20128}"
MODEL="${1:-${OMNIROUTE_MODEL:-}}"
AUTH_TOKEN="${OMNIROUTE_AUTH_TOKEN:-dummy}"
FAILURES=0
if [[ -z "$MODEL" ]]; then
echo "usage: $0 <model-id>" >&2
exit 2
fi
require_tool() {
if ! command -v "$1" >/dev/null 2>&1; then
echo "missing required tool: $1" >&2
exit 2
fi
}
check_corruption() {
local value="$1"
grep -qE 'fifnd|grreep|lls|cacat|f{2,}ind|g{2,}rep' <<<"$value"
}
require_tool curl
require_tool jq
echo "=== OmniRoute Tool Call Integrity E2E Test ==="
echo "URL: $OMNIROUTE_URL | Model: $MODEL"
echo
echo "[TEST 1] Non-streaming tool call integrity..."
response=$(curl -fsS -X POST "$OMNIROUTE_URL/v1/chat/completions" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $AUTH_TOKEN" \
-d @- <<JSON
{
"model": "$MODEL",
"stream": false,
"messages": [
{"role": "user", "content": "Use shell tool: find /tmp -name test.txt -type f"}
],
"tools": [
{
"type": "function",
"function": {
"name": "shell",
"description": "Execute shell command",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"]
}
}
}
],
"tool_choice": "auto"
}
JSON
)
args=$(jq -r '.choices[0].message.tool_calls[0].function.arguments // empty' <<<"$response")
command_value=$(jq -r '.command // empty' <<<"${args:-{}}" 2>/dev/null || true)
tool_name=$(jq -r '.choices[0].message.tool_calls[0].function.name // empty' <<<"$response")
if [[ -z "$args" ]]; then
echo " WARN: model did not return a tool call in non-streaming mode"
elif check_corruption "$command_value"; then
echo " FAIL: duplicated characters detected: $command_value"
FAILURES=$((FAILURES + 1))
else
echo " PASS: arguments intact: $command_value"
fi
if [[ -n "$tool_name" && "$tool_name" != "shell" ]]; then
echo " FAIL: function.name corrupted or unexpected: $tool_name"
FAILURES=$((FAILURES + 1))
else
echo " PASS: function.name: ${tool_name:-empty/no tool call}"
fi
echo
echo "[TEST 2] Streaming tool call integrity..."
stream_output=$(curl -fsS -X POST "$OMNIROUTE_URL/v1/chat/completions" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $AUTH_TOKEN" \
-d @- <<JSON
{
"model": "$MODEL",
"stream": true,
"messages": [
{"role": "user", "content": "Use shell tool: grep -r pattern /var"}
],
"tools": [
{
"type": "function",
"function": {
"name": "shell",
"description": "Execute shell command",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"]
}
}
}
],
"tool_choice": "auto"
}
JSON
)
assembled_args=$(awk '/^data: / { sub(/^data: /, ""); if ($0 != "[DONE]") print }' \
<<<"$stream_output" \
| jq -sr '[.[].choices[0].delta.tool_calls?.[0]?.function?.arguments // empty] | join("")' \
2>/dev/null || true)
if [[ -z "$assembled_args" || "$assembled_args" == '""' ]]; then
echo " WARN: no streaming tool-call arguments observed"
elif check_corruption "$assembled_args"; then
echo " FAIL: duplicated characters in stream: $assembled_args"
FAILURES=$((FAILURES + 1))
else
echo " PASS: streaming arguments OK: $assembled_args"
fi
echo
echo "[TEST 3] API health check..."
models=$(curl -fsS "$OMNIROUTE_URL/v1/models" \
-H "Authorization: Bearer $AUTH_TOKEN" \
| jq -r '.data | length' 2>/dev/null || echo 0)
if [[ "${models:-0}" -gt 0 ]]; then
echo " PASS: $models models available"
else
echo " WARN: no models listed"
fi
echo
echo "=== Results: $FAILURES failure(s) ==="
exit "$FAILURES"

50
tests/unit/responses-transformer.test.ts
View File

@@ -234,3 +234,53 @@ test("createResponsesLogger returns null for invalid base paths and swallows flu
assert.ok(capturedLogs.some((entry) => entry.includes("[RESPONSES] Failed to write logs:")));
});
test("createResponsesApiTransformStream deduplicates repeated tool argument snapshots", async () => {
const args = JSON.stringify({ command: "find /tmp -name test.txt" });
const output = await runTransformStream([
`data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"shell","arguments":${JSON.stringify(args)}}}]}}]}\n\n`,
`data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":${JSON.stringify(args)}}]},"finish_reason":"tool_calls"}]}\n\n`,
]);
const events = parseSseOutput(output);
const completed = JSON.parse(
events.find((event) => event.event === "response.completed").data
).response;
const toolCall = completed.output.find((item) => item.type === "function_call");
assert.equal(toolCall.arguments, args);
assert.equal(JSON.parse(toolCall.arguments).command, "find /tmp -name test.txt");
// The streamed deltas must also reconstruct the arguments exactly once — a
// duplicated snapshot must not be re-emitted to the client.
const streamedArgs = events
.filter((event) => event.event === "response.function_call_arguments.delta")
.map((event) => JSON.parse(event.data).delta)
.join("");
assert.equal(streamedArgs, args);
});
test("createResponsesApiTransformStream concatenates incremental tool argument fragments without dropping repeated chars", async () => {
// Real providers stream `function.arguments` as small incremental fragments.
// A doubled char straddling a fragment boundary ("l" + "l -l") must survive
// — the previous fuzzy-dedup heuristic silently turned `ll -l` into `l -l`.
const output = await runTransformStream([
`data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"shell","arguments":"{\\"cmd\\":\\"l"}}]}}]}\n\n`,
`data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"l -l\\"}"}}]},"finish_reason":"tool_calls"}]}\n\n`,
]);
const events = parseSseOutput(output);
const completed = JSON.parse(
events.find((event) => event.event === "response.completed").data
).response;
const toolCall = completed.output.find((item) => item.type === "function_call");
assert.equal(toolCall.arguments, '{"cmd":"ll -l"}');
assert.equal(JSON.parse(toolCall.arguments).cmd, "ll -l");
const streamedArgs = events
.filter((event) => event.event === "response.function_call_arguments.delta")
.map((event) => JSON.parse(event.data).delta)
.join("");
assert.equal(streamedArgs, '{"cmd":"ll -l"}');
});

43
tests/unit/sse-parser.test.ts
View File

@@ -248,3 +248,46 @@ test("parseSSEToResponsesOutput treats response.canceled as terminal and reconst
assert.equal(parsed.output[0].type, "message");
assert.equal(parsed.output[0].content[0].text, "Bye");
});
test("parseSSEToOpenAIResponse deduplicates repeated tool call snapshots", () => {
const args = JSON.stringify({ command: "find /tmp -name test.txt" });
const first = {
id: "chatcmpl_tool",
choices: [
{
index: 0,
delta: {
tool_calls: [
{
index: 0,
id: "call_1",
type: "function",
function: { name: "shell", arguments: args },
},
],
},
},
],
};
const second = {
id: "chatcmpl_tool",
choices: [
{
index: 0,
delta: { tool_calls: [{ index: 0, function: { arguments: args } }] },
finish_reason: "tool_calls",
},
],
};
const rawSSE = [
`data: ${JSON.stringify(first)}`,
`data: ${JSON.stringify(second)}`,
"data: [DONE]",
].join("\n");
const parsed = parseSSEToOpenAIResponse(rawSSE, "fallback-model");
const toolCall = parsed.choices[0].message.tool_calls[0];
assert.equal(toolCall.function.arguments, args);
assert.equal(JSON.parse(toolCall.function.arguments).command, "find /tmp -name test.txt");
});

91
tests/unit/sseTextTransform.test.ts
View File

@@ -32,7 +32,9 @@ test("processor receives delta.content from OpenAI CC format", async () => {
return text.toUpperCase();
});
const output = await testTransform(transform, [`data: {"choices":[{"delta":{"content":"hello"}}]}\n\n`]);
const output = await testTransform(transform, [
`data: {"choices":[{"delta":{"content":"hello"}}]}\n\n`,
]);
assert.equal(received.length, 1);
assert.equal(received[0], "hello");
@@ -51,16 +53,33 @@ test("processor receives 'content' field category for delta.content", async () =
assert.equal(fields[0], "content");
});
test("processor receives tool_calls function.arguments", async () => {
test("processor does not mutate tool_calls function.arguments", async () => {
const received: string[] = [];
const transform = createSseTextTransform((text, field) => {
received.push(text);
return text;
received.push(`${field}:${text}`);
return "MUTATED";
});
const payload = {
choices: [
{
delta: {
tool_calls: [
{
function: {
arguments: JSON.stringify({ command: "find /tmp -name test.txt" }),
},
},
],
},
},
],
};
await testTransform(transform, [`data: {"choices":[{"delta":{"tool_calls":[{"function":{"arguments":"{\\"email\\":\\"test@example.com\\"}"}}]}}]}\n\n`]);
const output = await testTransform(transform, [`data: ${JSON.stringify(payload)}\n\n`]);
assert.ok(received.some(t => t.includes("test@example.com")), "should extract tool call arguments");
assert.equal(received.length, 0, "tool call arguments must bypass text processors");
assert.ok(output.includes("find /tmp -name test.txt"), "arguments should pass through unchanged");
assert.ok(!output.includes("MUTATED"), "processor output must not replace tool JSON");
});
test("processor receives delta.reasoning_content with 'reasoning' category", async () => {
@@ -70,7 +89,9 @@ test("processor receives delta.reasoning_content with 'reasoning' category", asy
return text;
});
await testTransform(transform, [`data: {"choices":[{"delta":{"reasoning_content":"thinking..."}}]}\n\n`]);
await testTransform(transform, [
`data: {"choices":[{"delta":{"reasoning_content":"thinking..."}}]}\n\n`,
]);
assert.ok(fields.includes("reasoning"));
});
@@ -107,32 +128,34 @@ test("handles data: line split across two chunks", async () => {
return text;
});
await testTransform(transform, [
`data: {"choices":[{"del`,
`ta":{"content":"split"}}]}\n\n`
]);
await testTransform(transform, [`data: {"choices":[{"del`, `ta":{"content":"split"}}]}\n\n`]);
assert.equal(received.length, 1);
assert.equal(received[0], "split");
});
test("processor receives Claude delta.text with 'content' category", async () => {
const received: Array<{text: string, field: FieldCategory}> = [];
const received: Array<{ text: string; field: FieldCategory }> = [];
const transform = createSseTextTransform((text, field) => {
received.push({text, field});
received.push({ text, field });
return text;
});
await testTransform(transform, [`data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hello claude"}}\n\n`]);
await testTransform(transform, [
`data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hello claude"}}\n\n`,
]);
assert.ok(received.some(r => r.text === "hello claude" && r.field === "content"));
assert.ok(received.some((r) => r.text === "hello claude" && r.field === "content"));
});
test("onFlush callback invoked at stream close", async () => {
let flushCalled = false;
const transform = createSseTextTransform(
(text) => text,
() => { flushCalled = true; return ""; },
() => {
flushCalled = true;
return "";
}
);
await testTransform(transform, [`data: {"choices":[{"delta":{"content":"x"}}]}\n\n`]);
@@ -156,7 +179,7 @@ test("unknown JSON format passes through without processing (no processDeep)", a
});
const output = await testTransform(transform, [
`data: {"model":"gpt-4","id":"chatcmpl-123","object":"chat.completion.chunk"}\n\n`
`data: {"model":"gpt-4","id":"chatcmpl-123","object":"chat.completion.chunk"}\n\n`,
]);
assert.equal(received.length, 0, "processor should NOT be called for unrecognized format");
@@ -167,29 +190,32 @@ test("onFlush called exactly once when [DONE] is present", async () => {
let flushCount = 0;
const transform = createSseTextTransform(
(text) => text,
() => { flushCount++; return null; },
() => {
flushCount++;
return null;
}
);
await testTransform(transform, [
`data: {"choices":[{"delta":{"content":"hi"}}]}\n\n`,
`data: [DONE]\n\n`
`data: [DONE]\n\n`,
]);
assert.equal(flushCount, 1, "onFlush should be called exactly once");
});
test("Gemini candidates[0].content.parts[0].text is processed", async () => {
const received: Array<{text: string, field: FieldCategory}> = [];
const received: Array<{ text: string; field: FieldCategory }> = [];
const transform = createSseTextTransform((text, field) => {
received.push({text, field});
received.push({ text, field });
return text.toUpperCase();
});
const output = await testTransform(transform, [
`data: {"candidates":[{"content":{"parts":[{"text":"gemini response"}]}}]}\n\n`
`data: {"candidates":[{"content":{"parts":[{"text":"gemini response"}]}}]}\n\n`,
]);
assert.ok(received.some(r => r.text === "gemini response" && r.field === "content"));
assert.ok(received.some((r) => r.text === "gemini response" && r.field === "content"));
assert.ok(output.includes("GEMINI RESPONSE"));
});
@@ -201,7 +227,7 @@ test("Responses API string delta is processed", async () => {
});
await testTransform(transform, [
`data: {"type":"response.output_text.delta","delta":"hello responses"}\n\n`
`data: {"type":"response.output_text.delta","delta":"hello responses"}\n\n`,
]);
assert.ok(received.includes("hello responses"));
@@ -216,7 +242,7 @@ test("recursive scanning sanitizes multiple nested format fields (no format bypa
// This JSON has both `choices` (OpenAI) AND top-level `content` (Generic)
await testTransform(transform, [
`data: {"choices":[{"delta":{"content":"hi"}}],"content":"generic"}\n\n`
`data: {"choices":[{"delta":{"content":"hi"}}],"content":"generic"}\n\n`,
]);
// Should sanitize both fields recursively to prevent format-based bypasses
@@ -225,20 +251,17 @@ test("recursive scanning sanitizes multiple nested format fields (no format bypa
test("Responses API snapshot text is identified as snapshot and bypasses delta buffering", async () => {
let isSnapshotReceived = false;
const transform = createSseTextTransform(
(text, field, isStopSignal, index, isSnapshot) => {
if (isSnapshot) {
isSnapshotReceived = true;
}
return text.toUpperCase();
const transform = createSseTextTransform((text, field, isStopSignal, index, isSnapshot) => {
if (isSnapshot) {
isSnapshotReceived = true;
}
);
return text.toUpperCase();
});
const output = await testTransform(transform, [
`data: {"type":"response.output_text.done","text":"hello snapshot"}\n\n`
`data: {"type":"response.output_text.done","text":"hello snapshot"}\n\n`,
]);
assert.ok(isSnapshotReceived, "should identify done event as snapshot");
assert.ok(output.includes("HELLO SNAPSHOT"), "output should contain sanitized snapshot text");
});

146
tests/unit/streamingPiiTransform.test.ts
View File

@@ -49,11 +49,12 @@ test("createPiiSseTransform redacts email in delta.content", async () => {
const output = await testTransform(transform, [input]);
// Should NOT contain the raw email
assert.ok(!output.includes("john@example.com"),
"raw email should be redacted from output");
assert.ok(!output.includes("john@example.com"), "raw email should be redacted from output");
// Should contain some form of redaction marker
assert.ok(output.includes("REDACTED") || output.includes("[EMAIL"),
"output should contain redaction marker");
assert.ok(
output.includes("REDACTED") || output.includes("[EMAIL"),
"output should contain redaction marker"
);
});
test("createPiiSseTransform passes non-PII content through unchanged", async () => {
@@ -62,8 +63,10 @@ test("createPiiSseTransform passes non-PII content through unchanged", async ()
const input = `data: {"choices":[{"delta":{"content":"hello world no secrets here"}}]}\n\n`;
const output = await testTransform(transform, [input]);
assert.ok(output.includes("hello world no secrets here"),
"non-PII content should pass through unchanged");
assert.ok(
output.includes("hello world no secrets here"),
"non-PII content should pass through unchanged"
);
});
test("createPiiSseTransform redacts PII split across chunk boundaries", async () => {
@@ -74,10 +77,11 @@ test("createPiiSseTransform redacts PII split across chunk boundaries", async ()
const output = await testTransform(transform, [chunk1, chunk2]);
assert.ok(!output.includes("john@example.com"),
"email split across chunks should be redacted");
assert.ok(output.includes("REDACTED") || output.includes("[EMAIL"),
"redaction marker should be present in final stream");
assert.ok(!output.includes("john@example.com"), "email split across chunks should be redacted");
assert.ok(
output.includes("REDACTED") || output.includes("[EMAIL"),
"redaction marker should be present in final stream"
);
});
test("createPiiSseTransform flushes final redacted content before [DONE] sentinel", async () => {
@@ -106,14 +110,19 @@ test("createPiiSseTransform flushes final redacted content before [DONE] sentine
await writePromise;
const fullOutput = outputChunks.join("");
const lines = fullOutput.split("\n").map(l => l.trim()).filter(Boolean);
const lines = fullOutput
.split("\n")
.map((l) => l.trim())
.filter(Boolean);
const doneIndex = lines.findIndex(l => l === "data: [DONE]");
const doneIndex = lines.findIndex((l) => l === "data: [DONE]");
assert.ok(doneIndex !== -1, "[DONE] sentinel should be in the stream");
assert.equal(doneIndex, lines.length - 1, "nothing should be enqueued after the [DONE] sentinel");
const redactedLine = lines.find((l, idx) => idx < doneIndex && (l.includes("REDACTED") || l.includes("[EMAIL")));
const redactedLine = lines.find(
(l, idx) => idx < doneIndex && (l.includes("REDACTED") || l.includes("[EMAIL"))
);
assert.ok(redactedLine, "redacted content chunk should be enqueued before the [DONE] sentinel");
});
@@ -126,8 +135,10 @@ test("content flushed when last chunk is metadata-only (no delta.content)", asyn
const output = await testTransform(transform, [chunk1, chunk2, chunk3]);
assert.ok(output.includes("hello world"),
"buffered content must be flushed even when last chunk has no delta.content");
assert.ok(
output.includes("hello world"),
"buffered content must be flushed even when last chunk has no delta.content"
);
});
test("no duplicate content when stream has [DONE] and normal close", async () => {
@@ -179,8 +190,10 @@ test("PII split across sliding window boundary is still redacted", async () => {
const done = `data: [DONE]\n\n`;
const output = await testTransform(transform, [chunk1, chunk2, done]);
assert.ok(!output.includes("user@example.com"),
"email spanning window boundary should be redacted");
assert.ok(
!output.includes("user@example.com"),
"email spanning window boundary should be redacted"
);
});
test("preserve event names when flushing buffered SSE text", async () => {
@@ -207,7 +220,11 @@ test("do not leak custom event name to subsequent default message events on flus
const output = await testTransform(transform, [eventLine + inputLine1 + inputLine2 + doneLine]);
const occurrences = (output.match(/event: response.output_text.delta/g) || []).length;
assert.strictEqual(occurrences, 1, "custom event name should only appear once and not leak to the flushed chunk of the default message");
assert.strictEqual(
occurrences,
1,
"custom event name should only appear once and not leak to the flushed chunk of the default message"
);
});
test("insert an SSE event separator before flushed chunks", async () => {
@@ -235,29 +252,41 @@ test("reset event line on empty line message boundary", async () => {
// The defaultLine is preceded by a blank line (\n\n), so it is a separate event.
// The event name "response.output_text.delta" must NOT leak into the second event.
const parts = output.split("\n\n");
// parts[0] should have the custom event name
assert.ok(parts[0].includes("event: response.output_text.delta"), "first block should have custom event name");
assert.ok(
parts[0].includes("event: response.output_text.delta"),
"first block should have custom event name"
);
// parts[1] should NOT have the custom event name
assert.ok(!parts[1].includes("event: response.output_text.delta"), "second block should reset event name and not leak it");
assert.ok(
!parts[1].includes("event: response.output_text.delta"),
"second block should reset event name and not leak it"
);
});
test("sanitize compressed IPv6 addresses", async () => {
const transform = createPiiSseTransform();
const inputLoopback = `data: {"choices":[{"delta":{"content":"server address is ::1"}}]}\n\n`;
const inputCompressed = `data: {"choices":[{"delta":{"content":"server address is 2001:db8::1"}}]}\n\n`;
const doneLine = `data: [DONE]\n\n`;
const outputLoopback = await testTransform(transform, [inputLoopback + doneLine]);
assert.ok(!outputLoopback.includes("::1"), "compressed loopback IPv6 should be redacted");
assert.ok(outputLoopback.includes("[IP_REDACTED]"), "redaction marker should be present for loopback IPv6");
assert.ok(
outputLoopback.includes("[IP_REDACTED]"),
"redaction marker should be present for loopback IPv6"
);
const transform2 = createPiiSseTransform();
const outputCompressed = await testTransform(transform2, [inputCompressed + doneLine]);
assert.ok(!outputCompressed.includes("2001:db8::1"), "compressed IPv6 should be redacted");
assert.ok(outputCompressed.includes("[IP_REDACTED]"), "redaction marker should be present for compressed IPv6");
assert.ok(
outputCompressed.includes("[IP_REDACTED]"),
"redaction marker should be present for compressed IPv6"
);
});
test("no event: prefix in flushed chunk when no event line was seen", async () => {
@@ -271,7 +300,10 @@ test("no event: prefix in flushed chunk when no event line was seen", async () =
const output = await testTransform(transform, [inputLine + doneLine]);
// The flushed chunk (last 10 chars "klmnopqrst") must NOT be preceded by any "event:" line.
assert.ok(!output.includes("event:"), "no event: prefix should appear when no event line was seen");
assert.ok(
!output.includes("event:"),
"no event: prefix should appear when no event line was seen"
);
});
test("event name preserved when stream closes without [DONE] sentinel", async () => {
@@ -303,7 +335,10 @@ test("event: line without trailing space is tracked as currentEventLine", async
const output = await testTransform(transform, [eventLine + inputLine + doneLine]);
assert.ok(output.includes("event:custom.event"), "no-space event: form should be tracked and prepended on flush");
assert.ok(
output.includes("event:custom.event"),
"no-space event: form should be tracked and prepended on flush"
);
});
test("lastEventLine is not updated when processing a stop-signal chunk", async () => {
@@ -402,7 +437,10 @@ test("verify event line flushed before other non-data lines (e.g. id, retry)", a
const inputLines = "event: foo\nid: 123\ndata: bar\n\n";
const output = await testTransform(transform, [inputLines]);
assert.ok(output.includes("event: foo\nid: 123\ndata: bar"), "event line must be flushed before non-data lines like id");
assert.ok(
output.includes("event: foo\nid: 123\ndata: bar"),
"event line must be flushed before non-data lines like id"
);
});
test("verify trailing event line is flushed on stream close", async () => {
@@ -411,7 +449,10 @@ test("verify trailing event line is flushed on stream close", async () => {
const inputLines = "event: some-trailing-event\n";
const output = await testTransform(transform, [inputLines]);
assert.ok(output.includes("event: some-trailing-event"), "trailing event line should be flushed on stream close");
assert.ok(
output.includes("event: some-trailing-event"),
"trailing event line should be flushed on stream close"
);
});
test("verify consecutive event lines without intervening data are both preserved", async () => {
@@ -435,3 +476,48 @@ test.after(async () => {
coreDb.resetDbInstance();
fs.rmSync(tmpDir, { recursive: true, force: true });
});
test("createPiiSseTransform preserves tool call arguments without buffering", async () => {
const transform = (createPiiSseTransform as any)({ windowSize: 10 });
const payload = {
choices: [
{
delta: {
tool_calls: [
{
index: 0,
function: {
arguments: JSON.stringify({ command: "find /tmp -name test.txt" }),
},
},
],
},
},
],
};
const input = `data: ${JSON.stringify(payload)}\n\n`;
const done = `data: [DONE]\n\n`;
const output = await testTransform(transform, [input, done]);
assert.ok(output.includes("find /tmp -name test.txt"));
assert.ok(!output.includes("REDACTED"));
});
test("createPiiSseTransform preserves Claude partial_json without buffering", async () => {
const transform = (createPiiSseTransform as any)({ windowSize: 10 });
const payload = {
type: "content_block_delta",
index: 1,
delta: {
type: "input_json_delta",
partial_json: JSON.stringify({ command: "grep -r pattern /var" }),
},
};
const input = `event: content_block_delta\ndata: ${JSON.stringify(payload)}\n\n`;
const done = `data: [DONE]\n\n`;
const output = await testTransform(transform, [input, done]);
assert.ok(output.includes("grep -r pattern /var"));
assert.ok(!output.includes("REDACTED"));
});

46
tests/unit/tool-call-arguments.test.ts Normal file
View File

@@ -0,0 +1,46 @@
import test from "node:test";
import assert from "node:assert/strict";
import { appendToolCallArgumentDelta } from "../../open-sse/utils/toolCallArguments.ts";
/**
* Helper: reduce a list of incremental delta fragments the way the streaming
* accumulators do (sseParser / responsesTransformer / translators).
*/
function accumulate(fragments: string[]): string {
return fragments.reduce<string>((acc, frag) => appendToolCallArgumentDelta(acc, frag), "");
}
test("appendToolCallArgumentDelta concatenates incremental fragments verbatim", () => {
assert.equal(accumulate(['{"q":"hel', 'lo"}']), '{"q":"hello"}');
});
test("appendToolCallArgumentDelta preserves repeated characters across fragment boundaries (#3701 anti-truncation)", () => {
// The previous fuzzy-dedup heuristic dropped a byte here, turning the buffer's
// trailing "x" + the next "x" into a single "x". Repeated chars must survive.
assert.equal(accumulate(['{"a":"', "x", "x", '"}']), '{"a":"xx"}');
// A shell command with a doubled letter must not be silently truncated
// (`ls -ll` → `ls -l` was the real-world corruption).
assert.equal(accumulate(['{"cmd":"l', 'l -l"}']), '{"cmd":"ll -l"}');
// A doubled letter that straddles a single-char fragment boundary survives.
assert.equal(accumulate(['{"path":"/a/bb', 'b/c"}']), '{"path":"/a/bbb/c"}');
});
test("appendToolCallArgumentDelta dedups an identical full-snapshot repeat", () => {
const args = JSON.stringify({ command: "find /tmp -name test.txt" });
assert.equal(appendToolCallArgumentDelta(args, args), args);
});
test("appendToolCallArgumentDelta replaces a growing full snapshot instead of concatenating", () => {
assert.equal(appendToolCallArgumentDelta('{"a"', '{"a":1}'), '{"a":1}');
assert.equal(appendToolCallArgumentDelta('{"command":"ec', '{"command":"echo hi"}'), '{"command":"echo hi"}');
});
test("appendToolCallArgumentDelta handles empty / non-string inputs", () => {
assert.equal(appendToolCallArgumentDelta("", "x"), "x");
assert.equal(appendToolCallArgumentDelta("x", ""), "x");
assert.equal(appendToolCallArgumentDelta(undefined, "x"), "x");
assert.equal(appendToolCallArgumentDelta("x", undefined), "x");
assert.equal(appendToolCallArgumentDelta(null, null), "");
assert.equal(appendToolCallArgumentDelta(42, "x"), "x");
});

42
tests/unit/translator-resp-openai-responses.test.ts
View File

@@ -446,3 +446,45 @@ test("Responses -> OpenAI: response.failed records upstream error", () => {
assert.equal(state.upstreamError.code, "rate_limit_exceeded");
assert.match(state.upstreamError.message, /Rate limit reached/);
});
test("OpenAI -> Responses: deduplicates repeated tool argument snapshots", () => {
const args = JSON.stringify({ command: "grep -r pattern /var" });
const events = collectEvents([
{
id: "chatcmpl-tool-snapshot",
model: "gpt-4.1",
choices: [
{
index: 0,
delta: {
tool_calls: [
{
index: 0,
id: "call_1",
type: "function",
function: { name: "shell", arguments: args },
},
],
},
finish_reason: null,
},
],
},
{
id: "chatcmpl-tool-snapshot",
model: "gpt-4.1",
choices: [
{
index: 0,
delta: { tool_calls: [{ index: 0, function: { arguments: args } }] },
finish_reason: "tool_calls",
},
],
},
]);
const done = events.find((event) => event.event === "response.function_call_arguments.done");
assert.equal(done.data.arguments, args);
assert.equal(JSON.parse(done.data.arguments).command, "grep -r pattern /var");
});