Thread previous_response_id through Tzafon and OpenAI CUA providers

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 () => {

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"`)

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<Api> {
 		throw new Error(`unsupported CUA model "${ref}"`);
 	}
 	const fromRegistry = getModel(provider as never, modelId as never) as Model<Api> | 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<Api>): Model<Api> {
+	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<Api>): CuaProvider {
 	if (!isCuaProvider(model.provider)) {
@@ -249,7 +259,7 @@ function cuaModel(provider: CuaProvider, id: string, name: string): Model<Api> {
 
 	switch (provider) {
 		case "openai":
-			return { ...base, api: "openai-responses", baseUrl: "https://api.openai.com/v1", contextWindow: 400_000, maxTokens: 32_768 } as Model<Api>;
+			return { ...base, api: OPENAI_CUA_RESPONSES_API, baseUrl: "https://api.openai.com/v1", contextWindow: 400_000, maxTokens: 32_768 } as Model<Api>;
 		case "anthropic":
 			return { ...base, api: "anthropic-messages", baseUrl: "https://api.anthropic.com", contextWindow: 200_000, maxTokens: 64_000 } as Model<Api>;
 		case "google":

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 };

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.
  *

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<string, unknown>;
-	if (current.store === true) return undefined;
-	return {
-		...current,
-		store: true,
-	};
-}
-
 export const providerModule = {
 	toolDefinitions: computerTools,
 	toolExecutors: computerToolExecutors,
 	coordinateSystem,
 	buildSystemPrompt: buildOpenAISystemPrompt,
-	onPayload: openaiResponsesStoreOnPayload,
 } satisfies CuaProviderModule;

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<StreamOptions["onPayload"]>;
+
+/**
+ * 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<string, unknown>),
+			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<typeof OPENAI_CUA_RESPONSES_API, OpenAIResponsesOptions> = (model, context, options) => {
+	const threaded = threadRequest(context, options);
+	return piStreamOpenAIResponses(model as never, threaded.context, { ...options, onPayload: threaded.onPayload });
+};
+
+export const streamSimpleOpenAIResponses: StreamFunction<typeof OPENAI_CUA_RESPONSES_API, SimpleStreamOptions> = (model, context, options) => {
+	const threaded = threadRequest(context, options);
+	return piStreamSimpleOpenAIResponses(model as never, threaded.context, { ...options, onPayload: threaded.onPayload });
+};

packages/ai/src/providers/tzafon/index.ts

@@ -12,14 +12,15 @@ export type {
 	ComputerToolsOptions,
 } from "../common";
 export {
+	buildTzafonRequestInput,
 	TZAFON_RESPONSES_API,
 	streamSimpleTzafonResponses,
 	streamTzafonResponses,
 	toCanonicalActions,
 	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

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<string>(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<Record<string, unknown>>;
+	tools: Array<Record<string, unknown>>;
+	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<Api>, 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<typeof TZAFON_RESPONSES_API, SimpleStreamOptions> = (model, context, options) => {
 	return streamTzafonResponses(model, context, options);
 };
@@ -43,14 +94,7 @@ export const streamTzafonResponses: StreamFunction<typeof TZAFON_RESPONSES_API,
 			const apiKey = options?.apiKey || process.env.TZAFON_API_KEY;
 			if (!apiKey) throw new Error(`No API key for provider: ${model.provider}`);
 			const client = new Lightcone({ apiKey });
-			const payload = {
-				model: model.id,
-				input: convertContextMessages(context),
-				tools: convertTools(context.tools ?? []),
-				instructions: context.systemPrompt,
-				temperature: options?.temperature ?? 0,
-				max_output_tokens: options?.maxTokens ?? model.maxTokens,
-			};
+			const payload = buildTzafonRequestInput(model as Model<Api>, context, options);
 			const tzafonPayload = tzafonComputerUseOnPayload(payload, model as Model<Api>, {
 				keepToolNames: [...keepToolNamesFromContext(context), ...(options?.keepToolNames ?? [])],
 			});
@@ -292,9 +336,9 @@ function readToolName(tool: unknown): string | undefined {
 	return getString(fn, "name");
 }
 
-function convertContextMessages(context: Context): Array<Record<string, unknown>> {
+function convertMessages(messages: readonly Message[]): Array<Record<string, unknown>> {
 	const items: Array<Record<string, unknown>> = [];
-	for (const message of context.messages) {
+	for (const message of messages) {
 		if (message.role === "user") {
 			items.push({ role: "user", content: convertUserContent(message.content) });
 			continue;

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 {