From 3a3f171b8df39018e250a132d54d99d54c8a887e Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Thu, 9 Jul 2026 20:01:44 +0000 Subject: [PATCH 1/3] Retry empty CUA responses once --- packages/agent/src/agent.ts | 53 +++++ packages/agent/test/agent.test.ts | 355 +++++++++++++++++++++++++++++- 2 files changed, 405 insertions(+), 3 deletions(-) diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 38ca37b..3568bdb 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -2,6 +2,7 @@ import { Agent, AgentHarness, type AgentHarnessOptions, + type AgentMessage, type AgentOptions, type AgentState, type AgentTool, @@ -269,6 +270,17 @@ class CuaRuntimeController { /** Default stream path: the shared CUA `Models` collection. */ const defaultCuaStream: StreamFn = (model, context, options) => cuaModels().streamSimple(model, context, options); +const EMPTY_RESPONSE_RECOVERY = "Continue working on the task."; + +function isEmptyAssistantResponse(message: AgentMessage): boolean { + return ( + message.role === "assistant" && + message.stopReason === "stop" && + message.content.length === 0 && + message.usage.output === 0 + ); +} + /** * Pi `Agent` configured for Kernel browser computer use. * @@ -281,6 +293,7 @@ export class CuaAgent extends Agent { private readonly runtime: CuaRuntimeController; private readonly ownsSystemPrompt: boolean; private runtimeDirty = false; + private emptyResponseRecoveryConsumed = false; private stateProxy?: CuaAgentState; private stateProxyTarget?: AgentState; @@ -331,6 +344,14 @@ export class CuaAgent extends Agent { this.runtime = runtime; this.ownsSystemPrompt = initialState.systemPrompt === undefined; + this.subscribe((event, signal) => { + if (event.type === "agent_start") { + this.emptyResponseRecoveryConsumed = false; + return; + } + if (event.type !== "turn_end" || !isEmptyAssistantResponse(event.message)) return; + this.recoverFromEmptyResponse(signal); + }); /** * pi's loop only re-reads model/tools/prompt between provider requests * through `prepareNextTurn`. The wrapper stays pass-through (returning @@ -405,6 +426,17 @@ export class CuaAgent extends Agent { return this.runtime.mode; } + private recoverFromEmptyResponse(signal: AbortSignal): void { + if (signal.aborted || this.emptyResponseRecoveryConsumed) return; + this.emptyResponseRecoveryConsumed = true; + if (this.hasQueuedMessages()) return; + super.followUp({ + role: "user", + content: [{ type: "text", text: EMPTY_RESPONSE_RECOVERY }], + timestamp: Date.now(), + }); + } + private applyRuntime(model: CuaRuntimeInput): void { this.runtime.setModel(model); this.runtimeDirty = true; @@ -432,6 +464,8 @@ export class CuaAgentHarness< > extends AgentHarness { private readonly runtime: CuaRuntimeController; private requestedActiveToolNames?: string[]; + private emptyResponseRecoveryConsumed = false; + private hasPendingCallerQueue = false; constructor(options: CuaAgentHarnessOptions) { const { @@ -471,6 +505,18 @@ export class CuaAgentHarness< this.runtime = runtime; this.requestedActiveToolNames = activeToolNames; + this.on("before_agent_start", () => { + this.emptyResponseRecoveryConsumed = false; + return undefined; + }); + this.subscribe(async (event, signal) => { + if (event.type === "queue_update") { + this.hasPendingCallerQueue = event.steer.length > 0 || event.followUp.length > 0; + return; + } + if (event.type !== "turn_end" || !isEmptyAssistantResponse(event.message)) return; + await this.recoverFromEmptyResponse(signal); + }); this.on("before_provider_payload", async ({ model, payload }: { model: Model; payload: unknown }) => { const onPayload = this.runtime.onPayload(); if (!onPayload) return { payload }; @@ -478,6 +524,13 @@ export class CuaAgentHarness< }); } + private async recoverFromEmptyResponse(signal?: AbortSignal): Promise { + if (signal?.aborted || this.emptyResponseRecoveryConsumed) return; + this.emptyResponseRecoveryConsumed = true; + if (this.hasPendingCallerQueue) return; + await super.followUp(EMPTY_RESPONSE_RECOVERY); + } + /** * Mirror pi `AgentHarness.setModel()` while accepting CUA model refs. * diff --git a/packages/agent/test/agent.test.ts b/packages/agent/test/agent.test.ts index d7d2273..ee51f05 100644 --- a/packages/agent/test/agent.test.ts +++ b/packages/agent/test/agent.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it, vi } from "vitest"; import { createAssistantMessageEventStream, type AssistantMessage } from "@earendil-works/pi-ai"; -import { resolveCuaRuntimeSpec } from "@onkernel/cua-ai"; +import { createCuaModels, resolveCuaRuntimeSpec } from "@onkernel/cua-ai"; import type Kernel from "@onkernel/sdk"; import { Agent, @@ -62,6 +62,40 @@ function createCustomTool(name = "custom"): AgentTool { }; } +function finishMessage(message: AssistantMessage, text?: string): AssistantMessage { + if (text !== undefined) message.content = [{ type: "text", text }]; + return message; +} + +function createScriptedStream(texts: Array, contexts?: Array<{ messages: Array<{ role: string }> }>) { + let providerCalls = 0; + const streamFn: StreamFn = (model, context) => { + contexts?.push(context as never); + const stream = createAssistantMessageEventStream(); + const message = finishMessage(createAssistantMessage(model), texts[providerCalls]); + providerCalls += 1; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + return stream; + }; + return { streamFn, calls: () => providerCalls }; +} + +function createScriptedModels(texts: Array, contexts?: Array<{ messages: Array<{ role: string }> }>) { + const scripted = createScriptedStream(texts, contexts); + const models = createCuaModels(); + models.setProvider({ + id: "openai", + name: "scripted openai", + auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, + getModels: () => [], + stream: scripted.streamFn, + streamSimple: scripted.streamFn, + } as never); + return { models, calls: scripted.calls }; +} + describe("CuaAgent", () => { it("extends pi Agent and resolves model refs in initialState", () => { const runtime = resolveCuaRuntimeSpec("openai:gpt-5.5"); @@ -237,7 +271,7 @@ describe("CuaAgent", () => { const stream = createAssistantMessageEventStream(); void (async () => { payloads.push(await options?.onPayload?.({ provider: model.provider }, model)); - const message = createAssistantMessage(model); + const message = finishMessage(createAssistantMessage(model), "done"); stream.push({ type: "start", partial: message }); stream.push({ type: "done", reason: "stop", message }); stream.end(message); @@ -286,7 +320,7 @@ describe("CuaAgent", () => { model, ), ); - const message = createAssistantMessage(model); + const message = finishMessage(createAssistantMessage(model), "done"); stream.push({ type: "start", partial: message }); stream.push({ type: "done", reason: "stop", message }); stream.end(message); @@ -394,9 +428,324 @@ describe("CuaAgent", () => { expect(fedBack, "second provider request should carry the tool result").toBeDefined(); expect(fedBack!.content.some((block) => block.type === "image" && block.mimeType === "image/png")).toBe(true); }); + + it("re-prompts once after an exact-empty response", async () => { + const contexts: Array<{ messages: Array<{ role: string }> }> = []; + const scripted = createScriptedStream([undefined, "finished"], contexts); + const agent = new CuaAgent({ + browser, + client, + streamFn: scripted.streamFn, + initialState: { model: "openai:gpt-5.5" }, + }); + + await agent.prompt("finish the task"); + + expect(scripted.calls()).toBe(2); + expect(contexts[1]?.messages.some((message) => message.role === "user")).toBe(true); + expect(agent.state.messages.at(-1)?.content).toEqual([{ type: "text", text: "finished" }]); + }); + + it("caps exact-empty recovery at one attempt", async () => { + const scripted = createScriptedStream([undefined, undefined]); + const agent = new CuaAgent({ + browser, + client, + streamFn: scripted.streamFn, + initialState: { model: "openai:gpt-5.5" }, + }); + + await agent.prompt("finish the task"); + + expect(scripted.calls()).toBe(2); + expect(agent.state.messages.at(-1)?.content).toEqual([]); + }); + + it("resets exact-empty recovery for each prompt", async () => { + const scripted = createScriptedStream([undefined, "first", undefined, "second"]); + const agent = new CuaAgent({ + browser, + client, + streamFn: scripted.streamFn, + initialState: { model: "openai:gpt-5.5" }, + }); + + await agent.prompt("first task"); + await agent.prompt("second task"); + + expect(scripted.calls()).toBe(4); + expect(agent.state.messages.at(-1)?.content).toEqual([{ type: "text", text: "second" }]); + }); + + it.each(["one-at-a-time", "all"] as const)("preserves queued caller follow-ups in %s mode", async (mode) => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const contexts: Array<{ messages: Array<{ role: string; content: Array<{ type: string; text?: string }> }> }> = []; + let calls = 0; + const streamFn: StreamFn = (model, context) => { + contexts.push(context as never); + const stream = createAssistantMessageEventStream(); + const message = finishMessage(createAssistantMessage(model), calls === 0 ? undefined : "finished"); + calls += 1; + void (async () => { + if (calls === 1) await gate; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + })(); + return stream; + }; + const agent = new CuaAgent({ + browser, + client, + streamFn, + followUpMode: mode, + initialState: { model: "openai:gpt-5.5" }, + }); + const prompt = agent.prompt("task"); + await vi.waitFor(() => expect(calls).toBe(1)); + agent.followUp({ role: "user", content: [{ type: "text", text: "caller one" }], timestamp: Date.now() }); + agent.followUp({ role: "user", content: [{ type: "text", text: "caller two" }], timestamp: Date.now() }); + release(); + await prompt; + + expect(calls).toBe(mode === "all" ? 2 : 3); + const callerCounts = contexts.slice(1).map((context) => + context.messages.filter( + (message) => + message.role === "user" && + message.content.some((block) => block.type === "text" && block.text?.startsWith("caller")), + ).length, + ); + expect(callerCounts).toEqual(mode === "all" ? [2] : [1, 2]); + }); + + it("does not recover an exact-empty response after abort", async () => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let calls = 0; + const streamFn: StreamFn = (model) => { + calls += 1; + const stream = createAssistantMessageEventStream(); + const message = createAssistantMessage(model); + void gate.then(() => { + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + }); + return stream; + }; + const agent = new CuaAgent({ browser, client, streamFn, initialState: { model: "openai:gpt-5.5" } }); + const prompt = agent.prompt("task"); + await vi.waitFor(() => expect(calls).toBe(1)); + agent.abort(); + release(); + await prompt; + + expect(calls).toBe(1); + expect(agent.hasQueuedMessages()).toBe(false); + }); }); describe("CuaAgentHarness", () => { + it("re-prompts after a successful empty response", async () => { + let providerCalls = 0; + const contexts: Array<{ messages: Array<{ role: string }> }> = []; + const models = createCuaModels(); + models.setProvider({ + id: "openai", + name: "scripted openai", + auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, + getModels: () => [], + stream: (_model, context) => { + contexts.push(context as never); + return stream(); + }, + streamSimple: (_model, context) => { + contexts.push(context as never); + return stream(); + }, + } as never); + + function stream() { + const response = createAssistantMessageEventStream(); + const message = createAssistantMessage(resolveCuaRuntimeSpec("openai:gpt-5.5").model); + if (providerCalls++ > 0) message.content = [{ type: "text", text: "finished" }]; + response.push({ type: "start", partial: message }); + response.push({ type: "done", reason: "stop", message }); + response.end(message); + return response; + } + + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models, + }); + + const response = await harness.prompt("Use the browser to finish the task."); + + expect(providerCalls).toBe(2); + expect(contexts[1]?.messages.some((message) => message.role === "user")).toBe(true); + expect(response.content).toEqual([{ type: "text", text: "finished" }]); + }); + + it("caps exact-empty recovery at one attempt", async () => { + const scripted = createScriptedModels([undefined, undefined]); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models: scripted.models, + }); + + const response = await harness.prompt("finish the task"); + + expect(scripted.calls()).toBe(2); + expect(response.content).toEqual([]); + }); + + it("resets exact-empty recovery for each prompt", async () => { + const scripted = createScriptedModels([undefined, "first", undefined, "second"]); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models: scripted.models, + }); + + await expect(harness.prompt("first task")).resolves.toMatchObject({ content: [{ text: "first" }] }); + await expect(harness.prompt("second task")).resolves.toMatchObject({ content: [{ text: "second" }] }); + expect(scripted.calls()).toBe(4); + }); + + it("does not retry a content-bearing whitespace response", async () => { + const scripted = createScriptedModels([" "]); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models: scripted.models, + }); + + const response = await harness.prompt("finish the task"); + + expect(scripted.calls()).toBe(1); + expect(response.content).toEqual([{ type: "text", text: " " }]); + }); + + it.each(["one-at-a-time", "all"] as const)("preserves queued caller follow-ups in %s mode", async (mode) => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const contexts: Array<{ messages: Array<{ role: string; content: Array<{ type: string; text?: string }> }> }> = []; + let calls = 0; + const models = createCuaModels(); + const streamFn: StreamFn = (model, context) => { + contexts.push(context as never); + const stream = createAssistantMessageEventStream(); + const message = finishMessage(createAssistantMessage(model), calls === 0 ? undefined : "finished"); + calls += 1; + void (async () => { + if (calls === 1) await gate; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + })(); + return stream; + }; + models.setProvider({ + id: "openai", + name: "scripted openai", + auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, + getModels: () => [], + stream: streamFn, + streamSimple: streamFn, + } as never); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models, + }); + await harness.setFollowUpMode(mode); + const prompt = harness.prompt("task"); + await vi.waitFor(() => expect(calls).toBe(1)); + await harness.followUp("caller one"); + await harness.followUp("caller two"); + release(); + await prompt; + + expect(calls).toBe(mode === "all" ? 2 : 3); + const callerCounts = contexts.slice(1).map((context) => + context.messages.filter( + (message) => + message.role === "user" && + message.content.some((block) => block.type === "text" && block.text?.startsWith("caller")), + ).length, + ); + expect(callerCounts).toEqual(mode === "all" ? [2] : [1, 2]); + }); + + it("does not recover an exact-empty response after abort", async () => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let calls = 0; + const models = createCuaModels(); + const streamFn: StreamFn = (model) => { + calls += 1; + const stream = createAssistantMessageEventStream(); + const message = createAssistantMessage(model); + void gate.then(() => { + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + }); + return stream; + }; + models.setProvider({ + id: "openai", + name: "scripted openai", + auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, + getModels: () => [], + stream: streamFn, + streamSimple: streamFn, + } as never); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models, + }); + let queued = false; + harness.subscribe((event) => { + if (event.type === "queue_update" && (event.steer.length > 0 || event.followUp.length > 0)) queued = true; + }); + const prompt = harness.prompt("task"); + await vi.waitFor(() => expect(calls).toBe(1)); + const abort = harness.abort(); + release(); + await abort; + await prompt; + + expect(calls).toBe(1); + expect(queued).toBe(false); + }); + it("extends pi AgentHarness and resolves model refs", async () => { const harness = new CuaAgentHarness({ ...(await createHarnessServices()), From 08fdeffe0c62abf1a40eadc94d39e4251a0184e9 Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Thu, 9 Jul 2026 21:19:04 +0000 Subject: [PATCH 2/3] Make empty response recovery explicit --- packages/agent/README.md | 26 ++ packages/agent/src/agent.ts | 111 ++++-- packages/agent/src/index.ts | 7 +- packages/agent/test/agent.test.ts | 599 +++++++++++++++++++++++------- 4 files changed, 558 insertions(+), 185 deletions(-) diff --git a/packages/agent/README.md b/packages/agent/README.md index 5371d01..7b484f3 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -85,6 +85,32 @@ active tool selection, compaction/tree workflows, and higher-level queue events. `CuaAgentHarness` extends pi `AgentHarness`, installs CUA defaults, and refreshes provider-specific runtime state when `setModel()` changes models. +## Empty Response Recovery + +pi treats an assistant message with empty content and a successful `stop` reason +as a completed turn. CUA preserves that behavior by default. Applications that +want another model turn can opt into a bounded continuation: + +```ts +const agent = new CuaAgent({ + browser, + client, + initialState: { model: "openai:gpt-5.5" }, + emptyResponseRecovery: { + followUp: "Continue working on the task.", + maxAttempts: 1, + }, +}); +``` + +Recovery uses pi's public `followUp()` queue. It preserves the empty assistant +message, appends the configured user message, and starts another model turn. +Each attempt therefore adds provider latency and a billed provider call. A +`maxAttempts` value of `1` is recommended when recovery is needed; omission or +`0` keeps stock pi completion semantics. Provider history converters and +response-ID threading may represent the preserved empty turn differently in a +later wire request. + ## Core Concepts ### Class-First API diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 3568bdb..4c6c2e4 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -57,6 +57,14 @@ type CuaAgentInitialState = Omit, "mod model: CuaRuntimeInput; }; +/** Explicit opt-in policy for continuing after a successful exact-empty assistant response. */ +export interface CuaEmptyResponseRecoveryOptions { + /** User message appended to request another model turn. */ + followUp: string; + /** Maximum recovery continuations per top-level prompt. */ + maxAttempts: number; +} + /** * Constructor options for {@link CuaAgent}. * @@ -79,6 +87,8 @@ export type CuaAgentOptions = Omit & { nativeTool?: CuaNativeToolSpec; /** Expose a tool that runs Playwright code against the browser session. */ playwright?: boolean; + /** Explicitly continue successful exact-empty responses with pi's follow-up queue. */ + emptyResponseRecovery?: CuaEmptyResponseRecoveryOptions; }; /** @@ -117,6 +127,8 @@ export type CuaAgentHarnessOptions< playwright?: boolean; /** Optional payload hook composed after the provider-specific CUA payload hook. */ onPayload?: SimpleStreamOptions["onPayload"]; + /** Explicitly continue successful exact-empty responses with pi's follow-up queue. */ + emptyResponseRecovery?: CuaEmptyResponseRecoveryOptions; }; /** @@ -270,15 +282,21 @@ class CuaRuntimeController { /** Default stream path: the shared CUA `Models` collection. */ const defaultCuaStream: StreamFn = (model, context, options) => cuaModels().streamSimple(model, context, options); -const EMPTY_RESPONSE_RECOVERY = "Continue working on the task."; +function resolveEmptyResponseRecovery( + options: CuaEmptyResponseRecoveryOptions | undefined, +): CuaEmptyResponseRecoveryOptions | undefined { + if (!options) return undefined; + if (options.followUp.trim().length === 0) { + throw new Error("emptyResponseRecovery.followUp must not be blank"); + } + if (!Number.isInteger(options.maxAttempts) || options.maxAttempts < 0) { + throw new Error("emptyResponseRecovery.maxAttempts must be a non-negative finite integer"); + } + return { followUp: options.followUp, maxAttempts: options.maxAttempts }; +} function isEmptyAssistantResponse(message: AgentMessage): boolean { - return ( - message.role === "assistant" && - message.stopReason === "stop" && - message.content.length === 0 && - message.usage.output === 0 - ); + return message.role === "assistant" && message.stopReason === "stop" && message.content.length === 0; } /** @@ -293,7 +311,7 @@ export class CuaAgent extends Agent { private readonly runtime: CuaRuntimeController; private readonly ownsSystemPrompt: boolean; private runtimeDirty = false; - private emptyResponseRecoveryConsumed = false; + private emptyResponseRecoveryAttempts = 0; private stateProxy?: CuaAgentState; private stateProxyTarget?: AgentState; @@ -309,8 +327,10 @@ export class CuaAgent extends Agent { mode, nativeTool, playwright, + emptyResponseRecovery, ...agentOptions } = options; + const recovery = resolveEmptyResponseRecovery(emptyResponseRecovery); const runtime = new CuaRuntimeController({ browser, client, @@ -344,14 +364,16 @@ export class CuaAgent extends Agent { this.runtime = runtime; this.ownsSystemPrompt = initialState.systemPrompt === undefined; - this.subscribe((event, signal) => { - if (event.type === "agent_start") { - this.emptyResponseRecoveryConsumed = false; - return; - } - if (event.type !== "turn_end" || !isEmptyAssistantResponse(event.message)) return; - this.recoverFromEmptyResponse(signal); - }); + if (recovery && recovery.maxAttempts > 0) { + this.subscribe((event, signal) => { + if (event.type === "agent_start") { + this.emptyResponseRecoveryAttempts = 0; + return; + } + if (event.type !== "turn_end" || !isEmptyAssistantResponse(event.message)) return; + this.recoverFromEmptyResponse(recovery, signal); + }); + } /** * pi's loop only re-reads model/tools/prompt between provider requests * through `prepareNextTurn`. The wrapper stays pass-through (returning @@ -426,15 +448,16 @@ export class CuaAgent extends Agent { return this.runtime.mode; } - private recoverFromEmptyResponse(signal: AbortSignal): void { - if (signal.aborted || this.emptyResponseRecoveryConsumed) return; - this.emptyResponseRecoveryConsumed = true; - if (this.hasQueuedMessages()) return; + private recoverFromEmptyResponse(recovery: CuaEmptyResponseRecoveryOptions, signal: AbortSignal): void { + if (signal.aborted || this.emptyResponseRecoveryAttempts >= recovery.maxAttempts || this.hasQueuedMessages()) { + return; + } super.followUp({ role: "user", - content: [{ type: "text", text: EMPTY_RESPONSE_RECOVERY }], + content: [{ type: "text", text: recovery.followUp }], timestamp: Date.now(), }); + this.emptyResponseRecoveryAttempts += 1; } private applyRuntime(model: CuaRuntimeInput): void { @@ -464,8 +487,8 @@ export class CuaAgentHarness< > extends AgentHarness { private readonly runtime: CuaRuntimeController; private requestedActiveToolNames?: string[]; - private emptyResponseRecoveryConsumed = false; - private hasPendingCallerQueue = false; + private emptyResponseRecoveryAttempts = 0; + private hasPendingActiveQueue = false; constructor(options: CuaAgentHarnessOptions) { const { @@ -480,8 +503,10 @@ export class CuaAgentHarness< systemPrompt, onPayload, activeToolNames, + emptyResponseRecovery, ...harnessOptions } = options; + const recovery = resolveEmptyResponseRecovery(emptyResponseRecovery); const runtime = new CuaRuntimeController({ browser, client, @@ -505,18 +530,20 @@ export class CuaAgentHarness< this.runtime = runtime; this.requestedActiveToolNames = activeToolNames; - this.on("before_agent_start", () => { - this.emptyResponseRecoveryConsumed = false; - return undefined; - }); - this.subscribe(async (event, signal) => { - if (event.type === "queue_update") { - this.hasPendingCallerQueue = event.steer.length > 0 || event.followUp.length > 0; - return; - } - if (event.type !== "turn_end" || !isEmptyAssistantResponse(event.message)) return; - await this.recoverFromEmptyResponse(signal); - }); + if (recovery && recovery.maxAttempts > 0) { + this.on("before_agent_start", () => { + this.emptyResponseRecoveryAttempts = 0; + return undefined; + }); + this.subscribe(async (event, signal) => { + if (event.type === "queue_update") { + this.hasPendingActiveQueue = event.steer.length > 0 || event.followUp.length > 0; + return; + } + if (event.type !== "turn_end" || !isEmptyAssistantResponse(event.message)) return; + await this.recoverFromEmptyResponse(recovery, signal); + }); + } this.on("before_provider_payload", async ({ model, payload }: { model: Model; payload: unknown }) => { const onPayload = this.runtime.onPayload(); if (!onPayload) return { payload }; @@ -524,11 +551,15 @@ export class CuaAgentHarness< }); } - private async recoverFromEmptyResponse(signal?: AbortSignal): Promise { - if (signal?.aborted || this.emptyResponseRecoveryConsumed) return; - this.emptyResponseRecoveryConsumed = true; - if (this.hasPendingCallerQueue) return; - await super.followUp(EMPTY_RESPONSE_RECOVERY); + private async recoverFromEmptyResponse( + recovery: CuaEmptyResponseRecoveryOptions, + signal?: AbortSignal, + ): Promise { + if (signal?.aborted || this.emptyResponseRecoveryAttempts >= recovery.maxAttempts || this.hasPendingActiveQueue) { + return; + } + await super.followUp(recovery.followUp); + this.emptyResponseRecoveryAttempts += 1; } /** diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 2f79b50..89332b1 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -16,4 +16,9 @@ export type { PlaywrightDetails, } from "./tools"; export { CuaAgent, CuaAgentHarness } from "./agent"; -export type { CuaAgentHarnessOptions, CuaAgentOptions, CuaAgentState } from "./agent"; +export type { + CuaAgentHarnessOptions, + CuaAgentOptions, + CuaAgentState, + CuaEmptyResponseRecoveryOptions, +} from "./agent"; diff --git a/packages/agent/test/agent.test.ts b/packages/agent/test/agent.test.ts index ee51f05..6e945ba 100644 --- a/packages/agent/test/agent.test.ts +++ b/packages/agent/test/agent.test.ts @@ -82,18 +82,33 @@ function createScriptedStream(texts: Array, contexts?: Array return { streamFn, calls: () => providerCalls }; } -function createScriptedModels(texts: Array, contexts?: Array<{ messages: Array<{ role: string }> }>) { - const scripted = createScriptedStream(texts, contexts); +function createModelsFromStream(streamFn: StreamFn) { const models = createCuaModels(); models.setProvider({ id: "openai", name: "scripted openai", - auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, + auth: { + apiKey: { + name: "test key", + resolve: async () => ({ auth: { apiKey: "test-key" } }), + }, + }, getModels: () => [], - stream: scripted.streamFn, - streamSimple: scripted.streamFn, + stream: streamFn, + streamSimple: streamFn, } as never); - return { models, calls: scripted.calls }; + return models; +} + +function createScriptedModels( + texts: Array, + contexts?: Array<{ messages: Array<{ role: string }> }>, +) { + const scripted = createScriptedStream(texts, contexts); + return { + models: createModelsFromStream(scripted.streamFn), + calls: scripted.calls, + }; } describe("CuaAgent", () => { @@ -271,7 +286,7 @@ describe("CuaAgent", () => { const stream = createAssistantMessageEventStream(); void (async () => { payloads.push(await options?.onPayload?.({ provider: model.provider }, model)); - const message = finishMessage(createAssistantMessage(model), "done"); + const message = createAssistantMessage(model); stream.push({ type: "start", partial: message }); stream.push({ type: "done", reason: "stop", message }); stream.end(message); @@ -320,7 +335,7 @@ describe("CuaAgent", () => { model, ), ); - const message = finishMessage(createAssistantMessage(model), "done"); + const message = createAssistantMessage(model); stream.push({ type: "start", partial: message }); stream.push({ type: "done", reason: "stop", message }); stream.end(message); @@ -417,6 +432,7 @@ describe("CuaAgent", () => { browser, client: screenshotClient, streamFn, + emptyResponseRecovery: { followUp: "continue", maxAttempts: 1 }, initialState: { model: "openai:gpt-5.5" }, }); @@ -429,9 +445,10 @@ describe("CuaAgent", () => { expect(fedBack!.content.some((block) => block.type === "image" && block.mimeType === "image/png")).toBe(true); }); - it("re-prompts once after an exact-empty response", async () => { - const contexts: Array<{ messages: Array<{ role: string }> }> = []; - const scripted = createScriptedStream([undefined, "finished"], contexts); + const recovery = { followUp: " Please continue exactly. ", maxAttempts: 1 }; + + it("preserves pi completion semantics when recovery is omitted", async () => { + const scripted = createScriptedStream([undefined]); const agent = new CuaAgent({ browser, client, @@ -441,88 +458,217 @@ describe("CuaAgent", () => { await agent.prompt("finish the task"); - expect(scripted.calls()).toBe(2); - expect(contexts[1]?.messages.some((message) => message.role === "user")).toBe(true); - expect(agent.state.messages.at(-1)?.content).toEqual([{ type: "text", text: "finished" }]); + expect(scripted.calls()).toBe(1); + expect(agent.state.messages.at(-1)?.content).toEqual([]); + expect(agent.state.messages.filter((message) => message.role === "user")).toHaveLength(1); }); - it("caps exact-empty recovery at one attempt", async () => { - const scripted = createScriptedStream([undefined, undefined]); + it("uses pi followUp for explicit empty-response recovery", async () => { + const contexts: Array<{ + messages: Array<{ + role: string; + content: Array<{ type: string; text?: string }>; + }>; + }> = []; + const scripted = createScriptedStream([undefined, "finished"], contexts); const agent = new CuaAgent({ browser, client, streamFn: scripted.streamFn, + emptyResponseRecovery: recovery, initialState: { model: "openai:gpt-5.5" }, }); await agent.prompt("finish the task"); expect(scripted.calls()).toBe(2); - expect(agent.state.messages.at(-1)?.content).toEqual([]); + const messages = contexts[1]!.messages; + expect(messages.at(-2)).toMatchObject({ role: "assistant", content: [] }); + expect(messages.at(-1)).toMatchObject({ + role: "user", + content: [{ type: "text", text: recovery.followUp }], + }); }); - it("resets exact-empty recovery for each prompt", async () => { - const scripted = createScriptedStream([undefined, "first", undefined, "second"]); + it("enforces and resets the configured recovery budget", async () => { + const scripted = createScriptedStream([undefined, undefined, undefined, "done"]); const agent = new CuaAgent({ browser, client, streamFn: scripted.streamFn, + emptyResponseRecovery: recovery, initialState: { model: "openai:gpt-5.5" }, }); - await agent.prompt("first task"); - await agent.prompt("second task"); - + await agent.prompt("first"); + expect(scripted.calls()).toBe(2); + await agent.prompt("second"); expect(scripted.calls()).toBe(4); - expect(agent.state.messages.at(-1)?.content).toEqual([{ type: "text", text: "second" }]); }); - it.each(["one-at-a-time", "all"] as const)("preserves queued caller follow-ups in %s mode", async (mode) => { - let release!: () => void; - const gate = new Promise((resolve) => { - release = resolve; + it("makes exactly one additional call per configured attempt", async () => { + const scripted = createScriptedStream([undefined, undefined, undefined]); + const agent = new CuaAgent({ + browser, + client, + streamFn: scripted.streamFn, + emptyResponseRecovery: { followUp: "continue", maxAttempts: 2 }, + initialState: { model: "openai:gpt-5.5" }, }); - const contexts: Array<{ messages: Array<{ role: string; content: Array<{ type: string; text?: string }> }> }> = []; + + await agent.prompt("task"); + expect(scripted.calls()).toBe(3); + }); + + it("recovers exact-empty content regardless of usage accounting", async () => { let calls = 0; - const streamFn: StreamFn = (model, context) => { - contexts.push(context as never); + const streamFn: StreamFn = (model) => { const stream = createAssistantMessageEventStream(); - const message = finishMessage(createAssistantMessage(model), calls === 0 ? undefined : "finished"); + const message = finishMessage(createAssistantMessage(model), calls++ === 0 ? undefined : "done"); + if (message.content.length === 0) { + message.usage.output = 7; + message.usage.reasoning = 5; + } + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + return stream; + }; + const agent = new CuaAgent({ + browser, + client, + streamFn, + emptyResponseRecovery: recovery, + initialState: { model: "openai:gpt-5.5" }, + }); + + await agent.prompt("task"); + expect(calls).toBe(2); + }); + + it.each([ + ["whitespace text", [{ type: "text", text: " " }], "stop"], + ["thinking content", [{ type: "thinking", thinking: "working" }], "stop"], + ["length stop", [], "length"], + ["error stop", [], "error"], + ["aborted stop", [], "aborted"], + ] as const)("does not recover %s", async (_name, content, stopReason) => { + let calls = 0; + const streamFn: StreamFn = (model) => { calls += 1; - void (async () => { - if (calls === 1) await gate; - stream.push({ type: "start", partial: message }); - stream.push({ type: "done", reason: "stop", message }); - stream.end(message); - })(); + const stream = createAssistantMessageEventStream(); + const message = createAssistantMessage(model); + message.content = [...content] as AssistantMessage["content"]; + message.stopReason = stopReason; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: stopReason, message }); + stream.end(message); return stream; }; const agent = new CuaAgent({ browser, client, streamFn, - followUpMode: mode, + emptyResponseRecovery: recovery, initialState: { model: "openai:gpt-5.5" }, }); - const prompt = agent.prompt("task"); - await vi.waitFor(() => expect(calls).toBe(1)); - agent.followUp({ role: "user", content: [{ type: "text", text: "caller one" }], timestamp: Date.now() }); - agent.followUp({ role: "user", content: [{ type: "text", text: "caller two" }], timestamp: Date.now() }); - release(); - await prompt; - expect(calls).toBe(mode === "all" ? 2 : 3); - const callerCounts = contexts.slice(1).map((context) => - context.messages.filter( - (message) => - message.role === "user" && - message.content.some((block) => block.type === "text" && block.text?.startsWith("caller")), - ).length, - ); - expect(callerCounts).toEqual(mode === "all" ? [2] : [1, 2]); + await agent.prompt("task"); + expect(calls).toBe(1); + }); + + it.each([[-1], [1.5], [Number.POSITIVE_INFINITY], [Number.NaN]])("rejects invalid maxAttempts %s", (maxAttempts) => { + expect( + () => + new CuaAgent({ + browser, + client, + emptyResponseRecovery: { followUp: "continue", maxAttempts }, + initialState: { model: "openai:gpt-5.5" }, + }), + ).toThrow(/non-negative finite integer/); + }); + + it("rejects blank follow-up text and accepts zero attempts", async () => { + expect( + () => + new CuaAgent({ + browser, + client, + emptyResponseRecovery: { followUp: " ", maxAttempts: 1 }, + initialState: { model: "openai:gpt-5.5" }, + }), + ).toThrow(/must not be blank/); + const scripted = createScriptedStream([undefined]); + const agent = new CuaAgent({ + browser, + client, + streamFn: scripted.streamFn, + emptyResponseRecovery: { followUp: "continue", maxAttempts: 0 }, + initialState: { model: "openai:gpt-5.5" }, + }); + await agent.prompt("task"); + expect(scripted.calls()).toBe(1); }); - it("does not recover an exact-empty response after abort", async () => { + it.each(["one-at-a-time", "all"] as const)( + "lets queued caller work drain before recovery in %s mode", + async (mode) => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const contexts: Array<{ + messages: Array<{ + role: string; + content: Array<{ type: string; text?: string }>; + }>; + }> = []; + let calls = 0; + const streamFn: StreamFn = (model, context) => { + contexts.push(context as never); + const stream = createAssistantMessageEventStream(); + const message = finishMessage(createAssistantMessage(model), calls === 2 ? "recovered" : undefined); + calls += 1; + void (async () => { + if (calls === 1) await gate; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + })(); + return stream; + }; + const agent = new CuaAgent({ + browser, + client, + streamFn, + followUpMode: mode, + emptyResponseRecovery: recovery, + initialState: { model: "openai:gpt-5.5" }, + }); + const prompt = agent.prompt("task"); + await vi.waitFor(() => expect(calls).toBe(1)); + agent.followUp({ + role: "user", + content: [{ type: "text", text: "caller" }], + timestamp: Date.now(), + }); + release(); + await prompt; + + expect(calls).toBe(3); + expect(contexts[1]!.messages.at(-1)).toMatchObject({ + role: "user", + content: [{ text: "caller" }], + }); + expect(contexts[2]!.messages.at(-1)).toMatchObject({ + role: "user", + content: [{ text: recovery.followUp }], + }); + }, + ); + + it("does not recover when the active signal is aborted", async () => { let release!: () => void; const gate = new Promise((resolve) => { release = resolve; @@ -539,123 +685,307 @@ describe("CuaAgent", () => { }); return stream; }; - const agent = new CuaAgent({ browser, client, streamFn, initialState: { model: "openai:gpt-5.5" } }); + const agent = new CuaAgent({ + browser, + client, + streamFn, + emptyResponseRecovery: recovery, + initialState: { model: "openai:gpt-5.5" }, + }); const prompt = agent.prompt("task"); await vi.waitFor(() => expect(calls).toBe(1)); agent.abort(); release(); await prompt; - expect(calls).toBe(1); - expect(agent.hasQueuedMessages()).toBe(false); }); }); describe("CuaAgentHarness", () => { - it("re-prompts after a successful empty response", async () => { - let providerCalls = 0; - const contexts: Array<{ messages: Array<{ role: string }> }> = []; - const models = createCuaModels(); - models.setProvider({ - id: "openai", - name: "scripted openai", - auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, - getModels: () => [], - stream: (_model, context) => { - contexts.push(context as never); - return stream(); - }, - streamSimple: (_model, context) => { - contexts.push(context as never); - return stream(); - }, - } as never); - - function stream() { - const response = createAssistantMessageEventStream(); - const message = createAssistantMessage(resolveCuaRuntimeSpec("openai:gpt-5.5").model); - if (providerCalls++ > 0) message.content = [{ type: "text", text: "finished" }]; - response.push({ type: "start", partial: message }); - response.push({ type: "done", reason: "stop", message }); - response.end(message); - return response; - } + const recovery = { followUp: " Please continue exactly. ", maxAttempts: 1 }; + it("preserves pi completion semantics when recovery is omitted", async () => { + const scripted = createScriptedModels([undefined]); const harness = new CuaAgentHarness({ ...(await createHarnessServices()), browser, client, model: "openai:gpt-5.5", - models, + models: scripted.models, }); - const response = await harness.prompt("Use the browser to finish the task."); - - expect(providerCalls).toBe(2); - expect(contexts[1]?.messages.some((message) => message.role === "user")).toBe(true); - expect(response.content).toEqual([{ type: "text", text: "finished" }]); + const response = await harness.prompt("finish the task"); + expect(scripted.calls()).toBe(1); + expect(response.content).toEqual([]); }); - it("caps exact-empty recovery at one attempt", async () => { - const scripted = createScriptedModels([undefined, undefined]); + it("uses pi followUp for explicit empty-response recovery", async () => { + const contexts: Array<{ + messages: Array<{ + role: string; + content: Array<{ type: string; text?: string }>; + }>; + }> = []; + const scripted = createScriptedModels([undefined, "finished"], contexts); const harness = new CuaAgentHarness({ ...(await createHarnessServices()), browser, client, model: "openai:gpt-5.5", models: scripted.models, + emptyResponseRecovery: recovery, }); const response = await harness.prompt("finish the task"); expect(scripted.calls()).toBe(2); - expect(response.content).toEqual([]); + expect(response.content).toEqual([{ type: "text", text: "finished" }]); + expect(contexts[1]!.messages.at(-2)).toMatchObject({ + role: "assistant", + content: [], + }); + expect(contexts[1]!.messages.at(-1)).toMatchObject({ + role: "user", + content: [{ type: "text", text: recovery.followUp }], + }); }); - it("resets exact-empty recovery for each prompt", async () => { - const scripted = createScriptedModels([undefined, "first", undefined, "second"]); + it("enforces and resets the configured recovery budget", async () => { + const scripted = createScriptedModels([undefined, undefined, undefined, "done"]); const harness = new CuaAgentHarness({ ...(await createHarnessServices()), browser, client, model: "openai:gpt-5.5", models: scripted.models, + emptyResponseRecovery: recovery, }); - await expect(harness.prompt("first task")).resolves.toMatchObject({ content: [{ text: "first" }] }); - await expect(harness.prompt("second task")).resolves.toMatchObject({ content: [{ text: "second" }] }); + await harness.prompt("first"); + expect(scripted.calls()).toBe(2); + await harness.prompt("second"); expect(scripted.calls()).toBe(4); }); - it("does not retry a content-bearing whitespace response", async () => { - const scripted = createScriptedModels([" "]); + it("makes exactly one additional call per configured attempt", async () => { + const scripted = createScriptedModels([undefined, undefined, undefined]); const harness = new CuaAgentHarness({ ...(await createHarnessServices()), browser, client, model: "openai:gpt-5.5", models: scripted.models, + emptyResponseRecovery: { followUp: "continue", maxAttempts: 2 }, }); - const response = await harness.prompt("finish the task"); + await harness.prompt("task"); + expect(scripted.calls()).toBe(3); + }); + + it("recovers exact-empty content with nonzero usage", async () => { + let calls = 0; + const streamFn: StreamFn = (model) => { + const stream = createAssistantMessageEventStream(); + const message = finishMessage(createAssistantMessage(model), calls++ === 0 ? undefined : "done"); + if (message.content.length === 0) { + message.usage.output = 9; + message.usage.reasoning = 6; + } + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + return stream; + }; + const models = createModelsFromStream(streamFn); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models, + emptyResponseRecovery: recovery, + }); + + await harness.prompt("task"); + expect(calls).toBe(2); + }); + + it.each([ + ["whitespace text", [{ type: "text", text: " " }], "stop"], + ["thinking content", [{ type: "thinking", thinking: "working" }], "stop"], + ["length stop", [], "length"], + ["error stop", [], "error"], + ["aborted stop", [], "aborted"], + ] as const)("does not recover %s", async (_name, content, stopReason) => { + let calls = 0; + const streamFn: StreamFn = (model) => { + calls += 1; + const stream = createAssistantMessageEventStream(); + const message = createAssistantMessage(model); + message.content = [...content] as AssistantMessage["content"]; + message.stopReason = stopReason; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: stopReason, message }); + stream.end(message); + return stream; + }; + const models = createModelsFromStream(streamFn); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models, + emptyResponseRecovery: recovery, + }); + await harness.prompt("task"); + expect(calls).toBe(1); + }); + + it("does not treat a tool-use turn as an empty response", async () => { + let calls = 0; + const streamFn: StreamFn = (model) => { + const stream = createAssistantMessageEventStream(); + const message = createAssistantMessage(model); + if (calls++ === 0) { + message.content = [{ type: "toolCall", id: "tool-1", name: "custom", arguments: {} }]; + message.stopReason = "toolUse"; + } else { + message.content = [{ type: "text", text: "done" }]; + } + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: message.stopReason, message }); + stream.end(message); + return stream; + }; + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models: createModelsFromStream(streamFn), + extraTools: [createCustomTool()], + emptyResponseRecovery: recovery, + }); + + await harness.prompt("task"); + expect(calls).toBe(2); + }); + + it.each([[-1], [1.5], [Number.POSITIVE_INFINITY], [Number.NaN]])( + "rejects invalid maxAttempts %s", + async (maxAttempts) => { + const services = await createHarnessServices(); + expect( + () => + new CuaAgentHarness({ + ...services, + browser, + client, + model: "openai:gpt-5.5", + emptyResponseRecovery: { followUp: "continue", maxAttempts }, + }), + ).toThrow(/non-negative finite integer/); + }, + ); + + it("rejects blank follow-up text and accepts zero attempts", async () => { + const services = await createHarnessServices(); + expect( + () => + new CuaAgentHarness({ + ...services, + browser, + client, + model: "openai:gpt-5.5", + emptyResponseRecovery: { followUp: " ", maxAttempts: 1 }, + }), + ).toThrow(/must not be blank/); + const scripted = createScriptedModels([undefined]); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models: scripted.models, + emptyResponseRecovery: { followUp: "continue", maxAttempts: 0 }, + }); + await harness.prompt("task"); expect(scripted.calls()).toBe(1); - expect(response.content).toEqual([{ type: "text", text: " " }]); }); - it.each(["one-at-a-time", "all"] as const)("preserves queued caller follow-ups in %s mode", async (mode) => { + it.each(["one-at-a-time", "all"] as const)( + "lets queued caller work drain before recovery in %s mode", + async (mode) => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const contexts: Array<{ + messages: Array<{ + role: string; + content: Array<{ type: string; text?: string }>; + }>; + }> = []; + let calls = 0; + const streamFn: StreamFn = (model, context) => { + contexts.push(context as never); + const stream = createAssistantMessageEventStream(); + const message = finishMessage(createAssistantMessage(model), calls === 2 ? "recovered" : undefined); + calls += 1; + void (async () => { + if (calls === 1) await gate; + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + stream.end(message); + })(); + return stream; + }; + const models = createModelsFromStream(streamFn); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models, + emptyResponseRecovery: recovery, + }); + await harness.setFollowUpMode(mode); + const prompt = harness.prompt("task"); + await vi.waitFor(() => expect(calls).toBe(1)); + await harness.followUp("caller"); + release(); + await prompt; + + expect(calls).toBe(3); + expect(contexts[1]!.messages.at(-1)).toMatchObject({ + role: "user", + content: [{ text: "caller" }], + }); + expect(contexts[2]!.messages.at(-1)).toMatchObject({ + role: "user", + content: [{ text: recovery.followUp }], + }); + }, + ); + + it("does not let nextTurn suppress active-run recovery", async () => { let release!: () => void; const gate = new Promise((resolve) => { release = resolve; }); - const contexts: Array<{ messages: Array<{ role: string; content: Array<{ type: string; text?: string }> }> }> = []; + const contexts: Array<{ + messages: Array<{ + role: string; + content: Array<{ type: string; text?: string }>; + }>; + }> = []; let calls = 0; - const models = createCuaModels(); const streamFn: StreamFn = (model, context) => { contexts.push(context as never); const stream = createAssistantMessageEventStream(); - const message = finishMessage(createAssistantMessage(model), calls === 0 ? undefined : "finished"); - calls += 1; + const message = finishMessage(createAssistantMessage(model), calls++ === 0 ? undefined : "done"); void (async () => { if (calls === 1) await gate; stream.push({ type: "start", partial: message }); @@ -664,47 +994,40 @@ describe("CuaAgentHarness", () => { })(); return stream; }; - models.setProvider({ - id: "openai", - name: "scripted openai", - auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, - getModels: () => [], - stream: streamFn, - streamSimple: streamFn, - } as never); + const models = createModelsFromStream(streamFn); const harness = new CuaAgentHarness({ ...(await createHarnessServices()), browser, client, model: "openai:gpt-5.5", models, + emptyResponseRecovery: recovery, }); - await harness.setFollowUpMode(mode); - const prompt = harness.prompt("task"); + const prompt = harness.prompt("first"); await vi.waitFor(() => expect(calls).toBe(1)); - await harness.followUp("caller one"); - await harness.followUp("caller two"); + await harness.nextTurn("later"); release(); await prompt; + expect(calls).toBe(2); + expect(contexts[1]!.messages.at(-1)).toMatchObject({ + role: "user", + content: [{ text: recovery.followUp }], + }); - expect(calls).toBe(mode === "all" ? 2 : 3); - const callerCounts = contexts.slice(1).map((context) => - context.messages.filter( - (message) => - message.role === "user" && - message.content.some((block) => block.type === "text" && block.text?.startsWith("caller")), - ).length, - ); - expect(callerCounts).toEqual(mode === "all" ? [2] : [1, 2]); + await harness.prompt("second"); + expect( + contexts[2]!.messages.some( + (message) => message.role === "user" && message.content.some((block) => block.text === "later"), + ), + ).toBe(true); }); - it("does not recover an exact-empty response after abort", async () => { + it("does not recover when the active signal is aborted", async () => { let release!: () => void; const gate = new Promise((resolve) => { release = resolve; }); let calls = 0; - const models = createCuaModels(); const streamFn: StreamFn = (model) => { calls += 1; const stream = createAssistantMessageEventStream(); @@ -716,24 +1039,14 @@ describe("CuaAgentHarness", () => { }); return stream; }; - models.setProvider({ - id: "openai", - name: "scripted openai", - auth: { apiKey: { name: "test key", resolve: async () => ({ auth: { apiKey: "test-key" } }) } }, - getModels: () => [], - stream: streamFn, - streamSimple: streamFn, - } as never); + const models = createModelsFromStream(streamFn); const harness = new CuaAgentHarness({ ...(await createHarnessServices()), browser, client, model: "openai:gpt-5.5", models, - }); - let queued = false; - harness.subscribe((event) => { - if (event.type === "queue_update" && (event.steer.length > 0 || event.followUp.length > 0)) queued = true; + emptyResponseRecovery: recovery, }); const prompt = harness.prompt("task"); await vi.waitFor(() => expect(calls).toBe(1)); @@ -741,9 +1054,7 @@ describe("CuaAgentHarness", () => { release(); await abort; await prompt; - expect(calls).toBe(1); - expect(queued).toBe(false); }); it("extends pi AgentHarness and resolves model refs", async () => { From e943ef29ca7dfab6bdba600c3bb1b66f4863419d Mon Sep 17 00:00:00 2001 From: rgarcia <72655+rgarcia@users.noreply.github.com> Date: Thu, 9 Jul 2026 22:46:42 +0000 Subject: [PATCH 3/3] Reset harness queue state between prompts --- packages/agent/src/agent.ts | 1 + packages/agent/test/agent.test.ts | 18 ++++++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 4c6c2e4..498bb88 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -533,6 +533,7 @@ export class CuaAgentHarness< if (recovery && recovery.maxAttempts > 0) { this.on("before_agent_start", () => { this.emptyResponseRecoveryAttempts = 0; + this.hasPendingActiveQueue = false; return undefined; }); this.subscribe(async (event, signal) => { diff --git a/packages/agent/test/agent.test.ts b/packages/agent/test/agent.test.ts index 6e945ba..73d090d 100644 --- a/packages/agent/test/agent.test.ts +++ b/packages/agent/test/agent.test.ts @@ -767,6 +767,24 @@ describe("CuaAgentHarness", () => { expect(scripted.calls()).toBe(4); }); + it("clears stale queue snapshots before a new prompt starts", async () => { + const scripted = createScriptedModels([undefined, "finished"]); + const harness = new CuaAgentHarness({ + ...(await createHarnessServices()), + browser, + client, + model: "openai:gpt-5.5", + models: scripted.models, + emptyResponseRecovery: recovery, + }); + + (harness as unknown as { hasPendingActiveQueue: boolean }).hasPendingActiveQueue = true; + const response = await harness.prompt("finish the task"); + + expect(scripted.calls()).toBe(2); + expect(response.content).toEqual([{ type: "text", text: "finished" }]); + }); + it("makes exactly one additional call per configured attempt", async () => { const scripted = createScriptedModels([undefined, undefined, undefined]); const harness = new CuaAgentHarness({