diff --git a/packages/harness/src/agent-tools.integration.test.ts b/packages/harness/src/agent-tools.integration.test.ts index aef724a..50b877f 100644 --- a/packages/harness/src/agent-tools.integration.test.ts +++ b/packages/harness/src/agent-tools.integration.test.ts @@ -125,4 +125,59 @@ describe("AgentToolSet", () => { await harness.close(); } }); + + test("returns connected tool failures to the model instead of failing the run", async () => { + const dataDirectory = await mkdtemp(join(tmpdir(), "opengui-agent-tool-failure-")); + temporaryDirectories.push(dataDirectory); + const model = new FakeModel([ + { toolCalls: [{ id: "stale-call", name: "mcp_call_tool", input: {} }] }, + { text: "Continued without the unavailable tool" }, + ]); + const harness = createOpenGuiHarness({ + dataDirectory, + model, + agentTools: { + async resolve() { + return { + generation: "catalog-generation", + definitions: [ + { + name: "mcp_call_tool", + description: "Call a connected tool.", + parameters: { type: "object", properties: {} }, + }, + ], + async invoke() { + throw new Error("Unknown MCP connection: exa"); + }, + }; + }, + }, + }); + + try { + const session = await harness.createSession({ + projectDirectory: dataDirectory, + model: { connectionId: "fake", modelId: "fake-model" }, + reasoning: "none", + }); + for await (const _event of session.run({ text: "Use the stale tool" })) { + // Drain the Run. + } + + const snapshot = await session.read(); + expect(snapshot.entries.find((entry) => entry.kind === "tool_result")?.payload).toMatchObject( + { + output: { status: "error", summary: "Unknown MCP connection: exa" }, + }, + ); + expect(snapshot.entries.some((entry) => entry.kind === "run_failed")).toBe(false); + expect(snapshot.entries.at(-2)).toMatchObject({ + kind: "assistant_message", + payload: { text: "Continued without the unavailable tool" }, + }); + } finally { + await harness.close(); + } + }); }); diff --git a/packages/harness/src/harness.test.ts b/packages/harness/src/harness.test.ts index 3425eb4..6553874 100644 --- a/packages/harness/src/harness.test.ts +++ b/packages/harness/src/harness.test.ts @@ -832,6 +832,8 @@ description: Review code changes and pull requests. Use when reviewing diffs or }); await session.reorderFollowUp(third.id, 0); await session.removeFollowUp(fourth.id); + // Removal can race with dispatch in the UI and must therefore be safe to retry. + await session.removeFollowUp(fourth.id); expect((await session.read()).followUps.map((item) => item.prompt.text)).toEqual([ "Third edited", @@ -949,7 +951,10 @@ description: Review code changes and pull requests. Use when reviewing diffs or { id: "call-timeout", name: "shell", - input: { command: `node -e "setTimeout(() => {}, 5000)"`, timeout: 0.05 }, + input: { + command: `nohup node -e "setTimeout(() => {}, 5000)" >/dev/null 2>&1 & wait`, + timeout: 0.05, + }, }, ], }, diff --git a/packages/harness/src/models/transport-contracts.test.ts b/packages/harness/src/models/transport-contracts.test.ts index bcae8b9..de85074 100644 --- a/packages/harness/src/models/transport-contracts.test.ts +++ b/packages/harness/src/models/transport-contracts.test.ts @@ -148,6 +148,14 @@ describe("durable transport cache contract", () => { retryable: false, }); }); + + test.each([ + ["You have hit your ChatGPT usage limit (pro plan).", "rate_limit"], + ["WebSocket connection closed unexpectedly", "provider_unavailable"], + ["TypeError: fetch failed", "provider_unavailable"], + ])("normalizes transient provider failure: %s", (message, code) => { + expect(normalizeModelError(new Error(message))).toMatchObject({ code, retryable: true }); + }); }); test("provider replay metadata survives restart and deltas remain ordered", async () => { diff --git a/packages/harness/src/models/transport.ts b/packages/harness/src/models/transport.ts index 9312097..df51cd5 100644 --- a/packages/harness/src/models/transport.ts +++ b/packages/harness/src/models/transport.ts @@ -363,7 +363,12 @@ export function normalizeModelError(error: unknown, signal?: AbortSignal): Norma retryable: false, status, }; - if (status === 429) + if ( + status === 429 || + /usage limit|usage_limit|rate limit|rate_limit|insufficient[_ -]?quota|quota exceeded/i.test( + message, + ) + ) return { code: "rate_limit", message: "Model provider rate limit reached", @@ -386,7 +391,12 @@ export function normalizeModelError(error: unknown, signal?: AbortSignal): Norma retryable: false, status, }; - if ((status !== undefined && status >= 500) || /overloaded|unavailable/i.test(message)) + if ( + (status !== undefined && status >= 500) || + /overloaded|unavailable|fetch failed|network error|econn\w*|socket|websocket|connection (?:closed|failed|lost|reset|refused)/i.test( + message, + ) + ) return { code: "provider_unavailable", message: "Model provider is unavailable", diff --git a/packages/harness/src/open-gui-harness.ts b/packages/harness/src/open-gui-harness.ts index 1a86008..47cabe9 100644 --- a/packages/harness/src/open-gui-harness.ts +++ b/packages/harness/src/open-gui-harness.ts @@ -31,6 +31,7 @@ import { DEFAULT_MODEL_DELIVERY, ModelTransportError, normalizeModelError, + redactProviderText, type ModelRequest, type ModelToolName, type ProviderResponseMetadata, @@ -56,25 +57,28 @@ class RandomIdGenerator implements IdGenerator { } } -async function nextWithAbort(iterator: AsyncIterator, signal: AbortSignal) { - return await new Promise>((resolveNext, rejectNext) => { - const aborted = () => rejectNext(signal.reason ?? new DOMException("Aborted", "AbortError")); +async function promiseWithAbort(pending: Promise, signal: AbortSignal) { + return await new Promise((resolvePending, rejectPending) => { + const aborted = () => rejectPending(signal.reason ?? new DOMException("Aborted", "AbortError")); signal.addEventListener("abort", aborted, { once: true }); - const pending = iterator.next(); if (signal.aborted) aborted(); void pending.then( (result) => { signal.removeEventListener("abort", aborted); - resolveNext(result); + resolvePending(result); }, (error: unknown) => { signal.removeEventListener("abort", aborted); - rejectNext(error); + rejectPending(error); }, ); }); } +async function nextWithAbort(iterator: AsyncIterator, signal: AbortSignal) { + return await promiseWithAbort(iterator.next(), signal); +} + function selectedModel(entries: SessionSnapshot["entries"]): ModelSelection | null { for (let index = entries.length - 1; index >= 0; index -= 1) { const entry = entries[index]; @@ -1207,17 +1211,30 @@ class OpenGuiHarnessImpl implements OpenGuiHarness { executionPolicy, shellExecutor: this.#shellExecutor, }; - const output = agentToolSet?.definitions.some( - (definition) => definition.name === toolCall.name, - ) - ? await limitToolResult( - toolContext, - await agentToolSet.invoke( - { name: toolCall.name, input: toolCall.input }, + let output: unknown; + try { + output = agentToolSet?.definitions.some( + (definition) => definition.name === toolCall.name, + ) + ? await promiseWithAbort( + agentToolSet + .invoke( + { name: toolCall.name, input: toolCall.input }, + abortController.signal, + ) + .then((result) => limitToolResult(toolContext, result)), abortController.signal, - ), - ) - : await executeTool(toolContext, toolCall.name, toolCall.input); + ) + : await executeTool(toolContext, toolCall.name, toolCall.input); + } catch (error) { + if (abortController.signal.aborted) throw error; + output = { + status: "error", + summary: redactProviderText( + error instanceof Error ? error.message : "Connected tool failed", + ), + }; + } await revalidate(nextPrompt.actor, current.projectDirectory); yield { type: "entry_appended", diff --git a/packages/harness/src/storage/sqlite-store.ts b/packages/harness/src/storage/sqlite-store.ts index 0a58310..5573ca6 100644 --- a/packages/harness/src/storage/sqlite-store.ts +++ b/packages/harness/src/storage/sqlite-store.ts @@ -343,13 +343,15 @@ export class SqliteSessionStore { async removeFollowUp(sessionId: string, followUpId: string) { await this.#ready; - const result = await this.#database + // Deleting a Follow-up is intentionally idempotent. A completing Run may + // dispatch the item between the frontend rendering it and this request + // arriving; in that race the user's desired end state is already true. + await this.#database .deleteFrom("session_follow_ups") .where("session_id", "=", sessionId) .where("id", "=", followUpId) .where("state", "=", "pending") .executeTakeFirst(); - if (result.numDeletedRows !== 1n) throw new Error(`Pending follow-up not found: ${followUpId}`); } /** Atomically claim a pending follow-up for immediate dispatch (send-now). */ diff --git a/packages/harness/src/tools/shell.ts b/packages/harness/src/tools/shell.ts index 6afa31e..b08e2cc 100644 --- a/packages/harness/src/tools/shell.ts +++ b/packages/harness/src/tools/shell.ts @@ -110,7 +110,6 @@ export async function executeShellTool(context: ShellToolContext, rawInput: unkn terminateProcessTree(child, force); if (!force && !forceKillTimer) { forceKillTimer = setTimeout(() => terminateProcessTree(child, true), FORCE_KILL_DELAY_MS); - forceKillTimer.unref(); } }; const onAbort = () => { @@ -122,7 +121,6 @@ export async function executeShellTool(context: ShellToolContext, rawInput: unkn timedOut = true; stop(); }, timeoutMs); - timeout.unref(); const result = await new Promise<{ exitCode: number | null; signal: NodeJS.Signals | null }>( (resolve, reject) => {