From 9ce89ac3b1efb2b926c01bd1ea799a820abaf98d Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 05:43:32 +0800 Subject: [PATCH 1/7] test: cover response-scoped tool call aggregation contract --- src/__tests__/stream.test.ts | 326 +++++++++++++++++++++++++++++++++++ 1 file changed, 326 insertions(+) diff --git a/src/__tests__/stream.test.ts b/src/__tests__/stream.test.ts index 3ebf7e4..2380266 100644 --- a/src/__tests__/stream.test.ts +++ b/src/__tests__/stream.test.ts @@ -612,6 +612,332 @@ describe("streamQoder", () => { }); }); +// Synthetic wire inputs only; snapshots are copied synchronously at push(), +// because SDK partial messages are live references, not historical snapshots. +describe("streamQoder response aggregation contract", () => { + const originalFetch = globalThis.fetch; + afterEach(() => { + globalThis.fetch = originalFetch; + vi.restoreAllMocks(); + }); + + function tool(index: number | undefined, id?: string, name?: string, args?: string): object { + return { index, id, function: { name, arguments: args } }; + } + + async function run(deltas: object[], finish = "tool_calls", byteSize?: number) { + const snapshots: AssistantMessageEvent[] = []; + const StreamCtor = (PiAi as unknown as { AssistantMessageEventStream: new () => AssistantMessageEventStream }) + .AssistantMessageEventStream; + const prototype = StreamCtor.prototype; + const push = prototype.push; + vi.spyOn(prototype, "push").mockImplementation(function (this: AssistantMessageEventStream, event) { + snapshots.push(structuredClone(event) as AssistantMessageEvent); + return push.call(this, event); + }); + const frames = [...deltas.map((delta) => sseEnvelope(chunk(delta))), sseEnvelope(finishChunk(finish)), DONE_SSE]; + const bytes = new TextEncoder().encode(frames.join("")); + let cancelled = false; + const body = new ReadableStream({ + start(controller) { + if (byteSize) { + for (let offset = 0; offset < bytes.length; offset += byteSize) { + controller.enqueue(bytes.slice(offset, offset + byteSize)); + } + } else { + for (const frame of frames) controller.enqueue(new TextEncoder().encode(frame)); + } + // Leave open: DONE and all error paths must release the reader. + }, + cancel() { + cancelled = true; + }, + }); + globalThis.fetch = vi.fn(async () => new Response(body)) as unknown as typeof fetch; + const events = await consume(streamQoder(makeModel(), makeContext(), { apiKey: "synthetic" })); + return { events, snapshots, body, cancelled }; + } + + type Result = Awaited>; + function success( + result: Result, + expected: Array<{ id: string; name: string; arguments: object }>, + reason = "toolUse", + ) { + expect(result.events.filter((event) => event.type === "error")).toEqual([]); + const done = result.events.filter((event) => event.type === "done"); + expect(done).toHaveLength(1); + if (done[0].type !== "done") throw new Error("missing done"); + expect(done[0].message.stopReason).toBe(reason); + expect(done[0].reason).toBe(reason); + expect(done[0].message.content.filter((block) => block.type === "toolCall")).toEqual( + expected.map((call) => ({ type: "toolCall", ...call })), + ); + expect(result.snapshots[0].type).toBe("start"); + expect(result.snapshots.filter((event) => event.type === "toolcall_start")).toHaveLength(expected.length); + const ends = result.snapshots.filter((event) => event.type === "toolcall_end"); + expect(ends.map((event) => event.toolCall)).toEqual(expected.map((call) => ({ type: "toolCall", ...call }))); + for (const event of result.snapshots) { + if (event.type === "toolcall_start" || event.type === "toolcall_delta" || event.type === "toolcall_end") { + expect(event.partial.content[event.contentIndex]?.type).toBe("toolCall"); + if (event.type === "toolcall_end") expect(event.partial.content[event.contentIndex]).toEqual(event.toolCall); + } + } + expect(result.cancelled).toBe(true); + } + + function failure(result: Result, category: RegExp) { + expect(result.events.filter((event) => event.type === "done")).toEqual([]); + expect(result.events.filter((event) => event.type === "error")).toHaveLength(1); + const error = result.events.find((event) => event.type === "error"); + expect(error?.reason).toBe("error"); + expect(error?.error.stopReason).toBe("error"); + expect(error?.error.errorMessage).toMatch(category); + expect(error?.error.errorMessage).not.toMatch(/server error|synthetic-private/); + expect(result.snapshots.filter((event) => event.type === "toolcall_end")).toEqual([]); + expect(result.cancelled).toBe(true); + } + + const a = { id: "synthetic_a", name: "lookup", arguments: { path: "/a" } }; + const b = { id: "synthetic_b", name: "lookup", arguments: { path: "/b" } }; + + it("C01 preserves stable parallel indexes and ID-less continuations", async () => { + const result = await run([ + { tool_calls: [tool(7, a.id, a.name, '{"path":'), tool(2, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(2, undefined, undefined, '"/b"}'), tool(7, undefined, undefined, '"/a"}')] }, + ]); + success(result, [a, b]); + expect(result.snapshots.filter((event) => event.type === "toolcall_delta")).toHaveLength(4); + }); + + it("C01 positive control preserves ordinary stable parallel indexes", async () => { + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":'), tool(1, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(1, undefined, undefined, '"/b"}'), tool(0, undefined, undefined, '"/a"}')] }, + ]), + [a, b], + ); + }); + + it("C02 C03 separates same-index same-name IDs after a complete object", async () => { + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":"/a"}')] }, + { tool_calls: [tool(0, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(0, undefined, undefined, '"/b"}')] }, + ]), + [a, b], + ); + }); + + it("C04 routes repeated IDs across index changes without duplicating blocks", async () => { + success( + await run([ + { tool_calls: [tool(3, a.id, a.name, '{"path":')] }, + { tool_calls: [tool(9, a.id, a.name, '"/a"}')] }, + { tool_calls: [tool(3, a.id, undefined, " \n")] }, + ]), + [a], + ); + }); + + it.each(["arguments-first", "name-first"])("C05 binds a unique provisional state: %s", async (order) => { + const prefix = order === "arguments-first" ? tool(4, undefined, undefined, '{"path":') : tool(4, undefined, a.name); + const middle = + order === "arguments-first" ? tool(4, a.id, a.name, '"/a"}') : tool(4, undefined, undefined, '{"path":"/a"}'); + const result = await run([ + { content: "prefix" }, + { tool_calls: [prefix] }, + { tool_calls: [middle] }, + { tool_calls: [tool(4, a.id, a.name)] }, + ]); + success(result, [a]); + const deltas = result.snapshots.filter((event) => event.type === "toolcall_delta"); + expect(deltas.map((event) => event.delta).join("")).toBe('{"path":"/a"}'); + }); + + it.each([ + ["missing ID", [tool(0, undefined, "lookup", "{}")], /identity|unresolved|missing.*id/i], + ["missing name", [tool(0, a.id, undefined, "{}")], /identity|unresolved|missing.*name/i], + ["name conflict", [tool(0, a.id, "lookup", "{"), tool(0, a.id, "other", "}")], /identity|conflict/i], + ] as const)("C06 rejects %s", async (_label, calls, category) => { + failure(await run(calls.map((call) => ({ tool_calls: [call] }))), category); + }); + + it("C07 routes explicitly identified interleaving on a reused index", async () => { + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":'), tool(0, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(0, a.id, undefined, '"/a"}'), tool(0, b.id, undefined, '"/b"}')] }, + ]), + [a, b], + ); + }); + + it.each(["{", ""])("C08 rejects ambiguous continuation even if later ID arrives (prefix=%j)", async (prefix) => { + failure( + await run([ + { tool_calls: [tool(0, a.id, a.name, prefix), tool(0, b.id, b.name, prefix)] }, + { tool_calls: [tool(0, undefined, undefined, '"synthetic-private":1}')] }, + { tool_calls: [tool(0, b.id, undefined, "{}")] }, + ]), + /ambiguous tool-call continuation/i, + ); + }); + + it("C09 continues a unique call without index", async () => { + success( + await run([ + { tool_calls: [tool(5, a.id, a.name, '{"path":')] }, + { tool_calls: [tool(undefined, undefined, undefined, '"/a"}')] }, + ]), + [a], + ); + }); + + it("C09 routes explicit IDs without indexes among parallel calls", async () => { + success( + await run([ + { tool_calls: [tool(undefined, a.id, a.name, '{"path":'), tool(undefined, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(undefined, b.id, undefined, '"/b"}'), tool(undefined, a.id, undefined, '"/a"}')] }, + ]), + [a, b], + ); + }); + + it("C09 rejects absent index with multiple candidates", async () => { + failure( + await run([ + { tool_calls: [tool(0, a.id, a.name, "{"), tool(1, b.id, b.name, "{")] }, + { tool_calls: [tool(undefined, undefined, undefined, "}")] }, + ]), + /ambiguous tool-call continuation/i, + ); + }); + + it.each([undefined, "", "{}"])("C10 preserves independent no-argument calls (%j)", async (args) => { + success(await run([{ tool_calls: [tool(0, a.id, a.name, args), tool(0, b.id, b.name, args)] }]), [ + { ...a, arguments: {} }, + { ...b, arguments: {} }, + ]); + }); + + it.each(["{", '{"path":', "{}{}", "{}garbage", "[]", "null", "1", '"text"', "true", " \n\t", '{"x":}', '{"x":1,}'])( + "C11 rejects non-object or invalid nonempty JSON %j without fallback", + async (args) => { + failure(await run([{ tool_calls: [tool(0, a.id, a.name, args)] }]), /json|argument|object|incomplete/i); + }, + ); + + it("C11 accepts trailing whitespace on one stable object", async () => { + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":"/a"}')] }, + { tool_calls: [tool(0, undefined, undefined, " \n\t")] }, + ]), + [a], + ); + }); + + it("C12 ignores a completely empty placeholder", async () => { + const result = await run([{ content: "synthetic text" }, { tool_calls: [tool(0)] }], "stop"); + success(result, [], "stop"); + const done = result.events.find((event) => event.type === "done"); + expect(done?.message.content).toEqual([{ type: "text", text: "synthetic text" }]); + }); + + it.each([false, true])( + "C13 preserves block identity after tag thinking insertion (API reasoning=%s)", + async (apiReasoning) => { + const result = await run([ + ...(apiReasoning ? [{ reasoning_content: "api thought" }] : []), + { content: "before" }, + { tool_calls: [tool(0, a.id, a.name, '{"path":')] }, + { content: "tag thoughtafter" }, + { tool_calls: [tool(0, a.id, undefined, '"/a"}')] }, + ]); + success(result, [a]); + const done = result.events.find((event) => event.type === "done"); + expect(done?.message.content.filter((block) => block.type !== "toolCall")).toEqual([ + ...(apiReasoning ? [{ type: "thinking", thinking: "api thought" }] : []), + { type: "thinking", thinking: "tag thought" }, + { type: "text", text: "before" }, + { type: "text", text: "after" }, + ]); + for (const event of result.snapshots) { + if (event.type === "toolcall_start" || event.type === "toolcall_delta") { + expect(event.partial.content[event.contentIndex]).toMatchObject({ type: "toolCall", id: a.id, name: a.name }); + } + } + const first = result.snapshots.find((event) => event.type === "toolcall_start"); + expect( + first?.partial.content.some((block) => block.type === "thinking" && block.thinking === "tag thought"), + ).toBe(false); + }, + ); + + it.each([undefined, 100000, 17, 1])("C14 preserves UTF-8 and JSON escapes with byte size %s", async (size) => { + const args = { path: "/合成/😀", value: 'quote" slash\\ newline\n } {' }; + const json = JSON.stringify(args); + const split = json.indexOf('\\"') + 1; + success( + await run( + [ + { tool_calls: [tool(0, a.id, a.name, json.slice(0, split))] }, + { tool_calls: [tool(0, undefined, undefined, json.slice(split))] }, + ], + "tool_calls", + size, + ), + [{ ...a, arguments: args }], + ); + }); + + it("C15 validates the whole batch before any successful toolcall_end", async () => { + failure( + await run([{ tool_calls: [tool(0, a.id, a.name, "{}"), tool(1, b.id, b.name, '{"synthetic-private":')] }]), + /json|argument|incomplete/i, + ); + }); + + it("C15 preserves length for a complete tool object", async () => { + success(await run([{ tool_calls: [tool(0, a.id, a.name, '{"path":"/a"}')] }], "length"), [a], "length"); + }); + + it("C15 rejects truncated tool JSON even with length", async () => { + failure(await run([{ tool_calls: [tool(0, a.id, a.name, '{"path":')] }], "length"), /json|argument|incomplete/i); + }); + + it("C15 aborts an in-progress tool response and releases its reader", async () => { + const controller = new AbortController(); + let cancelled = false; + const body = new ReadableStream({ + start(target) { + target.enqueue(new TextEncoder().encode(sseEnvelope(chunk({ tool_calls: [tool(0, a.id, a.name, "{")] })))); + controller.signal.addEventListener("abort", () => target.error(new DOMException("Aborted", "AbortError"))); + }, + cancel() { + cancelled = true; + }, + }); + globalThis.fetch = vi.fn(async () => new Response(body)) as unknown as typeof fetch; + const stream = streamQoder(makeModel(), makeContext(), { apiKey: "synthetic", signal: controller.signal }); + const events: AssistantMessageEvent[] = []; + for await (const event of stream) { + events.push(event); + if (event.type === "toolcall_delta") controller.abort(); + } + expect(events.filter((event) => event.type === "done" || event.type === "toolcall_end")).toEqual([]); + const errors = events.filter((event) => event.type === "error"); + expect(errors).toHaveLength(1); + expect(errors[0].error.stopReason).toBe("aborted"); + // An errored ReadableStream need not call its underlying cancel callback, + // but parser ownership of the reader must still be released. + expect(cancelled || !body.locked).toBe(true); + }); +}); + describe("resolveRequestContext", () => { const flatContext = { systemPrompt: "legacy prompt", From 85f04b74aa8f0b1a94f35524c4935b3001c51710 Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 06:12:16 +0800 Subject: [PATCH 2/7] test: add offline real SDK tool-call flow regression --- scripts/verify-tool-call-flow.ts | 499 +++++++++++++++++++++++++++++++ 1 file changed, 499 insertions(+) create mode 100644 scripts/verify-tool-call-flow.ts diff --git a/scripts/verify-tool-call-flow.ts b/scripts/verify-tool-call-flow.ts new file mode 100644 index 0000000..d3bb31f --- /dev/null +++ b/scripts/verify-tool-call-flow.ts @@ -0,0 +1,499 @@ +/** Synthetic, offline-only regression. Build first; no live mode or credential input is supported. */ +import assert from "node:assert/strict"; +import childProcess from "node:child_process"; +import crypto from "node:crypto"; +import dgram from "node:dgram"; +import dns from "node:dns"; +import fs from "node:fs"; +import http from "node:http"; +import http2 from "node:http2"; +import https from "node:https"; +import { syncBuiltinESMExports } from "node:module"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; +import tls from "node:tls"; +import { pathToFileURL } from "node:url"; +import type { AssistantMessage, Message, ToolCall } from "@earendil-works/pi-ai"; +import type { ToolDefinition } from "@earendil-works/pi-coding-agent"; + +type WireMessage = { + role: string; + content?: unknown; + tool_calls?: { id: string; function: { name: string; arguments: string } }[]; + tool_call_id?: string; +}; +type Invocation = { id: string; name: string; args: unknown }; +type Fixture = { + name: string; + deltas: object[]; + expected: Invocation[]; + providerError?: boolean; + schemaError?: boolean; +}; + +function option(name: string): string { + const index = process.argv.indexOf(name); + assert(index > 0 && process.argv[index + 1], `Required option: ${name}`); + return path.resolve(process.argv[index + 1]); +} +const packageRoot = option("--package-root"); +const sdkRoot = option("--sdk-root"); +const outputPath = option("--output"); +assert.equal(process.argv.length, 8, "Only --package-root, --sdk-root and --output are supported"); +const temporary = fs.mkdtempSync(path.join(os.tmpdir(), "qoder-offline-flow-")); +const originalCwd = process.cwd(); +for (const key of Object.keys(process.env)) delete process.env[key]; +Object.assign(process.env, { + HOME: path.join(temporary, "home"), + XDG_CONFIG_HOME: path.join(temporary, "config"), + XDG_CACHE_HOME: path.join(temporary, "cache"), + XDG_DATA_HOME: path.join(temporary, "data"), + XDG_STATE_HOME: path.join(temporary, "state"), + PI_CODING_AGENT_DIR: path.join(temporary, "home/.pi/agent"), + TMPDIR: path.join(temporary, "tmp"), + JITI_FS_CACHE: "false", +}); +for (const directory of [...Object.values(process.env), path.join(temporary, "workspace")]) { + fs.mkdirSync(directory!, { recursive: true, mode: 0o700 }); +} +process.chdir(path.join(temporary, "workspace")); +const safety = { deniedNetwork: 0, deniedSubprocess: 0, selfTestsPassed: 0, notOsSandbox: true }; +const denyNetwork = (): never => { + safety.deniedNetwork++; + throw new Error("offline-network-denied"); +}; +const denySubprocess = (): never => { + safety.deniedSubprocess++; + throw new Error("offline-subprocess-denied"); +}; +// These guards precede every SDK/provider import. No loopback exception is needed. +globalThis.fetch = denyNetwork; +http.request = http.get = https.request = https.get = denyNetwork; +http2.connect = denyNetwork; +net.connect = net.createConnection = net.Socket.prototype.connect = tls.connect = denyNetwork; +dgram.Socket.prototype.send = denyNetwork; +for (const target of [dns, dns.promises]) { + for (const key of Object.keys(target)) { + if (key === "lookup" || key.startsWith("resolve")) Object.assign(target, { [key]: denyNetwork }); + } +} +for (const key of ["exec", "execSync", "execFile", "execFileSync", "spawn", "spawnSync", "fork"]) { + Object.assign(childProcess, { [key]: denySubprocess }); +} +syncBuiltinESMExports(); +for (const probe of [ + () => fetch("https://example.invalid"), + () => net.connect(443, "192.0.2.1"), + () => dns.promises.lookup("example.invalid"), + () => dns.lookup("example.invalid", () => {}), + () => tls.connect(443, "example.invalid"), + () => http.request("http://example.invalid"), + () => https.get("https://example.invalid"), + () => http2.connect("https://example.invalid"), + () => childProcess.spawnSync("sh", ["-c", "exit 0"]), + () => childProcess.exec("true"), +]) { + assert.throws(probe, /offline-.*-denied/); + safety.selfTestsPassed++; +} +const expectedDenials = { network: safety.deniedNetwork, subprocess: safety.deniedSubprocess }; +const sdkEntry = path.join(sdkRoot, "@earendil-works/pi-coding-agent/dist/index.js"); +const aiEntry = path.join(sdkRoot, "@earendil-works/pi-ai/dist/compat.js"); +const bundle = path.join(packageRoot, "dist/index.js"); +// Load a production-shaped copy with no adjacent development node_modules. Jiti's native-first +// path would otherwise bypass host aliases and accidentally import the project's pi-ai 1.0.4. +const loadRoot = path.join(temporary, "package"); +fs.mkdirSync(loadRoot); +fs.copyFileSync(path.join(packageRoot, "package.json"), path.join(loadRoot, "package.json")); +fs.cpSync(path.join(packageRoot, "dist"), path.join(loadRoot, "dist"), { recursive: true }); +const hash = (file: string) => crypto.createHash("sha256").update(fs.readFileSync(file)).digest("hex"); +const report: { + passed: boolean; + sdkEntry: string; + aiEntry: string; + sdkVersion?: string; + aiVersion?: string; + bundleSha256?: string; + loadedBundleSha256?: string; + safety: typeof safety; + cases: object[]; + failure?: string; +} = { + passed: false, + sdkEntry, + aiEntry, + safety, + cases: [], +}; +const delta = (index: number, id: string | undefined, name: string | undefined, args: string) => ({ + tool_calls: [{ index, ...(id ? { id } : {}), function: { ...(name ? { name } : {}), arguments: args } }], +}); +const read: Invocation = { id: "fixture_A", name: "fixture_read", args: { path: "synthetic/a.txt" } }; +const list: Invocation = { id: "fixture_B", name: "fixture_ls", args: { path: "synthetic/b" } }; +const fixtures: Fixture[] = [ + { + name: "stable-indices-positive", + deltas: [ + delta(0, read.id, read.name, JSON.stringify(read.args)), + delta(1, list.id, list.name, '{"path":"synthetic/'), + delta(1, undefined, undefined, 'b"}'), + ], + expected: [read, list], + }, + { + name: "reused-index-required-path", + deltas: [ + delta(0, read.id, read.name, JSON.stringify(read.args)), + delta(0, list.id, list.name, '{"path":"synthetic/'), + delta(0, undefined, undefined, 'b"}'), + ], + expected: [read, list], + }, + { + name: "optional-path-positive", + deltas: [delta(0, "optional_A", "fixture_optional", '{"path":"synthetic/optional"}')], + expected: [{ id: "optional_A", name: "fixture_optional", args: { path: "synthetic/optional" } }], + }, + { + name: "reused-index-optional-path", + deltas: [ + delta(0, "optional_A", "fixture_optional", '{"path":"synthetic/a"}'), + delta(0, "optional_B", "fixture_optional", '{"path":"synthetic/'), + delta(0, undefined, undefined, 'b"}'), + ], + expected: [ + { id: "optional_A", name: "fixture_optional", args: { path: "synthetic/a" } }, + { id: "optional_B", name: "fixture_optional", args: { path: "synthetic/b" } }, + ], + }, + { + name: "noargs-positive", + deltas: [delta(0, "noargs_A", "fixture_noargs", "")], + expected: [{ id: "noargs_A", name: "fixture_noargs", args: {} }], + }, + { + name: "reused-index-noargs", + deltas: [delta(0, "noargs_A", "fixture_noargs", "{}"), delta(0, "noargs_B", "fixture_noargs", "{}")], + expected: [ + { id: "noargs_A", name: "fixture_noargs", args: {} }, + { id: "noargs_B", name: "fixture_noargs", args: {} }, + ], + }, + { + name: "schema-error-sdk-positive", + deltas: [delta(0, "schema_A", "fixture_read", '{"path":{"invalid":true}}')], + expected: [{ id: "schema_A", name: "fixture_read", args: { path: { invalid: true } } }], + schemaError: true, + }, + { + name: "invalid-json-reject-whole-batch", + deltas: [ + delta(0, read.id, read.name, JSON.stringify(read.args)), + delta(1, "bad_A", "fixture_optional", '{"path":'), + ], + expected: [], + providerError: true, + }, + { + name: "ambiguous-idless-reject-whole-batch", + deltas: [ + delta(0, "ambiguous_A", "fixture_optional", '{"path":"synthetic/'), + delta(0, "ambiguous_B", "fixture_optional", '{"path":"synthetic/'), + delta(0, undefined, undefined, 'b"}'), + ], + expected: [], + providerError: true, + }, +]; +function envelope(body: unknown): string { + return `data:${JSON.stringify({ + headers: {}, + body: typeof body === "string" ? body : JSON.stringify(body), + statusCodeValue: 200, + statusCode: "OK", + })}\n\n`; +} +function sse(deltas: object[], reason: string): string { + return ( + deltas.map((d) => envelope({ choices: [{ index: 0, delta: d }] })).join("") + + envelope({ choices: [{ index: 0, finish_reason: reason }], usage: { prompt_tokens: 1, completion_tokens: 1 } }) + + envelope("[DONE]") + ); +} +// Independent decoder: reverse the protocol block permutation, then translate the custom alphabet. +function decodeBody(body: BodyInit | null | undefined): { messages: WireMessage[] } { + const custom = "_doRTgHZBKcGVjlvpC,@aFSx#DPuNJme&i*MzLOEn)sUrthbf%Y^w.(kIQyXqWA!$"; + const standard = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="; + const raw = Buffer.isBuffer(body) ? body.toString("ascii") : String(body); + const translated = [...raw].map((c) => standard[custom.indexOf(c)] ?? c).join(""); + const third = Math.floor(translated.length / 3); + const reordered = translated.slice(-third) + translated.slice(third, -third) + translated.slice(0, third); + return JSON.parse(Buffer.from(reordered, "base64").toString("utf8")); +} +try { + const sdk: typeof import("@earendil-works/pi-coding-agent") = await import(pathToFileURL(sdkEntry).href); + const ai: typeof import("@earendil-works/pi-ai/compat") = await import(pathToFileURL(aiEntry).href); + report.sdkVersion = JSON.parse( + fs.readFileSync(path.join(sdkRoot, "@earendil-works/pi-coding-agent/package.json"), "utf8"), + ).version; + report.aiVersion = JSON.parse( + fs.readFileSync(path.join(sdkRoot, "@earendil-works/pi-ai/package.json"), "utf8"), + ).version; + report.bundleSha256 = hash(bundle); + report.loadedBundleSha256 = hash(path.join(loadRoot, "dist/index.js")); + assert.equal(report.loadedBundleSha256, report.bundleSha256); + // Fresh synthetic metadata suppresses startup discovery without mocking the factory or catalog reader. + for (const name of ["qoder-models-cache.json", "qoder-cn-models-cache.json"]) { + fs.writeFileSync( + path.join(process.env.PI_CODING_AGENT_DIR!, name), + JSON.stringify({ + updatedAt: Date.now(), + models: [], + configs: { Lite: { key: "lite", display_name: "Lite", is_reasoning: false } }, + }), + ); + } + for (const fixture of fixtures) { + const invocations: Invocation[] = []; + const requests: { messages: WireMessage[] }[] = []; + const finalMessages: Message[] = []; + const parserEvents: string[] = []; + let settled = 0; + let executionStarts = 0; + let fCalls = 0; + let refresh = 0; + let sdkDerivedAuth = true; + let streamIdentity = true; + let identityRequests = 0; + globalThis.fetch = async (input, init) => { + const url = new URL(String(input)); + if (url.hostname === "openapi.qoder.sh" && url.pathname === "/api/v1/userinfo") { + identityRequests++; + return Response.json({ id: "synthetic-user", name: "Synthetic User", email: "synthetic@example.invalid" }); + } + if (url.hostname !== "api3.qoder.sh" || url.pathname !== "/algo/api/v2/service/pro/sse/agent_chat_generation") + return denyNetwork(); + requests.push(decodeBody(init?.body)); + assert(requests.length <= 2, "No retry or third model request is allowed"); + const text = + requests.length === 1 ? sse(fixture.deltas, "tool_calls") : sse([{ content: "SYNTHETIC_DONE" }], "stop"); + const bytes = new TextEncoder().encode(text); + return new Response( + new ReadableStream({ + start(controller) { + for (let offset = 0; offset < bytes.length; offset += 17) + controller.enqueue(bytes.slice(offset, offset + 17)); + controller.close(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + }; + const record = { + type: "oauth" as const, + access: "synthetic-access-not-a-token", + refresh: "synthetic-refresh-not-a-token", + expires: Date.now() + 3600000, + }; + const memory = new ai.InMemoryCredentialStore(); + await memory.modify("qoder", async () => record); + const runtime = await sdk.ModelRuntime.create({ + credentials: { + read: memory.read.bind(memory), + list: memory.list.bind(memory), + modify: async () => { + refresh++; + throw new Error("refresh-disabled"); + }, + delete: async () => { + throw new Error("delete-disabled"); + }, + }, + modelsPath: null, + allowModelNetwork: false, + refreshOnCreate: false, + }); + const settings = sdk.SettingsManager.inMemory({ + retry: { enabled: false }, + compaction: { enabled: false }, + cacheWarming: "off", + }); + const loader = new sdk.DefaultResourceLoader({ + cwd: process.cwd(), + agentDir: process.env.PI_CODING_AGENT_DIR!, + settingsManager: settings, + additionalExtensionPaths: [loadRoot], + noExtensions: true, + noSkills: true, + noPromptTemplates: true, + noThemes: true, + noContextFiles: true, + }); + await loader.reload(); + assert.deepEqual(loader.getExtensions().errors, [], "Official loader must load the actual built package"); + const pending = loader.getExtensions().runtime.pendingProviderRegistrations.filter((item) => item.name === "qoder"); + assert.equal(pending.length, 1); + const config = pending[0].config; + const F = config.streamSimple; + assert(F && config.oauth && !config.apiKey, "Original factory and OAuth handler required"); + const tracedF: NonNullable = (model, context, options) => { + fCalls++; + sdkDerivedAuth &&= options?.apiKey === record.access; + const stream = F(model, context, options); + // The constructor exists at runtime but is absent from the 1.0.4 public declarations. + const Stream = ( + ai as unknown as { + AssistantMessageEventStream: new () => ReturnType; + } + ).AssistantMessageEventStream; + streamIdentity &&= stream instanceof Stream; + return stream; + }; + config.streamSimple = tracedF; + runtime.registerProvider("qoder", config); + const model = runtime.getModel("qoder", "Lite"); + assert(model && model.api === "qoder-api"); + const tools: ToolDefinition[] = ["fixture_read", "fixture_ls", "fixture_optional", "fixture_noargs"].map( + (name) => ({ + name, + label: name, + description: "Synthetic in-memory fixture only", + parameters: { + type: "object", + properties: name === "fixture_noargs" ? {} : { path: { type: "string" } }, + required: name === "fixture_read" || name === "fixture_ls" ? ["path"] : [], + additionalProperties: false, + } as ToolDefinition["parameters"], + async execute(id, args) { + invocations.push({ id, name, args: structuredClone(args) }); + return { content: [{ type: "text", text: `RESULT:${id}:${JSON.stringify(args)}` }], details: {} }; + }, + }), + ); + const initializationErrors: string[] = []; + const { session } = await sdk.createAgentSession({ + cwd: process.cwd(), + agentDir: process.env.PI_CODING_AGENT_DIR!, + modelRuntime: runtime, + model, + thinkingLevel: "off", + settingsManager: settings, + resourceLoader: loader, + sessionManager: sdk.SessionManager.inMemory(process.cwd()), + customTools: tools, + tools: tools.map((tool) => tool.name), + }); + const unsubscribe = session.subscribe((event) => { + if (event.type === "agent_settled") settled++; + if (event.type === "tool_execution_start") executionStarts++; + if (event.type === "message_end") finalMessages.push(structuredClone(event.message) as Message); + if (event.type === "message_update") parserEvents.push(event.assistantMessageEvent.type); + }); + const caseReport: Record = { + name: fixture.name, + passed: false, + loadedExtensionPaths: loader.getExtensions().extensions.map((extension) => extension.resolvedPath), + }; + try { + await session.bindExtensions({ onError: () => initializationErrors.push("extension-initialization-error") }); + assert.deepEqual(initializationErrors, []); + assert.equal(runtime.getRegisteredProviderConfig("qoder")?.streamSimple, tracedF); + assert.equal(session.model?.id, model.id); + assert.deepEqual(session.getActiveToolNames().sort(), tools.map((tool) => tool.name).sort()); + const auth = await runtime.getAuth(model); + assert(auth?.source === "OAuth" && auth.auth.apiKey === record.access); + await session.prompt("Run the synthetic fixture tools, then return SYNTHETIC_DONE."); + const assistants = finalMessages.filter((m): m is AssistantMessage => m.role === "assistant"); + const calls = assistants[0]?.content.filter((c): c is ToolCall => c.type === "toolCall") ?? []; + Object.assign(caseReport, { + parserEvents, + assistant: assistants[0], + invocations, + executionStarts, + toolResults: finalMessages.filter((m) => m.role === "toolResult"), + requests, + settled, + fCalls, + sdkDerivedAuth, + streamIdentity, + identityRequests, + refresh, + terminal: assistants.at(-1)?.stopReason, + }); + assert.equal(refresh, 0); + assert(sdkDerivedAuth && streamIdentity, "The actual F must receive SDK OAuth and return this host's stream"); + assert.equal(settled, 1, "The SDK must emit the real settled event"); + if (fixture.providerError) { + assert.equal(assistants[0]?.stopReason, "error"); + assert.deepEqual(invocations, [], "An invalid/ambiguous batch must not execute any implementation"); + assert.equal(requests.length, 1); + } else { + assert.deepEqual( + calls.map((c) => ({ id: c.id, name: c.name, args: c.arguments })), + fixture.expected, + "Final SDK assistant preserves IDs/names/complete arguments", + ); + assert.deepEqual( + invocations, + fixture.schemaError ? [] : fixture.expected, + "Count actual implementations, not tool_execution_start", + ); + const results = finalMessages.filter((m) => m.role === "toolResult"); + assert.deepEqual( + results.map((m) => m.toolCallId), + fixture.expected.map((c) => c.id), + ); + assert(results.every((m) => m.isError === !!fixture.schemaError)); + assert.deepEqual( + results.map((m) => m.toolName), + fixture.expected.map((c) => c.name), + ); + assert.equal(requests.length, 2); + const replay = requests[1].messages.find((m) => m.role === "assistant" && m.tool_calls); + assert.deepEqual( + replay?.tool_calls?.map((c) => ({ id: c.id, name: c.function.name, args: JSON.parse(c.function.arguments) })), + fixture.expected, + ); + assert.deepEqual( + requests[1].messages.filter((m) => m.role === "tool").map((m) => m.tool_call_id), + fixture.expected.map((c) => c.id), + ); + if (!fixture.schemaError) + assert.deepEqual( + requests[1].messages.filter((m) => m.role === "tool").map((m) => m.content), + fixture.expected.map((c) => `RESULT:${c.id}:${JSON.stringify(c.args)}`), + ); + assert.equal(assistants.at(-1)?.stopReason, "stop"); + assert.equal(session.getLastAssistantText(), "SYNTHETIC_DONE"); + } + caseReport.passed = true; + } catch (error) { + caseReport.failure = error instanceof Error ? error.message : "synthetic-flow-failed"; + } finally { + unsubscribe(); + session.dispose(); + report.cases.push(caseReport); + } + } + assert.equal(safety.deniedNetwork, expectedDenials.network, "Unexpected network attempt outside mock transport"); + assert.equal(safety.deniedSubprocess, expectedDenials.subprocess, "Unexpected child process"); + report.passed = report.cases.every((c) => (c as { passed: boolean }).passed); +} catch (error) { + report.failure = error instanceof Error ? error.message : "offline-flow-failed"; +} finally { + globalThis.fetch = denyNetwork; + process.chdir(originalCwd); + fs.rmSync(temporary, { recursive: true, force: true }); + fs.mkdirSync(path.dirname(outputPath), { recursive: true, mode: 0o700 }); + fs.writeFileSync(outputPath, `${JSON.stringify(report, null, 2)}\n`, { mode: 0o600 }); +} +process.stdout.write( + `${JSON.stringify({ + passed: report.passed, + sdkVersion: report.sdkVersion, + cases: report.cases.map((c) => ({ name: (c as Fixture).name, passed: (c as { passed: boolean }).passed })), + failure: report.failure, + })}\n`, +); +process.exitCode = report.passed ? 0 : 1; From 3efdf07dcd0dcc10039311622354ecfa6f8facaa Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 06:24:10 +0800 Subject: [PATCH 3/7] fix: preserve streamed tool-call identity and strict arguments --- src/protocol/stream.ts | 608 ++++++++++++++++++++++++++--------------- 1 file changed, 390 insertions(+), 218 deletions(-) diff --git a/src/protocol/stream.ts b/src/protocol/stream.ts index 2e26983..5c08896 100644 --- a/src/protocol/stream.ts +++ b/src/protocol/stream.ts @@ -24,13 +24,101 @@ import { qoderEncodeBody } from "./encoding.js"; import { isDegenerateDsmlTurn, stripDsmlResidue, stripThinkingTags, ThinkingTagParser } from "./thinking.js"; import { transformMessagesForQoder, transformTools } from "./transform.js"; +interface ToolCallCandidates { + states: Set; + complete: Set; +} + interface ToolCallState { - arguments: string; + argumentChunks: string[]; + argumentLength: number; + objectRoot?: boolean; + depth: number; + inString: boolean; + escaped: boolean; + complete: boolean; + parsedArguments?: ToolCall["arguments"]; id: string; name: string; - emittedStart?: boolean; - emittedEnd?: boolean; + block?: ToolCall; contentIndex: number; + candidates: Set; + unindexed: boolean; +} + +const TOOL_ARGUMENT_STRUCTURE = /["\\{}[\]]/g; +const NON_JSON_WHITESPACE = /[^ \t\r\n]/; + +function invalidToolArguments(state: ToolCallState): Error { + return new Error(`Invalid or incomplete tool-call JSON (characters=${state.argumentLength})`); +} + +// Only locate a possible root-object boundary here. JSON.parse remains the +// authority on grammar; scanning each new fragment avoids parsing prefixes. +function appendToolArguments(state: ToolCallState, fragment: string): void { + state.argumentChunks.push(fragment); + state.argumentLength += fragment.length; + if (state.complete) { + if (NON_JSON_WHITESPACE.test(fragment)) throw invalidToolArguments(state); + return; + } + let start = 0; + if (state.objectRoot === undefined) { + start = fragment.search(NON_JSON_WHITESPACE); + if (start === -1) return; + state.objectRoot = fragment[start] === "{"; + } + if (!state.objectRoot) return; + if (state.escaped) { + start = 1; + state.escaped = false; + } + TOOL_ARGUMENT_STRUCTURE.lastIndex = start; + while (true) { + const match = TOOL_ARGUMENT_STRUCTURE.exec(fragment); + if (!match) break; + const character = match[0]; + if (state.inString) { + if (character === "\\") { + if (TOOL_ARGUMENT_STRUCTURE.lastIndex === fragment.length) state.escaped = true; + else TOOL_ARGUMENT_STRUCTURE.lastIndex++; + } else if (character === '"') { + state.inString = false; + } + } else if (character === '"') { + state.inString = true; + } else if (character === "{" || character === "[") { + state.depth++; + } else if (character === "}" || character === "]") { + state.depth--; + if (state.depth === 0) { + state.complete = true; + if (NON_JSON_WHITESPACE.test(fragment.slice(TOOL_ARGUMENT_STRUCTURE.lastIndex))) { + throw invalidToolArguments(state); + } + return; + } + } + } +} + +function parseToolArguments(state: ToolCallState): ToolCall["arguments"] { + if (state.parsedArguments) return state.parsedArguments; + let args: unknown = {}; + if (state.argumentLength > 0) { + try { + args = JSON.parse(state.argumentChunks.join("")); + } catch { + // Do not let the malformed-SSE SyntaxError handler swallow this failure, + // and never include the parser's message (which can expose arguments). + throw invalidToolArguments(state); + } + } + if (args === null || typeof args !== "object" || Array.isArray(args)) { + throw new Error(`Non-object tool-call arguments (characters=${state.argumentLength})`); + } + state.parsedArguments = args as ToolCall["arguments"]; + return state.parsedArguments; } function stableHash(prefix: string, ...inputs: string[]): string { @@ -409,6 +497,35 @@ export function streamQoder( // this stays bounded instead of growing with the reply. let rawReasoningTail = ""; const toolCallsState: ToolCallState[] = []; + const toolCallsById = new Map(); + const toolCandidatesByIndex = new Map(); + const unscopedToolCandidates = new Set(); + const toolStatesByBlock = new Map(); + let lastToolState: ToolCallState | undefined; + + const refreshToolIndexes = () => { + if (!lastToolState || output.content[lastToolState.contentIndex] === lastToolState.block) return; + // ThinkingTagParser.emitThinking() can splice only when creating its + // first thinking block. All other layout changes append. Check the + // last tool in O(1), then refresh the shifted suffix in one scan. + for (let index = 0; index < output.content.length; index++) { + const block = output.content[index]; + if (block.type !== "toolCall") continue; + const state = toolStatesByBlock.get(block); + if (state) state.contentIndex = index; + } + }; + + const retireCompleteCandidates = (candidates: ToolCallCandidates) => { + for (const state of candidates.complete) { + // Structural closure alone is not sufficient evidence of a boundary. + parseToolArguments(state); + candidates.states.delete(state); + state.candidates.delete(candidates); + if (state.candidates.size === 0 && !state.unindexed) unscopedToolCandidates.delete(state); + } + candidates.complete.clear(); + }; const thinkingEnabled = (options?.reasoning as unknown) !== false && (options?.reasoning as unknown) !== "off"; const thinkingParser = thinkingEnabled ? new ThinkingTagParser(output, stream) : null; @@ -422,228 +539,290 @@ export function streamQoder( // already been streamed by then, so the agent looked stuck with no error. let sawDone = false; - while (!sawDone) { - const { done, value } = await reader.read(); - if (done) break; - - // Drop consumed prefix before appending so we do not keep growing a - // dead head of the string across chunks. - if (bufferStart > 0) { - buffer = buffer.substring(bufferStart); - bufferStart = 0; - } - buffer += decoder.decode(value, { stream: true }); - - while (true) { - const lineEnd = buffer.indexOf("\n", bufferStart); - if (lineEnd === -1) break; - - const line = buffer.substring(bufferStart, lineEnd).trim(); - bufferStart = lineEnd + 1; + try { + while (!sawDone) { + const { done, value } = await reader.read(); + if (done) break; + + // Drop consumed prefix before appending so we do not keep growing a + // dead head of the string across chunks. + if (bufferStart > 0) { + buffer = buffer.substring(bufferStart); + bufferStart = 0; + } + buffer += decoder.decode(value, { stream: true }); - if (!line.startsWith("data:")) continue; + while (true) { + const lineEnd = buffer.indexOf("\n", bufferStart); + if (lineEnd === -1) break; - const dataStr = line.substring(5).trim(); - if (dataStr === "[DONE]") { - sawDone = true; - break; - } + const line = buffer.substring(bufferStart, lineEnd).trim(); + bufferStart = lineEnd + 1; - try { - const envelope = JSON.parse(dataStr); - if (envelope.statusCodeValue && envelope.statusCodeValue !== 200) { - throw new Error(`Upstream status ${envelope.statusCodeValue}: ${envelope.body}`); - } + if (!line.startsWith("data:")) continue; - const innerStr = envelope.body; - // The gateway sends the sentinel wrapped in an envelope - // (`body: "[DONE]"`) as well as bare, and both mean the reply is - // over, so both have to end the read loop. - if (innerStr === "[DONE]") { + const dataStr = line.substring(5).trim(); + if (dataStr === "[DONE]") { sawDone = true; break; } - if (!innerStr) continue; - - const inner = JSON.parse(innerStr); - if (inner.id) output.responseId = inner.id as string; - if (inner.model) output.responseModel = inner.model as string; - if (inner.usage) { - applyQoderCredits(output.usage, inner.usage, model.provider === "qoder-cn" ? "qoder-cn" : "qoder"); - const u = inner.usage as { - prompt_tokens?: number; - completion_tokens?: number; - total_tokens?: number; - completion_tokens_details?: { reasoning_tokens?: number }; - prompt_tokens_details?: { - cacheable_tokens?: number; - cached_tokens?: number; - cache_write_tokens?: number; - }; - }; - // pi-core computes `promptTokens = input + cacheRead + cacheWrite` - // (Anthropic convention: `input` EXCLUDES cached/written tokens). - // Qoder follows OpenAI semantics where `prompt_tokens` INCLUDES - // `cached_tokens`, so subtract cacheRead (and cache_write_tokens - // when reported) to match the contract pi-ai's own OpenAI - // provider uses. `cacheable_tokens` is a capacity metric, not a - // write count (it is 0 even on first-turn writes), so it is NOT - // mapped to cacheWrite. - const promptTokens = u.prompt_tokens ?? 0; - const cacheReadTokens = u.prompt_tokens_details?.cached_tokens ?? 0; - const cacheWriteTokens = u.prompt_tokens_details?.cache_write_tokens ?? 0; - output.usage.input = Math.max(0, promptTokens - cacheReadTokens - cacheWriteTokens); - output.usage.output = u.completion_tokens ?? 0; - output.usage.totalTokens = u.total_tokens ?? 0; - output.usage.cacheRead = cacheReadTokens; - output.usage.cacheWrite = cacheWriteTokens; - } - if (inner.choices && inner.choices.length > 0) { - const choice = inner.choices[0]; - const delta = choice.delta; - - if (delta) { - // 1. Process reasoning/thinking content (API reasoning) - if (delta.reasoning_content) { - // Qoder's backend sometimes routes a literal `` - // opener into reasoning_content (with the matching - // `` closer landing in the content stream), and - // sometimes leaks unparsed DSML tool-call markup there. Strip - // both so the thinking block stays clean, matching the SDK's - // ContentBlock model. - rawReasoningTail = (rawReasoningTail + delta.reasoning_content).slice(-512); - const reasoningChunk = stripDsmlResidue(stripThinkingTags(delta.reasoning_content)); - if (reasoningChunk) { - if (thinkingBlockIndex === -1) { - thinkingBlockIndex = output.content.length; - output.content.push({ type: "thinking", thinking: "" }); - stream.push({ type: "thinking_start", contentIndex: thinkingBlockIndex, partial: output }); - } - const block = output.content[thinkingBlockIndex] as ThinkingContent; - block.thinking += reasoningChunk; - stream.push({ - type: "thinking_delta", - contentIndex: thinkingBlockIndex, - delta: reasoningChunk, - partial: output, - }); - } - } - // 2. Process text content - if (delta.content) { - // End API thinking block if active - if (thinkingBlockIndex !== -1) { - const block = output.content[thinkingBlockIndex] as ThinkingContent; - stream.push({ - type: "thinking_end", - contentIndex: thinkingBlockIndex, - content: block.thinking, - partial: output, - }); - thinkingBlockIndex = -1; - } + try { + const envelope = JSON.parse(dataStr); + if (envelope.statusCodeValue && envelope.statusCodeValue !== 200) { + throw new Error(`Upstream status ${envelope.statusCodeValue}: ${envelope.body}`); + } - if (thinkingParser) { - thinkingParser.processChunk(delta.content); - } else { - if (contentBlockIndex === -1) { - contentBlockIndex = output.content.length; - output.content.push({ type: "text", text: "" }); - stream.push({ type: "text_start", contentIndex: contentBlockIndex, partial: output }); + const innerStr = envelope.body; + // The gateway sends the sentinel wrapped in an envelope + // (`body: "[DONE]"`) as well as bare, and both mean the reply is + // over, so both have to end the read loop. + if (innerStr === "[DONE]") { + sawDone = true; + break; + } + if (!innerStr) continue; + + const inner = JSON.parse(innerStr); + if (inner.id) output.responseId = inner.id as string; + if (inner.model) output.responseModel = inner.model as string; + if (inner.usage) { + applyQoderCredits(output.usage, inner.usage, model.provider === "qoder-cn" ? "qoder-cn" : "qoder"); + const u = inner.usage as { + prompt_tokens?: number; + completion_tokens?: number; + total_tokens?: number; + completion_tokens_details?: { reasoning_tokens?: number }; + prompt_tokens_details?: { + cacheable_tokens?: number; + cached_tokens?: number; + cache_write_tokens?: number; + }; + }; + // pi-core computes `promptTokens = input + cacheRead + cacheWrite` + // (Anthropic convention: `input` EXCLUDES cached/written tokens). + // Qoder follows OpenAI semantics where `prompt_tokens` INCLUDES + // `cached_tokens`, so subtract cacheRead (and cache_write_tokens + // when reported) to match the contract pi-ai's own OpenAI + // provider uses. `cacheable_tokens` is a capacity metric, not a + // write count (it is 0 even on first-turn writes), so it is NOT + // mapped to cacheWrite. + const promptTokens = u.prompt_tokens ?? 0; + const cacheReadTokens = u.prompt_tokens_details?.cached_tokens ?? 0; + const cacheWriteTokens = u.prompt_tokens_details?.cache_write_tokens ?? 0; + output.usage.input = Math.max(0, promptTokens - cacheReadTokens - cacheWriteTokens); + output.usage.output = u.completion_tokens ?? 0; + output.usage.totalTokens = u.total_tokens ?? 0; + output.usage.cacheRead = cacheReadTokens; + output.usage.cacheWrite = cacheWriteTokens; + } + if (inner.choices && inner.choices.length > 0) { + const choice = inner.choices[0]; + const delta = choice.delta; + + if (delta) { + // 1. Process reasoning/thinking content (API reasoning) + if (delta.reasoning_content) { + // Qoder's backend sometimes routes a literal `` + // opener into reasoning_content (with the matching + // `` closer landing in the content stream), and + // sometimes leaks unparsed DSML tool-call markup there. Strip + // both so the thinking block stays clean, matching the SDK's + // ContentBlock model. + rawReasoningTail = (rawReasoningTail + delta.reasoning_content).slice(-512); + const reasoningChunk = stripDsmlResidue(stripThinkingTags(delta.reasoning_content)); + if (reasoningChunk) { + if (thinkingBlockIndex === -1) { + thinkingBlockIndex = output.content.length; + output.content.push({ type: "thinking", thinking: "" }); + stream.push({ type: "thinking_start", contentIndex: thinkingBlockIndex, partial: output }); + } + const block = output.content[thinkingBlockIndex] as ThinkingContent; + block.thinking += reasoningChunk; + stream.push({ + type: "thinking_delta", + contentIndex: thinkingBlockIndex, + delta: reasoningChunk, + partial: output, + }); } - const block = output.content[contentBlockIndex] as TextContent; - block.text += delta.content; - stream.push({ - type: "text_delta", - contentIndex: contentBlockIndex, - delta: delta.content, - partial: output, - }); } - } - - // 3. Process tool calls - if (delta.tool_calls && Array.isArray(delta.tool_calls)) { - for (const tc of delta.tool_calls) { - const idx = tc.index ?? 0; - if (!toolCallsState[idx]) { - toolCallsState[idx] = { arguments: "", id: "", name: "", contentIndex: 0 }; - } - const state = toolCallsState[idx]; - if (tc.id) state.id = tc.id; - if (tc.function?.name) state.name = tc.function.name; - - // Open the block as soon as the call is IDENTIFIABLE, not - // when its first argument byte arrives. A call whose - // arguments are absent or an empty string — a no-argument - // tool, or a model that sends id+name and then stops — used - // to create a toolCallsState entry and no content block, so - // the finalizer below saw a non-empty state array, set - // stopReason "toolUse", and handed back a message with no - // tool call in it. The agent loop then had nothing to run - // and the turn simply ended, mid-task and without an error. - if (state.emittedStart === undefined && (state.id || state.name)) { - state.emittedStart = true; - state.contentIndex = output.content.length; - output.content.push({ - type: "toolCall", - id: state.id, - name: state.name, - arguments: {}, - } satisfies ToolCall); - stream.push({ type: "toolcall_start", contentIndex: state.contentIndex, partial: output }); - } - // id and name can arrive after the block is open; keep it - // in step, since the finalizer only rewrites `arguments`. - if (state.emittedStart) { - const block = output.content[state.contentIndex] as ToolCall; - block.id = state.id; - block.name = state.name; + // 2. Process text content + if (delta.content) { + // End API thinking block if active + if (thinkingBlockIndex !== -1) { + const block = output.content[thinkingBlockIndex] as ThinkingContent; + stream.push({ + type: "thinking_end", + contentIndex: thinkingBlockIndex, + content: block.thinking, + partial: output, + }); + thinkingBlockIndex = -1; } - if (tc.function?.arguments) { - const argDelta = tc.function.arguments; - state.arguments += argDelta; + if (thinkingParser) { + thinkingParser.processChunk(delta.content); + refreshToolIndexes(); + } else { + if (contentBlockIndex === -1) { + contentBlockIndex = output.content.length; + output.content.push({ type: "text", text: "" }); + stream.push({ type: "text_start", contentIndex: contentBlockIndex, partial: output }); + } + const block = output.content[contentBlockIndex] as TextContent; + block.text += delta.content; stream.push({ - type: "toolcall_delta", - contentIndex: state.contentIndex, - delta: argDelta, + type: "text_delta", + contentIndex: contentBlockIndex, + delta: delta.content, partial: output, }); } } + + // 3. Process tool calls + if (delta.tool_calls && Array.isArray(delta.tool_calls)) { + for (const tc of delta.tool_calls) { + const id = tc.id || ""; + const name = tc.function?.name || ""; + const argDelta = tc.function?.arguments ?? ""; + if (typeof id !== "string" || typeof name !== "string") { + throw new Error("Tool-call identity conflict: invalid metadata"); + } + if (typeof argDelta !== "string") throw new Error("Invalid tool-call argument fragment"); + if (!id && !name && !argDelta) continue; + + const index = typeof tc.index === "number" ? tc.index : undefined; + let candidates: ToolCallCandidates | undefined; + if (index !== undefined) { + candidates = toolCandidatesByIndex.get(index); + if (!candidates) { + candidates = { states: new Set(), complete: new Set() }; + toolCandidatesByIndex.set(index, candidates); + } + } + let state = id ? toolCallsById.get(id) : undefined; + if (id && candidates && (!state || !candidates.states.has(state))) { + retireCompleteCandidates(candidates); + } + const available = candidates?.states ?? unscopedToolCandidates; + if (!state) { + const only = available.size === 1 ? available.values().next().value : undefined; + if (!id && available.size > 1) { + throw new Error(`Ambiguous tool-call continuation (candidates=${available.size})`); + } + // An explicit new ID never overwrites another identity. + // Only a unique, still-unidentified provisional can bind. + if (only && (!id || !only.id)) state = only; + else { + state = { + argumentChunks: [], + argumentLength: 0, + depth: 0, + inString: false, + escaped: false, + complete: false, + id: "", + name: "", + contentIndex: -1, + candidates: new Set(), + unindexed: index === undefined, + }; + toolCallsState.push(state); + } + } + if (name && state.name && name !== state.name) { + throw new Error("Tool-call identity conflict: inconsistent name"); + } + const newlyIdentified = !!id && !state.id; + if (id) { + state.id = id; + toolCallsById.set(id, state); + } + if (name) state.name = name; + unscopedToolCandidates.add(state); + if (candidates) { + state.unindexed = false; + state.candidates.add(candidates); + candidates.states.add(state); + if (state.complete && state.id) candidates.complete.add(state); + } + + const wasComplete = state.complete; + if (argDelta) appendToolArguments(state, argDelta); + if (state.complete && state.id && (!wasComplete || newlyIdentified)) { + // Each association is registered at most once on closure, + // not revisited on every fragment or repeated ID. + for (const scope of state.candidates) scope.complete.add(state); + } + + if (!state.block && (state.id || state.name)) { + state.block = { type: "toolCall", id: state.id, name: state.name, arguments: {} }; + state.contentIndex = output.content.length; + output.content.push(state.block); + toolStatesByBlock.set(state.block, state); + lastToolState = state; + stream.push({ type: "toolcall_start", contentIndex: state.contentIndex, partial: output }); + // Argument-only provisional fragments have no content + // position until this block exists. Replay them in order. + for (const fragment of state.argumentChunks) { + stream.push({ + type: "toolcall_delta", + contentIndex: state.contentIndex, + delta: fragment, + partial: output, + }); + } + } else if (state.block) { + state.block.id = state.id; + state.block.name = state.name; + if (argDelta) { + stream.push({ + type: "toolcall_delta", + contentIndex: state.contentIndex, + delta: argDelta, + partial: output, + }); + } + } + } + } } - } - if (choice.finish_reason) { - // Preserve the real upstream finish_reason (e.g. "length", - // "content_filter") instead of forcing "stop" later. - output.stopReason = choice.finish_reason as AssistantMessage["stopReason"]; + if (choice.finish_reason) { + // Preserve the real upstream finish_reason (e.g. "length", + // "content_filter") instead of forcing "stop" later. + output.stopReason = choice.finish_reason as AssistantMessage["stopReason"]; + } } - } - } catch (e) { - // A single malformed SSE line shouldn't kill the stream — skip it. - // But a genuine upstream error (thrown below) must propagate to the - // outer catch and surface as stopReason="error", not be swallowed. - if (e instanceof SyntaxError) { - if (process.env.QODER_DEBUG) { - console.error("[pi-provider-qoder] skipping malformed SSE line:", dataStr.slice(0, 200)); + } catch (e) { + // A single malformed SSE line shouldn't kill the stream — skip it. + // But a genuine upstream error (thrown below) must propagate to the + // outer catch and surface as stopReason="error", not be swallowed. + if (e instanceof SyntaxError) { + if (process.env.QODER_DEBUG) { + console.error("[pi-provider-qoder] skipping malformed SSE line:", dataStr.slice(0, 200)); + } + continue; } - continue; + throw e; } - throw e; } } + } finally { + // Release on DONE, abort, and identity/argument failures alike. + try { + await reader.cancel().catch(() => {}); + } finally { + reader.releaseLock(); + } } - // Stop reading and let the connection go once the reply is complete. - // Without this the body stays open until the server times it out. - await reader.cancel().catch(() => {}); - if (thinkingParser) { thinkingParser.finalize(); + refreshToolIndexes(); } if (thinkingBlockIndex !== -1) { @@ -656,33 +835,26 @@ export function streamQoder( }); } + // Validate the whole batch before publishing any successful end event. for (const state of toolCallsState) { - if (state?.emittedStart && !state.emittedEnd) { - state.emittedEnd = true; - let args = {}; - try { - args = JSON.parse(state.arguments || "{}"); - } catch {} - const block = output.content[state.contentIndex] as ToolCall; - block.arguments = args; - stream.push({ - type: "toolcall_end", - contentIndex: state.contentIndex, - toolCall: { - type: "toolCall", - id: state.id, - name: state.name, - arguments: args, - }, - partial: output, - }); + if (!state.id || !state.name) { + throw new Error(`Unresolved tool-call identity (characters=${state.argumentLength})`); } + parseToolArguments(state); + } + // Map insertion order is the order calls first became identifiable, + // independent of sparse/reused upstream indexes and provisional buffers. + for (const [block, state] of toolStatesByBlock) { + block.arguments = parseToolArguments(state); + stream.push({ + type: "toolcall_end", + contentIndex: state.contentIndex, + toolCall: block, + partial: output, + }); } - // Guarded on blocks that actually reached the message, not on the state - // array being non-empty. Claiming "toolUse" for a message carrying no - // tool call is what turned a malformed stream into a silent dead end. - if (toolCallsState.some((state) => state?.emittedStart)) { + if (toolStatesByBlock.size > 0 && output.stopReason !== "length") { output.stopReason = "toolUse"; } // A turn the gateway stripped the tool call out of has no text and no tool From 4f9a400902d3b0260cd314329de36147b00708dc Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 09:12:52 +0800 Subject: [PATCH 4/7] test: avoid redundant SSE fixture encoding --- src/__tests__/stream.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/__tests__/stream.test.ts b/src/__tests__/stream.test.ts index 2380266..30b0ef8 100644 --- a/src/__tests__/stream.test.ts +++ b/src/__tests__/stream.test.ts @@ -636,16 +636,17 @@ describe("streamQoder response aggregation contract", () => { return push.call(this, event); }); const frames = [...deltas.map((delta) => sseEnvelope(chunk(delta))), sseEnvelope(finishChunk(finish)), DONE_SSE]; - const bytes = new TextEncoder().encode(frames.join("")); + const encoder = new TextEncoder(); let cancelled = false; const body = new ReadableStream({ start(controller) { if (byteSize) { + const bytes = encoder.encode(frames.join("")); for (let offset = 0; offset < bytes.length; offset += byteSize) { controller.enqueue(bytes.slice(offset, offset + byteSize)); } } else { - for (const frame of frames) controller.enqueue(new TextEncoder().encode(frame)); + for (const frame of frames) controller.enqueue(encoder.encode(frame)); } // Leave open: DONE and all error paths must release the reader. }, From d8a3f6d12942406e9fe6c3f4802f8e566a872815 Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 09:45:59 +0800 Subject: [PATCH 5/7] test: guard tool candidate lifecycle during cache optimization --- src/__tests__/stream.test.ts | 91 ++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/src/__tests__/stream.test.ts b/src/__tests__/stream.test.ts index 30b0ef8..41631f0 100644 --- a/src/__tests__/stream.test.ts +++ b/src/__tests__/stream.test.ts @@ -910,6 +910,97 @@ describe("streamQoder response aggregation contract", () => { failure(await run([{ tool_calls: [tool(0, a.id, a.name, '{"path":')] }], "length"), /json|argument|incomplete/i); }); + it("C12 C05 initializes provisional arguments after empty placeholders and tag thinking", async () => { + const result = await run([ + { tool_calls: [tool(0)] }, + { content: "initial thoughtready" }, + { tool_calls: [tool(0, undefined, undefined, '{"path":')] }, + { tool_calls: [tool(0, a.id, a.name, '"/a"}')] }, + ]); + success(result, [a]); + const done = result.events.find((event) => event.type === "done"); + expect(done?.message.content.filter((block) => block.type !== "toolCall")).toEqual([ + { type: "thinking", thinking: "initial thought" }, + { type: "text", text: "ready" }, + ]); + const deltas = result.snapshots.filter((event) => event.type === "toolcall_delta"); + expect(deltas.map((event) => event.delta).join("")).toBe('{"path":"/a"}'); + }); + + it("C12 C06 rejects unresolved argument-only state after an empty placeholder", async () => { + const result = await run([{ tool_calls: [tool(0)] }, { tool_calls: [tool(0, undefined, undefined, "{}")] }]); + failure(result, /identity|unresolved|missing.*id/i); + expect( + result.snapshots.filter((event) => event.type === "toolcall_start" || event.type === "toolcall_delta"), + ).toEqual([]); + }); + + it("C07 C02 restores a unique index candidate after retiring completed interleaved calls", async () => { + const c = { id: "synthetic_c", name: "lookup", arguments: { path: "/c" } }; + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":'), tool(0, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(0, a.id, undefined, '"/a"}'), tool(0, b.id, undefined, '"/b"}')] }, + { tool_calls: [tool(0, c.id, c.name, '{"path":')] }, + { tool_calls: [tool(0, undefined, undefined, '"/c"}')] }, + ]), + [a, b, c], + ); + }); + + it("C04 C09 re-registers a retired ID before subsequent index and global continuations", async () => { + const c = { id: "synthetic_c", name: "lookup", arguments: { path: "/c" } }; + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":"/a"}')] }, + { tool_calls: [tool(0, b.id, b.name, '{"path":"/b"}')] }, + { tool_calls: [tool(0, a.id, undefined, " ")] }, + { tool_calls: [tool(0, c.id, c.name, '{"path":')] }, + { tool_calls: [tool(0, undefined, undefined, '"/c"}')] }, + { tool_calls: [tool(undefined, undefined, undefined, "\n")] }, + ]), + [a, b, c], + ); + }); + + it("C05 C02 retires an already-closed provisional after its ID arrives", async () => { + success( + await run([ + { tool_calls: [tool(0, undefined, undefined, '{"path":"/a"}')] }, + { tool_calls: [tool(0, a.id, a.name)] }, + { tool_calls: [tool(0, b.id, b.name, '{"path":')] }, + { tool_calls: [tool(0, undefined, undefined, '"/b"}')] }, + ]), + [a, b], + ); + }); + + it("C04 C05 binds a provisional uniquely after retiring a completed alias from its index", async () => { + success( + await run([ + { tool_calls: [tool(0, a.id, a.name, '{"path":"/a"}')] }, + { tool_calls: [tool(1, undefined, undefined, '{"path":')] }, + { tool_calls: [tool(1, a.id, undefined, " ")] }, + { tool_calls: [tool(1, b.id, b.name)] }, + { tool_calls: [tool(1, undefined, undefined, '"/b"}')] }, + ]), + [a, b], + ); + }); + + it("C04 C09 keeps a globally reactivated retired ID in ambiguity checks", async () => { + failure( + await run([ + { tool_calls: [tool(0, a.id, a.name, "{}")] }, + { tool_calls: [tool(0, b.id, b.name, "{}")] }, + { tool_calls: [tool(undefined, a.id, undefined, " ")] }, + { tool_calls: [tool(0, "synthetic_c", "lookup", "{}")] }, + { tool_calls: [tool(undefined, undefined, undefined, " ")] }, + ]), + /ambiguous tool-call continuation/i, + ); + }); + it("C15 aborts an in-progress tool response and releases its reader", async () => { const controller = new AbortController(); let cancelled = false; From 2d6849e957ebe0483bb93a33ec8dcfa002db2b8b Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 09:58:36 +0800 Subject: [PATCH 6/7] perf: initialize tool aggregation lazily and reuse stable links --- src/protocol/stream.ts | 179 ++++++++++++++++++++++++----------------- 1 file changed, 104 insertions(+), 75 deletions(-) diff --git a/src/protocol/stream.ts b/src/protocol/stream.ts index 5c08896..89fda8f 100644 --- a/src/protocol/stream.ts +++ b/src/protocol/stream.ts @@ -27,6 +27,7 @@ import { transformMessagesForQoder, transformTools } from "./transform.js"; interface ToolCallCandidates { states: Set; complete: Set; + unique: ToolCallState | undefined; } interface ToolCallState { @@ -46,6 +47,40 @@ interface ToolCallState { unindexed: boolean; } +interface ToolCallAggregation { + states: ToolCallState[]; + byId: Map; + byIndex: Map; + unscopedCandidates: Set; + statesByBlock: Map; + lastToolState?: ToolCallState; +} + +function refreshToolIndexes(output: AssistantMessage, toolCalls: ToolCallAggregation): void { + const lastToolState = toolCalls.lastToolState; + if (!lastToolState || output.content[lastToolState.contentIndex] === lastToolState.block) return; + // ThinkingTagParser.emitThinking() can splice only when creating its first + // thinking block. All other layout changes append. Refresh in one scan. + for (let index = 0; index < output.content.length; index++) { + const block = output.content[index]; + if (block.type !== "toolCall") continue; + const state = toolCalls.statesByBlock.get(block); + if (state) state.contentIndex = index; + } +} + +function retireCompleteCandidates(toolCalls: ToolCallAggregation, candidates: ToolCallCandidates): void { + for (const state of candidates.complete) { + // Structural closure alone is not sufficient evidence of a boundary. + parseToolArguments(state); + candidates.states.delete(state); + state.candidates.delete(candidates); + if (state.candidates.size === 0 && !state.unindexed) toolCalls.unscopedCandidates.delete(state); + } + candidates.complete.clear(); + candidates.unique = candidates.states.size === 1 ? candidates.states.values().next().value : undefined; +} + const TOOL_ARGUMENT_STRUCTURE = /["\\{}[\]]/g; const NON_JSON_WHITESPACE = /[^ \t\r\n]/; @@ -496,36 +531,7 @@ export function streamQoder( // tail matters (the degenerate-turn check looks for trailing markup), so // this stays bounded instead of growing with the reply. let rawReasoningTail = ""; - const toolCallsState: ToolCallState[] = []; - const toolCallsById = new Map(); - const toolCandidatesByIndex = new Map(); - const unscopedToolCandidates = new Set(); - const toolStatesByBlock = new Map(); - let lastToolState: ToolCallState | undefined; - - const refreshToolIndexes = () => { - if (!lastToolState || output.content[lastToolState.contentIndex] === lastToolState.block) return; - // ThinkingTagParser.emitThinking() can splice only when creating its - // first thinking block. All other layout changes append. Check the - // last tool in O(1), then refresh the shifted suffix in one scan. - for (let index = 0; index < output.content.length; index++) { - const block = output.content[index]; - if (block.type !== "toolCall") continue; - const state = toolStatesByBlock.get(block); - if (state) state.contentIndex = index; - } - }; - - const retireCompleteCandidates = (candidates: ToolCallCandidates) => { - for (const state of candidates.complete) { - // Structural closure alone is not sufficient evidence of a boundary. - parseToolArguments(state); - candidates.states.delete(state); - state.candidates.delete(candidates); - if (state.candidates.size === 0 && !state.unindexed) unscopedToolCandidates.delete(state); - } - candidates.complete.clear(); - }; + let toolCalls: ToolCallAggregation | undefined; const thinkingEnabled = (options?.reasoning as unknown) !== false && (options?.reasoning as unknown) !== "off"; const thinkingParser = thinkingEnabled ? new ThinkingTagParser(output, stream) : null; @@ -664,7 +670,7 @@ export function streamQoder( if (thinkingParser) { thinkingParser.processChunk(delta.content); - refreshToolIndexes(); + if (toolCalls?.lastToolState) refreshToolIndexes(output, toolCalls); } else { if (contentBlockIndex === -1) { contentBlockIndex = output.content.length; @@ -694,29 +700,44 @@ export function streamQoder( if (typeof argDelta !== "string") throw new Error("Invalid tool-call argument fragment"); if (!id && !name && !argDelta) continue; + toolCalls ??= { + states: [], + byId: new Map(), + byIndex: new Map(), + unscopedCandidates: new Set(), + statesByBlock: new Map(), + }; const index = typeof tc.index === "number" ? tc.index : undefined; let candidates: ToolCallCandidates | undefined; if (index !== undefined) { - candidates = toolCandidatesByIndex.get(index); + candidates = toolCalls.byIndex.get(index); if (!candidates) { - candidates = { states: new Set(), complete: new Set() }; - toolCandidatesByIndex.set(index, candidates); + candidates = { states: new Set(), complete: new Set(), unique: undefined }; + toolCalls.byIndex.set(index, candidates); } } - let state = id ? toolCallsById.get(id) : undefined; - if (id && candidates && (!state || !candidates.states.has(state))) { - retireCompleteCandidates(candidates); + let state = id ? toolCalls.byId.get(id) : undefined; + let associated = !!state && !!candidates && candidates.states.has(state); + if (id && candidates && !associated && candidates.complete.size > 0) { + retireCompleteCandidates(toolCalls, candidates); } - const available = candidates?.states ?? unscopedToolCandidates; + const available = candidates?.states ?? toolCalls.unscopedCandidates; if (!state) { - const only = available.size === 1 ? available.values().next().value : undefined; + const only = + available.size === 1 + ? candidates + ? candidates.unique + : available.values().next().value + : undefined; if (!id && available.size > 1) { throw new Error(`Ambiguous tool-call continuation (candidates=${available.size})`); } // An explicit new ID never overwrites another identity. // Only a unique, still-unidentified provisional can bind. - if (only && (!id || !only.id)) state = only; - else { + if (only && (!id || !only.id)) { + state = only; + associated = candidates !== undefined; + } else { state = { argumentChunks: [], argumentLength: 0, @@ -730,24 +751,30 @@ export function streamQoder( candidates: new Set(), unindexed: index === undefined, }; - toolCallsState.push(state); + toolCalls.states.push(state); } } if (name && state.name && name !== state.name) { throw new Error("Tool-call identity conflict: inconsistent name"); } const newlyIdentified = !!id && !state.id; - if (id) { + const newlyNamed = !!name && !state.name; + if (newlyIdentified) { state.id = id; - toolCallsById.set(id, state); + toolCalls.byId.set(id, state); } - if (name) state.name = name; - unscopedToolCandidates.add(state); - if (candidates) { - state.unindexed = false; - state.candidates.add(candidates); - candidates.states.add(state); - if (state.complete && state.id) candidates.complete.add(state); + if (newlyNamed) state.name = name; + if (!associated) { + // Indexed continuations already have all three links. + // Unindexed explicit IDs can reactivate retired states. + toolCalls.unscopedCandidates.add(state); + if (candidates) { + state.unindexed = false; + state.candidates.add(candidates); + candidates.states.add(state); + candidates.unique = candidates.states.size === 1 ? state : undefined; + if (state.complete && state.id) candidates.complete.add(state); + } } const wasComplete = state.complete; @@ -762,8 +789,8 @@ export function streamQoder( state.block = { type: "toolCall", id: state.id, name: state.name, arguments: {} }; state.contentIndex = output.content.length; output.content.push(state.block); - toolStatesByBlock.set(state.block, state); - lastToolState = state; + toolCalls.statesByBlock.set(state.block, state); + toolCalls.lastToolState = state; stream.push({ type: "toolcall_start", contentIndex: state.contentIndex, partial: output }); // Argument-only provisional fragments have no content // position until this block exists. Replay them in order. @@ -776,8 +803,8 @@ export function streamQoder( }); } } else if (state.block) { - state.block.id = state.id; - state.block.name = state.name; + if (newlyIdentified) state.block.id = state.id; + if (newlyNamed) state.block.name = state.name; if (argDelta) { stream.push({ type: "toolcall_delta", @@ -822,7 +849,7 @@ export function streamQoder( if (thinkingParser) { thinkingParser.finalize(); - refreshToolIndexes(); + if (toolCalls?.lastToolState) refreshToolIndexes(output, toolCalls); } if (thinkingBlockIndex !== -1) { @@ -835,27 +862,29 @@ export function streamQoder( }); } - // Validate the whole batch before publishing any successful end event. - for (const state of toolCallsState) { - if (!state.id || !state.name) { - throw new Error(`Unresolved tool-call identity (characters=${state.argumentLength})`); + if (toolCalls) { + // Validate the whole batch before publishing any successful end event. + for (const state of toolCalls.states) { + if (!state.id || !state.name) { + throw new Error(`Unresolved tool-call identity (characters=${state.argumentLength})`); + } + parseToolArguments(state); + } + // Map insertion order is the order calls first became identifiable, + // independent of sparse/reused upstream indexes and provisional buffers. + for (const [block, state] of toolCalls.statesByBlock) { + block.arguments = parseToolArguments(state); + stream.push({ + type: "toolcall_end", + contentIndex: state.contentIndex, + toolCall: block, + partial: output, + }); } - parseToolArguments(state); - } - // Map insertion order is the order calls first became identifiable, - // independent of sparse/reused upstream indexes and provisional buffers. - for (const [block, state] of toolStatesByBlock) { - block.arguments = parseToolArguments(state); - stream.push({ - type: "toolcall_end", - contentIndex: state.contentIndex, - toolCall: block, - partial: output, - }); - } - if (toolStatesByBlock.size > 0 && output.stopReason !== "length") { - output.stopReason = "toolUse"; + if (toolCalls.statesByBlock.size > 0 && output.stopReason !== "length") { + output.stopReason = "toolUse"; + } } // A turn the gateway stripped the tool call out of has no text and no tool // call, so `stop` reads as a finished task: the agent loop ends the turn From 65bcb1047e84bece7ecc5ad6f6d8f783649e0f65 Mon Sep 17 00:00:00 2001 From: summer-tt Date: Tue, 20 Oct 2026 15:08:26 +0800 Subject: [PATCH 7/7] test: fix SDK flow isolation and required empty arguments --- scripts/verify-tool-call-flow.ts | 35 +++++++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/scripts/verify-tool-call-flow.ts b/scripts/verify-tool-call-flow.ts index d3bb31f..f7bafc6 100644 --- a/scripts/verify-tool-call-flow.ts +++ b/scripts/verify-tool-call-flow.ts @@ -54,7 +54,16 @@ Object.assign(process.env, { TMPDIR: path.join(temporary, "tmp"), JITI_FS_CACHE: "false", }); -for (const directory of [...Object.values(process.env), path.join(temporary, "workspace")]) { +for (const directory of [ + process.env.HOME, + process.env.XDG_CONFIG_HOME, + process.env.XDG_CACHE_HOME, + process.env.XDG_DATA_HOME, + process.env.XDG_STATE_HOME, + process.env.PI_CODING_AGENT_DIR, + process.env.TMPDIR, + path.join(temporary, "workspace"), +]) { fs.mkdirSync(directory!, { recursive: true, mode: 0o700 }); } process.chdir(path.join(temporary, "workspace")); @@ -205,6 +214,18 @@ const fixtures: Fixture[] = [ expected: [], providerError: true, }, + { + name: "required-path-zero-bytes-sdk-error", + deltas: [delta(0, "required_empty_bytes_A", "fixture_read", "")], + expected: [{ id: "required_empty_bytes_A", name: "fixture_read", args: {} }], + schemaError: true, + }, + { + name: "required-path-explicit-empty-object-sdk-error", + deltas: [delta(0, "required_empty_object_A", "fixture_read", "{}")], + expected: [{ id: "required_empty_object_A", name: "fixture_read", args: {} }], + schemaError: true, + }, ]; function envelope(body: unknown): string { return `data:${JSON.stringify({ @@ -464,6 +485,18 @@ try { requests[1].messages.filter((m) => m.role === "tool").map((m) => m.content), fixture.expected.map((c) => `RESULT:${c.id}:${JSON.stringify(c.args)}`), ); + else { + const errorTexts = results.map((result) => + result.content.filter((content) => content.type === "text").map((content) => content.text).join(""), + ); + assert(errorTexts.every((text, index) => + text.startsWith(`Validation failed for tool "${fixture.expected[index].name}":`), + )); + assert.deepEqual( + requests[1].messages.filter((m) => m.role === "tool").map((m) => m.content), + errorTexts, + ); + } assert.equal(assistants.at(-1)?.stopReason, "stop"); assert.equal(session.getLastAssistantText(), "SYNTHETIC_DONE"); }