mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-04 14:22:09 +03:00
refactor(sse): resolve open-sse utils/translator type diagnostics for TS 7 (#8483)
First slice of the TypeScript 7 migration split requested on #7697: resolve the type diagnostics under `open-sse/tsconfig.json` in the lowest-risk modules, with no toolchain change. 12 diagnostics across 8 files, all outside the hot path — `chatCore.ts` and `stream.ts` are deliberately left for a later, standalone slice. Fixes, by cause: * `Transformer.cancel` (progressTracker, sseHeartbeat, and stream.ts's existing handler) — the WHATWG Streams standard defines `transformer.cancel(reason)` and Node implements it (verified on v24: cancelling the readable side invokes it), but `lib.dom.d.ts` still omits it from `Transformer`, so every such handler was TS2353. These handlers clear the heartbeat/progress intervals when an SSE client disconnects, so deleting them to satisfy the checker would leak a timer per abandoned stream. The interface is patched in `open-sse/types.d.ts` instead. * `earlyStreamKeepalive` — `SettledHandler` was discriminated by `ok: true | false`. 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 (the two `.response` reads elsewhere in the file did, which is why only one site errored). Retagged with a string discriminant, which narrows both branches under the same settings. * `toolCallShim` / `openai-responses` — assigning back to a property declared `unknown` resets the `typeof` narrowing, so the following comparison no longer saw a number/array. Both now read through a local. The `Read` limit clamp is behavior-identical: its two branches are mutually exclusive at READ_MAX_LIMIT 2000. * `sanitizeToolResultId` — takes `unknown` but forwards to a `string` parameter; a non-string id previously reached `.replace()` and threw. Coerced instead. * `openaiHelper` — `opts = {}` inferred `{}`; typed as `FilterToOpenAIFormatOptions`. * `cursorAgentProtobuf` — `Buffer.alloc(0)` infers `Buffer<ArrayBuffer>` under @types/node 26 while the decoded field is `Buffer<ArrayBufferLike>`; the locals now use bare `Buffer`, matching `requestMetadata` a few lines above. Validation: 335 -> 321 diagnostics with zero new errors (full tsc error-set diff against the base config). typecheck:core clean, lint clean, check:type-coverage 92.17% -> 94.17%. All 114 existing test files that import a touched module pass; `plan3-p0.test.ts` fails identically with and without this change (it reads the developer's real ~/.omniroute DB instead of a test-scoped DATA_DIR). The new test covers the three behavioral surfaces rather than the refactors the existing keepalive/heartbeat suites already hold: that `transformer.cancel()` really fires and can clear an interval, the id coercion, and the limit-clamp bounds.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
|
||||
@@ -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
open-sse/types.d.ts
vendored
21
open-sse/types.d.ts
vendored
@@ -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>;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
|
||||
@@ -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>,
|
||||
@@ -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;
|
||||
@@ -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.
|
||||
@@ -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 {
|
||||
|
||||
151
tests/unit/ts7-open-sse-type-fixes.test.ts
Normal file
151
tests/unit/ts7-open-sse-type-fixes.test.ts
Normal file
@@ -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);
|
||||
});
|
||||
Reference in New Issue
Block a user