From 1d5b6c49f8022f36995e007567a32b9c2037e408 Mon Sep 17 00:00:00 2001 From: Banna Date: Tue, 6 Oct 2026 09:38:12 -0400 Subject: [PATCH] agent: retry a dropped model connection and say why it failed A chat turn died at the first transient connection failure. pi-ai 1.0.2 sends every provider request with the SDK's own retries off (maxRetries: 0) and defaults its own retry count to 0; archdraw never set it, and pi-agent-core has no option for it. So one reset, DNS miss or refused connection on the request that carries a tool result ended the turn with the SDK's bare "Connection error.". pi-ai also keeps only the error's message, so the cause (err.cause.cause: ECONNRESET, ENOTFOUND, UND_ERR_SOCKET) was lost and the failure was undiagnosable. streamWithRetries (models.ts) sends each model request with two retries, the SDKs' default, and passes a fetch that records the transport cause of a failed request (not for Google, whose adapter refuses a custom fetch). The conversation shows "Connection error ()." when the error remains. The scheduled change check and its drafts get the same retries. The test drives a Conversation over a real socket against a local Responses API stand-in: request 1 answers with a tool call, request 2's connection is dropped once (retried, the agent answers) or every time (the error names "other side closed, UND_ERR_SOCKET"). Signed-off-by: Banna Agent: Banna --- CHANGELOG.md | 6 ++ app/core/src/agent.ts | 11 ++- app/core/src/models.ts | 43 ++++++++++ app/core/src/sync.ts | 6 +- app/core/test/agent-network.test.ts | 125 ++++++++++++++++++++++++++++ 5 files changed, 185 insertions(+), 6 deletions(-) create mode 100644 app/core/test/agent-network.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c93b5f..0442920 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/app/core/src/agent.ts b/app/core/src/agent.ts index 655202d..4ecfd58 100644 --- a/app/core/src/agent.ts +++ b/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() // message id → text so far private demo: ReturnType | 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 diff --git a/app/core/src/models.ts b/app/core/src/models.ts index 6d386f5..642e876 100644 --- a/app/core/src/models.ts +++ b/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 +} diff --git a/app/core/src/sync.ts b/app/core/src/sync.ts index 512780d..5defd87 100644 --- a/app/core/src/sync.ts +++ b/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", }) diff --git a/app/core/test/agent-network.test.ts b/app/core/test/agent-network.test.ts new file mode 100644 index 0000000..94990fb --- /dev/null +++ b/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() + 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(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 + }) +})