fix: propagate provider finish errors to headless runs

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.

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:

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.

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)

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,
             })

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

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