Arch Agent: retry a dropped model connection, and say why it failed

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to archdraw. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versions follow [Semantic Versioning](https://semver.org) (in 0.x, a minor version may change behavior).
 
+## Unreleased
+
+### Fixed
+
+- The agent no longer stops at "Connection error." when one connection to the AI provider drops: a request to the model is sent again up to twice, as the OpenAI and Anthropic SDKs do by default (the scheduled change check and its drafts too). An error that remains says why, for example "Connection error (getaddrinfo ENOTFOUND api.openai.com)."
+
 ## 0.1.0 — 2026-10-06
 
 The first public release.

app/core/src/agent.ts

@@ -11,7 +11,7 @@ import { Agent } from "@earendil-works/pi-agent-core"
 import type { Model, Models } from "@earendil-works/pi-ai"
 import { AppError, type Store } from "./config.js"
 import type { Library } from "./library.js"
-import { type KeyStore, pickModel } from "./models.js"
+import { type KeyStore, pickModel, streamWithRetries, withCause } from "./models.js"
 import type { Spend } from "./spend.js"
 import { makeTools, type Proposal } from "./tools.js"
 import { demoAgent } from "./sample.js"
@@ -75,6 +75,7 @@ export class Conversation {
   private waiters: (() => void)[] = []
   private live = new Map<string, string>() // message id → text so far
   private demo: ReturnType<typeof demoAgent> | null = null // the sample's scripted agent
+  private transport: string | null = null // why the current model request's connection failed, if it did
 
   constructor(
     readonly project: string,
@@ -133,9 +134,13 @@ export class Conversation {
       skills: skills(),
       onProposal: (pr: Proposal) => this.emit({ type: "proposal", kind: "archdraw", name: pr.name, source: pr.source, doc: pr.doc, error: null }),
     })
+    const stream = streamWithRetries(models, why => (this.transport = why))
     const agent = new Agent({
       initialState: { systemPrompt: systemPrompt({ slug: p.slug, title: p.title }), model, tools, thinkingLevel: "medium" },
-      streamFn: models.streamSimple.bind(models),
+      streamFn: (m, context, options) => {
+        this.transport = null
+        return stream(m, context, options)
+      },
       getApiKey: async () => (this.demo ? "demo" : d.keys.get(s.provider)),
       sessionId: `archdraw-${this.id}`,
       toolExecution: "sequential",
@@ -161,7 +166,7 @@ export class Conversation {
       const mid = this.live.get("current") ?? `m${this.events.length}`
       const text = (e.message.content ?? []).filter((c: any) => c.type === "text").map((c: any) => c.text).join("")
       if (text.trim()) this.emit({ type: "assistant", id: mid, text })
-      if (e.message.errorMessage) this.emit({ type: "error", text: String(e.message.errorMessage).slice(0, 400) })
+      if (e.message.errorMessage) this.emit({ type: "error", text: withCause(String(e.message.errorMessage), this.transport).slice(0, 400) })
       const usd = Number(e.message.usage?.cost?.total) || 0
       if (usd > 0) {
         this.cost += usd

app/core/src/models.ts

@@ -4,6 +4,7 @@
 // app uses the OS keychain).
 
 import { readFileSync, statSync } from "node:fs"
+import type { StreamFn } from "@earendil-works/pi-agent-core"
 import { createModels, type Model, type Models } from "@earendil-works/pi-ai"
 import { anthropicProvider } from "@earendil-works/pi-ai/providers/anthropic"
 import { googleProvider } from "@earendil-works/pi-ai/providers/google"
@@ -101,3 +102,45 @@ export function pickModel(m: Models, provider: string, id: string, role: "chat"
   if (!model) throw new Error(`no model ${want || "(none)"} for ${provider}; pick one in Settings`)
   return model
 }
+
+/**
+ * How many times a model request is sent again after a dropped connection, a 408/409/429 or a 5xx: the OpenAI and
+ * Anthropic SDKs' own default. pi-ai turns the SDKs' retries off and its own default is none, so without this a single
+ * dropped connection ended the agent's turn with "Connection error.".
+ */
+export const MODEL_RETRIES = 2
+
+// the APIs whose SDK takes a custom fetch (Google's adapter refuses one)
+const FETCH_APIS = new Set(["openai-responses", "openai-completions", "anthropic-messages"])
+
+/** Why a request's fetch failed, from its deepest cause: "other side closed, UND_ERR_SOCKET", "getaddrinfo ENOTFOUND …". */
+export function transportCause(err: unknown): string {
+  let e = err as { cause?: unknown; message?: string; code?: unknown; name?: string; errors?: { message?: string }[] } | undefined
+  for (let i = 0; i < 6 && e?.cause; i++) e = e.cause as typeof e
+  const parts = e instanceof AggregateError && e.errors?.length ? e.errors.map(x => x?.message) : [e?.message]
+  const text = parts.filter(Boolean).join("; ") || String(e?.name ?? e)
+  const code = typeof e?.code === "string" && !text.includes(e.code) ? `, ${e.code}` : ""
+  return (text + code).slice(0, 200)
+}
+
+/**
+ * The stream function for a pi Agent: transient failures are retried, and `onFailure` hears why a request's connection
+ * failed (the SDKs keep that in the error's `cause`; pi-ai passes on only the message).
+ */
+export function streamWithRetries(m: Models, onFailure?: (why: string) => void): StreamFn {
+  const fetchWithCause: typeof fetch = async (input, init) => {
+    try {
+      return await fetch(input, init)
+    } catch (err) {
+      if (!init?.signal?.aborted) onFailure?.(transportCause(err))
+      throw err
+    }
+  }
+  return (model, context, options) =>
+    m.streamSimple(model, context, { ...options, maxRetries: MODEL_RETRIES, ...(onFailure && FETCH_APIS.has(model.api) ? { fetch: fetchWithCause } : {}) })
+}
+
+/** The SDKs' bare "Connection error." with the reason it happened, when one was heard. */
+export function withCause(message: string, cause: string | null): string {
+  return cause && /^Connection error\.?$/.test(message.trim()) ? `Connection error (${cause}).` : message
+}

app/core/src/sync.ts

@@ -16,7 +16,7 @@ import { systemPrompt, skills } from "./agent.js"
 import { git } from "./git.js"
 import { graph, outline } from "./graph.js"
 import type { Library } from "./library.js"
-import { type KeyStore, type ModelSet, pickModel } from "./models.js"
+import { type KeyStore, MODEL_RETRIES, type ModelSet, pickModel, streamWithRetries } from "./models.js"
 import type { Spend } from "./spend.js"
 import { makeTools, type Proposal } from "./tools.js"
 
@@ -215,7 +215,7 @@ export class Syncer {
     const msg = await models.completeSimple(model, {
       systemPrompt: TRIAGE,
       messages: [{ role: "user", content: `# The diagrams now\n\n${joined.slice(0, 12_000)}\n\n# New commits (${commits.length}${shown < commits.length ? `, the first ${shown} shown` : ""})\n\n${listing}`, timestamp: Date.now() }],
-    } as any, { apiKey: key } as any)
+    } as any, { apiKey: key, maxRetries: MODEL_RETRIES } as any)
     spend.add(slug, Number((msg as any).usage?.cost?.total) || 0)
     const text = ((msg as any).content ?? []).filter((c: any) => c.type === "text").map((c: any) => c.text).join("")
     let j: { changed?: unknown; why?: unknown; touches?: unknown }
@@ -253,7 +253,7 @@ export class Syncer {
     })
     const agent = new Agent({
       initialState: { systemPrompt: systemPrompt({ slug: p.slug, title: p.title }), model, tools, thinkingLevel: "medium" },
-      streamFn: models.streamSimple.bind(models),
+      streamFn: streamWithRetries(models),
       getApiKey: async () => keys.get(s.provider),
       toolExecution: "sequential",
     })

app/core/test/agent-network.test.ts

@@ -0,0 +1,125 @@
+// The agent over a real socket: a local stand-in for OpenAI's Responses API answers the first request with a tool call
+// (as a real model does for "draw this project's system overview") and drops the connection of the second, the one
+// that carries the tool result. A drop that happens once is retried; one that persists is reported with its cause.
+
+import { mkdirSync, mkdtempSync, writeFileSync } from "node:fs"
+import { createServer, type Server } from "node:http"
+import type { AddressInfo } from "node:net"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { afterEach, describe, expect, it } from "vitest"
+import { Conversations } from "../src/agent.js"
+import { Store } from "../src/config.js"
+import { Library } from "../src/library.js"
+import { MemoryKeys, models } from "../src/models.js"
+import { Spend } from "../src/spend.js"
+
+type Seen = { turn: number; dropped: boolean }
+
+/** A Responses API stand-in. `drop(turn, attempt)` says whether to destroy the socket instead of answering. */
+async function fakeOpenAI(drop: (turn: number, attempt: number) => boolean): Promise<{ base: string; seen: Seen[]; server: Server }> {
+  const seen: Seen[] = []
+  const attempts = new Map<number, number>()
+  const server = createServer((req, res) => {
+    let raw = ""
+    req.on("data", c => (raw += c))
+    req.on("end", () => {
+      const body = JSON.parse(raw)
+      const turn = (body.input as { type?: string }[]).some(i => i.type === "function_call_output") ? 2 : 1
+      const attempt = (attempts.get(turn) ?? 0) + 1
+      attempts.set(turn, attempt)
+      const dropped = drop(turn, attempt)
+      seen.push({ turn, dropped })
+      if (dropped) return req.socket.destroy()
+      const item =
+        turn === 1
+          ? { id: "fc_1", type: "function_call", call_id: "call_1", name: "read_skill", arguments: JSON.stringify({ path: "archdraw/SKILL.md" }), status: "completed" }
+          : { id: "msg_2", type: "message", role: "assistant", status: "completed", content: [{ type: "output_text", text: "Read it; drawing now.", annotations: [] }] }
+      const events = [
+        { type: "response.created", response: { id: `resp_${turn}`, status: "in_progress", output: [] } },
+        { type: "response.output_item.added", output_index: 0, item: turn === 1 ? { ...item, arguments: "" } : { ...item, content: [] } },
+        { type: "response.output_item.done", output_index: 0, item },
+        { type: "response.completed", response: { id: `resp_${turn}`, status: "completed", output: [item], usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 } } },
+      ]
+      res.writeHead(200, { "content-type": "text/event-stream" })
+      for (const [i, e] of events.entries()) res.write(`event: ${e.type}\ndata: ${JSON.stringify({ ...e, sequence_number: i })}\n\n`)
+      res.end()
+    })
+  })
+  await new Promise<void>(r => server.listen(0, "127.0.0.1", r))
+  return { base: `http://127.0.0.1:${(server.address() as AddressInfo).port}/v1`, seen, server }
+}
+
+/** archdraw's own model set (the four real providers), with OpenAI's endpoint moved to the stand-in. */
+function pointedAt(base: string) {
+  const real = models()
+  return new Proxy(real, {
+    get(t, p) {
+      if (p === "getModel") return (provider: string, id: string) => ({ ...t.getModel(provider as never, id), baseUrl: base })
+      const v = Reflect.get(t, p)
+      return typeof v === "function" ? v.bind(t) : v
+    },
+  })
+}
+
+function conversation(base: string) {
+  const root = mkdtempSync(join(tmpdir(), "archdraw-net-"))
+  const dir = join(root, "proj")
+  mkdirSync(join(dir, "src"), { recursive: true })
+  writeFileSync(join(dir, "src", "main.ts"), "export const app = 1\n")
+  const store = new Store(join(root, "home"))
+  store.update(c => {
+    c.settings.provider = "openai"
+  })
+  const lib = new Library(store, null)
+  lib.addFolder(dir, { slug: "proj" })
+  const keys = new MemoryKeys()
+  keys.set("openai", "sk-test-0000000000000000000000")
+  return new Conversations({ store, lib, keys, models: pointedAt(base) as never, spend: new Spend(store) }).open("proj")
+}
+
+async function settle(c: { events: { type: string }[] }, ms = 15_000) {
+  const end = Date.now() + ms
+  while (Date.now() < end) {
+    if (c.events.some(e => e.type === "settled")) return
+    await new Promise(r => setTimeout(r, 20))
+  }
+  throw new Error(`not settled: ${c.events.map(e => e.type).join(",")}`)
+}
+
+const PROMPT = "Draw this project's system overview: read the code first, then propose the diagram and its explanation."
+let server: Server | null = null
+afterEach(() => {
+  server?.close()
+  server = null
+})
+
+describe("the agent's connection to the model", () => {
+  it("sends a request again when its connection drops once, so the agent still answers", async () => {
+    const fake = await fakeOpenAI((turn, attempt) => turn === 2 && attempt === 1)
+    server = fake.server
+    const c = conversation(fake.base)
+    c.say(PROMPT)
+    await settle(c)
+    const errors = c.events.filter(e => e.type === "error")
+    expect(errors).toEqual([])
+    expect(c.events.find(e => e.type === "tool")).toMatchObject({ name: "read_skill" })
+    expect(c.events.find(e => e.type === "assistant")).toMatchObject({ text: "Read it; drawing now." })
+    expect(fake.seen).toEqual([
+      { turn: 1, dropped: false },
+      { turn: 2, dropped: true },
+      { turn: 2, dropped: false },
+    ])
+  })
+
+  it("says why a connection failed, not only that it did", async () => {
+    const fake = await fakeOpenAI(turn => turn === 2)
+    server = fake.server
+    const c = conversation(fake.base)
+    c.say(PROMPT)
+    await settle(c)
+    const error = c.events.find(e => e.type === "error") as { text: string } | undefined
+    expect(error?.text).toMatch(/^Connection error \(other side closed, UND_ERR_SOCKET\)/)
+    expect(fake.seen.filter(s => s.turn === 2).length).toBe(3) // the first try and two retries
+  })
+})