diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 1b6b93ef9e..881678f41b 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -11,6 +11,7 @@ ## [0.14.1] - 2026-08-18 - Compaction pruning no longer kills the turn when a persisted `toolCall.arguments` is `null`. Sessions written by an earlier cold-spill eviction path store `null` where the spill sentinel belongs, and the staleness index dereferenced that payload unguarded, so reloading such a session threw `null is not an object (evaluating 'args.path')` as a turn-fatal error instead of skipping the one unusable call. `ToolCall.arguments` is typed non-nullable, so no type check flagged the gap. Every read of a persisted argument bag — path extraction, `apply_patch` header parsing, idempotent-bash keys, and search target keys — now treats a non-object payload as absent. The original arguments are not lost: the eviction marker still names the blob and rehydration restores them. +- Managed fallback attempt snapshots no longer fail the whole run on benign provider shape variations: an assistant message whose `content` is a bare string, is missing, or is a primitive scalar (null/number/boolean) now degrades to an empty content array; staged `*_delta`/`*_end` events whose `delta`/`content` is missing or a primitive scalar degrade to an empty string; and staged assistant events with out-of-vocabulary `done`/`error` reasons or an unknown string `type` degrade to schema-valid values instead of throwing a non-retryable `ManagedAttemptSnapshotError`. Object-shaped or other plain-object `content`/`delta` stays fail-closed under the named `shell.content`/`event.delta`/`event.content` diagnostic, as does sanitizer-sentinel string content (`[unserializable]`/`[accessor]`/`[truncated]`/`[Circular]`, which marks a non-cloneable original rather than provider string variance — degrading those would silently drop real tool-call or streamed content behind a successful empty turn), and hostile inputs keep failing fast with no retry authority: a live proxy root, a throwing `get`/`getOwnPropertyDescriptor` trap, and a non-string event `type` all remain local snapshot failures. ### Added diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index c7b0498912..c400d5a42d 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -735,6 +735,26 @@ const SANITIZER_SENTINELS: ReadonlySet = new Set([ "[truncated]", "[Circular]", ]); +/** + * Bounded diagnostic for a degraded primitive at the shared managed-snapshot + * boundary. Every provider/custom stream that still forwards a malformed + * primitive increment degrades here (to "" / []), so the degradation stays + * observable. The caller supplies a run-scoped set so repeated malformed + * increments emit at most one payload-free warning per field name, naming + * only the field and the received typeof — never the payload. + */ +function warnManagedDegradedPrimitive( + field: string, + received: unknown, + diagnostics: Set = new Set(), +): void { + if (diagnostics.has(field)) return; + diagnostics.add(field); + logger.warn("agent: managed snapshot degraded a non-string primitive to an empty value", { + field, + receivedType: received === null ? "null" : typeof received, + }); +} /** * Cycle-aware deep clone that always returns a detached, JSON-serializable @@ -887,16 +907,16 @@ function managedSnapshotJsonBytes(value: unknown): number | undefined { } } -function managedAttemptSnapshotDetailed(value: T): { snapshot: T; jsonBytes?: number } { +function managedAttemptSnapshotDetailed(value: T): { snapshot: T; jsonBytes?: number; sanitized: boolean } { try { const snapshot = structuredClone(value); const jsonBytes = managedSnapshotJsonBytes(snapshot); - if (jsonBytes !== undefined) return { snapshot, jsonBytes }; + if (jsonBytes !== undefined) return { snapshot, jsonBytes, sanitized: false }; const sanitized = sanitizedDetachedClone(snapshot); - return { snapshot: sanitized, jsonBytes: managedSnapshotJsonBytes(sanitized) }; + return { snapshot: sanitized, jsonBytes: managedSnapshotJsonBytes(sanitized), sanitized: true }; } catch { const snapshot = sanitizedDetachedClone(value); - return { snapshot, jsonBytes: managedSnapshotJsonBytes(snapshot) }; + return { snapshot, jsonBytes: managedSnapshotJsonBytes(snapshot), sanitized: true }; } } @@ -997,7 +1017,11 @@ function losslessDetachedClone(value: T): T { * are read, and executable content is retained only when it has its complete * discriminant shape. */ -function managedAssistantShell(value: unknown, model: AgentLoopConfig["model"]): AssistantMessage { +function managedAssistantShell( + value: unknown, + model: AgentLoopConfig["model"], + degradedFieldDiagnostics: Set = new Set(), +): AssistantMessage { const detailed = managedAttemptSnapshotDetailed(value); const snapshotRecord = isManagedPlainRecord(detailed.snapshot) ? detailed.snapshot : undefined; // Two benign root degradations are repaired by reading through the @@ -1015,22 +1039,26 @@ function managedAssistantShell(value: unknown, model: AgentLoopConfig["model"]): snapshotRecord !== undefined && managedProperty(snapshotRecord, "role") === "assistant" ? snapshotRecord : value; if (managedProperty(source, "role") !== "assistant") throw new ManagedAttemptSnapshotError("shell.role"); const rawContent = managedAttemptSnapshot(managedProperty(source, "content")); - // Benign providers occasionally deliver a string or missing content value. - // Degrade those to an empty content array — an empty assistant turn — - // instead of failing the whole managed run: the staged shell must stay - // schema-valid, and empty content is the neutral, side-effect-free - // degradation. A string is benign ONLY when the provider actually sent a - // string: when the whole-message snapshot degraded, the sanitizer replaces - // a non-cloneable content value (proxy, function, accessor) with one of - // its own sentinel strings, and mistaking that sentinel for provider - // variance would silently drop real content (tool calls) behind a - // successful empty turn. Sentinel-string content therefore stays - // fail-closed, as does every other non-array shape, so the named-site - // diagnostic can report shell.content. + // Providers may deliver `content` as a string, missing value, or a primitive + // scalar — all benign variance that degrades to an empty content array + // (an empty, side-effect-free assistant turn). A plain-object `content` + // is NOT degraded: it can carry array-like toolCall payloads + // (`{0:{type:"toolCall"}}`) and silently dropping them would lose + // executable content behind a successful empty turn. Only sentinel + // strings produced by the sanitizer itself (`[unserializable]` etc.) + // also stay fail-closed for the same reason, plus any plain object. const rawArray = Array.isArray(rawContent) ? rawContent : undefined; - const benignContent = - rawContent === undefined || (typeof rawContent === "string" && !SANITIZER_SENTINELS.has(rawContent)); - if (rawArray === undefined && !benignContent) throw new ManagedAttemptSnapshotError("shell.content"); + if (rawArray === undefined) { + if (typeof rawContent === "string" && SANITIZER_SENTINELS.has(rawContent)) { + throw new ManagedAttemptSnapshotError("shell.content"); + } + if (rawContent !== null && typeof rawContent === "object") { + throw new ManagedAttemptSnapshotError("shell.content"); + } + if (rawArray === undefined && rawContent !== undefined && !SANITIZER_SENTINELS.has(rawContent as string)) { + warnManagedDegradedPrimitive("shell.content", rawContent, degradedFieldDiagnostics); + } + } const content = rawArray === undefined ? [] : rawArray.flatMap(managedContentBlock); const usage = managedAssistantUsage(managedAttemptSnapshot(managedProperty(source, "usage"))); const api = managedProperty(source, "api"); @@ -1153,8 +1181,45 @@ function managedAssistantUsage(value: unknown): AssistantMessage["usage"] { export function managedAssistantEventSnapshot( event: AssistantMessageEvent, message: AssistantMessage, + degradedFieldDiagnostics: Set = new Set(), ): AssistantMessageEvent { - const detached = managedAttemptSnapshot(event); + const directType = managedProperty(event, "type"); + if ( + directType === "text_delta" || + directType === "thinking_delta" || + directType === "reasoning_summary_delta" || + directType === "toolcall_delta" + ) { + // Delta events are snapshotted field-by-field so unrelated event metadata + // cannot erase provenance. The delta is read once, detached once, then the + // same captured value is both validated and emitted. + const contentIndex = managedAttemptSnapshot(managedProperty(event, "contentIndex")); + if (!Number.isInteger(contentIndex) || (contentIndex as number) < 0) { + throw new ManagedAttemptSnapshotError("event.contentIndex"); + } + const deltaSnapshot = managedAttemptSnapshotDetailed(managedProperty(event, "delta")); + const delta = deltaSnapshot.snapshot; + if (directType === "toolcall_delta" && (deltaSnapshot.sanitized || typeof delta !== "string")) { + throw new ManagedAttemptSnapshotError("event.delta"); + } + if (deltaSnapshot.sanitized && typeof delta === "string" && SANITIZER_SENTINELS.has(delta)) { + throw new ManagedAttemptSnapshotError("event.delta"); + } + if (delta !== undefined && delta !== null && typeof delta === "object") { + throw new ManagedAttemptSnapshotError("event.delta"); + } + if (typeof delta !== "string") { + warnManagedDegradedPrimitive("event.delta", delta, degradedFieldDiagnostics); + } + return { + type: directType, + contentIndex: contentIndex as number, + delta: typeof delta === "string" ? delta : "", + partial: message, + }; + } + const eventSnapshot = managedAttemptSnapshotDetailed(event); + const detached = eventSnapshot.snapshot; const record = isManagedPlainRecord(detached) ? detached : undefined; // Root repair, mirroring the shell: two benign degradations are re-read // through the original event with guarded reads (`managedProperty`) — @@ -1184,20 +1249,20 @@ export function managedAssistantEventSnapshot( type === "toolcall_start" ) return { type, contentIndex: indexed(), partial: message }; - if ( - type === "text_delta" || - type === "thinking_delta" || - type === "reasoning_summary_delta" || - type === "toolcall_delta" - ) { - const delta = managedProperty(source, "delta"); - if (typeof delta !== "string") throw new ManagedAttemptSnapshotError("event.delta"); - return { type, contentIndex: indexed(), delta, partial: message }; - } if (type === "text_end" || type === "thinking_end" || type === "reasoning_summary_end") { - const content = managedProperty(source, "content"); - if (typeof content !== "string") throw new ManagedAttemptSnapshotError("event.content"); - return { type, contentIndex: indexed(), content, partial: message }; + const contentSnapshot = managedAttemptSnapshotDetailed(managedProperty(source, "content")); + const content = contentSnapshot.snapshot; + if (contentSnapshot.sanitized && typeof content === "string" && SANITIZER_SENTINELS.has(content)) { + throw new ManagedAttemptSnapshotError("event.content"); + } + if (content !== undefined && content !== null && typeof content === "object") { + throw new ManagedAttemptSnapshotError("event.content"); + } + if (typeof content !== "string") { + warnManagedDegradedPrimitive("event.content", content, degradedFieldDiagnostics); + } + const safeContent = typeof content === "string" ? content : ""; + return { type, contentIndex: indexed(), content: safeContent, partial: message }; } if (type === "toolcall_end") { const toolCall = managedAssistantContent(managedAttemptSnapshot(managedProperty(source, "toolCall"))); @@ -1304,6 +1369,7 @@ class ManagedAttemptTransaction { #lastStagedShape: { stagedEventCount: number; stagedBytes: number; contentBlockCount: number } | undefined; #discarded = false; #committed = false; + #degradedFieldDiagnostics = new Set(); constructor( private readonly stream: EventStream, @@ -1596,22 +1662,28 @@ class ManagedAttemptTransaction { if (this.snapshotMode === "lossless") return event; if (event.type === "message_start" || event.type === "message_end" || event.type === "turn_end") { return event.message.role === "assistant" - ? { ...event, message: managedAssistantShell(event.message, this.model) } + ? { ...event, message: managedAssistantShell(event.message, this.model, this.#degradedFieldDiagnostics) } : event; } if (event.type === "message_update") { - const message = managedAssistantShell(event.message, this.model); + const message = managedAssistantShell(event.message, this.model, this.#degradedFieldDiagnostics); return { ...event, message, - assistantMessageEvent: managedAssistantEventSnapshot(event.assistantMessageEvent, message), + assistantMessageEvent: managedAssistantEventSnapshot( + event.assistantMessageEvent, + message, + this.#degradedFieldDiagnostics, + ), }; } if (event.type === "agent_end") { return { ...event, messages: event.messages.map(message => - message.role === "assistant" ? managedAssistantShell(message, this.model) : message, + message.role === "assistant" + ? managedAssistantShell(message, this.model, this.#degradedFieldDiagnostics) + : message, ), }; } @@ -1647,11 +1719,13 @@ class ManagedAttemptTransaction { #assistantSnapshot(message: AssistantMessage): AssistantMessage { return this.snapshotMode === "lossless" ? this.#losslessSnapshot(message) - : managedAssistantShell(message, this.model); + : managedAssistantShell(message, this.model, this.#degradedFieldDiagnostics); } #assistantEventSnapshot(event: AssistantMessageEvent, message: AssistantMessage): AssistantMessageEvent { - if (this.snapshotMode === "managed") return managedAssistantEventSnapshot(event, message); + if (this.snapshotMode === "managed") { + return managedAssistantEventSnapshot(event, message, this.#degradedFieldDiagnostics); + } const snapshot = this.#losslessSnapshot(event); if (snapshot.type === "done") return { ...snapshot, message }; if (snapshot.type === "error") return { ...snapshot, error: message }; @@ -2819,6 +2893,7 @@ async function streamAssistantResponse( provisionalToolTransaction?: ManagedAttemptTransaction, toolChoiceOverride?: { value: ToolChoice | undefined }, ): Promise { + const managedDegradedFieldDiagnostics = new Set(); // Apply context transform if configured (AgentMessage[] → AgentMessage[]) let messages = context.messages; if (config.transformContext) { @@ -3143,7 +3218,7 @@ async function streamAssistantResponse( switch (event.type) { case "start": partialMessage = config.fallbackManaged - ? managedAssistantShell(event.partial, config.model) + ? managedAssistantShell(event.partial, config.model, managedDegradedFieldDiagnostics) : event.partial; context.messages.push(partialMessage); addedPartial = true; @@ -3171,7 +3246,7 @@ async function streamAssistantResponse( case "toolcall_end": if (partialMessage) { partialMessage = config.fallbackManaged - ? managedAssistantShell(event.partial, config.model) + ? managedAssistantShell(event.partial, config.model, managedDegradedFieldDiagnostics) : event.partial; // Normalize through the managed event snapshot instead of a // naive `{ ...event }` spread: spreading copies only own @@ -3181,7 +3256,7 @@ async function streamAssistantResponse( // downstream. The snapshot repairs benign class/prototype shapes // and keeps the named fail-fast diagnostics for hostile ones. const partialEvent = config.fallbackManaged - ? managedAssistantEventSnapshot(event, partialMessage) + ? managedAssistantEventSnapshot(event, partialMessage, managedDegradedFieldDiagnostics) : event; context.messages[context.messages.length - 1] = partialMessage; if (provisionalToolTransaction) { @@ -3210,7 +3285,7 @@ async function streamAssistantResponse( case "done": case "error": { const finalMessage = config.fallbackManaged - ? managedAssistantShell(await finishResponse(), config.model) + ? managedAssistantShell(await finishResponse(), config.model, managedDegradedFieldDiagnostics) : await finishResponse(); promoteTypedEmptyResponseStop(finalMessage); if (addedPartial) { @@ -3233,7 +3308,7 @@ async function streamAssistantResponse( } const trailing = config.fallbackManaged - ? managedAssistantShell(await finishResponse(), config.model) + ? managedAssistantShell(await finishResponse(), config.model, managedDegradedFieldDiagnostics) : await finishResponse(); await finishChat(trailing); return trailing; diff --git a/packages/agent/test/agent-loop-anthropic-truncated-toolcall.test.ts b/packages/agent/test/agent-loop-anthropic-truncated-toolcall.test.ts index 64b7ed337f..071ba7b71a 100644 --- a/packages/agent/test/agent-loop-anthropic-truncated-toolcall.test.ts +++ b/packages/agent/test/agent-loop-anthropic-truncated-toolcall.test.ts @@ -182,7 +182,7 @@ describe("agentLoop with Anthropic truncated tool calls", () => { expect(toolResults[1]).toEqual({ isError: false, text: "wrote" }); }); - it("executes only the complete same-ID replacement after a malformed duplicate index", async () => { + it("fails closed before executing a malformed duplicate index", async () => { const responses = [duplicateToolResponse("tool_shared"), textResponse("done")]; let responseIndex = 0; vi.spyOn(Messages.prototype, "create").mockImplementation(() => { @@ -223,11 +223,9 @@ describe("agentLoop with Anthropic truncated tool calls", () => { } } - expect(responseIndex).toBe(2); - expect(executed).toEqual([{ path: "b.ts", content: "ok" }]); + expect(responseIndex).toBe(1); + expect(executed).toHaveLength(0); expect(toolResults).toHaveLength(2); - expect(toolResults[0].isError).toBe(true); - expect(toolResults[0].text).toContain("cut off"); - expect(toolResults[1]).toEqual({ isError: false, text: "wrote" }); + expect(toolResults.every(result => result.isError)).toBe(true); }); }); diff --git a/packages/agent/test/managed-attempt-transaction.test.ts b/packages/agent/test/managed-attempt-transaction.test.ts index cbe89bbe85..9a63686652 100644 --- a/packages/agent/test/managed-attempt-transaction.test.ts +++ b/packages/agent/test/managed-attempt-transaction.test.ts @@ -1360,8 +1360,9 @@ describe("managed attempt transaction", () => { streamFn: () => { const stream = new AssistantMessageEventStream(); queueMicrotask(() => { - // A non-array `content` cannot be normalized into the managed - // assistant shell, so staging rejects it at the content stage. + // A plain-object `content` (e.g. {0:{type:"text"}}) can hide + // array-like toolCalls — it stays fail-closed at shell.content. + // This is the blocker from the red-team review. const malformed = assistantMessage(mock.model) as unknown as { content: unknown }; malformed.content = { 0: { type: "text", text: "not an array" } }; stream.push({ type: "start", partial: malformed as unknown as AssistantMessage }); @@ -1423,6 +1424,58 @@ describe("managed attempt transaction", () => { expect(diagnostics).toHaveLength(1); expect(diagnostics[0]).toMatchObject({ stage: "shell.content", errorKind: "local_snapshot_failure" }); }); + it("degrades benign primitive content to an empty turn (null/number/boolean/string)", async () => { + const cases: Array<{ content: unknown; label: string }> = [ + { content: null, label: "null" }, + { content: 42, label: "number" }, + { content: true, label: "boolean true" }, + { content: false, label: "boolean false" }, + { content: "hello", label: "benign string" }, + { content: undefined, label: "undefined" }, + ]; + for (const { content, label } of cases) { + const diagnostics = captureSnapshotDiagnostics(); + const mock = createMockModel(); + const base = assistantMessage(mock.model); + const malformed: AssistantMessage = + content === undefined + ? (() => { + const c = { ...base } as unknown as Record; + delete c.content; + return c as unknown as AssistantMessage; + })() + : ({ ...base, content } as unknown as AssistantMessage); + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + stream.push({ type: "start", partial: malformed }); + stream.push({ type: "done", reason: "stop", message: malformed }); + }); + return stream; + }, + }); + let outcomes = 0; + await (agent.prompt as (input: string, opts: unknown) => Promise)("run", { + fallbackManaged: true, + onManagedAttemptOutcome: () => { + outcomes += 1; + return { + type: "retry", + continuation: (() => ({})) as unknown as () => AssistantMessage, + }; + }, + }); + expect(agent.state.error, label).toBeUndefined(); + expect(diagnostics, label).toHaveLength(0); + const committed = agent.state.messages.at(-1) as AssistantMessage; + expect(committed.role, label).toBe("assistant"); + expect(committed.content, label).toEqual([]); + expect(outcomes, label).toBe(0); + vi.restoreAllMocks(); + } + }); it("ignores a foreign error that self-labels a local failure kind", async () => { const diagnostics = captureSnapshotDiagnostics(); const mock = createMockModel(); @@ -1558,7 +1611,7 @@ describe("managed attempt transaction", () => { }); }); - it("rejects managed events with hidden required fields as local failures", async () => { + it("rejects managed events with object-shaped deltas as local failures", async () => { const mock = createMockModel(); const agent = new Agent({ initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, @@ -1567,12 +1620,15 @@ describe("managed attempt transaction", () => { queueMicrotask(() => { const partial = assistantMessage(mock.model); stream.push({ type: "start", partial }); - stream.push( - new Proxy( - { type: "text_delta", contentIndex: 0, partial }, - { get: (target, key) => (key === "delta" ? undefined : Reflect.get(target, key)) }, - ) as AssistantMessageEvent, - ); + // An object-shaped delta can hide real streamed text/thinking + // (or a tool-argument fragment). Degrading it to "" would + // drop that payload behind a successful empty increment. + stream.push({ + type: "text_delta", + contentIndex: 0, + delta: { chunks: ["hidden"] } as unknown as string, + partial, + }); stream.push({ type: "done", reason: "stop", message: partial }); }); return stream; @@ -1580,6 +1636,211 @@ describe("managed attempt transaction", () => { }); await agent.prompt("run", { fallbackManaged: true }); expect(agent.state.error).toContain("local snapshot"); + expect((agent.state.messages.at(-1) as AssistantMessage).errorKind).toBe("local_snapshot_failure"); + }); + it("degrades missing or primitive deltas to an empty increment instead of killing the turn", async () => { + const cases: Array<{ delta: unknown; label: string }> = [ + { delta: undefined, label: "undefined" }, + { delta: null, label: "null" }, + { delta: 42, label: "number" }, + { delta: true, label: "boolean" }, + ]; + for (const { delta, label } of cases) { + const diagnostics = captureSnapshotDiagnostics(); + const mock = createMockModel(); + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + const partial = assistantMessage(mock.model); + partial.content.push({ type: "thinking", thinking: "" }); + stream.push({ type: "start", partial }); + stream.push({ type: "thinking_start", contentIndex: 0, partial }); + stream.push({ + type: "thinking_delta", + contentIndex: 0, + delta: delta as string, + partial, + }); + stream.push({ type: "done", reason: "stop", message: partial }); + }); + return stream; + }, + }); + await agent.prompt("run", { fallbackManaged: true }); + expect(agent.state.error, label).toBeUndefined(); + expect(diagnostics, label).toHaveLength(0); + const committed = agent.state.messages.at(-1) as AssistantMessage; + expect(committed.role, label).toBe("assistant"); + expect(committed.content, label).toEqual([{ type: "thinking", thinking: "" }]); + vi.restoreAllMocks(); + } + }); + it("warns once per run when repeated partial shells degrade primitive content", async () => { + const mock = createMockModel(); + const warnings: unknown[] = []; + vi.spyOn(logger, "warn").mockImplementation((message, payload) => { + if (message === "agent: managed snapshot degraded a non-string primitive to an empty value") { + warnings.push(payload); + } + }); + const malformed = assistantMessage(mock.model) as unknown as { content: number }; + malformed.content = 42; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + stream.push({ type: "start", partial: malformed as unknown as AssistantMessage }); + for (let index = 0; index < 3; index++) { + stream.push({ + type: "thinking_delta", + contentIndex: 0, + delta: undefined as unknown as string, + partial: malformed as unknown as AssistantMessage, + }); + } + stream.push({ type: "done", reason: "stop", message: malformed as unknown as AssistantMessage }); + }); + return stream; + }, + }); + + await agent.prompt("run", { fallbackManaged: true }); + expect(warnings.filter(payload => (payload as { field?: string }).field === "shell.content")).toHaveLength(1); + }); + it("warns once per run for degraded prose and fails closed on primitive executable deltas", () => { + const mock = createMockModel(); + const message = assistantMessage(mock.model); + const warnings: Array<{ message: string; payload: unknown }> = []; + vi.spyOn(logger, "warn").mockImplementation((warning, payload) => { + warnings.push({ message: warning, payload }); + }); + const diagnostics = new Set(); + + for (const delta of [undefined, 42, true]) { + const snapshot = managedAssistantEventSnapshot( + { type: "thinking_delta", contentIndex: 0, delta: delta as unknown as string, partial: message }, + message, + diagnostics, + ); + expect(snapshot).toMatchObject({ type: "thinking_delta", delta: "" }); + } + expect(warnings).toEqual([ + { + message: "agent: managed snapshot degraded a non-string primitive to an empty value", + payload: { field: "event.delta", receivedType: "undefined" }, + }, + ]); + + for (const delta of [undefined, null, 42, true, 1n]) { + expect(() => + managedAssistantEventSnapshot( + { type: "toolcall_delta", contentIndex: 0, delta: delta as unknown as string, partial: message }, + message, + diagnostics, + ), + ).toThrow(/snapshot/i); + } + }); + it("uses one captured tool delta value when an accessor changes between reads", () => { + const mock = createMockModel(); + const message = assistantMessage(mock.model); + let reads = 0; + const event = { + type: "toolcall_delta", + contentIndex: 0, + partial: message, + get delta(): unknown { + reads += 1; + return reads === 1 ? 42 : '{"laundered":true}'; + }, + } as unknown as AssistantMessageEvent; + + expect(() => managedAssistantEventSnapshot(event, message)).toThrow(/snapshot/i); + expect(reads).toBe(1); + }); + it("preserves a prototype delta when the event type is an own property", () => { + const mock = createMockModel(); + const message = assistantMessage(mock.model); + let reads = 0; + const prototype = { + get delta(): string { + reads += 1; + return '{"path":"prototype.ts"}'; + }, + }; + const event = Object.assign(Object.create(prototype) as Record, { + type: "toolcall_delta", + contentIndex: 0, + partial: message, + }) as unknown as AssistantMessageEvent; + + expect(managedAssistantEventSnapshot(event, message)).toMatchObject({ + type: "toolcall_delta", + delta: '{"path":"prototype.ts"}', + }); + expect(reads).toBe(1); + }); + it("preserves a valid tool delta from a readable proxy event", () => { + const mock = createMockModel(); + const message = assistantMessage(mock.model); + const event = new Proxy( + { + type: "toolcall_delta", + contentIndex: 0, + delta: '{"path":"proxy.ts"}', + partial: message, + } as AssistantMessageEvent, + {}, + ); + + expect(managedAssistantEventSnapshot(event, message)).toMatchObject({ + type: "toolcall_delta", + delta: '{"path":"proxy.ts"}', + }); + }); + it("preserves a valid tool delta when unrelated metadata requires sanitization", () => { + const mock = createMockModel(); + const message = assistantMessage(mock.model); + const event = { + type: "toolcall_delta", + contentIndex: 0, + delta: '{"path":"metadata.ts"}', + partial: message, + providerMetadata: { sequence: 1n }, + } as unknown as AssistantMessageEvent; + + expect(managedAssistantEventSnapshot(event, message)).toMatchObject({ + type: "toolcall_delta", + delta: '{"path":"metadata.ts"}', + }); + }); + it("preserves a literal sentinel-looking delta when no sanitizer produced it", async () => { + const diagnostics = captureSnapshotDiagnostics(); + const mock = createMockModel(); + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: () => { + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + const partial = assistantMessage(mock.model); + stream.push({ type: "start", partial }); + stream.push({ + type: "thinking_delta", + contentIndex: 0, + delta: "[unserializable]", + partial, + }); + stream.push({ type: "done", reason: "stop", message: partial }); + }); + return stream; + }, + }); + await agent.prompt("run", { fallbackManaged: true }); + expect(agent.state.error).toBeUndefined(); + expect(diagnostics).toHaveLength(0); }); it("normalizes invalid stop reasons and rejects invalid event indices", async () => { const mock = createMockModel(); diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 926fab98b4..339053ad32 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -4,6 +4,8 @@ - OpenAI-family streams now give xAI Grok and the Grok Build (`grok-cli-responses`) wrapper the same 300-second default idle window as Anthropic, so long Grok reasoning gaps no longer surface as `OpenAI responses stream stalled while waiting for the next event` under the 120-second OpenAI default. Env overrides still win. The observed stall was `grok-build/grok-4.6` on `openai-responses`; keying only `xai` would have left that path on 120s because `streamGrokCli` keeps `model.provider === "grok-build"`. - `getCachedUsageReport` now surfaces provider-level cached usage reports for stored API-key credentials, not only OAuth rows. `checkCredentials` fetches and caches usage for API-key providers (for example `zai`, whose login flow stores an API key by design), but the display lookup rejected every non-OAuth row, so `/usage` and account listings could never show usage data that had been successfully fetched and cached. The lookup builds the same cache identity `checkCredentials` writes, and the returned observation stays redacted — credential bytes never appear in the cached report. - Anthropic clients now set an SDK request `timeout` derived from the first-event window (`resolveAnthropicSdkRequestTimeoutMs`; 300s by default for Anthropic, floored at the env/default first-event window, disabled by an explicit `streamFirstEventTimeoutMs: 0`). The Anthropic first-event watchdog deliberately arms only after response headers arrive, so a connection that silently died before headers — the exact failure mode of recent Anthropic stream instability right after a completed tool call — was previously bounded only by the SDK's 10-minute default per attempt multiplied by its internal retry budget, observable as an endless "Working…" spinner for up to an hour with no error, no retry indicator, and no automatic recovery. This mirrors the existing `resolveOpenAISdkRequestTimeoutMs` stalled-before-headers bound on the OpenAI family. +- Anthropic `input_json_delta` and Codex `function_call`/`custom_tool_call` argument increments now fail the turn closed for every non-string value instead of continuing with missing, default, or silently altered tool arguments. Primitive thinking/text/signature anomalies still degrade to an empty string, and each stream emits at most one payload-free diagnostic per increment type naming only the envelope (`deltaType`/`eventType` and `receivedType`). +- Anthropic-compatible and Codex stream handlers now coerce non-string prose, thinking, and signature increments to an empty string before emitting `*_delta` events, so a Z.AI or Codex thinking delta that arrives as `undefined` or a numeric token count cannot produce a non-string `delta` that managed snapshot staging rejects as `event.delta`. Executable tool-argument fragments remain fail-closed for every non-string shape. Anthropic `signature_delta` likewise appends only string signatures, so a numeric or missing signature cannot pollute `thinkingSignature` as `"1"` / `"[object Object]"`. - Added the `ask-round-zero-metadata-requires-full-topology-fields` raw-argument rejection code so the ask tool's Round-0 deep-interview validator can name the omitted topology fields and their correction; previously the incomplete-object failure surfaced only as generic zod issues with a full payload echo (#4649). - oMLX OpenAI-compatible completions now send `chat_template_kwargs.reasoning_effort` with `enable_thinking` when `thinkingFormat` is `qwen-chat-template`. Discovered oMLX models are treated as reasoning models with `low`/`medium`/`high` effort so local Qwen presets can differentiate roles without swapping weights. - Fixed a resume-breaking HTTP 400 on `google-gemini-cli`/`google-antigravity` replay: assistant thinking blocks whose `thinkingSignature` is missing, empty (persistence clears oversized signatures to `""`), or invalid no longer emit an unsigned `{"thought": true}` part. Cloud Code Assist maps such parts to Anthropic `thinking` blocks and rejects the whole request with `messages.N.content.0.thinking.signature: Field required`, permanently bricking resumed sessions (#4630). Unsigned thinking now degrades to plain text — the same treatment cross-model reasoning already gets — while validly signed thinking still replays natively as a thought part with its `thoughtSignature`. diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 0faaa502ab..065e01fb3d 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -2023,7 +2023,21 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( const blocks = output.content as Block[]; const blocksByAnthropicIndex = new Map(); const truncatedToolCalls = new Set(); - let sawTerminalStopReason = false; + // Bounded diagnostic for degraded primitive increments: at most one + // warning per delta type per stream invocation, naming only the + // envelope shape (delta type and received typeof) — never the payload. + const degradedIncrementDiagnostics = new Set(); + const noteDegradedIncrement = (deltaType: string, received: unknown): void => { + if (degradedIncrementDiagnostics.has(deltaType)) return; + degradedIncrementDiagnostics.add(deltaType); + logger.warn("anthropic: degraded non-string stream increment to empty string", { + model: model.id, + provider: model.provider, + deltaType, + receivedType: received === null ? "null" : typeof received, + }); + }; + // Derive from the ACTUAL request shape, not the option default: the request // only sends `display: "summarized"` on specific paths (adaptive display is // omitted for models where supportsAdaptiveThinkingDisplay is false). Defaulting @@ -2037,25 +2051,9 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( return { block, contentIndex: blocks.indexOf(block) }; }; const trackBlockByAnthropicIndex = (anthropicIndex: number, block: Block) => { - // A duplicate start for an active index is a provider-envelope violation; - // finalize the orphaned block so no internal stream fields leak into output. const orphaned = blocksByAnthropicIndex.get(anthropicIndex); if (orphaned) { - if (orphaned.type === "toolCall") { - if (!isCompleteJson(orphaned.partialJson)) { - orphaned.incompleteArguments = true; - orphaned.incompleteArgumentsReason = "truncated"; - truncatedToolCalls.add(orphaned); - } - if (orphaned.partialJson.trim()) { - orphaned.arguments = parseStreamingJson(orphaned.partialJson); - if (findUnnecessaryUnicodeEscape(orphaned.partialJson)) { - orphaned.escapedNonAsciiArguments = true; - } - } - } - delete (orphaned as { index?: number }).index; - delete (orphaned as { partialJson?: string }).partialJson; + throw new Error("Anthropic stream reused an active content block index"); } blocksByAnthropicIndex.set(anthropicIndex, block); }; @@ -2070,7 +2068,6 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( output.stopReason = "stop"; firstTokenTime = undefined; truncatedToolCalls.clear(); - sawTerminalStopReason = false; }; const idleTimeoutMs = options?.streamIdleTimeoutMs ?? @@ -2105,7 +2102,6 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( // Retries reset output.content; drop stale block correlations from the aborted attempt. blocksByAnthropicIndex.clear(); truncatedToolCalls.clear(); - sawTerminalStopReason = false; activeAbortTracker = createAbortSourceTracker(options?.signal); let firstEventTimeoutAbortError: FirstEventTimeoutError | undefined; const idleTimeoutAbortError = new Error("Anthropic stream stalled while waiting for the next event"); @@ -2253,13 +2249,21 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( trackBlockByAnthropicIndex(event.index, block); } else if (event.content_block.type === "tool_use") { streamedReplayUnsafeContent = true; + const initialArguments: unknown = event.content_block.input; + if ( + initialArguments === null || + typeof initialArguments !== "object" || + Array.isArray(initialArguments) + ) { + throw new Error("Anthropic tool_use started with non-object arguments"); + } const block: Block = { type: "toolCall", id: event.content_block.id, name: isOAuthToken ? stripClaudeToolPrefix(event.content_block.name) : event.content_block.name, - arguments: (event.content_block.input as Record) ?? {}, + arguments: initialArguments as Record, partialJson: "", index: event.index, }; @@ -2275,32 +2279,42 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( if (event.delta.type === "text_delta") { const { block, contentIndex: index } = getBlockByAnthropicIndex(event.index); if (block && block.type === "text") { - block.text += event.delta.text; + const rawTextDelta: unknown = event.delta.text; + if (typeof rawTextDelta !== "string") { + noteDegradedIncrement("text_delta", rawTextDelta); + } + const textDelta = typeof rawTextDelta === "string" ? rawTextDelta : ""; + block.text += textDelta; stream.push({ type: "text_delta", contentIndex: index, - delta: event.delta.text, + delta: textDelta, partial: output, }); } } else if (event.delta.type === "thinking_delta") { const { block, contentIndex: index } = getBlockByAnthropicIndex(event.index); if (block && block.type === "thinking") { - block.thinking += event.delta.thinking; + const rawThinkingDelta: unknown = event.delta.thinking; + if (typeof rawThinkingDelta !== "string") { + noteDegradedIncrement("thinking_delta", rawThinkingDelta); + } + const thinkingDelta = typeof rawThinkingDelta === "string" ? rawThinkingDelta : ""; + block.thinking += thinkingDelta; if (summarizedThinking) { - const summary = (reasoningBuffers.get(block) ?? "") + event.delta.thinking; + const summary = (reasoningBuffers.get(block) ?? "") + thinkingDelta; reasoningBuffers.set(block, summary); stream.push({ type: "reasoning_summary_delta", contentIndex: index, - delta: event.delta.thinking, + delta: thinkingDelta, partial: output, }); } else { stream.push({ type: "thinking_delta", contentIndex: index, - delta: event.delta.thinking, + delta: thinkingDelta, partial: output, }); } @@ -2308,12 +2322,26 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( } else if (event.delta.type === "input_json_delta") { const { block, contentIndex: index } = getBlockByAnthropicIndex(event.index); if (block && block.type === "toolCall") { - block.partialJson += event.delta.partial_json; + const rawJsonDelta: unknown = event.delta.partial_json; + if (typeof rawJsonDelta !== "string") { + // Tool-argument fragments are positional JSON text: erasing or + // coercing any malformed increment (primitive OR object/function) + // assembles valid-but-wrong arguments — e.g. `{"n":1` + numeric + // primitive erased to "" + `3}` parses as {"n":13} and executes. + // Prose/thinking/signature anomalies are safe to degrade; tool + // arguments fail the turn closed. The payload never enters the + // error. + throw new Error( + "Anthropic stream sent a non-string input_json_delta tool-argument increment; failing the turn instead of assembling wrong tool arguments", + ); + } + const jsonDelta = rawJsonDelta; + block.partialJson += jsonDelta; block.arguments = parseStreamingJson(block.partialJson); stream.push({ type: "toolcall_delta", contentIndex: index, - delta: event.delta.partial_json, + delta: jsonDelta, partial: output, }); } @@ -2321,7 +2349,12 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( const { block } = getBlockByAnthropicIndex(event.index); if (block && block.type === "thinking") { block.thinkingSignature = block.thinkingSignature || ""; - block.thinkingSignature += event.delta.signature; + const rawSignatureDelta: unknown = event.delta.signature; + if (typeof rawSignatureDelta === "string") { + block.thinkingSignature += rawSignatureDelta; + } else { + noteDegradedIncrement("signature_delta", rawSignatureDelta); + } } } } else if (event.type === "content_block_stop") { @@ -2361,7 +2394,15 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( } else if (block.type === "toolCall") { if (!isCompleteJson(block.partialJson)) truncatedToolCalls.add(block); if (block.partialJson.trim()) { - block.arguments = parseStreamingJson(block.partialJson); + const parsedArguments: unknown = parseStreamingJson(block.partialJson); + if ( + parsedArguments === null || + typeof parsedArguments !== "object" || + Array.isArray(parsedArguments) + ) { + throw new Error("Anthropic tool_use completed with non-object arguments"); + } + block.arguments = parsedArguments as Record; if (findUnnecessaryUnicodeEscape(block.partialJson)) { block.escapedNonAsciiArguments = true; } @@ -2383,7 +2424,6 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( if (rawStopReason) { output.stopReason = isProviderSafetyStop ? "error" : mapStopReason(rawStopReason); sawTerminalEnvelope = true; - sawTerminalStopReason = true; } if (isProviderSafetyStop) { sawProviderSafetyStop = true; @@ -2862,12 +2902,10 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( } } blocksByAnthropicIndex.clear(); - if (output.stopReason === "length" || !sawTerminalStopReason) { - for (const block of output.content) { - if (block.type === "toolCall" && truncatedToolCalls.has(block)) { - block.incompleteArguments = true; - block.incompleteArgumentsReason = "truncated"; - } + for (const block of output.content) { + if (block.type === "toolCall" && truncatedToolCalls.has(block)) { + block.incompleteArguments = true; + block.incompleteArgumentsReason = "truncated"; } } output.duration = Date.now() - startTime; diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index b09c7d9a51..a187f05932 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -163,10 +163,21 @@ const CODEX_PROGRESS_EVENT_TYPES = new Set([ "error", ]); +/** + * A progress event must carry real semantic payload, matching the Anthropic + * predicate: a recognized envelope whose `delta` is absent or not a non-empty + * string is NOT progress — otherwise repeated malformed/no-op deltas reset the + * idle watchdog indefinitely and a managed attempt need never terminate. + * Non-delta envelope types (lifecycle, item boundaries, terminal events) count + * as progress by type alone, as before. + */ function isCodexStreamProgressEvent(event: unknown): boolean { if (!event || typeof event !== "object") return false; const type = (event as { type?: unknown }).type; - return typeof type === "string" && CODEX_PROGRESS_EVENT_TYPES.has(type); + if (typeof type !== "string" || !CODEX_PROGRESS_EVENT_TYPES.has(type)) return false; + if (!type.endsWith(".delta")) return true; + const delta = (event as { delta?: unknown }).delta; + return typeof delta === "string" && delta.length > 0; } type CodexTransport = "sse" | "websocket"; interface CodexInitialTransport { @@ -177,7 +188,7 @@ interface CodexInitialTransport { } type CodexEventItem = ResponseReasoningItem | ResponseOutputMessage | ResponseFunctionToolCall | ResponseCustomToolCall; type CodexThinkingBlock = ThinkingContent & { summaryBuffer: string; rawBuffer: string; summaryStarted: boolean }; -type CodexOutputBlock = CodexThinkingBlock | TextContent | (ToolCall & { partialJson: string }); +type CodexOutputBlock = CodexThinkingBlock | TextContent | (ToolCall & { partialJson: string; doneInput?: string }); export interface OpenAICodexWebSocketDebugStats { fullContextRequests: number; deltaRequests: number; @@ -282,6 +293,8 @@ interface CodexStreamRuntime { canSafelyReplayWebsocketOverSse: boolean; /** Ids of tool calls that received their terminal `output_item.done`. */ finalizedToolCallIds: Set; + /** Event types whose degraded non-string increment was already diagnosed. */ + degradedIncrementDiagnostics: Set; } interface CodexStreamProcessingContext { @@ -592,6 +605,10 @@ function resetOutputState(output: AssistantMessage): void { function removeTransientBlockIndices(output: AssistantMessage): void { for (const block of output.content) { delete (block as { index?: number }).index; + if (block.type === "toolCall") { + delete (block as { partialJson?: string }).partialJson; + delete (block as { doneInput?: string }).doneInput; + } } } @@ -965,6 +982,7 @@ function createCodexStreamRuntime(initial: { sawTerminalEvent: false, canSafelyReplayWebsocketOverSse: true, finalizedToolCallIds: new Set(), + degradedIncrementDiagnostics: new Set(), }; } @@ -1019,6 +1037,13 @@ function handleCodexStreamEvent(args: { runtime.currentItem = item; runtime.currentBlock = createOutputBlockForItem(item); if (!runtime.currentBlock) return firstTokenTime; + const currentBlock = runtime.currentBlock; + if ( + currentBlock.type === "toolCall" && + output.content.some(block => block.type === "toolCall" && block.id === currentBlock.id) + ) { + throw new Error("Codex stream reused an active tool-call identifier"); + } output.content.push(runtime.currentBlock); stream.push({ type: getOutputBlockStartEventType(runtime.currentBlock), @@ -1034,7 +1059,13 @@ function handleCodexStreamEvent(args: { } if (eventType === "response.reasoning_summary_text.delta") { - handleReasoningSummaryTextDelta(runtime.currentItem, runtime.currentBlock, rawEvent, stream, output, blockIndex); + const delta = normalizeCodexIncrement( + rawEvent, + "response.reasoning_summary_text.delta", + model, + runtime.degradedIncrementDiagnostics, + ); + handleReasoningSummaryTextDelta(runtime.currentItem, runtime.currentBlock, delta, stream, output, blockIndex); return firstTokenTime; } @@ -1044,7 +1075,13 @@ function handleCodexStreamEvent(args: { } if (eventType === "response.reasoning_text.delta") { - handleReasoningTextDelta(runtime.currentItem, runtime.currentBlock, rawEvent, stream, output, blockIndex); + const delta = normalizeCodexIncrement( + rawEvent, + "response.reasoning_text.delta", + model, + runtime.degradedIncrementDiagnostics, + ); + handleReasoningTextDelta(runtime.currentItem, runtime.currentBlock, delta, stream, output, blockIndex); return firstTokenTime; } @@ -1054,10 +1091,16 @@ function handleCodexStreamEvent(args: { } if (eventType === "response.output_text.delta") { + const delta = normalizeCodexIncrement( + rawEvent, + "response.output_text.delta", + model, + runtime.degradedIncrementDiagnostics, + ); handleMessageTextDelta( runtime.currentItem, runtime.currentBlock, - rawEvent, + delta, stream, output, blockIndex, @@ -1067,20 +1110,19 @@ function handleCodexStreamEvent(args: { } if (eventType === "response.refusal.delta") { - handleMessageTextDelta( - runtime.currentItem, - runtime.currentBlock, + const delta = normalizeCodexIncrement( rawEvent, - stream, - output, - blockIndex, - "refusal", + "response.refusal.delta", + model, + runtime.degradedIncrementDiagnostics, ); + handleMessageTextDelta(runtime.currentItem, runtime.currentBlock, delta, stream, output, blockIndex, "refusal"); return firstTokenTime; } if (eventType === "response.function_call_arguments.delta") { - handleToolCallArgumentsDelta(runtime.currentItem, runtime.currentBlock, rawEvent, stream, output, blockIndex); + const delta = assertStringToolArgumentIncrement(rawEvent, "response.function_call_arguments.delta"); + handleToolCallArgumentsDelta(runtime.currentItem, runtime.currentBlock, delta, stream, output, blockIndex); return firstTokenTime; } @@ -1090,7 +1132,8 @@ function handleCodexStreamEvent(args: { } if (eventType === "response.custom_tool_call_input.delta") { - handleCustomToolCallInputDelta(runtime.currentItem, runtime.currentBlock, rawEvent, stream, output, blockIndex); + const delta = assertStringToolArgumentIncrement(rawEvent, "response.custom_tool_call_input.delta"); + handleCustomToolCallInputDelta(runtime.currentItem, runtime.currentBlock, delta, stream, output, blockIndex); return firstTokenTime; } @@ -1137,6 +1180,10 @@ function createOutputBlockForItem(item: CodexEventItem): CodexOutputBlock | null }; } if (item.type === "custom_tool_call") { + const initialInput: unknown = item.input; + if (typeof initialInput !== "string") { + throw new Error("Codex custom_tool_call started with non-string input"); + } // Wire name flows through unchanged; the agent-loop dispatcher also // matches `Tool.customWireName`. Reuse `partialJson` as the // accumulation buffer for the raw input string. @@ -1144,9 +1191,9 @@ function createOutputBlockForItem(item: CodexEventItem): CodexOutputBlock | null type: "toolCall", id: encodeResponsesToolCallId(item.call_id, item.id), name: item.name, - arguments: { input: item.input ?? "" }, + arguments: { input: initialInput }, customWireName: item.name, - partialJson: item.input ?? "", + partialJson: initialInput, }; } return null; @@ -1164,10 +1211,52 @@ function handleReasoningSummaryPartAdded(currentItem: CodexEventItem | null, raw currentItem.summary.push((rawEvent as { part: ResponseReasoningItem["summary"][number] }).part); } +/** + * Primitive anomalies (undefined, null, numbers, booleans) stay coerced to + * an empty string by the increment handlers. Diagnose at most once per event + * type per stream, naming only the envelope shape — never the payload. + */ +function normalizeCodexIncrement( + rawEvent: Record, + eventType: string, + model: Model<"openai-codex-responses">, + degradedIncrementDiagnostics: Set, +): string { + const raw = (rawEvent as { delta?: unknown }).delta; + if (typeof raw === "string") return raw; + if (!degradedIncrementDiagnostics.has(eventType)) { + degradedIncrementDiagnostics.add(eventType); + logger.warn("codex: degraded non-string stream increment to empty string", { + model: model.id, + provider: model.provider, + eventType, + receivedType: raw === null ? "null" : typeof raw, + }); + } + return ""; +} + +/** + * Tool-argument fragments are positional JSON text. Erasing or coercing ANY + * malformed increment — primitive, object, or function — assembles + * valid-but-wrong arguments (e.g. `{"n":1` + numeric primitive erased to "" + * + `3}` parses as {"n":13} and executes), so the turn fails closed on every + * non-string delta. The payload never enters the error message. + */ +function assertStringToolArgumentIncrement(rawEvent: Record, eventType: string): string { + const raw = (rawEvent as { delta?: unknown }).delta; + if (typeof raw !== "string") { + throw new Error( + `Codex stream sent a non-string ${eventType} tool-argument increment; failing the turn instead of assembling wrong tool arguments`, + ); + } + return raw; +} + function handleReasoningSummaryTextDelta( currentItem: CodexEventItem | null, currentBlock: CodexOutputBlock | null, - rawEvent: Record, + delta: string, stream: AssistantMessageEventStream, output: AssistantMessage, blockIndex: () => number, @@ -1180,7 +1269,6 @@ function handleReasoningSummaryTextDelta( currentItem.summary = currentItem.summary || []; const lastPart = currentItem.summary[currentItem.summary.length - 1]; if (!lastPart) return; - const delta = (rawEvent as { delta?: string }).delta || ""; currentBlock.thinking += delta; currentBlock.summaryBuffer += delta; lastPart.text += delta; @@ -1207,13 +1295,12 @@ function handleReasoningSummaryPartDone( function handleReasoningTextDelta( currentItem: CodexEventItem | null, currentBlock: CodexOutputBlock | null, - rawEvent: Record, + delta: string, stream: AssistantMessageEventStream, output: AssistantMessage, blockIndex: () => number, ): void { if (currentItem?.type !== "reasoning" || currentBlock?.type !== "thinking") return; - const delta = (rawEvent as { delta?: string }).delta || ""; currentBlock.thinking += delta; currentBlock.rawBuffer += delta; stream.push({ type: "thinking_delta", contentIndex: blockIndex(), delta, partial: output }); @@ -1231,7 +1318,7 @@ function handleContentPartAdded(currentItem: CodexEventItem | null, rawEvent: Re function handleMessageTextDelta( currentItem: CodexEventItem | null, currentBlock: CodexOutputBlock | null, - rawEvent: Record, + delta: string, stream: AssistantMessageEventStream, output: AssistantMessage, blockIndex: () => number, @@ -1241,7 +1328,6 @@ function handleMessageTextDelta( if (!currentItem.content || currentItem.content.length === 0) return; const lastPart = currentItem.content[currentItem.content.length - 1]; if (!lastPart || lastPart.type !== partType) return; - const delta = (rawEvent as { delta?: string }).delta || ""; currentBlock.text += delta; if (lastPart.type === "output_text") { lastPart.text += delta; @@ -1254,13 +1340,12 @@ function handleMessageTextDelta( function handleToolCallArgumentsDelta( currentItem: CodexEventItem | null, currentBlock: CodexOutputBlock | null, - rawEvent: Record, + delta: string, stream: AssistantMessageEventStream, output: AssistantMessage, blockIndex: () => number, ): void { if (currentItem?.type !== "function_call" || currentBlock?.type !== "toolCall") return; - const delta = (rawEvent as { delta?: string }).delta || ""; currentBlock.partialJson += delta; currentBlock.arguments = parseStreamingJson(currentBlock.partialJson); stream.push({ type: "toolcall_delta", contentIndex: blockIndex(), delta, partial: output }); @@ -1283,13 +1368,12 @@ function handleToolCallArgumentsDone( function handleCustomToolCallInputDelta( currentItem: CodexEventItem | null, currentBlock: CodexOutputBlock | null, - rawEvent: Record, + delta: string, stream: AssistantMessageEventStream, output: AssistantMessage, blockIndex: () => number, ): void { if (currentItem?.type !== "custom_tool_call" || currentBlock?.type !== "toolCall") return; - const delta = (rawEvent as { delta?: string }).delta || ""; currentBlock.partialJson += delta; currentBlock.arguments = { input: currentBlock.partialJson }; stream.push({ type: "toolcall_delta", contentIndex: blockIndex(), delta, partial: output }); @@ -1301,11 +1385,17 @@ function handleCustomToolCallInputDone( rawEvent: Record, ): void { if (currentItem?.type !== "custom_tool_call" || currentBlock?.type !== "toolCall") return; - const input = (rawEvent as { input?: string }).input; - if (typeof input === "string") { - currentBlock.partialJson = input; - currentBlock.arguments = { input }; + const input = (rawEvent as { input?: unknown }).input; + if (typeof input !== "string") { + throw new Error("Codex stream sent non-string input in custom_tool_call_input.done"); } + if (currentBlock.partialJson && currentBlock.partialJson !== input) { + throw new Error( + "Codex custom_tool_call input.done disagrees with the streamed input buffer; failing the turn instead of executing corrupted input", + ); + } + currentBlock.doneInput = input; + currentBlock.arguments = { input }; } function handleOutputItemDone( @@ -1385,36 +1475,90 @@ function handleOutputItemDone( } if (item.type === "function_call") { + if (typeof item.arguments !== "string") { + throw new Error("Codex function_call completed with non-string terminal arguments"); + } + let terminalArguments: unknown; + try { + terminalArguments = JSON.parse(item.arguments); + } catch { + throw new Error("Codex function_call completed with malformed terminal arguments"); + } + if (!terminalArguments || typeof terminalArguments !== "object" || Array.isArray(terminalArguments)) { + throw new Error("Codex function_call terminal arguments were not a JSON object"); + } const id = encodeResponsesToolCallId(item.call_id, item.id); + if (runtime.currentBlock?.type !== "toolCall" || runtime.currentBlock.id !== id) { + throw new Error("Codex function_call terminal item did not match the active tool call"); + } runtime.finalizedToolCallIds.add(id); const toolCall: ToolCall = { type: "toolCall", id, name: codexToolCanonicalName(item.name), - arguments: parseStreamingJson(item.arguments || "{}"), + arguments: terminalArguments as Record, ...(findUnnecessaryUnicodeEscape(item.arguments || "") ? { escapedNonAsciiArguments: true } : {}), }; + Object.assign(runtime.currentBlock, toolCall); + delete (runtime.currentBlock as { partialJson?: string }).partialJson; + delete (runtime.currentBlock as { doneInput?: string }).doneInput; runtime.canSafelyReplayWebsocketOverSse = false; stream.push({ type: "toolcall_end", contentIndex: blockIndex(), toolCall, partial: output }); + runtime.currentItem = null; + runtime.currentBlock = null; return; } if (item.type === "custom_tool_call") { + const terminalInput: unknown = item.input; + if (typeof terminalInput !== "string") { + throw new Error( + "Codex custom_tool_call completed with non-string terminal input; failing the turn instead of finalizing from the streamed buffer", + ); + } const id = encodeResponsesToolCallId(item.call_id, item.id); + if (runtime.currentBlock?.type !== "toolCall" || runtime.currentBlock.id !== id) { + throw new Error("Codex custom_tool_call terminal item did not match the active tool call"); + } runtime.finalizedToolCallIds.add(id); - const rawInput = + // The terminal `output_item.done.item.input` is the authoritative + // complete input; the streamed `partialJson` buffer is advisory. If both + // exist and disagree, the stream was corrupted (dropped/malformed + // increments), so fail closed instead of executing either variant — + // matching function-call finalization, which always trusts the terminal + // `item.arguments`. + const streamedInput = runtime.currentBlock?.type === "toolCall" && runtime.currentBlock.partialJson ? runtime.currentBlock.partialJson - : (item.input ?? ""); + : undefined; + if (streamedInput !== undefined && streamedInput !== terminalInput) { + throw new Error( + "Codex custom_tool_call terminal input disagrees with the streamed input buffer; failing the turn instead of executing a corrupted tool call", + ); + } + if ( + runtime.currentBlock?.type === "toolCall" && + runtime.currentBlock.doneInput !== undefined && + runtime.currentBlock.doneInput !== terminalInput + ) { + throw new Error( + "Codex custom_tool_call terminal input disagrees with input.done; failing the turn instead of executing conflicting input", + ); + } const toolCall: ToolCall = { type: "toolCall", id, name: item.name, - arguments: { input: rawInput }, + arguments: { input: terminalInput }, customWireName: item.name, }; + Object.assign(runtime.currentBlock, toolCall); + delete (runtime.currentBlock as { partialJson?: string }).partialJson; + delete (runtime.currentBlock as { doneInput?: string }).doneInput; runtime.canSafelyReplayWebsocketOverSse = false; stream.push({ type: "toolcall_end", contentIndex: blockIndex(), toolCall, partial: output }); + runtime.currentItem = null; + runtime.currentBlock = null; return; } @@ -1476,6 +1620,12 @@ function handleResponseCompleted( // call that never received its `output_item.done` so the agent loop rejects // the truncated arguments instead of executing a best-effort partial parse. flagTruncatedToolCalls(output, output.stopReason, block => runtime.finalizedToolCallIds.has(block.id)); + if ( + output.stopReason === "stop" && + output.content.some(block => block.type === "toolCall" && !runtime.finalizedToolCallIds.has(block.id)) + ) { + throw new Error("Codex response completed with an unfinalized tool call"); + } if (output.content.some(block => block.type === "toolCall") && output.stopReason === "stop") { output.stopReason = "toolUse"; } @@ -1540,6 +1690,7 @@ async function tryRetryWithoutForcedToolChoice( runtime.currentBlock = null; runtime.sawTerminalEvent = false; runtime.nativeOutputItems.length = 0; + runtime.finalizedToolCallIds.clear(); resetOutputState(context.output); context.firstTokenTime = undefined; @@ -1635,6 +1786,7 @@ async function tryReconnectCodexWebSocketOnConnectionLimit( runtime.currentItem = null; runtime.currentBlock = null; runtime.nativeOutputItems.length = 0; + runtime.finalizedToolCallIds.clear(); resetOutputState(context.output); context.firstTokenTime = undefined; recordCodexWebSocketFailure(websocketState, true); @@ -1681,6 +1833,7 @@ async function tryRecoverCodexPreviousResponseNotFound( runtime.currentBlock = null; runtime.sawTerminalEvent = false; runtime.nativeOutputItems.length = 0; + runtime.finalizedToolCallIds.clear(); resetOutputState(context.output); context.firstTokenTime = undefined; @@ -1739,6 +1892,7 @@ async function tryReplayWebsocketFailureOverSse( runtime.currentItem = null; runtime.currentBlock = null; runtime.nativeOutputItems.length = 0; + runtime.finalizedToolCallIds.clear(); resetOutputState(context.output); context.firstTokenTime = undefined; } @@ -1779,6 +1933,8 @@ async function tryRetryCodexProviderError( runtime.currentItem = null; runtime.currentBlock = null; runtime.sawTerminalEvent = false; + runtime.nativeOutputItems.length = 0; + runtime.finalizedToolCallIds.clear(); resetOutputState(context.output); context.firstTokenTime = undefined; await scheduler.wait(CODEX_RETRY_DELAY_MS * runtime.providerRetryAttempt, { @@ -1821,6 +1977,7 @@ function finalizeCodexResponse( throw new Error("Codex response failed"); } + removeTransientBlockIndices(output); output.providerPayload = createOpenAIResponsesHistoryPayload(context.model.provider, runtime.nativeOutputItems); output.duration = Date.now() - context.startTime; if (completion.firstTokenTime) { @@ -2290,10 +2447,15 @@ class CodexWebSocketConnection { this.#socket = socket; let settled = false; let timeout: NodeJS.Timeout | undefined; + const clearPending = () => { + if (timeout) clearTimeout(timeout); + if (signal) signal.removeEventListener("abort", onAbort); + }; const onAbort = () => { socket.close(1000, "aborted"); if (!settled) { settled = true; + clearPending(); reject(createCodexWebSocketTransportError("request was aborted")); } }; @@ -2304,17 +2466,16 @@ class CodexWebSocketConnection { signal.addEventListener("abort", onAbort, { once: true }); } } - const clearPending = () => { - if (timeout) clearTimeout(timeout); - if (signal) signal.removeEventListener("abort", onAbort); - }; - timeout = setTimeout(() => { - socket.close(1000, "connect-timeout"); - if (!settled) { - settled = true; - reject(createCodexWebSocketTransportError("connection timeout")); - } - }, CODEX_WEBSOCKET_CONNECT_TIMEOUT_MS); + if (!settled) { + timeout = setTimeout(() => { + socket.close(1000, "connect-timeout"); + if (!settled) { + settled = true; + clearPending(); + reject(createCodexWebSocketTransportError("connection timeout")); + } + }, CODEX_WEBSOCKET_CONNECT_TIMEOUT_MS); + } socket.onopen = event => { if (!settled) { @@ -2352,6 +2513,7 @@ class CodexWebSocketConnection { }; socket.onmessage = event => { try { + if (!this.#activeRequest) return; const text = typeof event.data === "string" ? event.data : Buffer.from(event.data).toString("utf-8"); if (!text) return; const parsed = JSON.parse(text) as Record; @@ -2382,6 +2544,7 @@ class CodexWebSocketConnection { request: Record, signal?: AbortSignal, firstEventTimeoutMs?: number, + idleTimeoutMs = this.#idleTimeoutMs, ): AsyncGenerator> { if (!this.#socket || this.#socket.readyState !== WebSocket.OPEN) { throw createCodexWebSocketTransportError("websocket connection is unavailable"); @@ -2405,14 +2568,23 @@ class CodexWebSocketConnection { try { this.#socket.send(JSON.stringify(request)); let sawFirstProgress = false; - let lastProgressAt = Date.now(); + const startedAt = Date.now(); + let lastProgressAt = startedAt; while (true) { - let timeoutMs = firstEventTimeoutMs; + let timeoutMs = + firstEventTimeoutMs === undefined ? undefined : firstEventTimeoutMs - (Date.now() - startedAt); if (sawFirstProgress) { - timeoutMs = this.#idleTimeoutMs - (Date.now() - lastProgressAt); + timeoutMs = idleTimeoutMs - (Date.now() - lastProgressAt); if (timeoutMs <= 0) { + this.close("idle-timeout"); throw createCodexWebSocketTransportError("idle timeout waiting for websocket"); } + } else if (timeoutMs !== undefined && timeoutMs <= 0) { + this.close("first-event-timeout"); + throw createCodexWebSocketTransportError( + "timeout waiting for first websocket event", + STREAM_FIRST_EVENT_TIMEOUT_PROVIDER_CODE, + ); } const next = await this.#nextMessage( timeoutMs, @@ -2429,20 +2601,19 @@ class CodexWebSocketConnection { sawFirstProgress = true; lastProgressAt = Date.now(); } - yield next; const eventType = typeof next.type === "string" ? next.type : ""; - if ( + const terminal = eventType === "response.completed" || eventType === "response.done" || eventType === "response.incomplete" || eventType === "response.failed" || - eventType === "error" - ) { - break; - } + eventType === "error"; + yield next; + if (terminal) break; } } finally { this.#activeRequest = false; + this.#queue.length = 0; if (signal) { signal.removeEventListener("abort", onAbort); } @@ -2485,9 +2656,9 @@ class CodexWebSocketConnection { await promise; if (timeout) clearTimeout(timeout); if (timedOut && this.#queue.length === 0) { - if (providerCode === STREAM_FIRST_EVENT_TIMEOUT_PROVIDER_CODE) { - this.close("first-event-timeout"); - } + this.close( + providerCode === STREAM_FIRST_EVENT_TIMEOUT_PROVIDER_CODE ? "first-event-timeout" : "idle-timeout", + ); return createCodexWebSocketTransportError(timeoutReason, providerCode); } } @@ -2590,7 +2761,12 @@ async function openCodexWebSocketEventStream( firstEventTimeoutMs?: number, ): Promise>> { const connection = await getOrCreateCodexWebSocketConnection(state, url, headers, signal, options); - return connection.streamRequest(request, signal, firstEventTimeoutMs); + return connection.streamRequest( + request, + signal, + firstEventTimeoutMs, + getCodexWebSocketIdleTimeoutMs(options?.streamIdleTimeoutMs), + ); } function createCodexHeaders( diff --git a/packages/ai/test/anthropic-stream-envelope.test.ts b/packages/ai/test/anthropic-stream-envelope.test.ts index 656b1bfad7..c030ed1b69 100644 --- a/packages/ai/test/anthropic-stream-envelope.test.ts +++ b/packages/ai/test/anthropic-stream-envelope.test.ts @@ -607,12 +607,8 @@ describe("anthropic stream envelope handling", () => { } const result = await stream.result(); - // The orphaned block keeps its streamed arguments and sheds internal - // stream-only fields; the replacement block owns subsequent deltas. - expect(result.content).toEqual([ - { type: "toolCall", id: "tool_orphaned", name: "bash", arguments: { command: "pwd" } }, - { type: "toolCall", id: "tool_replacement", name: "bash", arguments: { command: "ls" } }, - ]); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/reused an active content block index/i); }); it("round-trips OAuth tool prefixes without stripping original tool names that contain the prefix", () => { @@ -1433,4 +1429,46 @@ describe("anthropic stream envelope handling", () => { // Non-Claude models on unknown compatible endpoints receive no generated caching. expect(cacheControls[3]).toBeUndefined(); }); + it("coerces a non-string thinking increment to an empty string instead of forwarding the live value", async () => { + const thinkingModel: Model<"anthropic-messages"> = { + ...model, + id: "claude-sonnet-4-6", + thinking: { mode: "anthropic-adaptive", minLevel: Effort.Minimal, maxLevel: Effort.Max }, + }; + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => + createMockRequest([ + { + type: "message_start", + message: { id: "msg_zai_thinking", usage: { input_tokens: 0, output_tokens: 0 } }, + }, + { type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: 1 } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta" } }, + { + type: "content_block_delta", + index: 0, + delta: { type: "thinking_delta", thinking: "later" }, + }, + { type: "content_block_delta", index: 0, delta: { type: "signature_delta", signature: 7 } }, + { + type: "content_block_delta", + index: 0, + delta: { type: "signature_delta", signature: "sig_ok" }, + }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) as never, + ); + + const stream = streamAnthropic(thinkingModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true }); + const events: AssistantMessageEvent[] = []; + for await (const event of stream) events.push(event); + const result = await stream.result(); + + const deltas = events.filter(event => event.type === "thinking_delta"); + expect(deltas.map(event => event.delta)).toEqual(["", "", "later"]); + expect(result.content).toEqual([{ type: "thinking", thinking: "later", thinkingSignature: "sig_ok" }]); + }); }); diff --git a/packages/ai/test/anthropic-toolcall-increment-guard.test.ts b/packages/ai/test/anthropic-toolcall-increment-guard.test.ts new file mode 100644 index 0000000000..d2da18c6b4 --- /dev/null +++ b/packages/ai/test/anthropic-toolcall-increment-guard.test.ts @@ -0,0 +1,261 @@ +import { afterEach, describe, expect, it, vi } from "bun:test"; +import { Messages } from "@anthropic-ai/sdk/resources/messages/messages"; +import * as utils from "@gajae-code/utils"; +import { Effort } from "../src/model-thinking"; +import { streamAnthropic } from "../src/providers/anthropic"; +import type { AssistantMessageEvent, Context, Model } from "../src/types"; +import type { AssistantMessageEventStream } from "../src/utils/event-stream"; + +// Review follow-up for the primitive-increment degradation (PR #4612): +// tool-argument increments carry executable intent, so a malformed +// every non-string `input_json_delta.partial_json` must fail the turn closed +// instead of being silently erased to "". Primitive prose/thinking anomalies +// still degrade with a bounded diagnostic; executable fragments never do. + +const model: Model<"anthropic-messages"> = { + id: "claude-sonnet-4-5", + name: "Claude Sonnet 4.5", + api: "anthropic-messages", + provider: "anthropic", + baseUrl: "https://api.anthropic.com", + reasoning: true, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8_192, +}; + +const context: Context = { + messages: [{ role: "user", content: "run the tool", timestamp: Date.now() }], +}; + +type MockEvent = Record; +type MockStream = AsyncIterable; +type MockRequest = { + withResponse(): Promise<{ + data: MockStream; + response: Response; + request_id: string | null; + }>; +}; + +function mockRequest(events: MockEvent[]): MockRequest { + const response = new Response(null, { status: 200, headers: { "request-id": "req_mock" } }); + const stream: MockStream = { + async *[Symbol.asyncIterator]() { + for (const event of events) yield event; + }, + }; + return { + async withResponse() { + return { data: stream, response, request_id: response.headers.get("request-id") }; + }, + }; +} + +function toolUseStreamEvents(jsonDeltas: unknown[]): MockEvent[] { + return [ + { + type: "message_start", + message: { id: "msg_tool", usage: { input_tokens: 10, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "toolu_1", name: "write_file", input: {} }, + }, + ...jsonDeltas.map(partialJson => ({ + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: partialJson }, + })), + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 5 } }, + { type: "message_stop" }, + ]; +} + +async function drain(stream: AssistantMessageEventStream): Promise { + const events: AssistantMessageEvent[] = []; + for await (const event of stream) events.push(event); + return events; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("anthropic tool-argument increment guard", () => { + it("fails the turn closed when an input_json_delta increment is object-shaped", async () => { + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => + mockRequest( + toolUseStreamEvents(['{"path":"a.ts",', { content: "injected object increment" }, "}"]), + ) as never, + ); + + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const events = await drain(stream); + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/tool-argument|input_json_delta/i); + // The malformed payload must never leak into the surfaced error. + expect(result.errorMessage ?? "").not.toContain("injected object increment"); + expect(events.find(event => event.type === "done")).toBeUndefined(); + }); + + it("fails the turn closed when an input_json_delta increment is function-shaped", async () => { + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + const fnIncrement = () => '{"path":"a.ts"}'; + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => mockRequest(toolUseStreamEvents(['{"path":', fnIncrement])) as never, + ); + + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + await drain(stream); + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/tool-argument|input_json_delta/i); + }); + + it("fails the turn closed when an input_json_delta increment is a primitive (valid-but-wrong assembly)", async () => { + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + // `{"n":1` + numeric primitive + `3}` would assemble as {"n":13} if the + // primitive were erased to "" — a silently different tool call. The turn + // must fail closed instead of executing assembled-wrong arguments. + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => mockRequest(toolUseStreamEvents(['{"n":1', 2, "3}"])) as never, + ); + + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const events = await drain(stream); + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string input_json_delta tool-argument/i); + expect(result.errorMessage ?? "").not.toContain('{"n":1'); + expect(events.find(event => event.type === "done")).toBeUndefined(); + }); + + it("fails the turn closed when an input_json_delta increment is missing", async () => { + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => mockRequest(toolUseStreamEvents(['{"path":"a.ts",', undefined])) as never, + ); + + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + await drain(stream); + const result = await stream.result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string input_json_delta tool-argument/i); + }); + + it("marks a tool call incomplete when message_stop arrives without content_block_stop", async () => { + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => + mockRequest([ + { + type: "message_start", + message: { id: "msg_unfinalized", usage: { input_tokens: 1, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "tool_unfinalized", name: "write_file", input: {} }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "input_json_delta", partial_json: '{"path":"a"}' }, + }, + { type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) as never, + ); + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + await drain(stream); + const result = await stream.result(); + const tool = result.content.find(block => block.type === "toolCall"); + expect(tool).toMatchObject({ incompleteArguments: true, incompleteArgumentsReason: "truncated" }); + }); + + it("fails closed on non-object completed tool arguments", async () => { + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => mockRequest(toolUseStreamEvents(["[]"])) as never, + ); + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + await drain(stream); + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-object arguments/i); + }); + + it("keeps primitive thinking-delta degradation and emits one bounded warning for it", async () => { + const warnSpy = vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + const thinkingModel: Model<"anthropic-messages"> = { + ...model, + id: "claude-sonnet-4-6", + thinking: { mode: "anthropic-adaptive", minLevel: Effort.Minimal, maxLevel: Effort.Max }, + }; + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => + mockRequest([ + { + type: "message_start", + message: { id: "msg_think", usage: { input_tokens: 10, output_tokens: 0 } }, + }, + { type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: 1 } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta" } }, + { type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "later" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) as never, + ); + + const stream = streamAnthropic(thinkingModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true }); + await drain(stream); + const result = await stream.result(); + + expect(result.stopReason).toBe("stop"); + expect(result.content).toEqual([{ type: "thinking", thinking: "later", thinkingSignature: "" }]); + const degradeWarns = warnSpy.mock.calls.filter( + ([message]) => typeof message === "string" && message.includes("degraded non-string stream increment"), + ); + expect(degradeWarns).toHaveLength(1); + expect(degradeWarns[0]?.[1]).toHaveProperty("deltaType", "thinking_delta"); + }); + + it("uses one captured text_delta value", async () => { + let reads = 0; + const delta = { type: "text_delta" } as Record; + Object.defineProperty(delta, "text", { + enumerable: true, + get() { + reads += 1; + return reads <= 3 ? "captured" : { injected: true }; + }, + }); + vi.spyOn(Messages.prototype, "create").mockImplementation( + () => + mockRequest([ + { type: "message_start", message: { id: "msg_text", usage: { input_tokens: 1, output_tokens: 0 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) as never, + ); + const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + await drain(stream); + const result = await stream.result(); + expect(reads).toBe(3); + expect(result.content).toEqual([{ type: "text", text: "captured" }]); + }); +}); diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index ab16953bd6..e8d797f0e9 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/packages/ai/test/openai-codex-stream.test.ts @@ -86,7 +86,7 @@ function getRequestSignal(input: string | URL | Request, init: RequestInit | und return undefined; } -function createNoProgressCodexSse(signal: AbortSignal | undefined): Response { +function createMalformedDeltaCodexSse(signal: AbortSignal | undefined): Response { const encoder = new TextEncoder(); let interval: NodeJS.Timeout | undefined; let abortListener: (() => void) | undefined; @@ -97,19 +97,24 @@ function createNoProgressCodexSse(signal: AbortSignal | undefined): Response { encode({ type: "response.output_item.added", item: { - type: "function_call", - id: "fc_stalled", - call_id: "call_stalled", - name: "todo_write", - arguments: "", + type: "message", + id: "msg_stalled", + role: "assistant", + status: "in_progress", + content: [], }, }), ); + controller.enqueue(encode({ type: "response.content_part.added", part: { type: "output_text", text: "" } })); + const malformedDeltas: unknown[] = [42, undefined, ""]; + let malformedDeltaIndex = 0; interval = setInterval(() => { controller.enqueue( encode({ - type: "response.in_progress", - response: { id: "resp_stalled", status: "in_progress" }, + type: "response.output_text.delta", + item_id: "msg_stalled", + output_index: 0, + delta: malformedDeltas[malformedDeltaIndex++ % malformedDeltas.length], }), ); }, 2); @@ -402,13 +407,13 @@ describe("openai-codex streaming", () => { expect(result.stopReason).toBe("error"); }); - it("times out SSE streams that only emit no-progress status events", async () => { + it("times out SSE streams whose repeated malformed deltas are not semantic progress", async () => { const tempDir = TempDir.createSync("@pi-codex-stream-"); setAgentDir(tempDir.path()); const token = createCodexTestToken(); const context = createCodexTestContext(); global.fetch = ((input: string | URL | Request, init?: RequestInit) => - Promise.resolve(createNoProgressCodexSse(getRequestSignal(input, init)))) as typeof fetch; + Promise.resolve(createMalformedDeltaCodexSse(getRequestSignal(input, init)))) as typeof fetch; const model = { ...createCodexTestModel("https://chatgpt.com/backend-api"), preferWebsockets: false }; const result = await streamOpenAICodexResponses(model, context, { @@ -418,15 +423,7 @@ describe("openai-codex streaming", () => { expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("OpenAI Codex SSE stream stalled while waiting for the next event"); - expect(result.content as unknown[]).toEqual([ - { - type: "toolCall", - id: "call_stalled|fc_stalled", - name: "todo_write", - arguments: {}, - partialJson: "", - }, - ]); + expect(result.content as unknown[]).toEqual([{ type: "text", text: "", textSignature: undefined }]); }); it("parses websocket JSON from non-string payloads", async () => { @@ -2449,9 +2446,11 @@ describe("openai-codex streaming", () => { let sendCount = 0; let interval: NodeJS.Timeout | undefined; + const sockets: NoProgressWebSocket[] = []; class NoProgressWebSocket extends MockWebSocket { constructor(url: string, options?: { headers?: WsHeaders }) { super(url, options); + sockets.push(this); this.scheduleOpen(); } @@ -2495,6 +2494,7 @@ describe("openai-codex streaming", () => { expect(result.stopReason).toBe("stop"); expect(result.errorMessage).toBeUndefined(); expect(fetchMock).toHaveBeenCalledTimes(1); + expect(sockets[0]?.readyState).toBe(MockWebSocket.CLOSED); const transportDetails = getOpenAICodexTransportDetails(model, { sessionId: "ws-no-progress-session", providerSessionState, @@ -2503,6 +2503,41 @@ describe("openai-codex streaming", () => { expect(transportDetails.websocketDisabled).toBe(true); }); + it("bounds the websocket first-progress window across repeated no-op deltas", async () => { + const tempDir = TempDir.createSync("@pi-codex-stream-"); + setAgentDir(tempDir.path()); + let interval: NodeJS.Timeout | undefined; + class NoFirstProgressWebSocket extends MockWebSocket { + constructor(url: string, options?: { headers?: WsHeaders }) { + super(url, options); + this.scheduleOpen(); + } + send(): void { + interval = setInterval(() => { + this.sendJson({ type: "response.output_text.delta", delta: 42 }); + }, 2); + } + close(): void { + if (interval) clearInterval(interval); + super.close(); + } + } + global.WebSocket = NoFirstProgressWebSocket as unknown as typeof WebSocket; + const result = await streamOpenAICodexResponses( + createCodexTestModel("https://chatgpt.com/backend-api"), + createCodexTestContext(), + { + apiKey: createCodexTestToken(), + sessionId: "ws-first-progress-noop", + providerSessionState: new Map(), + streamFirstEventTimeoutMs: 20, + streamIdleTimeoutMs: 500, + }, + ).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toContain("timeout waiting for first websocket event"); + }); + it("retries websocket stream closes before surfacing transport errors", async () => { const tempDir = TempDir.createSync("@pi-codex-stream-"); setAgentDir(tempDir.path()); @@ -3302,6 +3337,42 @@ describe("openai-codex streaming", () => { expect(transportDetails.canAppend).toBe(true); }); + it("applies each reused websocket request's idle timeout", async () => { + const tempDir = TempDir.createSync("@pi-codex-stream-"); + setAgentDir(tempDir.path()); + let sendCount = 0; + class RequestScopedIdleWebSocket extends MockWebSocket { + constructor(url: string, options?: { headers?: WsHeaders }) { + super(url, options); + this.scheduleOpen(); + } + send(): void { + sendCount += 1; + if (sendCount === 1) { + this.emitCodexResponse({ messageId: "msg_first", responseId: "resp_first", text: "first" }); + return; + } + this.sendJson({ type: "response.created", response: { id: "resp_stalled", status: "in_progress" } }); + } + } + global.WebSocket = RequestScopedIdleWebSocket as unknown as typeof WebSocket; + const model = createCodexTestModel("https://chatgpt.com/backend-api"); + const providerSessionState = new Map(); + const options = { apiKey: createCodexTestToken(), sessionId: "ws-request-idle", providerSessionState }; + await streamOpenAICodexResponses(model, createCodexTestContext(), { + ...options, + streamIdleTimeoutMs: 1_000, + }).result(); + const startedAt = Date.now(); + const stalled = await streamOpenAICodexResponses(model, createCodexTestContext(), { + ...options, + streamIdleTimeoutMs: 20, + streamMaxRetries: 0, + }).result(); + expect(stalled.stopReason).toBe("error"); + expect(Date.now() - startedAt).toBeLessThan(600); + }); + it("replays x-codex-turn-state on subsequent SSE requests", async () => { const tempDir = TempDir.createSync("@pi-codex-stream-"); setAgentDir(tempDir.path()); diff --git a/packages/ai/test/openai-codex-toolcall-increment-guard.test.ts b/packages/ai/test/openai-codex-toolcall-increment-guard.test.ts new file mode 100644 index 0000000000..749e468f85 --- /dev/null +++ b/packages/ai/test/openai-codex-toolcall-increment-guard.test.ts @@ -0,0 +1,686 @@ +import { afterEach, describe, expect, it, vi } from "bun:test"; +import * as utils from "@gajae-code/utils"; +import { getAgentDir, setAgentDir, TempDir } from "@gajae-code/utils"; +import { streamOpenAICodexResponses } from "../src/providers/openai-codex-responses"; +import type { Context, Model, ToolCall } from "../src/types"; + +// Review follow-up for the primitive-increment degradation (PR #4612): +// every non-string tool-argument increment on the Codex Responses stream must +// fail the turn closed instead of being silently erased to "". Primitive +// prose/reasoning anomalies still degrade with a bounded diagnostic. + +const originalFetch = global.fetch; +const originalAgentDir = getAgentDir(); +afterEach(() => { + global.fetch = originalFetch; + setAgentDir(originalAgentDir); + vi.restoreAllMocks(); +}); + +function token(): string { + const payload = Buffer.from( + JSON.stringify({ "https://api.openai.com/auth": { chatgpt_account_id: "acc_test" } }), + "utf8", + ).toBase64(); + return `aaa.${payload}.bbb`; +} + +function model(): Model<"openai-codex-responses"> { + return { + id: "gpt-5.3-codex-spark", + name: "Codex", + api: "openai-codex-responses", + provider: "openai-codex", + baseUrl: "https://chatgpt.com/backend-api", + reasoning: true, + preferWebsockets: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 128000, + maxTokens: 128000, + }; +} + +function context(): Context { + return { systemPrompt: ["You are helpful."], messages: [{ role: "user", content: "go", timestamp: Date.now() }] }; +} + +function sse(events: unknown[]): string { + return `${events.map(e => `data: ${JSON.stringify(e)}`).join("\n\n")}\n\n`; +} + +function mockFetchOnce(body: string): void { + const fn = async (): Promise => + new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }); + global.fetch = Object.assign(fn, { preconnect: originalFetch.preconnect }); +} + +const USAGE = { input_tokens: 5, output_tokens: 3, total_tokens: 8, input_tokens_details: { cached_tokens: 0 } }; + +describe("openai-codex: tool-argument increment guard", () => { + it("fails the turn closed when a function_call arguments delta is object-shaped", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "write_file", arguments: "" }, + }, + { + type: "response.function_call_arguments.delta", + item_id: "fc_1", + output_index: 0, + delta: { path: "a.ts", content: "injected object increment" }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { + apiKey: token(), + streamMaxRetries: 0, + }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/tool-argument|arguments.delta/i); + expect(result.errorMessage ?? "").not.toContain("injected object increment"); + }); + + it("fails the turn closed when a function_call arguments delta is function-shaped", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + const parse = JSON.parse; + const fnIncrement = () => '{"path":"a.ts"}'; + JSON.parse = ((source: string, reviver?: (key: string, value: unknown) => unknown) => { + const value = parse(source, reviver) as Record; + if (value.type === "response.function_call_arguments.delta" && value.delta === "__fn__") { + value.delta = fnIncrement; + } + return value; + }) as typeof JSON.parse; + try { + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "write_file", arguments: "" }, + }, + { + type: "response.function_call_arguments.delta", + item_id: "fc_1", + output_index: 0, + delta: "__fn__", + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { + apiKey: token(), + streamMaxRetries: 0, + }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/tool-argument|arguments.delta/i); + expect(result.errorMessage ?? "").not.toContain("a.ts"); + } finally { + JSON.parse = parse; + } + }); + + it("uses one captured function_call arguments delta value", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + const parse = JSON.parse; + let reads = 0; + JSON.parse = ((source: string, reviver?: (key: string, value: unknown) => unknown) => { + const value = parse(source, reviver) as Record; + if (value.type === "response.function_call_arguments.delta" && value.delta === "__getter__") { + Object.defineProperty(value, "delta", { + enumerable: true, + get() { + reads += 1; + return reads <= 2 ? '{"safe":true}' : { injected: true }; + }, + }); + } + return value; + }) as typeof JSON.parse; + try { + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_getter", call_id: "call_getter", name: "run", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", delta: "__getter__" }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_getter", + call_id: "call_getter", + name: "run", + arguments: '{"safe":true}', + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("toolUse"); + expect(reads).toBe(2); + } finally { + JSON.parse = parse; + } + }); + + it("fails the turn closed when a custom_tool_call input delta is object-shaped", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "custom_tool_call", id: "ct_1", call_id: "call_1", name: "apply_patch", input: "" }, + }, + { + type: "response.custom_tool_call_input.delta", + item_id: "ct_1", + output_index: 0, + delta: { patch: "injected object increment" }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { + apiKey: token(), + streamMaxRetries: 0, + }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/tool-argument|custom_tool_call_input.delta/i); + }); + + it("fails closed on non-string initial custom-tool input", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_initial", + call_id: "call_initial", + name: "deploy", + input: ["unsafe"], + }, + }, + { type: "response.custom_tool_call_input.delta", delta: "" }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string input/i); + }); + + it("uses one captured initial custom-tool input value", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + const parse = JSON.parse; + let reads = 0; + JSON.parse = ((source: string, reviver?: (key: string, value: unknown) => unknown) => { + const value = parse(source, reviver) as Record; + const item = value.item as Record | undefined; + if (value.type === "response.output_item.added" && item?.input === "__getter__") { + Object.defineProperty(item, "input", { + enumerable: true, + get() { + reads += 1; + return reads <= 2 ? "exact" : ["unsafe"]; + }, + }); + } + return value; + }) as typeof JSON.parse; + try { + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_getter", + call_id: "call_getter", + name: "deploy", + input: "__getter__", + }, + }, + { + type: "response.output_item.done", + item: { + type: "custom_tool_call", + id: "ct_getter", + call_id: "call_getter", + name: "deploy", + input: "exact", + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("toolUse"); + expect(reads).toBe(1); + } finally { + JSON.parse = parse; + } + }); + + it("fails closed when a completed custom tool lacks output_item.done", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_unfinalized", + call_id: "call_unfinalized", + name: "deploy", + input: "", + }, + }, + { type: "response.custom_tool_call_input.done", input: "unsafe" }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/unfinalized tool call/i); + }); + + it("removes internal custom-tool fields from truncated successful output", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_truncated", + call_id: "call_truncated", + name: "deploy", + input: "", + }, + }, + { type: "response.custom_tool_call_input.delta", delta: "partial" }, + { type: "response.incomplete", response: { status: "incomplete", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("length"); + const tools = result.content.filter((block): block is ToolCall => block.type === "toolCall"); + expect(tools[0]).not.toHaveProperty("partialJson"); + expect(tools[0]).not.toHaveProperty("doneInput"); + }); + + it("fails the turn closed when a function_call arguments delta is a primitive (valid-but-wrong assembly)", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + // `{"n":1` + numeric primitive + `3}` would assemble as {"n":13} if the + // primitive were erased — a silently different tool call. Terminal + // `item.arguments` must not launder a corrupted delta stream either. + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "run_job", arguments: "" }, + }, + { type: "response.function_call_arguments.delta", item_id: "fc_1", output_index: 0, delta: '{"n":1' }, + { type: "response.function_call_arguments.delta", item_id: "fc_1", output_index: 0, delta: 2 }, + { type: "response.function_call_arguments.delta", item_id: "fc_1", output_index: 0, delta: "3}" }, + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "run_job", + arguments: '{"n":13}', + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string response\.function_call_arguments\.delta/i); + }); + + it("finalizes a custom_tool_call from terminal input and fails closed on a buffer mismatch", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + // Streamed buffer assembles "corrupt" while terminal item.input says + // "authoritative": the mismatch must fail the turn instead of executing + // either variant. + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "custom_tool_call", id: "ct_1", call_id: "call_ct", name: "deploy", input: "" }, + }, + { type: "response.custom_tool_call_input.delta", item_id: "ct_1", output_index: 0, delta: "corru" }, + { type: "response.custom_tool_call_input.delta", item_id: "ct_1", output_index: 0, delta: "pt" }, + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "custom_tool_call", + id: "ct_1", + call_id: "call_ct", + name: "deploy", + input: "authoritative", + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/terminal input disagrees with the streamed input buffer/i); + }); + + it("does not let custom_tool_call_input.done erase a streamed buffer mismatch", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { type: "custom_tool_call", id: "ct_done", call_id: "call_done", name: "deploy", input: "" }, + }, + { type: "response.custom_tool_call_input.delta", delta: "corrupt" }, + { type: "response.custom_tool_call_input.done", input: "authoritative" }, + { + type: "response.output_item.done", + item: { + type: "custom_tool_call", + id: "ct_done", + call_id: "call_done", + name: "deploy", + input: "authoritative", + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/disagrees with the streamed input buffer/i); + }); + + it("retains done-only custom input and rejects terminal disagreement", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_done_only", + call_id: "call_done_only", + name: "deploy", + input: "", + }, + }, + { type: "response.custom_tool_call_input.done", input: "exact" }, + { + type: "response.output_item.done", + item: { + type: "custom_tool_call", + id: "ct_done_only", + call_id: "call_done_only", + name: "deploy", + input: "exact", + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const valid = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(valid.stopReason).toBe("toolUse"); + const tools = valid.content.filter((block): block is ToolCall => block.type === "toolCall"); + expect(tools[0]?.arguments).toEqual({ input: "exact" }); + expect(tools[0]).not.toHaveProperty("partialJson"); + expect(tools[0]).not.toHaveProperty("doneInput"); + + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_conflict", + call_id: "call_conflict", + name: "deploy", + input: "", + }, + }, + { type: "response.custom_tool_call_input.done", input: "safe" }, + { + type: "response.output_item.done", + item: { + type: "custom_tool_call", + id: "ct_conflict", + call_id: "call_conflict", + name: "deploy", + input: "dangerous", + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const conflict = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(conflict.stopReason).toBe("error"); + expect(conflict.errorMessage ?? "").toMatch(/input\.done|terminal input/i); + const conflictTools = conflict.content.filter((block): block is ToolCall => block.type === "toolCall"); + expect(conflictTools[0]).not.toHaveProperty("partialJson"); + expect(conflictTools[0]).not.toHaveProperty("doneInput"); + }); + + it("fails closed on non-string custom_tool_call_input.done", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "custom_tool_call", + id: "ct_bad_done", + call_id: "call_bad_done", + name: "deploy", + input: "", + }, + }, + { type: "response.custom_tool_call_input.done", input: 42 }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string.*input\.done/i); + }); + + it("finalizes a custom_tool_call from matching terminal input without error", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + vi.spyOn(utils.logger, "warn").mockImplementation(() => {}); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "custom_tool_call", id: "ct_2", call_id: "call_ct2", name: "deploy", input: "" }, + }, + { type: "response.custom_tool_call_input.delta", item_id: "ct_2", output_index: 0, delta: "exact" }, + { + type: "response.output_item.done", + output_index: 0, + item: { type: "custom_tool_call", id: "ct_2", call_id: "call_ct2", name: "deploy", input: "exact" }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("toolUse"); + const tools = result.content.filter((b): b is ToolCall => b.type === "toolCall"); + expect(tools).toHaveLength(1); + expect(tools[0].arguments).toEqual({ input: "exact" }); + }); + + it("commits terminal function identity and arguments to stored output", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_authority", + call_id: "call_authority", + name: "run_job", + arguments: "", + }, + }, + { type: "response.function_call_arguments.delta", delta: '{"n":1}' }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_authority", + call_id: "call_authority", + name: "run_job", + arguments: '{"n":1}', + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + const tools = result.content.filter((block): block is ToolCall => block.type === "toolCall"); + expect(tools).toEqual([ + { type: "toolCall", id: "call_authority|fc_authority", name: "run_job", arguments: { n: 1 } }, + ]); + }); + + it("fails closed on non-string terminal function arguments", async () => { + for (const terminalArguments of [undefined, 0, false, null, ["unsafe"], { unsafe: true }]) { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "function_call", + id: "fc_terminal", + call_id: "call_terminal", + name: "run_job", + arguments: terminalArguments, + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string terminal arguments/i); + } + }); + + it("fails closed on malformed string terminal function arguments", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_malformed", + call_id: "call_malformed", + name: "run", + arguments: "", + }, + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_malformed", + call_id: "call_malformed", + name: "run", + arguments: '{"command":"dangerous"', + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/malformed terminal arguments/i); + expect(result.errorMessage ?? "").not.toContain("dangerous"); + }); + + it("fails closed on empty terminal function arguments", async () => { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_empty", call_id: "call_empty", name: "run", arguments: "" }, + }, + { + type: "response.output_item.done", + item: { type: "function_call", id: "fc_empty", call_id: "call_empty", name: "run", arguments: "" }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/malformed terminal arguments/i); + }); + + it("fails closed on non-string terminal custom-tool input", async () => { + for (const terminalInput of [undefined, 0, false, null, ["unsafe"], { unsafe: true }]) { + setAgentDir(TempDir.createSync("@pi-codex-increment-").path()); + mockFetchOnce( + sse([ + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "custom_tool_call", + id: "ct_terminal", + call_id: "call_terminal", + name: "deploy", + input: terminalInput, + }, + }, + { type: "response.completed", response: { status: "completed", usage: USAGE } }, + ]), + ); + + const result = await streamOpenAICodexResponses(model(), context(), { apiKey: token() }).result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage ?? "").toMatch(/non-string terminal input/i); + } + }); +});