fix(ai): preserve stream transport failures

packages/ai/src/route/executor.ts

@@ -297,7 +297,7 @@ export const classifyHttpFailure = (input: {
   })
 }
 
-const toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: unknown) => {
+export const mapHttpError = (error: unknown, redactedNames: ReadonlyArray<string | RegExp>) => {
   const transportError = (input: {
     readonly message: string
     readonly kind?: string | undefined
@@ -314,23 +314,35 @@ const toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: u
       }),
     })
 
-  if (Cause.isTimeoutError(error)) {
-    return transportError({ message: error.message, kind: "Timeout" })
-  }
+  const cause =
+    HttpClientError.isHttpClientError(error) && "cause" in error.reason
+      ? error.reason.cause
+      : error instanceof Error
+        ? error.cause
+        : undefined
+  const code = [cause, error]
+    .map((value) => (typeof value === "object" && value !== null ? Reflect.get(value, "code") : undefined))
+    .find((value): value is string => typeof value === "string")
+  const request = HttpClientError.isHttpClientError(error) && "request" in error ? error.request : undefined
+  const raw = cause instanceof Error ? cause.message : error instanceof Error ? error.message : undefined
+  const detail = raw && request ? redactBody(raw, secretValues(request)) : raw
+  const message = code && detail && !detail.includes(code) ? `${code}: ${detail}` : detail
+
+  if (Cause.isTimeoutError(error) || Cause.isTimeoutError(cause))
+    return transportError({ message: message ?? "HTTP transport timed out", kind: code ?? "Timeout", request })
   if (!HttpClientError.isHttpClientError(error)) {
-    return transportError({ message: error instanceof Error ? error.message : "HTTP transport failed" })
+    return transportError({ message: message ?? "HTTP transport failed", kind: code, request })
   }
-  const request = "request" in error ? error.request : undefined
   if (error.reason._tag === "TransportError") {
     return transportError({
-      message: error.reason.description ?? "HTTP transport failed",
-      kind: error.reason._tag,
+      message: message ?? error.reason.description ?? "HTTP transport failed",
+      kind: code ?? error.reason._tag,
       request,
     })
   }
   return transportError({
-    message: `HTTP transport failed: ${error.reason._tag}`,
-    kind: error.reason._tag,
+    message: message ?? `HTTP transport failed: ${error.reason._tag}`,
+    kind: code ?? error.reason._tag,
     request,
   })
 }
@@ -343,15 +355,16 @@ export const layer: Layer.Layer<Service, never, HttpClient.HttpClient> = Layer.e
       Effect.gen(function* () {
         const redactedNames = yield* Headers.CurrentRedactedNames
         if (!middleware)
-          return yield* http
-            .execute(request)
-            .pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(request, redactedNames)))
+          return yield* http.execute(request).pipe(
+            Effect.mapError((error) => mapHttpError(error, redactedNames)),
+            Effect.flatMap(statusError(request, redactedNames)),
+          )
 
         const response = yield* middleware(request, (input) =>
           http
             .execute(input)
             .pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause))))),
-        ).pipe(Effect.mapError(toHttpError(redactedNames)))
+        ).pipe(Effect.mapError((error) => mapHttpError(error, redactedNames)))
         return yield* statusError(response.request, redactedNames)(response)
       })
     return Service.of({

packages/ai/src/route/transport/http.ts

@@ -2,6 +2,7 @@ import { Effect, Stream } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import { Auth } from "../auth"
 import { render as renderEndpoint } from "../endpoint"
+import { mapHttpError } from "../executor"
 import { Framing } from "../framing"
 import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index"
 import * as ProviderShared from "../../protocols/shared"
@@ -86,20 +87,16 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
         middleware: prepareInput.middleware,
       }
     }),
