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/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..65e8d4d8 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. +export 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/common.ts b/packages/ai/src/providers/common.ts index e829a6c0..31723d0f 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,57 @@ 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 { + /** The most recent assistant turn's `responseId`, or undefined when it has none. */ + previousResponseId?: string; + /** 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. + * + * Anchors on the most recent assistant turn: returns its `responseId` and the + * 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 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] }; +} + /** * Runtime configuration for a supported CUA model. * diff --git a/packages/ai/src/providers/openai/index.ts b/packages/ai/src/providers/openai/index.ts index 1b1a7a1f..21c293a8 100644 --- a/packages/ai/src/providers/openai/index.ts +++ b/packages/ai/src/providers/openai/index.ts @@ -15,6 +15,12 @@ export type { ComputerToolsOptions, CuaNavigationInput as OpenAIExtraInput, } from "../common"; +export { + OPENAI_CUA_RESPONSES_API, + streamOpenAIResponses, + streamSimpleOpenAIResponses, +} 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 @@ -29,20 +35,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..bfb31c21 --- /dev/null +++ b/packages/ai/src/providers/openai/provider.ts @@ -0,0 +1,53 @@ +import { + streamOpenAIResponses as piStreamOpenAIResponses, + streamSimpleOpenAIResponses as piStreamSimpleOpenAIResponses, + type Context, + type OpenAIResponsesOptions as PiOpenAIResponsesOptions, + type SimpleStreamOptions, + type StreamFunction, + type StreamOptions, +} from "@earendil-works/pi-ai"; +import { responseThreadingDelta, responseThreadingEnabled, type ResponseThreadingOptions } from "../common"; + +export const OPENAI_CUA_RESPONSES_API = "openai-cua-responses"; + +/** Stream options for the cua OpenAI Responses provider: pi-ai's options plus threading control. */ +export interface OpenAIResponsesOptions extends PiOpenAIResponsesOptions, ResponseThreadingOptions {} + +type OnPayload = NonNullable; + +/** + * 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 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; + }; + 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 }); +}; + +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/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/src/runtime-spec.ts b/packages/ai/src/runtime-spec.ts index fc9fdb5e..178de193 100644 --- a/packages/ai/src/runtime-spec.ts +++ b/packages/ai/src/runtime-spec.ts @@ -1,5 +1,5 @@ import type { CuaProvider } from "./models"; -import { getCuaModel, providerForModel } from "./models"; +import { getCuaModel, providerForModel, routeCuaApi } from "./models"; import { providerModule as anthropic } from "./providers/anthropic/index"; import { providerModule as gemini } from "./providers/gemini/index"; import { providerModule as openai } from "./providers/openai/index"; @@ -29,7 +29,7 @@ const PROVIDERS = { * executors to a supported subset. */ export function resolveCuaRuntimeSpec(input: CuaRuntimeSpecInput, options?: ComputerToolsOptions): CuaRuntimeSpec { - const model = typeof input === "string" ? getCuaModel(input) : input; + const model = typeof input === "string" ? getCuaModel(input) : routeCuaApi(input); const provider = providerForModel(model); const mod: CuaProviderModule = PROVIDERS[provider]; return { 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..7ba25602 --- /dev/null +++ b/packages/ai/test/openai-threading.test.ts @@ -0,0 +1,107 @@ +import { afterEach, describe, expect, it } from "vitest"; +import type { Context, Message, Model } from "@earendil-works/pi-ai"; +import { OPENAI_CUA_RESPONSES_API, threadRequest } from "../src/providers/openai/provider"; + +const TURNS = 6; +const model = {} as Model; + +/** 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_CUA_RESPONSES_API, + provider: "openai", + 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", + 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" }; +} + +describe("openai threadRequest", () => { + afterEach(() => { + delete process.env.CUA_DISABLE_RESPONSE_THREADING; + }); + + 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 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 full history when CUA_DISABLE_RESPONSE_THREADING is set", async () => { + process.env.CUA_DISABLE_RESPONSE_THREADING = "1"; + 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 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(); + }); + + 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 }), + }); + expect(await onPayload({ input: [] }, model)).toEqual({ + wrapped: { input: [], store: true, previous_response_id: `resp_${TURNS - 1}` }, + }); + }); +}); diff --git a/packages/ai/test/runtime-spec.test.ts b/packages/ai/test/runtime-spec.test.ts index 8be6fcf3..7e696d0a 100644 --- a/packages/ai/test/runtime-spec.test.ts +++ b/packages/ai/test/runtime-spec.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { CUA_NAVIGATION_TOOL_NAME, CUA_PROVIDERS, listCuaModels, resolveCuaRuntimeSpec } from "../src/index"; +import { CUA_NAVIGATION_TOOL_NAME, CUA_PROVIDERS, getCuaModel, listCuaModels, openai, resolveCuaRuntimeSpec } from "../src/index"; describe("resolveCuaRuntimeSpec", () => { it("resolves a runtime spec for every CUA provider", () => { @@ -37,11 +37,18 @@ 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(); }); + 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"]); diff --git a/packages/ai/test/tzafon-threading.test.ts b/packages/ai/test/tzafon-threading.test.ts new file mode 100644 index 00000000..e7ba1f15 --- /dev/null +++ b/packages/ai/test/tzafon-threading.test.ts @@ -0,0 +1,152 @@ +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(); + }); + + 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) => { + 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); + }); +}); 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" }] }, ]);