From 1a09ae5ca816513c440c515d25162167eaee2e41 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sat, 27 Jun 2026 22:08:23 +0000 Subject: [PATCH 1/8] Thread Tzafon requests with previous_response_id + delta input Add a shared response-threading capability (responseThreadingEnabled with CUA_DISABLE_RESPONSE_THREADING opt-out) and a pure responseThreadingDelta util that finds the most recent assistant responseId and returns the messages after it. Refactor the Tzafon request building into a pure buildTzafonRequestInput that, when threading is enabled and a prior responseId exists, chains via previous_response_id with store:true and sends only the delta screenshot instead of replaying the full screenshot history that overflows the window. Covered by a failure-mode test locking the off-path full-history growth and asserting the on-path delta + previous_response_id + store. Co-Authored-By: Claude Opus 4.7 --- packages/ai/src/providers/common.ts | 58 +++++++- packages/ai/src/providers/tzafon/index.ts | 3 +- packages/ai/src/providers/tzafon/provider.ts | 68 ++++++++-- packages/ai/test/tzafon-threading.test.ts | 131 +++++++++++++++++++ 4 files changed, 246 insertions(+), 14 deletions(-) create mode 100644 packages/ai/test/tzafon-threading.test.ts diff --git a/packages/ai/src/providers/common.ts b/packages/ai/src/providers/common.ts index e829a6c0..4aed8a2e 100644 --- a/packages/ai/src/providers/common.ts +++ b/packages/ai/src/providers/common.ts @@ -1,4 +1,14 @@ -import { Type, type Api, type Model, type SimpleStreamOptions, type Static, type TSchema, type Tool } from "@earendil-works/pi-ai"; +import { + Type, + type Api, + type AssistantMessage, + type Message, + type Model, + type SimpleStreamOptions, + type Static, + type TSchema, + type Tool, +} from "@earendil-works/pi-ai"; import type { CuaModelRef, CuaProvider } from "../models"; export const CUA_ACTION_TYPES = [ @@ -485,6 +495,52 @@ export interface CuaSimpleStreamOptions extends SimpleStreamOptions { keepToolNames?: readonly string[]; } +/** Environment variable that disables server-side `previous_response_id` threading when truthy. */ +export const CUA_DISABLE_RESPONSE_THREADING_ENV_VAR = "CUA_DISABLE_RESPONSE_THREADING"; + +/** Per-call control over `previous_response_id` threading for Responses API providers. */ +export interface ResponseThreadingOptions { + /** Force full-history replay for this request, overriding the environment default. */ + disableResponseThreading?: boolean; +} + +/** + * Whether a Responses API provider should thread requests with + * `previous_response_id` + delta input instead of replaying the full message + * history. Threading is on by default and disabled by an explicit option or a + * truthy {@link CUA_DISABLE_RESPONSE_THREADING_ENV_VAR}. + */ +export function responseThreadingEnabled(options?: ResponseThreadingOptions): boolean { + if (options?.disableResponseThreading) return false; + const flag = process.env[CUA_DISABLE_RESPONSE_THREADING_ENV_VAR]; + return !(flag && flag !== "0" && flag.toLowerCase() !== "false"); +} + +/** Result of {@link responseThreadingDelta}: the chaining id and the messages to send this turn. */ +export interface ResponseThreadingDelta { + /** Most recent assistant `responseId`, or undefined when no prior turn carries one. */ + previousResponseId?: string; + /** Messages to send: those after the latest assistant `responseId`, or all messages when none. */ + deltaMessages: Message[]; +} + +/** + * Derive the `previous_response_id` continuation from a message history. + * + * Scans for the most recent assistant message carrying a `responseId` and + * returns it alongside the messages that follow it (the turn's delta). When no + * assistant message carries a `responseId`, returns every message and no id. + */ +export function responseThreadingDelta(messages: readonly Message[]): ResponseThreadingDelta { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]!; + if (message.role === "assistant" && (message as AssistantMessage).responseId) { + return { previousResponseId: (message as AssistantMessage).responseId, deltaMessages: messages.slice(index + 1) }; + } + } + return { deltaMessages: [...messages] }; +} + /** * Runtime configuration for a supported CUA model. * diff --git a/packages/ai/src/providers/tzafon/index.ts b/packages/ai/src/providers/tzafon/index.ts index 11f3bfcc..4a5e029e 100644 --- a/packages/ai/src/providers/tzafon/index.ts +++ b/packages/ai/src/providers/tzafon/index.ts @@ -12,6 +12,7 @@ export type { ComputerToolsOptions, } from "../common"; export { + buildTzafonRequestInput, TZAFON_RESPONSES_API, streamSimpleTzafonResponses, streamTzafonResponses, @@ -19,7 +20,7 @@ export { tzafonComputerUseOnPayload, tzafonToolCallId, } from "./provider"; -export type { TzafonCanonicalAction, TzafonResponsesOptions } from "./provider"; +export type { TzafonCanonicalAction, TzafonRequestBody, TzafonRequestOptions, TzafonResponsesOptions } from "./provider"; // Provider-native action vocabulary. The model card lists supported actions; // the Responses API loop dispatches on `action.type` and adds terminal control diff --git a/packages/ai/src/providers/tzafon/provider.ts b/packages/ai/src/providers/tzafon/provider.ts index f39b511f..db64cc90 100644 --- a/packages/ai/src/providers/tzafon/provider.ts +++ b/packages/ai/src/providers/tzafon/provider.ts @@ -4,6 +4,7 @@ import { type AssistantMessage, type Context, type ImageContent, + type Message, type Model, type SimpleStreamOptions, type StreamFunction, @@ -13,7 +14,16 @@ import { type ToolCall, } from "@earendil-works/pi-ai"; import Lightcone from "@tzafon/lightcone"; -import { canonicalToolCallArguments, canonicalToolCallName, CUA_ACTION_TYPES, type CuaAction, type CuaPayloadContext } from "../common"; +import { + canonicalToolCallArguments, + canonicalToolCallName, + CUA_ACTION_TYPES, + responseThreadingDelta, + responseThreadingEnabled, + type CuaAction, + type CuaPayloadContext, + type ResponseThreadingOptions, +} from "../common"; export const TZAFON_RESPONSES_API = "tzafon-responses"; const TZAFON_COMPUTER_USE_TOOL = { @@ -25,11 +35,52 @@ const TZAFON_COMPUTER_USE_TOOL = { const TZAFON_LOCAL_ACTION_TOOL_NAMES = new Set(CUA_ACTION_TYPES); /** Stream options accepted by {@link streamTzafonResponses}. */ -export interface TzafonResponsesOptions extends StreamOptions { +export interface TzafonResponsesOptions extends StreamOptions, ResponseThreadingOptions { /** Tool names to keep in the outbound payload even though they collide with local CUA action tool names. */ keepToolNames?: readonly string[]; } +/** Inputs {@link buildTzafonRequestInput} reads to shape the Responses API request body. */ +export interface TzafonRequestOptions extends ResponseThreadingOptions { + temperature?: number; + maxTokens?: number; +} + +/** Responses API request body for {@link Lightcone.responses.create}, including optional threading fields. */ +export interface TzafonRequestBody { + model: string; + input: Array>; + tools: Array>; + instructions?: string; + temperature: number; + max_output_tokens?: number; + previous_response_id?: string; + store?: boolean; +} + +/** + * Build the Tzafon Responses API request body from a context. + * + * Pure and network-free. When response threading is enabled and a prior + * assistant `responseId` exists, the body chains via `previous_response_id` + * with `store: true` and sends only the delta messages; otherwise it replays + * the full message history. + */ +export function buildTzafonRequestInput(model: Model, context: Context, options?: TzafonRequestOptions): TzafonRequestBody { + const body: TzafonRequestBody = { + model: model.id, + input: convertMessages(context.messages), + tools: convertTools(context.tools ?? []), + instructions: context.systemPrompt, + temperature: options?.temperature ?? 0, + max_output_tokens: options?.maxTokens ?? model.maxTokens, + }; + if (!responseThreadingEnabled(options)) return body; + const { previousResponseId, deltaMessages } = responseThreadingDelta(context.messages); + if (!previousResponseId) return body; + return { ...body, input: convertMessages(deltaMessages), previous_response_id: previousResponseId, store: true }; +} + export const streamSimpleTzafonResponses: StreamFunction = (model, context, options) => { return streamTzafonResponses(model, context, options); }; @@ -43,14 +94,7 @@ export const streamTzafonResponses: StreamFunction, context, options); const tzafonPayload = tzafonComputerUseOnPayload(payload, model as Model, { keepToolNames: [...keepToolNamesFromContext(context), ...(options?.keepToolNames ?? [])], }); @@ -292,9 +336,9 @@ function readToolName(tool: unknown): string | undefined { return getString(fn, "name"); } -function convertContextMessages(context: Context): Array> { +function convertMessages(messages: readonly Message[]): Array> { const items: Array> = []; - for (const message of context.messages) { + for (const message of messages) { if (message.role === "user") { items.push({ role: "user", content: convertUserContent(message.content) }); continue; diff --git a/packages/ai/test/tzafon-threading.test.ts b/packages/ai/test/tzafon-threading.test.ts new file mode 100644 index 00000000..e0a0b03c --- /dev/null +++ b/packages/ai/test/tzafon-threading.test.ts @@ -0,0 +1,131 @@ +import { afterEach, describe, expect, it } from "vitest"; +import type { Context, Message, Model } from "@earendil-works/pi-ai"; +import { tzafon } from "../src/index"; + +const model = { id: "tzafon.northstar-cua-fast", maxTokens: 4096 } as Model; + +const TURNS = 6; + +/** Build a multi-turn context where each assistant turn carries a distinct responseId followed by a screenshot tool result. */ +function multiTurnContext(): Context { + const messages: Message[] = [{ role: "user", content: "book a flight", timestamp: 0 }]; + for (let turn = 0; turn < TURNS; turn += 1) { + messages.push({ + role: "assistant", + content: [{ type: "toolCall", id: `call_${turn}`, name: "click", arguments: { x: turn, y: turn } }], + api: tzafon.TZAFON_RESPONSES_API, + provider: "tzafon", + model: model.id, + responseId: `resp_${turn}`, + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "toolUse", + timestamp: 0, + }); + messages.push({ + role: "toolResult", + toolCallId: `call_${turn}`, + toolName: "click", + content: [{ type: "image", mimeType: "image/png", data: `screenshot-${turn}` }], + isError: false, + timestamp: 0, + }); + } + return { messages, tools: [], systemPrompt: "control the browser" }; +} + +function screenshotImageUrls(input: Array>): string[] { + const urls: string[] = []; + for (const item of input) { + const content = item.content; + if (!Array.isArray(content)) continue; + for (const part of content) { + if (part && typeof part === "object" && (part as { type?: string }).type === "input_image") { + urls.push((part as { image_url: string }).image_url); + } + } + } + return urls; +} + +describe("buildTzafonRequestInput response threading", () => { + afterEach(() => { + delete process.env.CUA_DISABLE_RESPONSE_THREADING; + }); + + // Threading ON is the fix: chain via previous_response_id and send only the + // latest screenshot. OFF replays every screenshot — the per-turn growth that + // overflows Tzafon's real 64K window after a few turns. + it("threads the latest delta when enabled (default)", () => { + const body = tzafon.buildTzafonRequestInput(model, multiTurnContext()); + const screenshots = screenshotImageUrls(body.input); + + expect(screenshots).toHaveLength(1); + expect(screenshots[0]).toBe(`data:image/png;base64,screenshot-${TURNS - 1}`); + expect(body.previous_response_id).toBe(`resp_${TURNS - 1}`); + expect(body.store).toBe(true); + }); + + it("replays the full screenshot history when threading is disabled by option (locks the failure mode)", () => { + const body = tzafon.buildTzafonRequestInput(model, multiTurnContext(), { disableResponseThreading: true }); + const screenshots = screenshotImageUrls(body.input); + + expect(screenshots).toHaveLength(TURNS); + expect(screenshots).toEqual(Array.from({ length: TURNS }, (_, turn) => `data:image/png;base64,screenshot-${turn}`)); + expect(body.previous_response_id).toBeUndefined(); + expect(body.store).toBeUndefined(); + }); + + it("replays the full screenshot history when CUA_DISABLE_RESPONSE_THREADING is set", () => { + process.env.CUA_DISABLE_RESPONSE_THREADING = "1"; + const body = tzafon.buildTzafonRequestInput(model, multiTurnContext()); + + expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); + expect(body.previous_response_id).toBeUndefined(); + }); + + it("falls back to full history when no prior turn carries a responseId", () => { + const context = multiTurnContext(); + for (const message of context.messages) { + if (message.role === "assistant") delete message.responseId; + } + + const body = tzafon.buildTzafonRequestInput(model, context); + expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); + expect(body.previous_response_id).toBeUndefined(); + }); + + // Off-path screenshot count scales with turn count; on-path stays constant at one. + it("grows the payload per turn when off but stays flat when on", () => { + const counts = (turns: number, disable: boolean) => { + const messages: Message[] = [{ role: "user", content: "task", timestamp: 0 }]; + for (let turn = 0; turn < turns; turn += 1) { + messages.push({ + role: "assistant", + content: [{ type: "toolCall", id: `c_${turn}`, name: "click", arguments: {} }], + api: tzafon.TZAFON_RESPONSES_API, + provider: "tzafon", + model: model.id, + responseId: `r_${turn}`, + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "toolUse", + timestamp: 0, + }); + messages.push({ + role: "toolResult", + toolCallId: `c_${turn}`, + toolName: "click", + content: [{ type: "image", mimeType: "image/png", data: `s-${turn}` }], + isError: false, + timestamp: 0, + }); + } + const body = tzafon.buildTzafonRequestInput(model, { messages, tools: [] }, { disableResponseThreading: disable }); + return screenshotImageUrls(body.input).length; + }; + + expect(counts(3, true)).toBe(3); + expect(counts(8, true)).toBe(8); + expect(counts(3, false)).toBe(1); + expect(counts(8, false)).toBe(1); + }); +}); From 8ea308ebceb0f0d418cc77bb179d6f74337e3c2f Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sat, 27 Jun 2026 22:21:24 +0000 Subject: [PATCH 2/8] Add cua openai-cua-responses stream provider with previous_response_id threading Give cua's OpenAI computer-use path its own Responses stream function instead of routing through pi-ai's builtin openai-responses adapter. streamOpenAIResponses calls the openai SDK client.responses.create directly and reuses the shared response-threading capability flag and delta util from the Tzafon phase: when a prior assistant responseId exists it chains via previous_response_id + store:true and sends only the delta screenshot, otherwise it replays the full history. Register it under OPENAI_CUA_RESPONSES_API and route every OpenAI CUA model to it in getCuaModel, including registry-resolved gpt-5.4/gpt-5.5 families that otherwise carry pi-ai's builtin api. Fold the store:true onPayload into the builder; pi-ai's openai-responses builtin is left untouched. Preserves the existing computer-use behavior: function-tool calls, pixel coordinates, system prompt, and store:true. Covered by a pure-builder threading test mirroring the Tzafon one and a routing assertion locking the api override. Co-Authored-By: Claude Opus 4.7 --- packages/ai/README.md | 5 +- packages/ai/src/models.ts | 16 +- packages/ai/src/providers.ts | 11 +- packages/ai/src/providers/openai/index.ts | 18 +- packages/ai/src/providers/openai/provider.ts | 355 +++++++++++++++++++ packages/ai/test/models.test.ts | 10 + packages/ai/test/openai-threading.test.ts | 135 +++++++ packages/ai/test/runtime-spec.test.ts | 4 +- 8 files changed, 537 insertions(+), 17 deletions(-) create mode 100644 packages/ai/src/providers/openai/provider.ts create mode 100644 packages/ai/test/openai-threading.test.ts diff --git a/packages/ai/README.md b/packages/ai/README.md index 6d0da7ae..4e7c2c43 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -484,7 +484,10 @@ Every provider namespace (`openai`, `anthropic`, `gemini`, `tzafon`, Provider-specific extras: -- `openai`: `openaiResponsesStoreOnPayload` payload hook, plus the +- `openai`: the `openai-cua-responses` stream adapter (`OPENAI_CUA_RESPONSES_API`, + `streamOpenAIResponses`, `streamSimpleOpenAIResponses`, `OpenAIResponsesOptions`), + the pure `buildOpenAIRequestInput` request builder (threads + `previous_response_id` + delta input with `store: true`), plus the `computer_use_extra` navigation aliases `OPENAI_EXTRA_TOOL_NAME`, `OPENAI_EXTRA_TOOL_DESCRIPTION`, `OpenAIExtraSchema`, `OpenAIExtraInput` - `anthropic`: `ANTHROPIC_BATCH_TOOL_NAME` (`"computer_batch"`) diff --git a/packages/ai/src/models.ts b/packages/ai/src/models.ts index 1119cc06..789e8ae6 100644 --- a/packages/ai/src/models.ts +++ b/packages/ai/src/models.ts @@ -4,6 +4,7 @@ import { getModel, getModels, } from "@earendil-works/pi-ai"; +import { OPENAI_CUA_RESPONSES_API } from "./providers/openai/provider"; /** Providers with curated computer-use model support. */ export type CuaProvider = "openai" | "anthropic" | "google" | "tzafon" | "yutori"; @@ -187,12 +188,21 @@ export function getCuaModel(ref: CuaModelRef): Model { throw new Error(`unsupported CUA model "${ref}"`); } const fromRegistry = getModel(provider as never, modelId as never) as Model | undefined; - if (fromRegistry) return fromRegistry; + if (fromRegistry) return routeCuaApi(fromRegistry); const override = CUA_MODEL_OVERRIDES[provider].find((m) => m.id === modelId); - if (override) return override; + if (override) return routeCuaApi(override); throw new Error(`CUA model "${ref}" is supported but not registered. Add it to pi-ai (models.dev) or CUA_MODEL_OVERRIDES.`); } +// Route OpenAI CUA models to cua's own openai-cua-responses stream provider, +// which threads previous_response_id. Registry-resolved models (gpt-5.4, +// gpt-5.4-mini, gpt-5.5) otherwise carry pi-ai's builtin "openai-responses" api. +function routeCuaApi(model: Model): Model { + return model.provider === "openai" && model.api !== OPENAI_CUA_RESPONSES_API + ? { ...model, api: OPENAI_CUA_RESPONSES_API } + : model; +} + /** Return the {@link CuaProvider} for a concrete model, or throw when it is not a CUA provider. */ export function providerForModel(model: Model): CuaProvider { if (!isCuaProvider(model.provider)) { @@ -249,7 +259,7 @@ function cuaModel(provider: CuaProvider, id: string, name: string): Model { switch (provider) { case "openai": - return { ...base, api: "openai-responses", baseUrl: "https://api.openai.com/v1", contextWindow: 400_000, maxTokens: 32_768 } as Model; + return { ...base, api: OPENAI_CUA_RESPONSES_API, baseUrl: "https://api.openai.com/v1", contextWindow: 400_000, maxTokens: 32_768 } as Model; case "anthropic": return { ...base, api: "anthropic-messages", baseUrl: "https://api.anthropic.com", contextWindow: 200_000, maxTokens: 64_000 } as Model; case "google": diff --git a/packages/ai/src/providers.ts b/packages/ai/src/providers.ts index ba47b2ae..75ed36b6 100644 --- a/packages/ai/src/providers.ts +++ b/packages/ai/src/providers.ts @@ -1,11 +1,14 @@ import { registerApiProvider } from "@earendil-works/pi-ai"; +import { OPENAI_CUA_RESPONSES_API, streamOpenAIResponses, streamSimpleOpenAIResponses } from "./providers/openai/provider"; import { streamSimpleTzafonResponses, streamTzafonResponses, TZAFON_RESPONSES_API } from "./providers/tzafon/provider"; import { streamSimpleYutori, streamYutori, YUTORI_CHAT_COMPLETIONS_API } from "./providers/yutori/provider"; // pi-ai eagerly registers openai-responses, anthropic-messages, and // google-generative-ai when its index module loads (see // node_modules/@earendil-works/pi-ai/dist/providers/register-builtins.js). -// CUA only needs to add the providers pi-ai does not ship: Tzafon and Yutori. +// CUA adds the providers pi-ai does not ship (Tzafon, Yutori) plus its own +// openai-cua-responses, which threads previous_response_id and is left +// alongside pi-ai's untouched openai-responses builtin. /** * Register the Yutori and Tzafon stream providers with pi-ai's global API @@ -27,7 +30,13 @@ export function registerCuaProviders(): void { stream: streamTzafonResponses, streamSimple: streamSimpleTzafonResponses, }); + registerApiProvider({ + api: OPENAI_CUA_RESPONSES_API, + stream: streamOpenAIResponses, + streamSimple: streamSimpleOpenAIResponses, + }); } +export { OPENAI_CUA_RESPONSES_API, streamOpenAIResponses, streamSimpleOpenAIResponses }; export { TZAFON_RESPONSES_API, streamSimpleTzafonResponses, streamTzafonResponses }; export { YUTORI_CHAT_COMPLETIONS_API, streamSimpleYutori, streamYutori }; diff --git a/packages/ai/src/providers/openai/index.ts b/packages/ai/src/providers/openai/index.ts index 1b1a7a1f..c5398de1 100644 --- a/packages/ai/src/providers/openai/index.ts +++ b/packages/ai/src/providers/openai/index.ts @@ -15,6 +15,13 @@ export type { ComputerToolsOptions, CuaNavigationInput as OpenAIExtraInput, } from "../common"; +export { + buildOpenAIRequestInput, + OPENAI_CUA_RESPONSES_API, + streamOpenAIResponses, + streamSimpleOpenAIResponses, +} from "./provider"; +export type { OpenAIRequestBody, OpenAIRequestOptions, OpenAIResponsesOptions } from "./provider"; // Provider-native action vocabulary emitted on `computer_call.action.type`: // click, double_click, drag, move, scroll, type, keypress, wait, screenshot @@ -29,20 +36,9 @@ export function buildOpenAISystemPrompt(opts: { suffix?: string } = {}): string return [OPENAI_COMPUTER_INSTRUCTIONS, opts.suffix].filter(Boolean).join("\n\n"); } -export function openaiResponsesStoreOnPayload(payload: unknown): unknown | undefined { - if (!payload || typeof payload !== "object") return undefined; - const current = payload as Record; - if (current.store === true) return undefined; - return { - ...current, - store: true, - }; -} - export const providerModule = { toolDefinitions: computerTools, toolExecutors: computerToolExecutors, coordinateSystem, buildSystemPrompt: buildOpenAISystemPrompt, - onPayload: openaiResponsesStoreOnPayload, } satisfies CuaProviderModule; diff --git a/packages/ai/src/providers/openai/provider.ts b/packages/ai/src/providers/openai/provider.ts new file mode 100644 index 00000000..d9a30c6b --- /dev/null +++ b/packages/ai/src/providers/openai/provider.ts @@ -0,0 +1,355 @@ +import OpenAI from "openai"; +import type { ResponseStreamEvent } from "openai/resources/responses/responses"; +import { + clampThinkingLevel, + createAssistantMessageEventStream, + parseStreamingJson, + type Api, + type AssistantMessage, + type Context, + type ImageContent, + type Message, + type Model, + type SimpleStreamOptions, + type StreamFunction, + type StreamOptions, + type TextContent, + type ToolCall, +} from "@earendil-works/pi-ai"; +import { responseThreadingDelta, responseThreadingEnabled, type ResponseThreadingOptions } from "../common"; + +export const OPENAI_CUA_RESPONSES_API = "openai-cua-responses"; + +/** Stream options accepted by {@link streamOpenAIResponses}. */ +export interface OpenAIResponsesOptions extends StreamOptions, ResponseThreadingOptions { + reasoningEffort?: "minimal" | "low" | "medium" | "high" | "xhigh"; + reasoningSummary?: "auto" | "detailed" | "concise" | null; +} + +/** Inputs {@link buildOpenAIRequestInput} reads to shape the Responses API request body. */ +export interface OpenAIRequestOptions extends ResponseThreadingOptions { + temperature?: number; + maxTokens?: number; + reasoningEffort?: OpenAIResponsesOptions["reasoningEffort"]; + reasoningSummary?: OpenAIResponsesOptions["reasoningSummary"]; +} + +/** Responses API request body for {@link OpenAI.responses.create}, including optional threading fields. */ +export interface OpenAIRequestBody { + model: string; + input: Array>; + tools: Array>; + instructions?: string; + stream: true; + store: boolean; + temperature?: number; + max_output_tokens?: number; + reasoning?: { effort?: string; summary?: string }; + include?: string[]; + previous_response_id?: string; +} + +/** + * Build the OpenAI Responses API request body from a context. + * + * Pure and network-free. The public OpenAI Responses API requires `store: true` + * for `previous_response_id` continuity, so the body always stores. When + * response threading is enabled and a prior assistant `responseId` exists, the + * body chains via `previous_response_id` and sends only the delta messages; + * otherwise it replays the full message history. + */ +export function buildOpenAIRequestInput(model: Model, context: Context, options?: OpenAIRequestOptions): OpenAIRequestBody { + const body: OpenAIRequestBody = { + model: model.id, + input: convertMessages(context.messages), + tools: convertTools(context.tools ?? []), + instructions: context.systemPrompt, + stream: true, + store: true, + max_output_tokens: options?.maxTokens ?? model.maxTokens, + }; + if (options?.temperature !== undefined) body.temperature = options.temperature; + if (model.reasoning && (options?.reasoningEffort || options?.reasoningSummary)) { + const effort = options.reasoningEffort + ? (model.thinkingLevelMap?.[options.reasoningEffort] ?? options.reasoningEffort) + : "medium"; + body.reasoning = { effort, summary: options.reasoningSummary ?? "auto" }; + body.include = ["reasoning.encrypted_content"]; + } + if (!responseThreadingEnabled(options)) return body; + const { previousResponseId, deltaMessages } = responseThreadingDelta(context.messages); + if (!previousResponseId) return body; + return { ...body, input: convertMessages(deltaMessages), previous_response_id: previousResponseId }; +} + +export const streamSimpleOpenAIResponses: StreamFunction = (model, context, options) => { + const clamped = options?.reasoning ? clampThinkingLevel(model, options.reasoning) : undefined; + const reasoningEffort = clamped && clamped !== "off" ? clamped : undefined; + return streamOpenAIResponses(model, context, { ...options, reasoningEffort }); +}; + +export const streamOpenAIResponses: StreamFunction = (model, context, options) => { + const stream = createAssistantMessageEventStream(); + const output = initialAssistantMessage(model); + + void (async () => { + try { + const apiKey = options?.apiKey || process.env.OPENAI_API_KEY; + if (!apiKey) throw new Error(`No API key for provider: ${model.provider}`); + const client = new OpenAI({ apiKey, baseURL: model.baseUrl, dangerouslyAllowBrowser: true, defaultHeaders: model.headers }); + const payload = buildOpenAIRequestInput(model as Model, context, options); + const nextPayload = await options?.onPayload?.(payload, model as Model); + if (options?.signal?.aborted) throw new Error("Request was aborted"); + const responseStream = await client.responses.create((nextPayload ?? payload) as never, { + signal: options?.signal, + ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), + maxRetries: options?.maxRetries ?? 0, + }); + + stream.push({ type: "start", partial: output }); + await processStream(responseStream as unknown as AsyncIterable, output, stream, options?.signal); + if (options?.signal?.aborted) throw new Error("Request was aborted"); + + stream.push({ type: "done", reason: output.stopReason as "stop" | "length" | "toolUse", message: output }); + stream.end(); + } catch (err) { + for (const block of output.content) { + delete (block as { partialJson?: string }).partialJson; + } + output.stopReason = options?.signal?.aborted ? "aborted" : "error"; + output.errorMessage = err instanceof Error ? err.message : String(err); + stream.push({ type: "error", reason: output.stopReason, error: output }); + stream.end(); + } + })(); + + return stream; +}; + +async function processStream( + events: AsyncIterable, + output: AssistantMessage, + stream: ReturnType, + signal?: AbortSignal, +): Promise { + let current: { kind: "text" | "toolCall"; index: number; partialJson: string } | null = null; + const blockIndex = () => output.content.length - 1; + for await (const event of events) { + if (signal?.aborted) throw new Error("Request was aborted"); + const type = getString(event, "type"); + if (type === "response.created") { + output.responseId = getString(getValue(event, "response"), "id") || output.responseId; + } else if (type === "response.output_item.added") { + const item = getValue(event, "item"); + const itemType = getString(item, "type"); + if (itemType === "message") { + output.content.push({ type: "text", text: "" }); + current = { kind: "text", index: blockIndex(), partialJson: "" }; + stream.push({ type: "text_start", contentIndex: current.index, partial: output }); + } else if (itemType === "function_call") { + const toolCall: ToolCall = { + type: "toolCall", + id: openaiToolCallId(item), + name: getString(item, "name"), + arguments: {}, + }; + (toolCall as ToolCall & { partialJson?: string }).partialJson = getString(item, "arguments"); + output.content.push(toolCall); + current = { kind: "toolCall", index: blockIndex(), partialJson: getString(item, "arguments") }; + stream.push({ type: "toolcall_start", contentIndex: current.index, partial: output }); + } + } else if (type === "response.output_text.delta") { + if (current?.kind === "text") { + const delta = getString(event, "delta"); + const block = output.content[current.index] as TextContent; + block.text += delta; + stream.push({ type: "text_delta", contentIndex: current.index, delta, partial: output }); + } + } else if (type === "response.function_call_arguments.delta") { + if (current?.kind === "toolCall") { + const delta = getString(event, "delta"); + current.partialJson += delta; + const block = output.content[current.index] as ToolCall; + block.arguments = parseStreamingJson(current.partialJson); + stream.push({ type: "toolcall_delta", contentIndex: current.index, delta, partial: output }); + } + } else if (type === "response.output_item.done") { + const item = getValue(event, "item"); + const itemType = getString(item, "type"); + if (itemType === "message" && current?.kind === "text") { + const block = output.content[current.index] as TextContent; + block.text = extractMessageText(item) || block.text; + stream.push({ type: "text_end", contentIndex: current.index, content: block.text, partial: output }); + current = null; + } else if (itemType === "function_call" && current?.kind === "toolCall") { + const block = output.content[current.index] as ToolCall & { partialJson?: string }; + block.arguments = parseStreamingJson(block.partialJson || getString(item, "arguments") || "{}"); + delete block.partialJson; + stream.push({ type: "toolcall_end", contentIndex: current.index, toolCall: block, partial: output }); + current = null; + } + } else if (type === "response.completed" || type === "response.incomplete") { + const response = getValue(event, "response"); + output.responseId = getString(response, "id") || output.responseId; + output.usage = usageFromOpenAI(getValue(response, "usage")); + output.stopReason = type === "response.incomplete" ? "length" : "stop"; + } else if (type === "error") { + throw new Error(getString(event, "message") || `OpenAI error code ${getString(event, "code")}`); + } else if (type === "response.failed") { + const error = getValue(getValue(event, "response"), "error"); + throw new Error(getString(error, "message") || "OpenAI response failed"); + } + } + if (output.content.some((part) => part.type === "toolCall") && output.stopReason === "stop") { + output.stopReason = "toolUse"; + } +} + +/** Pair an OpenAI function-call item's `call_id` with its item `id` so pi-ai can round-trip the Responses item. */ +function openaiToolCallId(item: unknown): string { + const callId = getString(item, "call_id"); + const id = getString(item, "id"); + return id ? `${callId}|${id}` : callId; +} + +function initialAssistantMessage(model: Model): AssistantMessage { + return { + role: "assistant", + content: [], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }; +} + +function convertTools(tools: { name: string; description?: string; parameters?: unknown }[]): Array> { + return tools.map((tool) => ({ + type: "function", + name: tool.name, + description: tool.description, + parameters: tool.parameters, + strict: false, + })); +} + +function convertMessages(messages: readonly Message[]): Array> { + const items: Array> = []; + for (const message of messages) { + if (message.role === "user") { + items.push({ role: "user", content: convertUserContent(message.content) }); + continue; + } + if (message.role === "assistant") { + for (const part of message.content) { + if (part.type === "text" && part.text.trim()) { + items.push({ + type: "message", + role: "assistant", + content: [{ type: "output_text", text: part.text, annotations: [] }], + status: "completed", + }); + } else if (part.type === "toolCall") { + const [callId, itemId] = part.id.split("|"); + items.push({ + type: "function_call", + ...(itemId ? { id: itemId } : {}), + call_id: callId, + name: part.name, + arguments: JSON.stringify(part.arguments ?? {}), + }); + } + } + continue; + } + if (message.role === "toolResult") { + const text = message.content + .filter((part): part is TextContent => part.type === "text") + .map((part) => part.text) + .join("\n") + .trim(); + const images = message.content.filter((part): part is ImageContent => part.type === "image"); + const [callId] = message.toolCallId.split("|"); + if (images.length > 0) { + const content: Array> = []; + if (text) content.push({ type: "input_text", text }); + for (const image of images) { + content.push({ type: "input_image", image_url: `data:${image.mimeType};base64,${image.data}`, detail: "auto" }); + } + items.push({ type: "function_call_output", call_id: callId, output: content }); + } else { + items.push({ + type: "function_call_output", + call_id: callId, + output: message.isError ? `Error: ${text || "tool execution failed"}` : text || "ok", + }); + } + } + } + return items; +} + +function convertUserContent(content: string | (TextContent | ImageContent)[]): unknown { + if (typeof content === "string") return [{ type: "input_text", text: content }]; + return content.map((part) => { + if (part.type === "text") return { type: "input_text", text: part.text }; + return { type: "input_image", image_url: `data:${part.mimeType};base64,${part.data}`, detail: "auto" }; + }); +} + +function extractMessageText(item: unknown): string { + return getArray(item, "content") + .map((block) => getString(block, "text")) + .filter(Boolean) + .join("\n") + .trim(); +} + +function usageFromOpenAI(usage: unknown): AssistantMessage["usage"] { + const input = readUsageNumber(usage, "input_tokens"); + const output = readUsageNumber(usage, "output_tokens"); + const cacheRead = readUsageNumber(getValue(usage, "input_tokens_details"), "cached_tokens"); + const totalTokens = readUsageNumber(usage, "total_tokens") || input + output; + return { + input: Math.max(0, input - cacheRead), + output, + cacheRead, + cacheWrite: 0, + totalTokens, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; +} + +function readUsageNumber(obj: unknown, key: string): number { + return readOptionalNumber(obj, key) ?? 0; +} + +function readOptionalNumber(obj: unknown, key: string): number | undefined { + if (!obj || typeof obj !== "object") return undefined; + const value = (obj as Record)[key]; + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + +function getArray(obj: unknown, key: string): unknown[] { + const value = getValue(obj, key); + return Array.isArray(value) ? value : []; +} + +function getString(obj: unknown, key: string): string { + const value = getValue(obj, key); + return typeof value === "string" ? value : ""; +} + +function getValue(obj: unknown, key: string): unknown { + if (!obj || typeof obj !== "object") return undefined; + return (obj as Record)[key]; +} diff --git a/packages/ai/test/models.test.ts b/packages/ai/test/models.test.ts index 2a6c3ad3..cc3b6829 100644 --- a/packages/ai/test/models.test.ts +++ b/packages/ai/test/models.test.ts @@ -6,6 +6,7 @@ import { formatCuaModelRef, getCuaModel, listCuaModels, + openai, parseCuaModelRef, } from "../src/index"; @@ -56,6 +57,15 @@ describe("CUA model refs", () => { expect(getCuaModel("yutori:n1.5-latest").api).toBe("yutori-chat-completions"); }); + it("routes OpenAI CUA models to the threading-aware openai-cua-responses api", () => { + // gpt-5.5 resolves from pi-ai's registry (carrying its builtin + // "openai-responses" api) while gpt-5.5-2026-04-23 is a local override; + // both must be routed to cua's own previous_response_id-threading provider. + expect(getCuaModel("openai:gpt-5.5").api).toBe(openai.OPENAI_CUA_RESPONSES_API); + expect(getCuaModel("openai:gpt-5.5-2026-04-23").api).toBe(openai.OPENAI_CUA_RESPONSES_API); + expect(getCuaModel("openai:gpt-5.4-mini").api).toBe(openai.OPENAI_CUA_RESPONSES_API); + }); + it("rejects supported model IDs that are not in pi-ai or overrides", () => { // Matches the openai allowlist but has no pi-ai or override entry. expect(() => getCuaModel("openai:gpt-5.4-2099-01-01")).toThrow( diff --git a/packages/ai/test/openai-threading.test.ts b/packages/ai/test/openai-threading.test.ts new file mode 100644 index 00000000..e2f9e6bc --- /dev/null +++ b/packages/ai/test/openai-threading.test.ts @@ -0,0 +1,135 @@ +import { afterEach, describe, expect, it } from "vitest"; +import type { Context, Message, Model } from "@earendil-works/pi-ai"; +import { openai } from "../src/index"; + +const model = { id: "gpt-5.5", maxTokens: 32_768 } as Model; + +const TURNS = 6; + +/** Build a multi-turn context where each assistant turn carries a distinct responseId followed by a screenshot tool result. */ +function multiTurnContext(): Context { + const messages: Message[] = [{ role: "user", content: "book a flight", timestamp: 0 }]; + for (let turn = 0; turn < TURNS; turn += 1) { + messages.push({ + role: "assistant", + content: [{ type: "toolCall", id: `call_${turn}`, name: "click", arguments: { x: turn, y: turn } }], + api: openai.OPENAI_CUA_RESPONSES_API, + provider: "openai", + model: model.id, + responseId: `resp_${turn}`, + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "toolUse", + timestamp: 0, + }); + messages.push({ + role: "toolResult", + toolCallId: `call_${turn}`, + toolName: "click", + content: [{ type: "image", mimeType: "image/png", data: `screenshot-${turn}` }], + isError: false, + timestamp: 0, + }); + } + return { messages, tools: [], systemPrompt: "control the browser" }; +} + +/** Collect every input_image data URL the request carries, across user content and function_call_output content. */ +function screenshotImageUrls(input: Array>): string[] { + const urls: string[] = []; + const scan = (content: unknown) => { + if (!Array.isArray(content)) return; + for (const part of content) { + if (part && typeof part === "object" && (part as { type?: string }).type === "input_image") { + urls.push((part as { image_url: string }).image_url); + } + } + }; + for (const item of input) { + scan(item.content); + scan(item.output); + } + return urls; +} + +describe("buildOpenAIRequestInput response threading", () => { + afterEach(() => { + delete process.env.CUA_DISABLE_RESPONSE_THREADING; + }); + + // Threading ON is the fix: chain via previous_response_id and send only the + // latest screenshot. OFF replays every screenshot — the per-turn growth that + // inflates the request before previous_response_id continuity was added. + it("threads the latest delta when enabled (default)", () => { + const body = openai.buildOpenAIRequestInput(model, multiTurnContext()); + const screenshots = screenshotImageUrls(body.input); + + expect(screenshots).toHaveLength(1); + expect(screenshots[0]).toBe(`data:image/png;base64,screenshot-${TURNS - 1}`); + expect(body.previous_response_id).toBe(`resp_${TURNS - 1}`); + // store is always true for the public Responses API (required for previous_response_id). + expect(body.store).toBe(true); + }); + + it("replays the full screenshot history when threading is disabled by option (locks the failure mode)", () => { + const body = openai.buildOpenAIRequestInput(model, multiTurnContext(), { disableResponseThreading: true }); + const screenshots = screenshotImageUrls(body.input); + + expect(screenshots).toHaveLength(TURNS); + expect(screenshots).toEqual(Array.from({ length: TURNS }, (_, turn) => `data:image/png;base64,screenshot-${turn}`)); + expect(body.previous_response_id).toBeUndefined(); + }); + + it("replays the full screenshot history when CUA_DISABLE_RESPONSE_THREADING is set", () => { + process.env.CUA_DISABLE_RESPONSE_THREADING = "1"; + const body = openai.buildOpenAIRequestInput(model, multiTurnContext()); + + expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); + expect(body.previous_response_id).toBeUndefined(); + }); + + it("falls back to full history when no prior turn carries a responseId", () => { + const context = multiTurnContext(); + for (const message of context.messages) { + if (message.role === "assistant") delete message.responseId; + } + + const body = openai.buildOpenAIRequestInput(model, context); + expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); + expect(body.previous_response_id).toBeUndefined(); + }); + + // Off-path screenshot count scales with turn count; on-path stays constant at one. + it("grows the payload per turn when off but stays flat when on", () => { + const counts = (turns: number, disable: boolean) => { + const messages: Message[] = [{ role: "user", content: "task", timestamp: 0 }]; + for (let turn = 0; turn < turns; turn += 1) { + messages.push({ + role: "assistant", + content: [{ type: "toolCall", id: `c_${turn}`, name: "click", arguments: {} }], + api: openai.OPENAI_CUA_RESPONSES_API, + provider: "openai", + model: model.id, + responseId: `r_${turn}`, + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "toolUse", + timestamp: 0, + }); + messages.push({ + role: "toolResult", + toolCallId: `c_${turn}`, + toolName: "click", + content: [{ type: "image", mimeType: "image/png", data: `s-${turn}` }], + isError: false, + timestamp: 0, + }); + } + const body = openai.buildOpenAIRequestInput(model, { messages, tools: [] }, { disableResponseThreading: disable }); + return screenshotImageUrls(body.input).length; + }; + + expect(counts(3, true)).toBe(3); + expect(counts(8, true)).toBe(8); + expect(counts(3, false)).toBe(1); + expect(counts(8, false)).toBe(1); + }); +}); diff --git a/packages/ai/test/runtime-spec.test.ts b/packages/ai/test/runtime-spec.test.ts index 8be6fcf3..899b8602 100644 --- a/packages/ai/test/runtime-spec.test.ts +++ b/packages/ai/test/runtime-spec.test.ts @@ -37,8 +37,10 @@ describe("resolveCuaRuntimeSpec", () => { const tzafonSpec = resolveCuaRuntimeSpec("tzafon:tzafon.northstar-cua-fast"); const anthropicSpec = resolveCuaRuntimeSpec("anthropic:claude-opus-4-7"); expect(yutoriSpec.onPayload).toBeTypeOf("function"); - expect(openaiSpec.onPayload).toBeTypeOf("function"); expect(tzafonSpec.onPayload).toBeTypeOf("function"); + // OpenAI and Anthropic need no payload middleware: openai-cua-responses + // sets store:true in its own request builder. + expect(openaiSpec.onPayload).toBeUndefined(); expect(anthropicSpec.onPayload).toBeUndefined(); }); From a37e84ed02e61cd2628ff3997606a08deb40efe1 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sun, 28 Jun 2026 06:25:14 +0000 Subject: [PATCH 3/8] Anchor response threading on the latest assistant turn; fix stale store-hook test - responseThreadingDelta anchors on the most recent assistant turn and replays full history when it has no responseId (a failed/aborted request never stored server-side), instead of chaining to a staler id and re-sending items past it. Adds a regression test. - OpenAI store:true moved into buildOpenAIRequestInput, so drop the removed onPayload hook's store expectation from the agent payload-composition test. Co-Authored-By: Claude Opus 4.7 --- packages/agent/test/agent.test.ts | 2 +- packages/ai/src/providers/common.ts | 18 ++++++++++-------- packages/ai/test/tzafon-threading.test.ts | 21 +++++++++++++++++++++ 3 files changed, 32 insertions(+), 9 deletions(-) diff --git a/packages/agent/test/agent.test.ts b/packages/agent/test/agent.test.ts index dfc75254..1e87c53e 100644 --- a/packages/agent/test/agent.test.ts +++ b/packages/agent/test/agent.test.ts @@ -223,7 +223,7 @@ describe("CuaAgent", () => { await agent.prompt("hello"); - expect(payloads).toEqual([{ payload: { provider: "openai", store: true }, userHook: true }]); + expect(payloads).toEqual([{ payload: { provider: "openai" }, userHook: true }]); }); it("uses yutori runtime hooks to append screenshots while stripping local executor tools", async () => { diff --git a/packages/ai/src/providers/common.ts b/packages/ai/src/providers/common.ts index 4aed8a2e..8caf3361 100644 --- a/packages/ai/src/providers/common.ts +++ b/packages/ai/src/providers/common.ts @@ -518,25 +518,27 @@ export function responseThreadingEnabled(options?: ResponseThreadingOptions): bo /** Result of {@link responseThreadingDelta}: the chaining id and the messages to send this turn. */ export interface ResponseThreadingDelta { - /** Most recent assistant `responseId`, or undefined when no prior turn carries one. */ + /** The most recent assistant turn's `responseId`, or undefined when it has none. */ previousResponseId?: string; - /** Messages to send: those after the latest assistant `responseId`, or all messages when none. */ + /** Messages to send this turn: those after the anchor assistant turn, or all messages when not threading. */ deltaMessages: Message[]; } /** * Derive the `previous_response_id` continuation from a message history. * - * Scans for the most recent assistant message carrying a `responseId` and - * returns it alongside the messages that follow it (the turn's delta). When no - * assistant message carries a `responseId`, returns every message and no id. + * Anchors on the most recent assistant turn: returns its `responseId` and the + * messages after it (the delta). If that turn has no `responseId` — a failed or + * aborted request the server never stored — or there is no assistant turn yet, + * returns every message and no id so the caller replays the full history, + * rather than chaining to a staler id and re-sending the items past it. */ export function responseThreadingDelta(messages: readonly Message[]): ResponseThreadingDelta { for (let index = messages.length - 1; index >= 0; index -= 1) { const message = messages[index]!; - if (message.role === "assistant" && (message as AssistantMessage).responseId) { - return { previousResponseId: (message as AssistantMessage).responseId, deltaMessages: messages.slice(index + 1) }; - } + if (message.role !== "assistant") continue; + const responseId = (message as AssistantMessage).responseId; + return responseId ? { previousResponseId: responseId, deltaMessages: messages.slice(index + 1) } : { deltaMessages: [...messages] }; } return { deltaMessages: [...messages] }; } diff --git a/packages/ai/test/tzafon-threading.test.ts b/packages/ai/test/tzafon-threading.test.ts index e0a0b03c..e7ba1f15 100644 --- a/packages/ai/test/tzafon-threading.test.ts +++ b/packages/ai/test/tzafon-threading.test.ts @@ -94,6 +94,27 @@ describe("buildTzafonRequestInput response threading", () => { expect(body.previous_response_id).toBeUndefined(); }); + it("replays full history when the latest assistant turn has no responseId (failed request)", () => { + const context = multiTurnContext(); + // A newer assistant turn whose request errored carries no responseId; threading must + // anchor on it and replay, not chain to the older id and re-send the items past it. + context.messages.push({ + role: "assistant", + content: [{ type: "text", text: "request failed" }], + api: tzafon.TZAFON_RESPONSES_API, + provider: "tzafon", + model: model.id, + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "error", + timestamp: 0, + }); + + const body = tzafon.buildTzafonRequestInput(model, context); + expect(body.previous_response_id).toBeUndefined(); + expect(body.store).toBeUndefined(); + expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); + }); + // Off-path screenshot count scales with turn count; on-path stays constant at one. it("grows the payload per turn when off but stays flat when on", () => { const counts = (turns: number, disable: boolean) => { From fad9354966293bd1c42ee786822743749217cf21 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sun, 28 Jun 2026 06:34:47 +0000 Subject: [PATCH 4/8] Clear responseId on OpenAI error turns; route concrete Model inputs to the threading api - Clear output.responseId on OpenAI stream error/abort so an incomplete response never anchors previous_response_id on the next turn. - resolveCuaRuntimeSpec now routes concrete Model inputs (not just string refs) through routeCuaApi, so an OpenAI model passed directly still threads instead of falling back to pi-ai's builtin openai-responses. Co-Authored-By: Claude Opus 4.7 --- packages/ai/src/models.ts | 2 +- packages/ai/src/providers/openai/provider.ts | 3 +++ packages/ai/src/runtime-spec.ts | 4 ++-- packages/ai/test/runtime-spec.test.ts | 7 ++++++- 4 files changed, 12 insertions(+), 4 deletions(-) diff --git a/packages/ai/src/models.ts b/packages/ai/src/models.ts index 789e8ae6..65e8d4d8 100644 --- a/packages/ai/src/models.ts +++ b/packages/ai/src/models.ts @@ -197,7 +197,7 @@ export function getCuaModel(ref: CuaModelRef): Model { // Route OpenAI CUA models to cua's own openai-cua-responses stream provider, // which threads previous_response_id. Registry-resolved models (gpt-5.4, // gpt-5.4-mini, gpt-5.5) otherwise carry pi-ai's builtin "openai-responses" api. -function routeCuaApi(model: Model): Model { +export function routeCuaApi(model: Model): Model { return model.provider === "openai" && model.api !== OPENAI_CUA_RESPONSES_API ? { ...model, api: OPENAI_CUA_RESPONSES_API } : model; diff --git a/packages/ai/src/providers/openai/provider.ts b/packages/ai/src/providers/openai/provider.ts index d9a30c6b..b512a8bb 100644 --- a/packages/ai/src/providers/openai/provider.ts +++ b/packages/ai/src/providers/openai/provider.ts @@ -116,6 +116,9 @@ export const streamOpenAIResponses: StreamFunction { it("resolves a runtime spec for every CUA provider", () => { @@ -44,6 +44,11 @@ describe("resolveCuaRuntimeSpec", () => { expect(anthropicSpec.onPayload).toBeUndefined(); }); + it("routes a concrete OpenAI Model input (not just a string ref) to openai-cua-responses", () => { + const raw = { ...getCuaModel("openai:gpt-5.5"), api: "openai-responses" } as Parameters[0]; + expect(resolveCuaRuntimeSpec(raw).model.api).toBe(openai.OPENAI_CUA_RESPONSES_API); + }); + it("threads tool options through to the provider module", () => { const openaiSpec = resolveCuaRuntimeSpec("openai:gpt-5.5", { actions: ["click"] }); expect(openaiSpec.toolDefinitions.map((tool) => tool.name)).toEqual(["click"]); From 15a2bda842e8f79238fac076e0856715c9376c5f Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sun, 28 Jun 2026 07:26:49 +0000 Subject: [PATCH 5/8] Update cli test fixtures to the routed openai-cua-responses api OpenAI CUA models now resolve to openai-cua-responses, so the scripted-provider fixtures must register under that api for the harness to find them. Co-Authored-By: Claude Opus 4.7 --- packages/cli/test/fixtures/harness.ts | 2 +- packages/cli/test/fixtures/tui-fixture-runner.ts | 2 +- packages/cli/test/fixtures/tui-fixtures/abort.json | 2 +- packages/cli/test/fixtures/tui-fixtures/error.json | 2 +- packages/cli/test/fixtures/tui-fixtures/multiline.json | 2 +- packages/cli/test/fixtures/tui-fixtures/resources.json | 2 +- packages/cli/test/fixtures/tui-fixtures/streaming.json | 2 +- packages/cli/test/harness-assembly.test.ts | 6 +++--- 8 files changed, 10 insertions(+), 10 deletions(-) diff --git a/packages/cli/test/fixtures/harness.ts b/packages/cli/test/fixtures/harness.ts index ec3051e6..a32b7583 100644 --- a/packages/cli/test/fixtures/harness.ts +++ b/packages/cli/test/fixtures/harness.ts @@ -30,7 +30,7 @@ export interface BuildTestHarnessOptions { } const DEFAULT_API_FOR_MODEL: Record = { - "openai:gpt-5.5": "openai-responses", + "openai:gpt-5.5": "openai-cua-responses", "anthropic:claude-opus-4-7": "anthropic-messages", "google:gemini-3-flash-preview": "google-generative-ai", }; diff --git a/packages/cli/test/fixtures/tui-fixture-runner.ts b/packages/cli/test/fixtures/tui-fixture-runner.ts index f2ced50f..f0f6bacf 100644 --- a/packages/cli/test/fixtures/tui-fixture-runner.ts +++ b/packages/cli/test/fixtures/tui-fixture-runner.ts @@ -24,7 +24,7 @@ interface TuiFixture { } const DEFAULT_API_FOR_MODEL: Record = { - "openai:gpt-5.5": "openai-responses", + "openai:gpt-5.5": "openai-cua-responses", "anthropic:claude-opus-4-7": "anthropic-messages", "google:gemini-3-flash-preview": "google-generative-ai", }; diff --git a/packages/cli/test/fixtures/tui-fixtures/abort.json b/packages/cli/test/fixtures/tui-fixtures/abort.json index 70b16471..b122696f 100644 --- a/packages/cli/test/fixtures/tui-fixtures/abort.json +++ b/packages/cli/test/fixtures/tui-fixtures/abort.json @@ -1,6 +1,6 @@ { "modelRef": "openai:gpt-5.5", - "api": "openai-responses", + "api": "openai-cua-responses", "turns": [ { "steps": [ diff --git a/packages/cli/test/fixtures/tui-fixtures/error.json b/packages/cli/test/fixtures/tui-fixtures/error.json index 9e572df8..e849d55d 100644 --- a/packages/cli/test/fixtures/tui-fixtures/error.json +++ b/packages/cli/test/fixtures/tui-fixtures/error.json @@ -1,6 +1,6 @@ { "modelRef": "openai:gpt-5.5", - "api": "openai-responses", + "api": "openai-cua-responses", "turns": [ { "steps": [ diff --git a/packages/cli/test/fixtures/tui-fixtures/multiline.json b/packages/cli/test/fixtures/tui-fixtures/multiline.json index 39c88a35..aa70cb60 100644 --- a/packages/cli/test/fixtures/tui-fixtures/multiline.json +++ b/packages/cli/test/fixtures/tui-fixtures/multiline.json @@ -1,6 +1,6 @@ { "modelRef": "openai:gpt-5.5", - "api": "openai-responses", + "api": "openai-cua-responses", "turns": [ { "steps": [ diff --git a/packages/cli/test/fixtures/tui-fixtures/resources.json b/packages/cli/test/fixtures/tui-fixtures/resources.json index f12c3366..61a3a0a0 100644 --- a/packages/cli/test/fixtures/tui-fixtures/resources.json +++ b/packages/cli/test/fixtures/tui-fixtures/resources.json @@ -1,6 +1,6 @@ { "modelRef": "openai:gpt-5.5", - "api": "openai-responses", + "api": "openai-cua-responses", "skills": [ { "name": "deploy-skill", diff --git a/packages/cli/test/fixtures/tui-fixtures/streaming.json b/packages/cli/test/fixtures/tui-fixtures/streaming.json index 96cd50cd..3f43ab95 100644 --- a/packages/cli/test/fixtures/tui-fixtures/streaming.json +++ b/packages/cli/test/fixtures/tui-fixtures/streaming.json @@ -1,6 +1,6 @@ { "modelRef": "openai:gpt-5.5", - "api": "openai-responses", + "api": "openai-cua-responses", "turns": [ { "steps": [ diff --git a/packages/cli/test/harness-assembly.test.ts b/packages/cli/test/harness-assembly.test.ts index ff7a8edc..ef6b7c50 100644 --- a/packages/cli/test/harness-assembly.test.ts +++ b/packages/cli/test/harness-assembly.test.ts @@ -40,7 +40,7 @@ describe("buildCuaHarness", () => { }); it("composes the cua-ai default system prompt with the skill block", async () => { - provider = registerScriptedProvider("openai-responses", [ + provider = registerScriptedProvider("openai-cua-responses", [ { steps: [{ type: "text", text: "ok" }] }, ]); const cwd = mkdtempSync(join(tmpdir(), "cua-cli-harness-")); @@ -75,7 +75,7 @@ describe("buildCuaHarness", () => { }); it("injects loaded context files into the system prompt", async () => { - provider = registerScriptedProvider("openai-responses", [ + provider = registerScriptedProvider("openai-cua-responses", [ { steps: [{ type: "text", text: "ok" }] }, ]); const cwd = mkdtempSync(join(tmpdir(), "cua-cli-harness-")); @@ -102,7 +102,7 @@ describe("buildCuaHarness", () => { }); it("delivers the first prompt with an image attached via harness.prompt({ images })", async () => { - provider = registerScriptedProvider("openai-responses", [ + provider = registerScriptedProvider("openai-cua-responses", [ { steps: [{ type: "text", text: "done" }] }, ]); From 3ba365ee5e68ce4432a4cf87c7a2520dde218601 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sun, 28 Jun 2026 08:58:02 +0000 Subject: [PATCH 6/8] Reuse pi-ai's builtin OpenAI Responses stream; thread via onPayload Replace the 355-line hand-rolled OpenAI provider with a ~52-line wrapper that delegates to pi-ai's streamOpenAIResponses/streamSimpleOpenAIResponses and only adds previous_response_id threading: prune the context to the delta and inject store + previous_response_id through pi-ai's onPayload hook (whose return replaces the request body). Drops the duplicated message/tool conversion, streaming parser, reasoning, and usage handling. Co-Authored-By: Claude Opus 4.7 --- packages/ai/src/providers/openai/index.ts | 3 +- packages/ai/src/providers/openai/provider.ts | 370 ++----------------- packages/ai/test/openai-threading.test.ts | 135 +++---- 3 files changed, 77 insertions(+), 431 deletions(-) diff --git a/packages/ai/src/providers/openai/index.ts b/packages/ai/src/providers/openai/index.ts index c5398de1..21c293a8 100644 --- a/packages/ai/src/providers/openai/index.ts +++ b/packages/ai/src/providers/openai/index.ts @@ -16,12 +16,11 @@ export type { CuaNavigationInput as OpenAIExtraInput, } from "../common"; export { - buildOpenAIRequestInput, OPENAI_CUA_RESPONSES_API, streamOpenAIResponses, streamSimpleOpenAIResponses, } from "./provider"; -export type { OpenAIRequestBody, OpenAIRequestOptions, OpenAIResponsesOptions } from "./provider"; +export type { OpenAIResponsesOptions } from "./provider"; // Provider-native action vocabulary emitted on `computer_call.action.type`: // click, double_click, drag, move, scroll, type, keypress, wait, screenshot diff --git a/packages/ai/src/providers/openai/provider.ts b/packages/ai/src/providers/openai/provider.ts index b512a8bb..33ddcab8 100644 --- a/packages/ai/src/providers/openai/provider.ts +++ b/packages/ai/src/providers/openai/provider.ts @@ -1,358 +1,52 @@ -import OpenAI from "openai"; -import type { ResponseStreamEvent } from "openai/resources/responses/responses"; import { - clampThinkingLevel, - createAssistantMessageEventStream, - parseStreamingJson, - type Api, - type AssistantMessage, + streamOpenAIResponses as piStreamOpenAIResponses, + streamSimpleOpenAIResponses as piStreamSimpleOpenAIResponses, type Context, - type ImageContent, - type Message, - type Model, + type OpenAIResponsesOptions as PiOpenAIResponsesOptions, type SimpleStreamOptions, type StreamFunction, type StreamOptions, - type TextContent, - type ToolCall, } from "@earendil-works/pi-ai"; import { responseThreadingDelta, responseThreadingEnabled, type ResponseThreadingOptions } from "../common"; export const OPENAI_CUA_RESPONSES_API = "openai-cua-responses"; -/** Stream options accepted by {@link streamOpenAIResponses}. */ -export interface OpenAIResponsesOptions extends StreamOptions, ResponseThreadingOptions { - reasoningEffort?: "minimal" | "low" | "medium" | "high" | "xhigh"; - reasoningSummary?: "auto" | "detailed" | "concise" | null; -} +/** Stream options for the cua OpenAI Responses provider: pi-ai's options plus threading control. */ +export interface OpenAIResponsesOptions extends PiOpenAIResponsesOptions, ResponseThreadingOptions {} -/** Inputs {@link buildOpenAIRequestInput} reads to shape the Responses API request body. */ -export interface OpenAIRequestOptions extends ResponseThreadingOptions { - temperature?: number; - maxTokens?: number; - reasoningEffort?: OpenAIResponsesOptions["reasoningEffort"]; - reasoningSummary?: OpenAIResponsesOptions["reasoningSummary"]; -} - -/** Responses API request body for {@link OpenAI.responses.create}, including optional threading fields. */ -export interface OpenAIRequestBody { - model: string; - input: Array>; - tools: Array>; - instructions?: string; - stream: true; - store: boolean; - temperature?: number; - max_output_tokens?: number; - reasoning?: { effort?: string; summary?: string }; - include?: string[]; - previous_response_id?: string; -} +type OnPayload = NonNullable; /** - * Build the OpenAI Responses API request body from a context. - * - * Pure and network-free. The public OpenAI Responses API requires `store: true` - * for `previous_response_id` continuity, so the body always stores. When - * response threading is enabled and a prior assistant `responseId` exists, the - * body chains via `previous_response_id` and sends only the delta messages; - * otherwise it replays the full message history. + * Prepare a request for pi-ai's builtin OpenAI Responses stream so it threads + * `previous_response_id`. The public Responses API requires `store: true` to + * chain, so the payload always stores; when a prior assistant `responseId` + * exists, only the delta messages are sent with `previous_response_id` set. + * Any caller `onPayload` runs on top of the threaded payload. */ -export function buildOpenAIRequestInput(model: Model, context: Context, options?: OpenAIRequestOptions): OpenAIRequestBody { - const body: OpenAIRequestBody = { - model: model.id, - input: convertMessages(context.messages), - tools: convertTools(context.tools ?? []), - instructions: context.systemPrompt, - stream: true, - store: true, - max_output_tokens: options?.maxTokens ?? model.maxTokens, +export function threadRequest( + context: Context, + options: (ResponseThreadingOptions & { onPayload?: OnPayload }) | undefined, +): { context: Context; onPayload: OnPayload } { + const delta = responseThreadingEnabled(options) ? responseThreadingDelta(context.messages) : undefined; + const previousResponseId = delta?.previousResponseId; + const messages = previousResponseId && delta ? delta.deltaMessages : context.messages; + const onPayload: OnPayload = async (payload, model) => { + const threaded = { + ...(payload as Record), + store: true, + ...(previousResponseId ? { previous_response_id: previousResponseId } : {}), + }; + return options?.onPayload ? ((await options.onPayload(threaded, model)) ?? threaded) : threaded; }; - if (options?.temperature !== undefined) body.temperature = options.temperature; - if (model.reasoning && (options?.reasoningEffort || options?.reasoningSummary)) { - const effort = options.reasoningEffort - ? (model.thinkingLevelMap?.[options.reasoningEffort] ?? options.reasoningEffort) - : "medium"; - body.reasoning = { effort, summary: options.reasoningSummary ?? "auto" }; - body.include = ["reasoning.encrypted_content"]; - } - if (!responseThreadingEnabled(options)) return body; - const { previousResponseId, deltaMessages } = responseThreadingDelta(context.messages); - if (!previousResponseId) return body; - return { ...body, input: convertMessages(deltaMessages), previous_response_id: previousResponseId }; + return { context: messages === context.messages ? context : { ...context, messages }, onPayload }; } -export const streamSimpleOpenAIResponses: StreamFunction = (model, context, options) => { - const clamped = options?.reasoning ? clampThinkingLevel(model, options.reasoning) : undefined; - const reasoningEffort = clamped && clamped !== "off" ? clamped : undefined; - return streamOpenAIResponses(model, context, { ...options, reasoningEffort }); -}; - export const streamOpenAIResponses: StreamFunction = (model, context, options) => { - const stream = createAssistantMessageEventStream(); - const output = initialAssistantMessage(model); - - void (async () => { - try { - const apiKey = options?.apiKey || process.env.OPENAI_API_KEY; - if (!apiKey) throw new Error(`No API key for provider: ${model.provider}`); - const client = new OpenAI({ apiKey, baseURL: model.baseUrl, dangerouslyAllowBrowser: true, defaultHeaders: model.headers }); - const payload = buildOpenAIRequestInput(model as Model, context, options); - const nextPayload = await options?.onPayload?.(payload, model as Model); - if (options?.signal?.aborted) throw new Error("Request was aborted"); - const responseStream = await client.responses.create((nextPayload ?? payload) as never, { - signal: options?.signal, - ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), - maxRetries: options?.maxRetries ?? 0, - }); - - stream.push({ type: "start", partial: output }); - await processStream(responseStream as unknown as AsyncIterable, output, stream, options?.signal); - if (options?.signal?.aborted) throw new Error("Request was aborted"); - - stream.push({ type: "done", reason: output.stopReason as "stop" | "length" | "toolUse", message: output }); - stream.end(); - } catch (err) { - for (const block of output.content) { - delete (block as { partialJson?: string }).partialJson; - } - // An errored/aborted turn may have captured a responseId from an incomplete - // response; drop it so it never anchors `previous_response_id` next turn. - output.responseId = undefined; - output.stopReason = options?.signal?.aborted ? "aborted" : "error"; - output.errorMessage = err instanceof Error ? err.message : String(err); - stream.push({ type: "error", reason: output.stopReason, error: output }); - stream.end(); - } - })(); - - return stream; + const threaded = threadRequest(context, options); + return piStreamOpenAIResponses(model as never, threaded.context, { ...options, onPayload: threaded.onPayload }); }; -async function processStream( - events: AsyncIterable, - output: AssistantMessage, - stream: ReturnType, - signal?: AbortSignal, -): Promise { - let current: { kind: "text" | "toolCall"; index: number; partialJson: string } | null = null; - const blockIndex = () => output.content.length - 1; - for await (const event of events) { - if (signal?.aborted) throw new Error("Request was aborted"); - const type = getString(event, "type"); - if (type === "response.created") { - output.responseId = getString(getValue(event, "response"), "id") || output.responseId; - } else if (type === "response.output_item.added") { - const item = getValue(event, "item"); - const itemType = getString(item, "type"); - if (itemType === "message") { - output.content.push({ type: "text", text: "" }); - current = { kind: "text", index: blockIndex(), partialJson: "" }; - stream.push({ type: "text_start", contentIndex: current.index, partial: output }); - } else if (itemType === "function_call") { - const toolCall: ToolCall = { - type: "toolCall", - id: openaiToolCallId(item), - name: getString(item, "name"), - arguments: {}, - }; - (toolCall as ToolCall & { partialJson?: string }).partialJson = getString(item, "arguments"); - output.content.push(toolCall); - current = { kind: "toolCall", index: blockIndex(), partialJson: getString(item, "arguments") }; - stream.push({ type: "toolcall_start", contentIndex: current.index, partial: output }); - } - } else if (type === "response.output_text.delta") { - if (current?.kind === "text") { - const delta = getString(event, "delta"); - const block = output.content[current.index] as TextContent; - block.text += delta; - stream.push({ type: "text_delta", contentIndex: current.index, delta, partial: output }); - } - } else if (type === "response.function_call_arguments.delta") { - if (current?.kind === "toolCall") { - const delta = getString(event, "delta"); - current.partialJson += delta; - const block = output.content[current.index] as ToolCall; - block.arguments = parseStreamingJson(current.partialJson); - stream.push({ type: "toolcall_delta", contentIndex: current.index, delta, partial: output }); - } - } else if (type === "response.output_item.done") { - const item = getValue(event, "item"); - const itemType = getString(item, "type"); - if (itemType === "message" && current?.kind === "text") { - const block = output.content[current.index] as TextContent; - block.text = extractMessageText(item) || block.text; - stream.push({ type: "text_end", contentIndex: current.index, content: block.text, partial: output }); - current = null; - } else if (itemType === "function_call" && current?.kind === "toolCall") { - const block = output.content[current.index] as ToolCall & { partialJson?: string }; - block.arguments = parseStreamingJson(block.partialJson || getString(item, "arguments") || "{}"); - delete block.partialJson; - stream.push({ type: "toolcall_end", contentIndex: current.index, toolCall: block, partial: output }); - current = null; - } - } else if (type === "response.completed" || type === "response.incomplete") { - const response = getValue(event, "response"); - output.responseId = getString(response, "id") || output.responseId; - output.usage = usageFromOpenAI(getValue(response, "usage")); - output.stopReason = type === "response.incomplete" ? "length" : "stop"; - } else if (type === "error") { - throw new Error(getString(event, "message") || `OpenAI error code ${getString(event, "code")}`); - } else if (type === "response.failed") { - const error = getValue(getValue(event, "response"), "error"); - throw new Error(getString(error, "message") || "OpenAI response failed"); - } - } - if (output.content.some((part) => part.type === "toolCall") && output.stopReason === "stop") { - output.stopReason = "toolUse"; - } -} - -/** Pair an OpenAI function-call item's `call_id` with its item `id` so pi-ai can round-trip the Responses item. */ -function openaiToolCallId(item: unknown): string { - const callId = getString(item, "call_id"); - const id = getString(item, "id"); - return id ? `${callId}|${id}` : callId; -} - -function initialAssistantMessage(model: Model): AssistantMessage { - return { - role: "assistant", - content: [], - api: model.api, - provider: model.provider, - model: model.id, - usage: { - input: 0, - output: 0, - cacheRead: 0, - cacheWrite: 0, - totalTokens: 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, - }, - stopReason: "stop", - timestamp: Date.now(), - }; -} - -function convertTools(tools: { name: string; description?: string; parameters?: unknown }[]): Array> { - return tools.map((tool) => ({ - type: "function", - name: tool.name, - description: tool.description, - parameters: tool.parameters, - strict: false, - })); -} - -function convertMessages(messages: readonly Message[]): Array> { - const items: Array> = []; - for (const message of messages) { - if (message.role === "user") { - items.push({ role: "user", content: convertUserContent(message.content) }); - continue; - } - if (message.role === "assistant") { - for (const part of message.content) { - if (part.type === "text" && part.text.trim()) { - items.push({ - type: "message", - role: "assistant", - content: [{ type: "output_text", text: part.text, annotations: [] }], - status: "completed", - }); - } else if (part.type === "toolCall") { - const [callId, itemId] = part.id.split("|"); - items.push({ - type: "function_call", - ...(itemId ? { id: itemId } : {}), - call_id: callId, - name: part.name, - arguments: JSON.stringify(part.arguments ?? {}), - }); - } - } - continue; - } - if (message.role === "toolResult") { - const text = message.content - .filter((part): part is TextContent => part.type === "text") - .map((part) => part.text) - .join("\n") - .trim(); - const images = message.content.filter((part): part is ImageContent => part.type === "image"); - const [callId] = message.toolCallId.split("|"); - if (images.length > 0) { - const content: Array> = []; - if (text) content.push({ type: "input_text", text }); - for (const image of images) { - content.push({ type: "input_image", image_url: `data:${image.mimeType};base64,${image.data}`, detail: "auto" }); - } - items.push({ type: "function_call_output", call_id: callId, output: content }); - } else { - items.push({ - type: "function_call_output", - call_id: callId, - output: message.isError ? `Error: ${text || "tool execution failed"}` : text || "ok", - }); - } - } - } - return items; -} - -function convertUserContent(content: string | (TextContent | ImageContent)[]): unknown { - if (typeof content === "string") return [{ type: "input_text", text: content }]; - return content.map((part) => { - if (part.type === "text") return { type: "input_text", text: part.text }; - return { type: "input_image", image_url: `data:${part.mimeType};base64,${part.data}`, detail: "auto" }; - }); -} - -function extractMessageText(item: unknown): string { - return getArray(item, "content") - .map((block) => getString(block, "text")) - .filter(Boolean) - .join("\n") - .trim(); -} - -function usageFromOpenAI(usage: unknown): AssistantMessage["usage"] { - const input = readUsageNumber(usage, "input_tokens"); - const output = readUsageNumber(usage, "output_tokens"); - const cacheRead = readUsageNumber(getValue(usage, "input_tokens_details"), "cached_tokens"); - const totalTokens = readUsageNumber(usage, "total_tokens") || input + output; - return { - input: Math.max(0, input - cacheRead), - output, - cacheRead, - cacheWrite: 0, - totalTokens, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, - }; -} - -function readUsageNumber(obj: unknown, key: string): number { - return readOptionalNumber(obj, key) ?? 0; -} - -function readOptionalNumber(obj: unknown, key: string): number | undefined { - if (!obj || typeof obj !== "object") return undefined; - const value = (obj as Record)[key]; - return typeof value === "number" && Number.isFinite(value) ? value : undefined; -} - -function getArray(obj: unknown, key: string): unknown[] { - const value = getValue(obj, key); - return Array.isArray(value) ? value : []; -} - -function getString(obj: unknown, key: string): string { - const value = getValue(obj, key); - return typeof value === "string" ? value : ""; -} - -function getValue(obj: unknown, key: string): unknown { - if (!obj || typeof obj !== "object") return undefined; - return (obj as Record)[key]; -} +export const streamSimpleOpenAIResponses: StreamFunction = (model, context, options) => { + const threaded = threadRequest(context, options); + return piStreamSimpleOpenAIResponses(model as never, threaded.context, { ...options, onPayload: threaded.onPayload }); +}; diff --git a/packages/ai/test/openai-threading.test.ts b/packages/ai/test/openai-threading.test.ts index e2f9e6bc..609073e3 100644 --- a/packages/ai/test/openai-threading.test.ts +++ b/packages/ai/test/openai-threading.test.ts @@ -1,21 +1,20 @@ import { afterEach, describe, expect, it } from "vitest"; import type { Context, Message, Model } from "@earendil-works/pi-ai"; -import { openai } from "../src/index"; - -const model = { id: "gpt-5.5", maxTokens: 32_768 } as Model; +import { OPENAI_CUA_RESPONSES_API, threadRequest } from "../src/providers/openai/provider"; const TURNS = 6; +const model = {} as Model; -/** Build a multi-turn context where each assistant turn carries a distinct responseId followed by a screenshot tool result. */ +/** Multi-turn context where each assistant turn carries a distinct responseId followed by a screenshot tool result. */ function multiTurnContext(): Context { const messages: Message[] = [{ role: "user", content: "book a flight", timestamp: 0 }]; for (let turn = 0; turn < TURNS; turn += 1) { messages.push({ role: "assistant", content: [{ type: "toolCall", id: `call_${turn}`, name: "click", arguments: { x: turn, y: turn } }], - api: openai.OPENAI_CUA_RESPONSES_API, + api: OPENAI_CUA_RESPONSES_API, provider: "openai", - model: model.id, + model: "gpt-5.5", responseId: `resp_${turn}`, usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, stopReason: "toolUse", @@ -33,103 +32,57 @@ function multiTurnContext(): Context { return { messages, tools: [], systemPrompt: "control the browser" }; } -/** Collect every input_image data URL the request carries, across user content and function_call_output content. */ -function screenshotImageUrls(input: Array>): string[] { - const urls: string[] = []; - const scan = (content: unknown) => { - if (!Array.isArray(content)) return; - for (const part of content) { - if (part && typeof part === "object" && (part as { type?: string }).type === "input_image") { - urls.push((part as { image_url: string }).image_url); - } - } - }; - for (const item of input) { - scan(item.content); - scan(item.output); - } - return urls; -} - -describe("buildOpenAIRequestInput response threading", () => { +describe("openai threadRequest", () => { afterEach(() => { delete process.env.CUA_DISABLE_RESPONSE_THREADING; }); - // Threading ON is the fix: chain via previous_response_id and send only the - // latest screenshot. OFF replays every screenshot — the per-turn growth that - // inflates the request before previous_response_id continuity was added. - it("threads the latest delta when enabled (default)", () => { - const body = openai.buildOpenAIRequestInput(model, multiTurnContext()); - const screenshots = screenshotImageUrls(body.input); - - expect(screenshots).toHaveLength(1); - expect(screenshots[0]).toBe(`data:image/png;base64,screenshot-${TURNS - 1}`); - expect(body.previous_response_id).toBe(`resp_${TURNS - 1}`); - // store is always true for the public Responses API (required for previous_response_id). - expect(body.store).toBe(true); + it("prunes to the delta and injects store + previous_response_id when threading (default)", async () => { + const { context, onPayload } = threadRequest(multiTurnContext(), undefined); + // Only the latest tool result (after the last assistant turn) is sent; the rest lives server-side. + expect(context.messages).toHaveLength(1); + expect((context.messages[0] as { toolCallId?: string }).toolCallId).toBe(`call_${TURNS - 1}`); + expect(await onPayload({ input: [] }, model)).toEqual({ input: [], store: true, previous_response_id: `resp_${TURNS - 1}` }); }); - it("replays the full screenshot history when threading is disabled by option (locks the failure mode)", () => { - const body = openai.buildOpenAIRequestInput(model, multiTurnContext(), { disableResponseThreading: true }); - const screenshots = screenshotImageUrls(body.input); - - expect(screenshots).toHaveLength(TURNS); - expect(screenshots).toEqual(Array.from({ length: TURNS }, (_, turn) => `data:image/png;base64,screenshot-${turn}`)); - expect(body.previous_response_id).toBeUndefined(); + it("replays full history with store but no previous_response_id when disabled by option", async () => { + const ctx = multiTurnContext(); + const { context, onPayload } = threadRequest(ctx, { disableResponseThreading: true }); + expect(context).toBe(ctx); + expect(await onPayload({}, model)).toEqual({ store: true }); }); - it("replays the full screenshot history when CUA_DISABLE_RESPONSE_THREADING is set", () => { + it("replays full history when CUA_DISABLE_RESPONSE_THREADING is set", async () => { process.env.CUA_DISABLE_RESPONSE_THREADING = "1"; - const body = openai.buildOpenAIRequestInput(model, multiTurnContext()); - - expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); - expect(body.previous_response_id).toBeUndefined(); + const ctx = multiTurnContext(); + const { context, onPayload } = threadRequest(ctx, undefined); + expect(context).toBe(ctx); + expect(((await onPayload({}, model)) as Record).previous_response_id).toBeUndefined(); }); - it("falls back to full history when no prior turn carries a responseId", () => { - const context = multiTurnContext(); - for (const message of context.messages) { - if (message.role === "assistant") delete message.responseId; - } - - const body = openai.buildOpenAIRequestInput(model, context); - expect(screenshotImageUrls(body.input)).toHaveLength(TURNS); - expect(body.previous_response_id).toBeUndefined(); + it("falls back to full history when the latest assistant turn lacks a responseId", async () => { + const ctx = multiTurnContext(); + ctx.messages.push({ + role: "assistant", + content: [{ type: "text", text: "request failed" }], + api: OPENAI_CUA_RESPONSES_API, + provider: "openai", + model: "gpt-5.5", + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "error", + timestamp: 0, + }); + const { context, onPayload } = threadRequest(ctx, undefined); + expect(context).toBe(ctx); + expect(((await onPayload({}, model)) as Record).previous_response_id).toBeUndefined(); }); - // Off-path screenshot count scales with turn count; on-path stays constant at one. - it("grows the payload per turn when off but stays flat when on", () => { - const counts = (turns: number, disable: boolean) => { - const messages: Message[] = [{ role: "user", content: "task", timestamp: 0 }]; - for (let turn = 0; turn < turns; turn += 1) { - messages.push({ - role: "assistant", - content: [{ type: "toolCall", id: `c_${turn}`, name: "click", arguments: {} }], - api: openai.OPENAI_CUA_RESPONSES_API, - provider: "openai", - model: model.id, - responseId: `r_${turn}`, - usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, - stopReason: "toolUse", - timestamp: 0, - }); - messages.push({ - role: "toolResult", - toolCallId: `c_${turn}`, - toolName: "click", - content: [{ type: "image", mimeType: "image/png", data: `s-${turn}` }], - isError: false, - timestamp: 0, - }); - } - const body = openai.buildOpenAIRequestInput(model, { messages, tools: [] }, { disableResponseThreading: disable }); - return screenshotImageUrls(body.input).length; - }; - - expect(counts(3, true)).toBe(3); - expect(counts(8, true)).toBe(8); - expect(counts(3, false)).toBe(1); - expect(counts(8, false)).toBe(1); + it("composes a caller onPayload on top of the threaded payload", async () => { + const { onPayload } = threadRequest(multiTurnContext(), { + onPayload: (payload) => ({ wrapped: payload }), + }); + expect(await onPayload({ input: [] }, model)).toEqual({ + wrapped: { input: [], store: true, previous_response_id: `resp_${TURNS - 1}` }, + }); }); }); From 0f93fd424a2a21995e117bba7cfe33bb23c13bb0 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sun, 28 Jun 2026 09:05:53 +0000 Subject: [PATCH 7/8] Ignore responseId from errored/aborted turns in response threading An error or abort after response.created captures a responseId for a response the server never stored. responseThreadingDelta keyed only on responseId being truthy, so it could anchor previous_response_id on that phantom id and prune history, making the next request reference a non-existent stored response. Skip the anchor's responseId when its stopReason is error or aborted so the caller replays full history. Co-Authored-By: Claude Opus 4.7 --- packages/ai/src/providers/common.ts | 13 ++++++++----- packages/ai/test/openai-threading.test.ts | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/packages/ai/src/providers/common.ts b/packages/ai/src/providers/common.ts index 8caf3361..31723d0f 100644 --- a/packages/ai/src/providers/common.ts +++ b/packages/ai/src/providers/common.ts @@ -528,16 +528,19 @@ export interface ResponseThreadingDelta { * Derive the `previous_response_id` continuation from a message history. * * Anchors on the most recent assistant turn: returns its `responseId` and the - * messages after it (the delta). If that turn has no `responseId` — a failed or - * aborted request the server never stored — or there is no assistant turn yet, - * returns every message and no id so the caller replays the full history, - * rather than chaining to a staler id and re-sending the items past it. + * messages after it (the delta). An errored or aborted turn may carry a + * `responseId` captured from an incomplete response the server never stored, so + * its id is ignored. When the anchor has no usable `responseId`, or there is no + * assistant turn yet, returns every message and no id so the caller replays the + * full history, rather than chaining to a phantom id and pruning past it. */ export function responseThreadingDelta(messages: readonly Message[]): ResponseThreadingDelta { for (let index = messages.length - 1; index >= 0; index -= 1) { const message = messages[index]!; if (message.role !== "assistant") continue; - const responseId = (message as AssistantMessage).responseId; + const assistant = message as AssistantMessage; + const failed = assistant.stopReason === "error" || assistant.stopReason === "aborted"; + const responseId = failed ? undefined : assistant.responseId; return responseId ? { previousResponseId: responseId, deltaMessages: messages.slice(index + 1) } : { deltaMessages: [...messages] }; } return { deltaMessages: [...messages] }; diff --git a/packages/ai/test/openai-threading.test.ts b/packages/ai/test/openai-threading.test.ts index 609073e3..7ba25602 100644 --- a/packages/ai/test/openai-threading.test.ts +++ b/packages/ai/test/openai-threading.test.ts @@ -77,6 +77,25 @@ describe("openai threadRequest", () => { expect(((await onPayload({}, model)) as Record).previous_response_id).toBeUndefined(); }); + it("ignores a responseId from an errored turn so it never anchors previous_response_id", async () => { + const ctx = multiTurnContext(); + // An error after response.created can capture a responseId for a response the server never stored. + ctx.messages.push({ + role: "assistant", + content: [{ type: "text", text: "request failed" }], + api: OPENAI_CUA_RESPONSES_API, + provider: "openai", + model: "gpt-5.5", + responseId: "resp_failed", + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + stopReason: "error", + timestamp: 0, + }); + const { context, onPayload } = threadRequest(ctx, undefined); + expect(context).toBe(ctx); + expect(((await onPayload({}, model)) as Record).previous_response_id).toBeUndefined(); + }); + it("composes a caller onPayload on top of the threaded payload", async () => { const { onPayload } = threadRequest(multiTurnContext(), { onPayload: (payload) => ({ wrapped: payload }), From c40fbe9e84d3fe64937c384976466b63784bea37 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Sun, 28 Jun 2026 09:09:14 +0000 Subject: [PATCH 8/8] Note why the model needs `as never` when reusing pi-ai's builtin Co-Authored-By: Claude Opus 4.7 --- packages/ai/src/providers/openai/provider.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/ai/src/providers/openai/provider.ts b/packages/ai/src/providers/openai/provider.ts index 33ddcab8..bfb31c21 100644 --- a/packages/ai/src/providers/openai/provider.ts +++ b/packages/ai/src/providers/openai/provider.ts @@ -41,6 +41,7 @@ export function threadRequest( return { context: messages === context.messages ? context : { ...context, messages }, onPayload }; } +// pi-ai's builtin stream fns are typed to the "openai-responses" api; we reuse them under our routed api, hence `as never` on the model. export const streamOpenAIResponses: StreamFunction = (model, context, options) => { const threaded = threadRequest(context, options); return piStreamOpenAIResponses(model as never, threaded.context, { ...options, onPayload: threaded.onPayload });