diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cc2381..6e515e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +### Fixes + +- **Provider error finishes fail headless execution** — Normal streams ending in `error` or `content-filter` now persist a structured provider failure, emit consistent failure events, and exit nonzero while retaining partial output and usage. Empty successful responses remain successful. (#108) + ### Features - **GPT-5.6 Codex models** — Added OpenAI's Sol, Terra, and Luna models with API and subscription-backed reasoning effort variants, including the Codex-only `ultra` alias for Sol and Terra. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index ad79177..7b67326 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -39,6 +39,16 @@ https://github.com/anomalyco/models.dev bun dev ``` +### Headless release regression checks + +From `packages/cli`, run: + +```bash +bun test test/cli/run-provider-finish.test.ts test/cli/run-signal-cancellation.test.ts test/cli/classify-session-error.test.ts +``` + +The provider fixture uses real Gemini SSE responses and the pinned SDK in a headless subprocess. It checks failed finishes, content filtering, empty success, tool calls, partial output, and output limits. The cancellation tests verify that signals keep their distinct failure classification and flush terminal events before exit. + ### Running against a different directory By default, `bun dev` runs Aictrl in the `packages/aictrl` directory. To run it against a different directory or repository: diff --git a/EVENTS.md b/EVENTS.md index 911798f..d2709c6 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -230,6 +230,26 @@ For compatibility with sessions written before usage provenance was persisted, a - `ratio` (number) — `used / limit` (≥0; may exceed 1 if usage exceeds the model's registered limit). A value approaching or exceeding 1 signals context-exhaustion risk. - `null` — emitted when the model's context limit is not known (e.g. unregistered custom endpoint), or usage is missing. +**Terminal reason semantics (schema v1)** + +A provider can finish an HTTP stream normally while reporting a failed model turn. `error` and `content-filter` finishes persist a nonretryable `APIError` with `data.metadata.finishReason`; they emit `message_complete.status: "error"`, `session_error.reason: "provider"`, a populated `session_complete.error`, and `invocation_complete.status: "error"`. Headless execution exits 1 after flushing output. Partial text, completed tools, and reported usage remain available. No automatic recovery is attempted. + +| Finish or termination | Behavior | +| ----------------------------- | -------------------------------------------------------------------------------------------- | +| `error` | Failed model turn; session/invocation failure and exit 1. | +| `content-filter` | Failed model turn with a visible content-filter message; exit 1. | +| `stop` | Completed turn, including empty output. | +| `tool-calls` | Completed model turn; run tools and continue the session loop. | +| `length` | Existing behavior: completed turn; preserve the reason so consumers can identify truncation. | +| `unknown` | Existing behavior: continue the session loop. | +| Other nonempty finish | Existing behavior: end the loop without inferring failure from an unfamiliar reason. | +| Thrown provider error | Existing retry/error handling; unrecoverable failures emit the failure lifecycle. | +| Cancellation / stream timeout | Existing cancellation and timeout lifecycle; not reclassified as a provider finish error. | + +With pinned Google SDK 2.0.54, `IMAGE_SAFETY`, `RECITATION`, `SAFETY`, `BLOCKLIST`, `PROHIBITED_CONTENT`, and `SPII` map to `content-filter`; `MALFORMED_FUNCTION_CALL` maps to `error`. `OTHER` and `FINISH_REASON_UNSPECIFIED` map to `other`, while `LANGUAGE` maps to `unknown`. These last mappings retain the behavior above; they do not prove successful task completion. Consumers can distinguish `other` from `stop` using `message_complete.finish`. + +Errors are attributed to the originating session. A child error alone does not change the primary session's exit status if the primary agent handles it successfully. + ### `text` Emitted when a text block from the assistant is complete. diff --git a/packages/cli/src/cli/cmd/run.errors.ts b/packages/cli/src/cli/cmd/run.errors.ts index 1b86c54..a0cbeaf 100644 --- a/packages/cli/src/cli/cmd/run.errors.ts +++ b/packages/cli/src/cli/cmd/run.errors.ts @@ -30,8 +30,8 @@ export function classifySessionError(err: unknown): ClassifiedSessionError { if (/heap out of memory|ENOMEM/i.test(message)) { return { reason: "oom", message } } - if (status && status >= 500 && status < 600) { - return { reason: "provider", code: String(status), message } + if ((name === "APIError" && status === undefined) || (status && status >= 500 && status < 600)) { + return { reason: "provider", code: status ? String(status) : undefined, message } } return { reason: "unknown", code: status ? String(status) : undefined, message } } @@ -50,7 +50,8 @@ function extractMessage(err: unknown): string { function extractStatus(err: unknown): number | undefined { if (err && typeof err === "object") { const e = err as { status?: unknown; statusCode?: unknown; response?: { status?: unknown }; data?: unknown } - const data = e.data && typeof e.data === "object" ? (e.data as { status?: unknown; statusCode?: unknown }) : undefined + const data = + e.data && typeof e.data === "object" ? (e.data as { status?: unknown; statusCode?: unknown }) : undefined const raw = e.status ?? e.statusCode ?? e.response?.status ?? data?.status ?? data?.statusCode if (typeof raw === "number") return raw if (typeof raw === "string" && /^\d+$/.test(raw)) return Number(raw) diff --git a/packages/cli/src/session/processor.ts b/packages/cli/src/session/processor.ts index 2577169..75c11d6 100644 --- a/packages/cli/src/session/processor.ts +++ b/packages/cli/src/session/processor.ts @@ -252,6 +252,24 @@ export namespace SessionProcessor { input.assistantMessage.cost += usage.cost input.assistantMessage.tokens = usage.tokens input.assistantMessage.usageStatus = usage.usageStatus + // Providers can end a successful HTTP stream with a failed + // model turn. Preserve its parts and usage, but use the same + // failure lifecycle as a thrown, nonretryable provider error. + if (value.finishReason === "error" || value.finishReason === "content-filter") { + log.error("provider finish", { + sessionID: input.sessionID, + messageID: input.assistantMessage.id, + finishReason: value.finishReason, + }) + input.assistantMessage.error = new MessageV2.APIError({ + message: + value.finishReason === "content-filter" + ? "The provider blocked the response with a content filter." + : "The provider ended the response with an error finish reason.", + isRetryable: false, + metadata: { finishReason: value.finishReason }, + }).toObject() + } await Session.updatePart({ id: Identifier.ascending("part"), reason: value.finishReason, @@ -263,6 +281,13 @@ export namespace SessionProcessor { cost: usage.cost, }) await Session.updateMessage(input.assistantMessage) + if (input.assistantMessage.error) { + await Bus.publish(Session.Event.Error, { + sessionID: input.sessionID, + error: input.assistantMessage.error, + }) + break + } if (snapshot) { const patch = await Snapshot.patch(snapshot) if (patch.files.length) { @@ -349,7 +374,7 @@ export namespace SessionProcessor { }) continue } - if (needsCompaction) break + if (needsCompaction || input.assistantMessage.error) break } } catch (e: any) { log.error("process", { @@ -368,7 +393,7 @@ export namespace SessionProcessor { input.assistantMessage.error = new NamedError.Unknown({ message: `Max retry attempts (${SessionRetry.MAX_RETRY_ATTEMPTS}) reached: ${retry}`, }).toObject() - Bus.publish(Session.Event.Error, { + await Bus.publish(Session.Event.Error, { sessionID: input.assistantMessage.sessionID, error: input.assistantMessage.error, }) @@ -385,7 +410,7 @@ export namespace SessionProcessor { continue } input.assistantMessage.error = error - Bus.publish(Session.Event.Error, { + await Bus.publish(Session.Event.Error, { sessionID: input.assistantMessage.sessionID, error: input.assistantMessage.error, }) diff --git a/packages/cli/test/cli/classify-session-error.test.ts b/packages/cli/test/cli/classify-session-error.test.ts index f88c532..544eff9 100644 --- a/packages/cli/test/cli/classify-session-error.test.ts +++ b/packages/cli/test/cli/classify-session-error.test.ts @@ -60,4 +60,22 @@ describe("classifySessionError (#63)", () => { expect(res.code).toBe("500") expect(res.message).toBe("internal") }) + + test.each([ + [undefined, "The provider ended the response with an error finish reason.", "provider"], + [429, "Rate limit exceeded", "rate_limit"], + [429, "Resource has been exhausted", "rate_limit"], + [429, "rate_limit_error", "rate_limit"], + [401, "Invalid API key", "auth"], + [401, "Unauthenticated", "auth"], + [403, "Permission denied", "auth"], + [400, "Invalid request", "unknown"], + [402, "Payment required", "unknown"], + [404, "Not found", "unknown"], + [undefined, "Stream timeout", "timeout"], + ] as const)("APIError preserves specific classifications (%s, %s) → %s", (statusCode, message, reason) => { + const res = classifySessionError({ name: "APIError", data: { statusCode, message, isRetryable: false } }) + expect(res.reason).toBe(reason) + expect(res.code).toBe(statusCode ? String(statusCode) : undefined) + }) }) diff --git a/packages/cli/test/cli/run-provider-finish.test.ts b/packages/cli/test/cli/run-provider-finish.test.ts new file mode 100644 index 0000000..6089c2e --- /dev/null +++ b/packages/cli/test/cli/run-provider-finish.test.ts @@ -0,0 +1,216 @@ +import { describe, expect, test } from "bun:test" +import path from "path" +import { createGoogleGenerativeAI } from "@ai-sdk/google" +import { tmpdir } from "../fixture/fixture" + +const entry = path.resolve(import.meta.dir, "../../src/index.ts") + +describe("headless provider finish reasons (#108)", () => { + test.each([ + ["MALFORMED_FUNCTION_CALL", "error", 1, false, false], + ["MALFORMED_FUNCTION_CALL", "error", 1, true, true], + ["SAFETY", "content-filter", 1, false, false], + ["RECITATION", "content-filter", 1, false, false], + ["BLOCKLIST", "content-filter", 1, false, false], + ["SPII", "content-filter", 1, false, false], + ["PROHIBITED_CONTENT", "content-filter", 1, false, false], + ["FINISH_REASON_UNSPECIFIED", "other", 0, false, false], + ["OTHER", "other", 0, false, false], + ["STOP", "stop", 0, false, false], + ["STOP", "stop", 0, true, false], + ["MAX_TOKENS", "length", 0, false, false], + ] as const)( + "normal Gemini stream ending %s", + async (reason, finish, code, tool, partial) => { + await using tmp = await tmpdir() + let calls = 0 + const server = Bun.serve({ + port: 0, + fetch(): Response { + calls++ + const chunks = + tool && calls === 1 + ? [ + { + candidates: [ + { + index: 0, + content: { + role: "model", + parts: [ + { functionCall: { name: "read", args: { filePath: path.join(tmp.path, "aictrl.json") } } }, + ], + }, + finishReason: "STOP", + }, + ], + usageMetadata: { promptTokenCount: 7, candidatesTokenCount: 3, totalTokenCount: 10 }, + }, + ] + : [ + { + candidates: [ + { + index: 0, + content: { + role: "model", + parts: [ + { text: "Checking the input.", thought: true }, + ...(partial ? [{ text: "Partial review." }] : []), + ], + }, + }, + ], + }, + { + candidates: [{ index: 0, content: { role: "model", parts: [] }, finishReason: reason }], + usageMetadata: { promptTokenCount: 7, candidatesTokenCount: 3, totalTokenCount: 10 }, + }, + ] + return new Response(chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join(""), { + headers: { "content-type": "text/event-stream" }, + }) + }, + }) + await Bun.write( + path.join(tmp.path, "aictrl.json"), + JSON.stringify({ + provider: { + fixture: { + npm: "@ai-sdk/google", + options: { apiKey: "fixture", baseURL: `http://127.0.0.1:${server.port}` }, + models: { "gemini-fixture": { name: "fixture", limit: { context: 100000, output: 1000 } } }, + }, + }, + agent: { title: { disable: true } }, + }), + ) + const proc = Bun.spawn( + [ + "bun", + "run", + "--conditions=browser", + entry, + "run", + "--format", + "json", + "--thinking", + "--model", + "fixture/gemini-fixture", + "Check this input.", + ], + { + cwd: tmp.path, + env: { + ...process.env, + AICTRL_DISABLE_DEFAULT_PLUGINS: "true", + AICTRL_DISABLE_MODELS_FETCH: "true", + AICTRL_DISABLE_AUTOCOMPACT: "true", + }, + stdout: "pipe", + stderr: "pipe", + }, + ) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15000) + try { + const [stdout, stderr, exit] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]) + const events = stdout + .split("\n") + .filter((line) => line.startsWith("{")) + .map((line) => JSON.parse(line)) + expect(exit, stderr + stdout).toBe(code) + expect(calls).toBe(tool ? 2 : 1) + const message = events.filter((event) => event.type === "message_complete") + expect(message, stdout).toHaveLength(tool ? 2 : 1) + if (tool) { + expect(message[0].finish).toBe("tool-calls") + expect(events.filter((event) => event.type === "tool_use")).toHaveLength(1) + } + expect(message.at(-1).finish).toBe(finish) + expect(message.at(-1).status).toBe(code ? "error" : "completed") + expect(message.at(-1).usageStatus).toBe("reported") + expect(message.at(-1).tokens).toMatchObject({ input: 7, output: 3 }) + expect(events.filter((event) => event.type === "reasoning")).toHaveLength(1) + expect(events.filter((event) => event.type === "text")).toHaveLength(partial ? 1 : 0) + expect(events.filter((event) => event.type === "session_complete")).toHaveLength(1) + expect(events.filter((event) => event.type === "invocation_complete")).toHaveLength(1) + const invocation = events.find((event) => event.type === "invocation_complete") + expect(invocation.status).toBe(code ? "error" : "completed") + for (const event of events.filter((event) => + ["message_complete", "session_error", "session_complete"].includes(event.type), + )) { + expect(event.sessionID).toBe(invocation.sessionID) + expect(event.invocationID).toBe(invocation.invocationID) + } + expect(events.filter((event) => event.type === "session_error")).toHaveLength(code ? 1 : 0) + if (code) { + expect(events.find((event) => event.type === "session_error").reason).toBe("provider") + expect(events.find((event) => event.type === "session_complete").error).toBeTruthy() + expect(events.filter((event) => event.type === "error")).toHaveLength(1) + const failure = events.find((event) => event.type === "error") + expect(failure.error).toMatchObject({ + name: "APIError", + data: { isRetryable: false, metadata: { finishReason: finish } }, + }) + expect(failure.sessionID).toBe(invocation.sessionID) + expect(failure.invocationID).toBe(invocation.invocationID) + expect(events.findIndex((event) => event.type === "session_error")).toBeLessThan( + events.findIndex((event) => event.type === "session_complete"), + ) + } + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + server.stop(true) + } + }, + 20000, + ) +}) + +describe("pinned Google adapter finish mappings", () => { + test.each([ + ["STOP", "stop"], + ["MAX_TOKENS", "length"], + ["IMAGE_SAFETY", "content-filter"], + ["RECITATION", "content-filter"], + ["SAFETY", "content-filter"], + ["BLOCKLIST", "content-filter"], + ["PROHIBITED_CONTENT", "content-filter"], + ["SPII", "content-filter"], + ["MALFORMED_FUNCTION_CALL", "error"], + ["OTHER", "other"], + ["FINISH_REASON_UNSPECIFIED", "other"], + ["LANGUAGE", "unknown"], + ])("%s → %s", async (raw, normalized) => { + const provider = createGoogleGenerativeAI({ + apiKey: "fixture", + fetch: Object.assign( + async () => + new Response( + `data: ${JSON.stringify({ + candidates: [{ index: 0, content: { role: "model", parts: [] }, finishReason: raw }], + usageMetadata: { promptTokenCount: 7, candidatesTokenCount: 3, totalTokenCount: 10 }, + })}\n\n`, + { headers: { "content-type": "text/event-stream" } }, + ), + { preconnect: globalThis.fetch.preconnect }, + ), + }) + const response = await provider("gemini-fixture").doStream({ + prompt: [{ role: "user", content: [{ type: "text", text: "Hello" }] }], + }) + const reader = response.stream.getReader() + let finish: string | undefined + while (true) { + const { done, value } = await reader.read() + if (done) break + if (value.type === "finish") finish = value.finishReason + } + expect(finish).toBe(normalized) + }) +}) diff --git a/packages/cli/test/session/processor-error-publication.test.ts b/packages/cli/test/session/processor-error-publication.test.ts new file mode 100644 index 0000000..2304691 --- /dev/null +++ b/packages/cli/test/session/processor-error-publication.test.ts @@ -0,0 +1,101 @@ +import { describe, expect, spyOn, test } from "bun:test" +import { APICallError } from "ai" +import { Agent } from "../../src/agent/agent" +import { Bus } from "../../src/bus" +import { Identifier } from "../../src/id/id" +import { Instance } from "../../src/project/instance" +import { Provider } from "../../src/provider/provider" +import { Session } from "../../src/session" +import { LLM } from "../../src/session/llm" +import { MessageV2 } from "../../src/session/message-v2" +import { SessionProcessor } from "../../src/session/processor" +import { SessionRetry } from "../../src/session/retry" +import { tmpdir } from "../fixture/fixture" + +describe("processor terminal error publication", () => { + test.each([false, true])("waits for subscribers before returning (retryable=%s)", async (retryable) => { + await using tmp = await tmpdir({ + config: { + provider: { + fixture: { + npm: "@ai-sdk/google", + options: { apiKey: "fixture" }, + models: { "gemini-fixture": { name: "fixture", limit: { context: 100000, output: 1000 } } }, + }, + }, + }, + }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const session = await Session.create({}) + const model = await Provider.getModel("fixture", "gemini-fixture") + const agent = await Agent.get("build") + const abort = new AbortController().signal + const user: MessageV2.User = { + id: Identifier.ascending("message"), + sessionID: session.id, + role: "user", + time: { created: Date.now() }, + agent: agent.name, + model: { providerID: model.providerID, modelID: model.id }, + } + const message: MessageV2.Assistant = { + id: Identifier.ascending("message"), + sessionID: session.id, + parentID: user.id, + role: "assistant", + time: { created: Date.now() }, + agent: agent.name, + mode: agent.name, + modelID: model.id, + providerID: model.providerID, + path: { cwd: tmp.path, root: tmp.path }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + } + await Session.updateMessage(user) + await Session.updateMessage(message) + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + const unsub = Bus.subscribe(Session.Event.Error, async () => { + entered.resolve() + await release.promise + }) + const stream = spyOn(LLM, "stream").mockRejectedValue( + new APICallError({ + url: "https://fixture.invalid", + requestBodyValues: {}, + message: "fixture provider error", + statusCode: retryable ? 500 : 400, + isRetryable: retryable, + }), + ) + const sleep = spyOn(SessionRetry, "sleep").mockResolvedValue() + let finished = false + const processing = SessionProcessor.create({ assistantMessage: message, sessionID: session.id, model, abort }) + .process({ user, sessionID: session.id, model, agent, abort, system: [], messages: [], tools: {} }) + .then((result) => { + finished = true + return result + }) + try { + await entered.promise + // Give the processor time to return if it accidentally fire-and-forgets the subscriber. + await Bun.sleep(30) + expect(finished).toBe(false) + release.resolve() + await processing + expect(message.error).toBeDefined() + expect(stream).toHaveBeenCalledTimes(retryable ? SessionRetry.MAX_RETRY_ATTEMPTS + 1 : 1) + } finally { + release.resolve() + await processing + stream.mockRestore() + sleep.mockRestore() + unsub() + } + }, + }) + }) +})