-  frames: (prepared, request, runtime) =>
+  frames: (prepared, _request, runtime) =>
     Stream.unwrap(
       runtime.http
         .execute(prepared.request, prepared.middleware)
         .pipe(
           Effect.map((response) =>
-            prepared.framing.frame(
-              response.stream.pipe(
-                Stream.mapError((error) =>
-                  ProviderShared.eventError(
-                    `${request.model.provider}/${request.model.route.id}`,
-                    `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
-                    ProviderShared.errorText(error),
-                  ),
+            Stream.unwrap(
+              Effect.map(Headers.CurrentRedactedNames, (redactedNames) =>
+                prepared.framing.frame(
+                  response.stream.pipe(Stream.mapError((error) => mapHttpError(error, redactedNames))),
                 ),
               ),
             ),

packages/ai/test/lib/http.ts

@@ -63,14 +63,14 @@ export const dynamicResponse = (handler: Handler) => runtimeLayer(handlerLayer(h
  * Layer that emits the supplied SSE chunks and then aborts mid-stream. Used to
  * exercise transport errors that surface during parsing.
  */
-export const truncatedStream = (chunks: ReadonlyArray<string>) =>
+export const truncatedStream = (chunks: ReadonlyArray<string>, error: Error = new Error("connection reset")) =>
   dynamicResponse((input) =>
     Effect.sync(() => {
       const encoder = new TextEncoder()
       const stream = new ReadableStream({
         start(controller) {
           for (const chunk of chunks) controller.enqueue(encoder.encode(chunk))
-          controller.error(new Error("connection reset"))
+          controller.error(error)
         },
       })
       return input.respond(stream, { headers: SSE_HEADERS })

packages/ai/test/provider/openai-chat.test.ts

@@ -1221,12 +1221,18 @@ describe("OpenAI Chat route", () => {
 
   it.effect("surfaces transport errors that occur mid-stream", () =>
     Effect.gen(function* () {
-      const layer = truncatedStream([
-        `data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`,
-      ])
+      const layer = truncatedStream(
+        [`data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`],
+        Object.assign(new Error("socket closed unexpectedly"), { code: "ECONNRESET" }),
+      )
       const error = yield* LLMClient.generate(request).pipe(Effect.provide(layer), Effect.flip)
 
-      expect(error.message).toContain("Failed to read openai/openai-chat stream")
+      expect(error.reason).toMatchObject({
+        _tag: "Transport",
+        message: "ECONNRESET: socket closed unexpectedly",
+        kind: "ECONNRESET",
+        url: "https://api.openai.test/v1/chat/completions",
+      })
     }),
   )
 

packages/core/src/database/v1-migration.ts

@@ -7,7 +7,7 @@ import { SessionV1 } from "@opencode-ai/schema/session-v1"
 import { SessionMessage } from "../session/message"
 import { SessionSchema } from "../session/schema"
 import { KVTable } from "../kv/sql"
-import { EventSequenceTable } from "../event/sql"
+import { EventSequenceTable, EventTable } from "../event/sql"
 import { eq, sql } from "drizzle-orm"
 import { Global } from "@opencode-ai/util/global"
 import { existsSync } from "node:fs"
@@ -161,7 +161,6 @@ type NextMessage = {
 
 const lock = Semaphore.makeUnsafe(1)
 const MIGRATION_STATE_KEY = "migration.v1-v2"
-const EVENT_DELETE_BATCH_SIZE = 1_000
 const decodeJson = Schema.decodeUnknownOption(Schema.UnknownFromJsonString)
 const decodeMessage = Schema.decodeUnknownOption(SessionV1.Info)
 const decodePart = Schema.decodeUnknownOption(SessionV1.Part)
@@ -486,15 +485,7 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
           yield* db
             .transaction((tx) =>
               Effect.gen(function* () {
-                while (true) {
-                  yield* tx.run(sql`
-                    DELETE FROM event
-                    WHERE rowid IN (SELECT rowid FROM event LIMIT ${EVENT_DELETE_BATCH_SIZE})
-                  `)
-                  const deleted = (yield* tx.get<{ value: number }>(sql`SELECT changes() AS value`))?.value ?? 0
-                  if (deleted < EVENT_DELETE_BATCH_SIZE) break
-                  yield* Effect.yieldNow
-                }
+                yield* tx.delete(EventTable).run()
                 yield* tx
                   .insert(KVTable)
                   .values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions" } })

packages/core/src/tool/plugin/question.ts

@@ -21,9 +21,7 @@ Usage notes:
 - If you recommend a specific option, make that the first option in the list and add "(Recommended)" at the end of the label`
 
 export const Input = Schema.Struct({
-  questions: Schema.Array(Question.Prompt)
-    .check(Schema.isNonEmpty())
-    .annotate({ description: "Questions to ask" }),
+  questions: Schema.NonEmptyArray(Question.Prompt).annotate({ description: "Questions to ask" }),
 })
 
 export const Output = Schema.Struct({

packages/core/test/tool-question.test.ts

@@ -89,30 +89,6 @@ const it = testEffect(
 )
 
 describe("QuestionTool", () => {
-  it.effect("emits one item schema for the nonempty questions array", () =>
-    Effect.gen(function* () {
-      captured = undefined
-      const registry = yield* Tool.Service
-      const definition = (yield* toolDefinitions(registry)).find((tool) => tool.name === QuestionTool.name)
-
-      expect(definition?.inputSchema).toHaveProperty("properties.questions.type", "array")
-      expect(definition?.inputSchema).toHaveProperty("properties.questions.minItems", 1)
-      expect(definition?.inputSchema).toHaveProperty("properties.questions.items")
-      expect(definition?.inputSchema).not.toHaveProperty("properties.questions.prefixItems")
-      expect(
-        yield* executeTool(registry, {
-          sessionID,
-          ...toolIdentity,
-          call: { type: "tool-call", id: "call-question-empty", name: QuestionTool.name, input: { questions: [] } },
-        }),
-      ).toMatchObject({
-        status: "error",
-        error: { type: "tool.execution", message: expect.stringContaining("Invalid tool input") },
-      })
-      expect(capturedInput()).toBeUndefined()
-    }),
-  )
-
   it.effect("omits a catalog-denied question and enforces its leaf permission", () =>
     Effect.gen(function* () {
       captured = undefined

packages/core/test/v1-migration.test.ts

@@ -11,7 +11,7 @@ import { Project } from "@opencode-ai/core/project"
 import { ProjectTable } from "@opencode-ai/core/project/sql"
 import { AbsolutePath } from "@opencode-ai/core/schema"
 import { Global } from "@opencode-ai/util/global"
-import { Effect, Fiber, Layer, Logger, Schedule, Schema, Scope } from "effect"
+import { Effect, Layer, Logger, Schedule, Schema, Scope } from "effect"
 import { eq, sql } from "drizzle-orm"
 import type { SqlClient } from "effect/unstable/sql/SqlClient"
 import { tmpdir } from "./fixture/tmpdir"
@@ -798,35 +798,6 @@ describe("V1Migration database workflow", () => {
     )
   })
 
-  test("yields while clearing stale events in batches", async () => {
-    await database(
-      Effect.gen(function* () {
-        const { db } = yield* Database.Service
-        yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES ('stale', 2500)`)
-        yield* db.run(sql`
-          WITH RECURSIVE rows(value) AS (
-            VALUES(1)
-            UNION ALL
-            SELECT value + 1 FROM rows WHERE value < 2500
-          )
-          INSERT INTO event (id, aggregate_id, seq, created, type, data)
-          SELECT printf('event_%04d', value), 'stale', value, 1, 'session.renamed.1', '{}'
-          FROM rows
-        `)
-        let yielded = false
-        const heartbeat = yield* Effect.yieldNow.pipe(
-          Effect.andThen(Effect.sync(() => (yielded = true))),
-          Effect.forkChild({ startImmediately: true }),
-        )
-
-        expect(yield* V1Migration.run()).toEqual({ status: "completed" })
-        expect(yielded).toBe(true)
-        yield* Fiber.join(heartbeat)
-        expect(yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM event`)).toEqual({ value: 0 })
-      }),
-    )
-  })
-
   test("imports previous V2 sessions and messages as part of the migration", async () => {
     await using tmp = await tmpdir()
     const filename = path.join(tmp.path, "opencode-next.db")