fix(harness): recover stalled and failed agent runs

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();
+    }
+  });
 });

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,
+            },
           },
         ],
       },

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

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",

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<T>(iterator: AsyncIterator<T>, signal: AbortSignal) {
-  return await new Promise<IteratorResult<T>>((resolveNext, rejectNext) => {
-    const aborted = () => rejectNext(signal.reason ?? new DOMException("Aborted", "AbortError"));
+async function promiseWithAbort<T>(pending: Promise<T>, signal: AbortSignal) {
+  return await new Promise<T>((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<T>(iterator: AsyncIterator<T>, 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",

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

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