Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion open-sse/translator/helpers/openaiHelper.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,16 @@ const CLAUDE_TOOL_CHOICE_REQUIRED = "an" + "y";
// Filter messages to OpenAI standard format
// Remove: redacted_thinking, and other non-OpenAI blocks
// Convert: thinking blocks → reasoning_content on the message
export function filterToOpenAIFormat(body, opts = {}) {
export interface FilterToOpenAIFormatOptions {
/** Keep `cache_control` on content blocks (providers that honor OpenAI-format breakpoints). */
preserveCacheControl?: boolean;
/** Keep Moonshot's non-standard `video_url` content block. */
preserveVideoUrl?: boolean;
/** Keep `reasoning_content` on tool-call assistant turns (reasoning-replay providers). */
preserveReasoningContent?: boolean;
}

export function filterToOpenAIFormat(body, opts: FilterToOpenAIFormatOptions = {}) {
// #2069 — when the routed provider honors OpenAI-format cache_control
// breakpoints (DashScope/alibaba, Xiaomi MiMo, etc.) and preservation was
// requested upstream, keep the `cache_control` field on each content block
Expand Down
9 changes: 7 additions & 2 deletions open-sse/translator/helpers/toolCallShim.ts
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,13 @@ function sanitizeReadArgs(args: Record<string, unknown>): void {
}

if (typeof args.limit === "number") {
if (args.limit > READ_MAX_LIMIT) args.limit = READ_MAX_LIMIT;
if (args.limit < 1) delete args.limit;
// Read into a local: assigning back to `args.limit` (declared `unknown`) resets the
// `typeof` narrowing, so the second comparison would no longer see a number. The two
// branches are mutually exclusive (READ_MAX_LIMIT is 2000), so testing the original
// value keeps the behavior identical.
const limit = args.limit;
if (limit > READ_MAX_LIMIT) args.limit = READ_MAX_LIMIT;
if (limit < 1) delete args.limit;
}
if (typeof args.offset === "number" && args.offset < 0) args.offset = 0;

Expand Down
7 changes: 5 additions & 2 deletions open-sse/translator/request/openai-responses.ts
Original file line number Diff line number Diff line change
Expand Up @@ -633,10 +633,13 @@ export function openaiResponsesToOpenAIRequest(
);
}

result.tools = chatTools.filter((toolValue) =>
// Keep the filtered array in a local: `result` is a Record<string, unknown>, so
// reading `result.tools` back gives `unknown` and `.length` does not type-check.
const allowedTools = chatTools.filter((toolValue) =>
allowedNames.has(toString(toRecord(toRecord(toolValue).function).name))
);
if (result.tools.length === 0) {
result.tools = allowedTools;
if (allowedTools.length === 0) {
throw unsupportedFeature(
"Unsupported Responses API feature: allowed_tools resolved to zero Chat Completions function tools"
);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,5 +7,7 @@ import { sanitizeToolId } from "../../helpers/schemaCoercion.ts";
// that guard and silently fabricate a tool_result that can never match a tool_use.
export function sanitizeToolResultId(rawId: unknown): string | null {
if (!rawId) return null;
return sanitizeToolId(rawId);
// sanitizeToolId() takes a string; a non-string id would previously reach `.replace()`
// and throw. Coerce instead so a numeric id (some clients send one) sanitizes normally.
return sanitizeToolId(typeof rawId === "string" ? rawId : String(rawId));
}
21 changes: 21 additions & 0 deletions open-sse/types.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -137,3 +137,24 @@ export interface UsageData {
completion_tokens: number;
total_tokens: number;
}

// ============ Lib gap: Transformer.cancel ============

declare global {
/**
* The WHATWG Streams standard defines `transformer.cancel(reason)`, invoked when
* the readable side is cancelled (for us: an SSE client disconnecting). Node
* implements it — verified on v24 — but `lib.dom.d.ts` still omits it from
* `Transformer`, so every `new TransformStream({ ..., cancel() {} })` in the
* codebase fails with TS2353 ("'cancel' does not exist in type 'Transformer'").
*
* These `cancel` handlers are load-bearing: they clear heartbeat/progress
* intervals and idle timers on disconnect. Deleting them to satisfy the checker
* would leak a timer per abandoned stream, so the type is patched instead.
*
* Remove once the bundled lib declares it.
*/
interface Transformer<I = unknown, O = unknown> {
cancel?: (reason?: unknown) => void | PromiseLike<void>;
}
}
6 changes: 3 additions & 3 deletions open-sse/utils/cursorAgentProtobuf.ts
Original file line number Diff line number Diff line change
Expand Up @@ -718,7 +718,7 @@ export function decodeKvServerEvent(payload: Buffer): KvServerEvent | null {

if (getBlobArgs) {
// GetBlobArgs { blob_id (1): bytes }
let blobId = Buffer.alloc(0);
let blobId: Buffer = Buffer.alloc(0);
for (const f of decodeFields(getBlobArgs)) {
if (f.fieldNumber === GBA_BLOB_ID && f.wireType === 2) {
blobId = f.bytes;
Expand All @@ -728,8 +728,8 @@ export function decodeKvServerEvent(payload: Buffer): KvServerEvent | null {
}
if (setBlobArgs) {
// SetBlobArgs { blob_id (1): bytes, blob_data (2): bytes }
let blobId = Buffer.alloc(0);
let blobData = Buffer.alloc(0);
let blobId: Buffer = Buffer.alloc(0);
let blobData: Buffer = Buffer.alloc(0);
for (const f of decodeFields(setBlobArgs)) {
if (f.fieldNumber === SBA_BLOB_ID && f.wireType === 2) {
blobId = f.bytes;
Expand Down
23 changes: 16 additions & 7 deletions open-sse/utils/earlyStreamKeepalive.ts
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,15 @@ export type EarlyStreamKeepaliveOptions = {
errorFrame?: Uint8Array;
};

type SettledHandler = { ok: true; response: Response } | { ok: false; error: unknown };
/**
* Tagged with a string rather than an `ok: true | false` boolean: this workspace compiles
* with `strictNullChecks: false`, where a boolean-literal discriminant narrows the positive
* branch but not the negative one — so reading `.error` off the rejected arm did not
* type-check. A string discriminant narrows both branches under the same settings.
*/
type SettledHandler =
| { status: "fulfilled"; response: Response }
| { status: "rejected"; error: unknown };

export async function withEarlyStreamKeepalive(
handlerPromise: Promise<Response>,
Expand All @@ -209,8 +217,8 @@ export async function withEarlyStreamKeepalive(
// Settle into a tagged result so neither race branch leaves an unhandled
// rejection when the threshold timer wins.
const settled: Promise<SettledHandler> = handlerPromise.then(
(response) => ({ ok: true as const, response }),
(error) => ({ ok: false as const, error })
(response) => ({ status: "fulfilled" as const, response }),
(error) => ({ status: "rejected" as const, error })
);

let timer: ReturnType<typeof setTimeout> | undefined;
Expand All @@ -224,8 +232,9 @@ export async function withEarlyStreamKeepalive(

if (raced.kind === "settled") {
// Fast path — return verbatim, or rethrow so the route's normal error handling runs.
if (raced.result.ok) return raced.result.response;
throw raced.result.error;
const result = raced.result;
if (result.status === "fulfilled") return result.response;
throw result.error;
}

// Slow path — open the SSE stream now and keep it warm until the handler resolves.
Expand Down Expand Up @@ -287,13 +296,13 @@ export async function withEarlyStreamKeepalive(
if (aborted) {
// The synthetic keepalive response can be cancelled before the handler resolves.
// Cancel the eventual real response so its upstream work and lifecycle hooks finish.
if (result.ok && result.response.body) {
if (result.status === "fulfilled" && result.response.body) {
await result.response.body.cancel().catch(() => undefined);
}
return;
}

if (!result.ok) {
if (result.status === "rejected") {
// Handler rejected — emit a generic error frame (never the raw error/stack).
controller.enqueue(errorFrame);
} else {
Expand Down
151 changes: 151 additions & 0 deletions tests/unit/ts7-open-sse-type-fixes.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
/**
* Behavioral guards for the TS7-readiness type fixes in `open-sse/utils` and
* `open-sse/translator` (slice 1 of the TypeScript 7 migration).
*
* Most of that change is behavior-preserving refactoring, already covered by the
* existing keepalive/heartbeat suites. Three things are NOT covered elsewhere and are
* exactly the parts a future "just make the checker happy" edit would silently break:
*
* 1. `transformer.cancel()` — the WHATWG Streams hook that clears heartbeat/progress
* intervals when an SSE client disconnects. `lib.dom.d.ts` omits it from
* `Transformer`, so it is patched in `open-sse/types.d.ts`. If someone deletes the
* handlers instead of the type patch, every abandoned stream leaks a timer.
* 2. `sanitizeToolResultId()` — now coerces a non-string id instead of throwing.
* 3. The `Read` tool-call shim's `limit` clamping — the narrowing fix rewrote the
* comparison to read a local, which must stay behavior-identical at the bounds.
*/
import test from "node:test";
import assert from "node:assert/strict";

import { sanitizeToolResultId } from "../../open-sse/translator/request/openai-to-claude/sanitizeToolResultId.ts";
import { applyToolCallShimToBuffer } from "../../open-sse/translator/helpers/toolCallShim.ts";

// ---------------------------------------------------------------------------
// 1. transformer.cancel() runtime contract
// ---------------------------------------------------------------------------

test("TransformStream invokes transformer.cancel() when the readable side is cancelled", async () => {
let cancelled = false;
let seenReason: unknown;

const ts = new TransformStream({
transform(chunk, controller) {
controller.enqueue(chunk);
},
cancel(reason) {
cancelled = true;
seenReason = reason;
},
});

const writer = ts.writable.getWriter();
const reader = ts.readable.getReader();
void writer.write("chunk");
await reader.read();

const reason = new Error("client disconnect");
await reader.cancel(reason);

assert.equal(
cancelled,
true,
"transformer.cancel() must fire on readable cancel — the heartbeat/progress " +
"interval cleanup in sseHeartbeat.ts and progressTracker.ts depends on it"
);
assert.equal(seenReason, reason, "cancel() should receive the cancellation reason");
});

test("a transformer cancel handler can clear an interval (the leak this guards)", async (t) => {
let ticks = 0;
let stopped = false;
// Held in a local, not on the stream: `start()` runs inside the TransformStream
// constructor, before the `const` binding is initialized.
let stop: (() => void) | undefined;

// Belt-and-braces: a stray interval keeps node:test's event loop alive forever.
t.after(() => stop?.());

const ts = new TransformStream({
start() {
const id = setInterval(() => {
ticks += 1;
}, 5);
stop = () => {
clearInterval(id);
stopped = true;
};
},
transform(chunk, controller) {
controller.enqueue(chunk);
},
cancel() {
stop?.();
},
});

const reader = ts.readable.getReader();
await reader.cancel(new Error("disconnect"));

assert.equal(stopped, true, "cancel() should have cleared the interval");

const before = ticks;
await new Promise((resolve) => setTimeout(resolve, 30));
assert.equal(ticks, before, "interval must not keep firing after cancel()");
});

// ---------------------------------------------------------------------------
// 2. sanitizeToolResultId()
// ---------------------------------------------------------------------------

test("sanitizeToolResultId returns null for falsy ids so orphan tool_results stay skipped", () => {
assert.equal(sanitizeToolResultId(undefined), null);
assert.equal(sanitizeToolResultId(null), null);
assert.equal(sanitizeToolResultId(""), null);
assert.equal(sanitizeToolResultId(0), null);
});

test("sanitizeToolResultId passes a well-formed string id through unchanged", () => {
assert.equal(sanitizeToolResultId("toolu_abc-123"), "toolu_abc-123");
});

test("sanitizeToolResultId replaces characters outside [A-Za-z0-9_-]", () => {
assert.equal(sanitizeToolResultId("call:with spaces//slashes"), "call_with_spaces__slashes");
});

test("sanitizeToolResultId coerces a non-string id instead of throwing", () => {
// Previously this reached `id.replace()` on a number and threw a TypeError.
assert.equal(sanitizeToolResultId(12345), "12345");
});

// ---------------------------------------------------------------------------
// 3. Read shim `limit` clamping
// ---------------------------------------------------------------------------

function readShim(args: Record<string, unknown>): Record<string, unknown> {
return JSON.parse(applyToolCallShimToBuffer("Read", JSON.stringify(args)));
}

test("Read shim clamps a limit above the 2000-line cap", () => {
assert.equal(readShim({ file_path: "/a.txt", limit: 5000 }).limit, 2000);
});

test("Read shim leaves an in-range limit untouched at both bounds", () => {
assert.equal(readShim({ file_path: "/a.txt", limit: 1 }).limit, 1);
assert.equal(readShim({ file_path: "/a.txt", limit: 2000 }).limit, 2000);
assert.equal(readShim({ file_path: "/a.txt", limit: 500 }).limit, 500);
});

test("Read shim drops a limit below 1", () => {
assert.equal("limit" in readShim({ file_path: "/a.txt", limit: 0 }), false);
assert.equal("limit" in readShim({ file_path: "/a.txt", limit: -10 }), false);
});

test("Read shim coerces numeric-string limit/offset before clamping", () => {
const out = readShim({ file_path: "/a.txt", limit: "9999", offset: "-5" });
assert.equal(out.limit, 2000);
assert.equal(out.offset, 0);
});

test("Read shim floors a negative offset at 0", () => {
assert.equal(readShim({ file_path: "/a.txt", offset: -1 }).offset, 0);
});
Loading