fix(ai): preserve stream transport failures

Co-authored-by: Dax Raad <d@ironbay.co>

packages/ai/src/route/executor.ts

@@ -1,4 +1,4 @@
-import { Cause, Context, Effect, Layer, Option, Schema } from "effect"
+import { Cause, Context, Effect, Layer, Option, Schema, Stream } from "effect"
 import {
   FetchHttpClient,
   Headers,
@@ -16,12 +16,17 @@ import {
   TransportReason,
 } from "../schema"
 import { classifyProviderFailure } from "../provider-error"
+import { isRecord } from "../utils/record"
 
 export interface Interface {
   readonly execute: (
     request: HttpClientRequest.HttpClientRequest,
     middleware?: HttpMiddleware,
   ) => Effect.Effect<HttpClientResponse.HttpClientResponse, AIError>
+  readonly stream: (
+    request: HttpClientRequest.HttpClientRequest,
+    middleware?: HttpMiddleware,
+  ) => Stream.Stream<Uint8Array, AIError>
 }
 
 export type HttpHandler = (
@@ -297,41 +302,51 @@ export const classifyHttpFailure = (input: {
   })
 }
 
-const toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: unknown) => {
-  const transportError = (input: {
-    readonly message: string
-    readonly kind?: string | undefined
-    readonly request?: HttpClientRequest.HttpClientRequest | undefined
-  }) =>
+type HttpOperation = "request" | "read"
+
+const httpError = (input: {
+  readonly error: unknown
+  readonly request: HttpClientRequest.HttpClientRequest
+  readonly operation: HttpOperation
+  readonly redactedNames: ReadonlyArray<string | RegExp>
+}) => {
+  const transportError = (failure: { readonly message: string; readonly code?: string | undefined }) =>
     new AIError({
       module: "RequestExecutor",
-      method: "execute",
+      method: input.operation,
       reason: new TransportReason({
-        message: input.message,
-        kind: input.kind,
-        url: input.request ? redactUrl(input.request.url) : undefined,
-        http: input.request ? new HttpContext({ request: requestDetails(input.request, redactedNames) }) : undefined,
+        message: failure.message,
+        transport: "http",
+        operation: input.operation,
+        code: failure.code,
+        url: redactUrl(input.request.url),
+        http: new HttpContext({ request: requestDetails(input.request, input.redactedNames) }),
       }),
     })
 
-  if (Cause.isTimeoutError(error)) {
-    return transportError({ message: error.message, kind: "Timeout" })
-  }
-  if (!HttpClientError.isHttpClientError(error)) {
-    return transportError({ message: error instanceof Error ? error.message : "HTTP transport failed" })
-  }
-  const request = "request" in error ? error.request : undefined
-  if (error.reason._tag === "TransportError") {
+  const source =
+    HttpClientError.isHttpClientError(input.error) && "cause" in input.error.reason
+      ? input.error.reason.cause
+      : input.error
+  const code = isRecord(source) && typeof source.code === "string" ? source.code : undefined
+  const request = HttpClientError.isHttpClientError(input.error) ? input.error.request : input.request
+  const raw = source instanceof Error ? source.message : input.error instanceof Error ? input.error.message : undefined
+  const detail = raw ? redactBody(raw, secretValues(request)) : undefined
+  const message = code && detail && !detail.includes(code) ? `${code}: ${detail}` : detail
+
+  if (Cause.isTimeoutError(input.error) || Cause.isTimeoutError(source))
+    return transportError({ message: message ?? "HTTP transport timed out", code: code ?? "Timeout" })
+  if (!HttpClientError.isHttpClientError(input.error))
+    return transportError({ message: message ?? "HTTP transport failed", code })
+  if (input.error.reason._tag === "TransportError") {
     return transportError({
-      message: error.reason.description ?? "HTTP transport failed",
-      kind: error.reason._tag,
-      request,
+      message: message ?? input.error.reason.description ?? "HTTP transport failed",
+      code: code ?? input.error.reason._tag,
     })
   }
   return transportError({
-    message: `HTTP transport failed: ${error.reason._tag}`,
-    kind: error.reason._tag,
-    request,
+    message: message ?? `HTTP transport failed: ${input.error.reason._tag}`,
+    code: code ?? input.error.reason._tag,
   })
 }
 
@@ -339,23 +354,41 @@ export const layer: Layer.Layer<Service, never, HttpClient.HttpClient> = Layer.e
   Service,
   Effect.gen(function* () {
     const http = yield* HttpClient.HttpClient
-    const executeOnce = (request: HttpClientRequest.HttpClientRequest, middleware?: HttpMiddleware) =>
+    const execute = (
+      request: HttpClientRequest.HttpClientRequest,
+      middleware: HttpMiddleware | undefined,
+      redactedNames: ReadonlyArray<string | RegExp>,
+    ) =>
       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) => httpError({ error, request, operation: "request", 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) => httpError({ error, request, operation: "request", redactedNames })))
         return yield* statusError(response.request, redactedNames)(response)
       })
+    const executeOnce = (request: HttpClientRequest.HttpClientRequest, middleware?: HttpMiddleware) =>
+      Effect.flatMap(Headers.CurrentRedactedNames, (redactedNames) => execute(request, middleware, redactedNames))
     return Service.of({
       execute: executeOnce,
+      stream: (request, middleware) =>
+        Stream.unwrap(
+          Effect.gen(function* () {
+            const redactedNames = yield* Headers.CurrentRedactedNames
+            const response = yield* execute(request, middleware, redactedNames)
+            return response.stream.pipe(
+              Stream.mapError((error) =>
+                httpError({ error, request: response.request, operation: "read", redactedNames }),
+              ),
+            )
+          }),
+        ),
     })
   }),
 )

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

@@ -1,4 +1,4 @@
-import { Effect, Stream } from "effect"
+import { Effect } from "effect"
 import { Headers, HttpClientRequest } from "effect/unstable/http"
 import { Auth } from "../auth"
 import { render as renderEndpoint } from "../endpoint"
@@ -86,26 +86,8 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
         middleware: prepareInput.middleware,
       }
     }),
-  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),
-                  ),
-                ),
-              ),
-            ),
-          ),
-        ),
-    ),
+  frames: (prepared, _request, runtime) =>
+    prepared.framing.frame(runtime.http.stream(prepared.request, prepared.middleware)),
 })
 
 export const sseJson = {

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

@@ -1,6 +1,6 @@
 import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
 import { Headers } from "effect/unstable/http"
-import { AIError, TransportReason } from "../../schema"
+import { AIError, TransportReason, type TransportOperation } from "../../schema"
 import * as HttpTransport from "./http"
 import type { Transport } from "./index"
 
@@ -29,12 +29,18 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/AI
 const transportError = (
   method: string,
   message: string,
-  input: { readonly url?: string; readonly kind?: string } = {},
+  input: { readonly operation: TransportOperation; readonly url?: string; readonly code?: string },
 ) =>
   new AIError({
     module: "WebSocketExecutor",
     method,
-    reason: new TransportReason({ message, url: input.url, kind: input.kind }),
+    reason: new TransportReason({
+      message,
+      transport: "websocket",
+      operation: input.operation,
+      url: input.url,
+      code: input.code,
+    }),
   })
 
 const eventMessage = (event: Event) => {
@@ -55,7 +61,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
     return Effect.fail(
       transportError("open", `WebSocket closed before opening (state ${ws.readyState})`, {
         url: input.url,
-        kind: "open",
+        operation: "request",
+        code: "closed",
       }),
     )
   }
@@ -79,7 +86,10 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
       cleanup()
       resume(
         Effect.fail(
-          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
+          transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
+            url: input.url,
+            operation: "request",
+          }),
         ),
       )
     }
@@ -89,7 +99,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
         Effect.fail(
           transportError("open", `WebSocket closed before opening with code ${event.code}`, {
             url: input.url,
-            kind: "open",
+            operation: "request",
+            code: String(event.code),
           }),
         ),
       )
@@ -118,7 +129,8 @@ const webSocketUrl = (value: string) =>
     catch: (error) =>
       transportError("prepare", error instanceof Error ? error.message : "Invalid WebSocket URL", {
         url: value,
-        kind: "websocket",
+        operation: "request",
+        code: "invalid-url",
       }),
   })
 
@@ -129,7 +141,7 @@ export const open = (input: WebSocketRequest) =>
     catch: (error) =>
       transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
         url: input.url,
-        kind: "open",
+        operation: "request",
       }),
   }).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
 
@@ -150,7 +162,10 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", "Unsupported WebSocket message payload", { url: input.url, kind: "message" }),
+          transportError("message", "Unsupported WebSocket message payload", {
+            url: input.url,
+            operation: "read",
+          }),
         ),
       )
     }
@@ -158,7 +173,10 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, kind: "message" }),
+          transportError("message", `WebSocket error: ${eventMessage(event)}`, {
+            url: input.url,
+            operation: "read",
+          }),
         ),
       )
     }
@@ -167,7 +185,11 @@ export const fromWebSocket = (
       Queue.failCauseUnsafe(
         messages,
         Cause.fail(
-          transportError("message", `WebSocket closed with code ${event.code}`, { url: input.url, kind: "close" }),
+          transportError("message", `WebSocket closed with code ${event.code}`, {
+            url: input.url,
+            operation: "read",
+            code: String(event.code),
+          }),
         ),
       )
     }
@@ -188,7 +210,7 @@ export const fromWebSocket = (
           catch: (error) =>
             transportError("sendText", error instanceof Error ? error.message : "Failed to send WebSocket message", {
               url: input.url,
-              kind: "write",
+              operation: "write",
             }),
         }),
       messages: Stream.fromQueue(messages),
@@ -243,7 +265,8 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
       return Stream.fail(
         transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
           url: prepared.url,
-          kind: "websocket",
+          operation: "request",
+          code: "unavailable",
         }),
       )
     }

packages/ai/src/schema/errors.ts

@@ -92,10 +92,18 @@ export class ProviderInternalReason extends Schema.Class<ProviderInternalReason>
   http: Schema.optional(HttpContext),
 }) {}
 
+export const TransportType = Schema.Literals(["http", "websocket"])
+export type TransportType = typeof TransportType.Type
+
+export const TransportOperation = Schema.Literals(["request", "read", "write"])
+export type TransportOperation = typeof TransportOperation.Type
+
 export class TransportReason extends Schema.Class<TransportReason>("AI.Error.Transport")({
   _tag: Schema.tag("Transport"),
   message: Schema.String,
-  kind: Schema.optional(Schema.String),
+  transport: TransportType,
+  operation: TransportOperation,
+  code: Schema.optional(Schema.String),
   url: Schema.optional(Schema.String),
   http: Schema.optional(HttpContext),
 }) {}

packages/ai/test/executor.test.ts

@@ -1,10 +1,10 @@
 import { describe, expect } from "bun:test"
-import { Effect, Layer, Ref } from "effect"
+import { Effect, Layer, Ref, Stream } from "effect"
 import { Headers, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 import { LLM, AIError } from "../src"
 import { LLMClient, RequestExecutor } from "../src/route"
 import * as OpenAIChat from "../src/protocols/openai-chat"
-import { dynamicResponse } from "./lib/http"
+import { dynamicResponse, systemError } from "./lib/http"
 import { deltaChunk } from "./lib/openai-chunks"
 import { sseRaw } from "./lib/sse"
 import { it } from "./lib/effect"
@@ -67,6 +67,35 @@ const expectAIError = (error: unknown) => {
 const errorHttp = (error: AIError) => ("http" in error.reason ? error.reason.http : undefined)
 
 describe("RequestExecutor", () => {
+  it.effect("parses response body failures at the executor seam", () =>
+    Effect.gen(function* () {
+      const executor = yield* RequestExecutor.Service
+      const error = yield* executor.stream(secretRequest).pipe(Stream.runDrain, Effect.flip)
+
+      expectAIError(error)
+      expect(error.reason).toMatchObject({
+        _tag: "Transport",
+        message: "ECONNRESET: disconnected <redacted> <redacted>",
+        transport: "http",
+        operation: "read",
+        code: "ECONNRESET",
+        url: "https://provider.test/v1/chat?api_key=%3Credacted%3E&debug=1",
+      })
+    }).pipe(
+      Effect.provide(
+        responsesLayer([
+          new Response(
+            new ReadableStream({
+              start(controller) {
+                controller.error(systemError("ECONNRESET", "disconnected query-secret-123 header-secret-456"))
+              },
+            }),
+          ),
+        ]),
+      ),
+    ),
+  )
+
   it.effect("preserves middleware error messages", () =>
     Effect.gen(function* () {
       const executor = yield* RequestExecutor.Service

packages/ai/test/lib/http.ts

@@ -34,6 +34,12 @@ const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
 
 export type RuntimeEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService
 
+export interface SystemError extends Error {
+  readonly code: string
+}
+
+export const systemError = (code: string, message: string): SystemError => Object.assign(new Error(message), { code })
+
 export const runtimeLayer = (layer: Layer.Layer<HttpClient.HttpClient>): Layer.Layer<RuntimeEnv> => {
   const requestExecutorLayer = RequestExecutor.layer.pipe(Layer.provide(layer))
   const deps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
@@ -63,14 +69,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

@@ -22,7 +22,7 @@ import { ProviderShared } from "../../src/protocols/shared"
 import { Auth, LLMClient } from "../../src/route"
 import { compileRequest } from "../../src/route/client"
 import { it } from "../lib/effect"
-import { dynamicResponse, fixedResponse, truncatedStream } from "../lib/http"
+import { dynamicResponse, fixedResponse, systemError, truncatedStream } from "../lib/http"
 import { deltaChunk, usageChunk } from "../lib/openai-chunks"
 import { sseEvents } from "../lib/sse"
 
@@ -1221,12 +1221,20 @@ 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`],
+        systemError("ECONNRESET", "socket closed unexpectedly"),
+      )
       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",
+        transport: "http",
+        operation: "read",
+        code: "ECONNRESET",
+        url: "https://api.openai.test/v1/chat/completions",
+      })
     }),
   )
 

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

@@ -243,6 +243,7 @@ describe("OpenAI Responses route", () => {
           RequestExecutor.Service,
           RequestExecutor.Service.of({
             execute: () => Effect.die("unexpected HTTP request"),
+            stream: () => Stream.die("unexpected HTTP request"),
           }),
         ),
         Layer.succeed(

packages/app/src/context/server-session-v2-reducer.test.ts

@@ -69,6 +69,63 @@ describe("v2 session reducer", () => {
     })
   })
 
+  test("prefers durable selection predecessors and derives them for older events", () => {
+    const source: SessionMessageInfo[] = [
+      { id: "msg_previous_agent", type: "agent-switched", agent: "build", time: { created: 1 } },
+      {
+        id: "msg_previous_model",
+        type: "model-switched",
+        model: { id: "old", providerID: "provider" },
+        time: { created: 1 },
+      },
+    ]
+    const reducer = createV2SessionReducer()
+
+    const agent = reducer.reduce(
+      source,
+      event({
+        ...base,
+        id: "evt_agent",
+        type: "session.agent.selected",
+        data: { sessionID: "ses_1", agent: "plan", previous: "review" },
+      }),
+    )
+    const model = reducer.reduce(
+      source,
+      event({
+        ...base,
+        id: "evt_model",
+        type: "session.model.selected",
+        data: {
+          sessionID: "ses_1",
+          model: { id: "new", providerID: "provider" },
+          previous: { id: "durable", providerID: "provider" },
+        },
+      }),
+    )
+    const legacyAgent = reducer.reduce(
+      source,
+      event({
+        ...base,
+        id: "evt_legacy_agent",
+        type: "session.agent.selected",
+        data: { sessionID: "ses_1", agent: "plan" },
+      }),
+    )
+
+    expect(agent?.messages.at(-1)).toMatchObject({ type: "agent-switched", agent: "plan", previous: "review" })
+    expect(model?.messages.at(-1)).toMatchObject({
+      type: "model-switched",
+      model: { id: "new" },
+      previous: { id: "durable" },
+    })
+    expect(legacyAgent?.messages.at(-1)).toMatchObject({
+      type: "agent-switched",
+      agent: "plan",
+      previous: "build",
+    })
+  })
+
   test("folds tool, retry, and completion events", () => {
     const reducer = createV2SessionReducer()
     let messages: SessionMessageInfo[] = []

packages/app/src/context/server-session-v2-reducer.ts

@@ -61,6 +61,12 @@ export function createV2SessionReducer() {
           type: "agent-switched",
           metadata: event.metadata,
           agent: event.data.agent,
+          previous:
+            event.data.previous ??
+            source.findLast(
+              (item): item is Extract<SessionMessageInfo, { type: "agent-switched" | "assistant" }> =>
+                item.type === "agent-switched" || item.type === "assistant",
+            )?.agent,
           time: { created: event.created },
         })
       case "session.model.selected":
@@ -69,10 +75,12 @@ export function createV2SessionReducer() {
           type: "model-switched",
           metadata: event.metadata,
           model: event.data.model,
-          previous: source.findLast(
-            (item): item is Extract<SessionMessageInfo, { type: "model-switched" | "assistant" }> =>
-              item.type === "model-switched" || item.type === "assistant",
-          )?.model,
+          previous:
+            event.data.previous ??
+            source.findLast(
+              (item): item is Extract<SessionMessageInfo, { type: "model-switched" | "assistant" }> =>
+                item.type === "model-switched" || item.type === "assistant",
+            )?.model,
           time: { created: event.created },
         })
       case "session.synthetic":

packages/client/src/effect/api/api.ts

@@ -339,7 +339,11 @@ export type Endpoint5_31Output =
           readonly type: "session.agent.selected"
           readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version }
           readonly location?: Location.Ref | undefined
-          readonly data: { readonly sessionID: Session.ID; readonly agent: Agent.ID }
+          readonly data: {
+            readonly sessionID: Session.ID
+            readonly agent: Agent.ID
+            readonly previous?: Agent.ID | undefined
+          }
         }
       | {
           readonly id: Event.ID
@@ -348,7 +352,11 @@ export type Endpoint5_31Output =
           readonly type: "session.model.selected"
           readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version }
           readonly location?: Location.Ref | undefined
-          readonly data: { readonly sessionID: Session.ID; readonly model: Model.Ref }
+          readonly data: {
+            readonly sessionID: Session.ID
+            readonly model: Model.Ref
+            readonly previous?: Model.Ref | undefined
+          }
         }
       | {
           readonly id: Event.ID

packages/client/src/promise/generated/types.ts

@@ -436,7 +436,7 @@ export type SessionAgentSelected = {
   type: "session.agent.selected"
   durable: { aggregateID: string; seq: number; version: 1 }
   location?: LocationRef
-  data: { sessionID: string; agent: string }
+  data: { sessionID: string; agent: string; previous?: string }
 }
 
 export type SessionModelSelected = {
@@ -446,7 +446,7 @@ export type SessionModelSelected = {
   type: "session.model.selected"
   durable: { aggregateID: string; seq: number; version: 1 }
   location?: LocationRef
-  data: { sessionID: string; model: ModelRef }
+  data: { sessionID: string; model: ModelRef; previous?: ModelRef }
 }
 
 export type SessionMoved = {

packages/core/src/aisdk.ts

@@ -763,7 +763,9 @@ function apiCallErrorReason(error: APICallError) {
   if (error.statusCode !== undefined || !error.isRetryable) return reason
   return new TransportReason({
     message: reason.message,
-    kind: error.name,
+    transport: "http",
+    operation: "request",
+    code: error.name,
     url: error.url,
     http: "http" in reason ? reason.http : undefined,
   })

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, EventTable } from "../event/sql"
+import { EventSequenceTable } from "../event/sql"
 import { eq, sql } from "drizzle-orm"
 import { Global } from "@opencode-ai/util/global"
 import { existsSync } from "node:fs"
@@ -161,6 +161,7 @@ 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)
@@ -485,7 +486,15 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
           yield* db
             .transaction((tx) =>
               Effect.gen(function* () {
-                yield* tx.delete(EventTable).run()
+                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
                   .insert(KVTable)
                   .values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions" } })

packages/core/src/session.ts

@@ -716,10 +716,11 @@ const layer = Layer.effect(
             .pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
       }),
       switchAgent: Effect.fn("Session.switchAgent")(function* (input) {
-        yield* result.get(input.sessionID)
+        const session = yield* result.get(input.sessionID)
         yield* bus.publish(SessionEvent.AgentSelected, {
           sessionID: input.sessionID,
           agent: input.agent,
+          previous: session.agent,
         })
       }),
       switchModel: Effect.fn("Session.switchModel")(function* (input) {
@@ -733,6 +734,7 @@ const layer = Layer.effect(
         yield* bus.publish(SessionEvent.ModelSelected, {
           sessionID: input.sessionID,
           model: input.model,
+          previous: session.model,
         })
       }),
       rename: Effect.fn("Session.rename")(function* (input) {

packages/core/src/session/message-updater.ts

@@ -61,7 +61,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
       "session.usage.recorded": () => Effect.void,
       "session.agent.selected": (event) => {
         return Effect.gen(function* () {
-          const previous = yield* adapter.getAgent()
+          const previous = event.data.previous ?? (yield* adapter.getAgent())
           yield* adapter.appendMessage(
             SessionMessage.AgentSelected.make({
               id: SessionMessage.ID.fromEvent(event.id),
@@ -76,7 +76,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
       },
       "session.model.selected": (event) => {
         return Effect.gen(function* () {
-          const previous = yield* adapter.getModel()
+          const previous = event.data.previous ?? (yield* adapter.getModel())
           yield* adapter.appendMessage(
             SessionMessage.ModelSelected.make({
               id: SessionMessage.ID.fromEvent(event.id),

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

@@ -21,7 +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.NonEmptyArray(Question.Prompt).annotate({ description: "Questions to ask" }),
+  questions: Schema.Array(Question.Prompt).check(Schema.isNonEmpty()).annotate({ description: "Questions to ask" }),
 })
 
 export const Output = Schema.Struct({

packages/core/test/aisdk.test.ts

@@ -9,7 +9,7 @@ import { LLM, AIError, LLMEvent, Message, isContextOverflowFailure } from "@open
 import { LLMClient, RequestExecutor } from "@opencode-ai/ai/route"
 import { compileRequest } from "@opencode-ai/ai/route/client"
 import { expect } from "bun:test"
-import { Effect, Layer } from "effect"
+import { Effect, Layer, Stream } from "effect"
 import { testEffect } from "./lib/effect"
 
 const it = testEffect(AISDK.locationLayer)
@@ -49,7 +49,10 @@ const client = LLMClient.layer.pipe(
   Layer.provide(
     Layer.succeed(
       RequestExecutor.Service,
-      RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }),
+      RequestExecutor.Service.of({
+        execute: () => Effect.die("Unexpected HTTP request"),
+        stream: () => Stream.die("Unexpected HTTP request"),
+      }),
     ),
   ),
 )
@@ -542,7 +545,12 @@ it.effect("retries status-less AI SDK transport failures", () =>
         isRetryable: true,
       }),
     )
-    expect(error.reason).toMatchObject({ _tag: "Transport", kind: "AI_APICallError" })
+    expect(error.reason).toMatchObject({
+      _tag: "Transport",
+      transport: "http",
+      operation: "request",
+      code: "AI_APICallError",
+    })
     expect(SessionRunnerRetry.isRetryable(error)).toBeTrue()
     expect("http" in error.reason ? error.reason.http?.request.url : undefined).toBe("https://api.example.com/chat")
   }),

packages/core/test/session-create.test.ts

@@ -654,7 +654,7 @@ describe("Session.create", () => {
       expect(yield* session.get(created.id)).toMatchObject({ agent: "plan" })
       expect(
         Array.from(yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect)),
-      ).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan" } }])
+      ).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan", previous: "build" } }])
       expect(yield* session.messages({ sessionID: created.id, order: "asc" })).toMatchObject([
         { type: "agent-switched", agent: "plan", previous: "build" },
       ])
@@ -678,7 +678,12 @@ describe("Session.create", () => {
   it.effect("switches the selected model through the durable Session event", () =>
     Effect.gen(function* () {
       const session = yield* Session.Service
-      const created = yield* session.create({ location })
+      const previous = Model.Ref.make({
+        id: Model.ID.make("haiku"),
+        providerID: Provider.ID.anthropic,
+        variant: Model.VariantID.make("default"),
+      })
+      const created = yield* session.create({ location, model: previous })
       const model = Model.Ref.make({
         id: Model.ID.make("sonnet"),
         providerID: Provider.ID.anthropic,
@@ -692,7 +697,10 @@ describe("Session.create", () => {
         yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect),
       )
       expect(bus).toMatchObject([{ type: "session.model.selected" }])
-      expect(bus[0]?.data).toEqual({ sessionID: created.id, model })
+      expect(bus[0]?.data).toEqual({ sessionID: created.id, model, previous })
+      expect(yield* session.messages({ sessionID: created.id, order: "asc" })).toMatchObject([
+        { type: "model-switched", model, previous },
+      ])
     }),
   )
 

packages/core/test/session-error.test.ts

@@ -39,7 +39,9 @@ describe("toSessionError", () => {
     )
     expect(toSessionError(llm(new QuotaExceededReason({ message: "quota" }))).type).toBe("provider.quota")
     expect(toSessionError(llm(new ContentPolicyReason({ message: "blocked" }))).type).toBe("provider.content-filter")
-    expect(toSessionError(llm(new TransportReason({ message: "transport" }))).type).toBe("provider.transport")
+    expect(
+      toSessionError(llm(new TransportReason({ message: "transport", transport: "http", operation: "request" }))).type,
+    ).toBe("provider.transport")
     expect(toSessionError(llm(new ProviderInternalReason({ message: "internal", status: 500 }))).type).toBe(
       "provider.internal",
     )
@@ -111,7 +113,7 @@ describe("toSessionError", () => {
     const eligible = [
       llm(new RateLimitReason({ message: "rate" })),
       llm(new ProviderInternalReason({ message: "internal", status: 500 })),
-      llm(new TransportReason({ message: "transport" })),
+      llm(new TransportReason({ message: "transport", transport: "http", operation: "request" })),
     ]
     const ineligible = [
       llm(new AuthenticationReason({ message: "auth", kind: "invalid" })),

packages/core/test/session-execution.test.ts

@@ -31,7 +31,7 @@ describe("SessionExecution lifecycle", () => {
           new AIError({
             module: "test",
             method: "stream",
-            reason: new TransportReason({ message: "Disconnected" }),
+            reason: new TransportReason({ message: "Disconnected", transport: "http", operation: "request" }),
           }),
         ),
       ),

packages/core/test/session-runner.test.ts

@@ -515,7 +515,11 @@ const providerUnavailable = () =>
   new AIError({
     module: "test",
     method: "stream",
-    reason: new TransportReason({ message: "Provider unavailable" }),
+    reason: new TransportReason({
+      message: "Provider unavailable",
+      transport: "http",
+      operation: "request",
+    }),
   })
 
 const incompleteStream = () =>

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

@@ -89,6 +89,30 @@ 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, Layer, Logger, Schedule, Schema, Scope } from "effect"
+import { Effect, Fiber, 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,6 +798,35 @@ 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")

packages/protocol/openapi.json

@@ -14373,6 +14373,9 @@
               },
               "agent": {
                 "type": "string"
+              },
+              "previous": {
+                "type": "string"
               }
             },
             "required": ["sessionID", "agent"],
@@ -14441,6 +14444,9 @@
               },
               "model": {
                 "$ref": "#/components/schemas/Model.Ref"
+              },
+              "previous": {
+                "$ref": "#/components/schemas/Model.Ref"
               }
             },
             "required": ["sessionID", "model"],

packages/schema/src/session-event.ts

@@ -69,6 +69,7 @@ export const AgentSelected = Event.durable({
   schema: {
     ...Base,
     agent: Agent.ID,
+    previous: Agent.ID.pipe(optional),
   },
 })
 export type AgentSelected = typeof AgentSelected.Type
@@ -79,6 +80,7 @@ export const ModelSelected = Event.durable({
   schema: {
     ...Base,
     model: Model.Ref,
+    previous: Model.Ref.pipe(optional),
   },
 })
 export type ModelSelected = typeof ModelSelected.Type

packages/server/src/handlers/model.ts

@@ -12,6 +12,19 @@ export const ModelHandler = HttpApiBuilder.group(Api, "server.model", (handlers)
       .handle(
         "model.list",
         Effect.fn(function* () {
+          const plugins = yield* PluginSupervisor.Service
+          yield* plugins.flush.pipe(
+            Effect.timeoutOrElse({
+              duration: "5 seconds",
+              orElse: () =>
+                Effect.fail(
+                  new ServiceUnavailableError({
+                    message: "Model catalog initialization timed out",
+                    service: "model.catalog",
+                  }),
+                ),
+            }),
+          )
           const catalog = yield* Catalog.Service
           return yield* response(catalog.model.available())
         }),

packages/server/test/model.test.ts

@@ -0,0 +1,57 @@
+import fs from "node:fs/promises"
+import path from "node:path"
+import { expect } from "bun:test"
+import { Effect } from "effect"
+import { HttpServer } from "effect/unstable/http"
+import { tmpdir } from "../../core/test/fixture/tmpdir"
+import { it } from "../../core/test/lib/effect"
+import { ServerProcess } from "../src/process"
+
+it.live("waits for plugin initialization before listing models", () =>
+  Effect.acquireUseRelease(
+    Effect.promise(() => tmpdir("opencode-model-endpoint-")),
+    (tmp) =>
+      Effect.gen(function* () {
+        yield* Effect.promise(() =>
+          fs.writeFile(
+            path.join(tmp.path, "opencode.json"),
+            JSON.stringify({
+              providers: {
+                custom: {
+                  package: "aisdk:@ai-sdk/openai-compatible",
+                  settings: { apiKey: "secret" },
+                  models: { chat: {} },
+                },
+              },
+            }),
+          ),
+        )
+        const server = yield* ServerProcess.start<never, never>({
+          hostname: "127.0.0.1",
+          port: 0,
+          password: "secret",
+          app: { version: "test-version" },
+          database: { path: ":memory:" },
+          config: { directory: tmp.path },
+          fs: { filewatcher: false },
+        })
+        const url = new URL("/api/model", HttpServer.formatAddress(server.address))
+        url.searchParams.set("location[directory]", tmp.path)
+        const response = yield* Effect.promise(() =>
+          fetch(url, { headers: { authorization: `Basic ${btoa("opencode:secret")}` } }),
+        )
+
+        expect(response.status).toBe(200)
+        const body: unknown = yield* Effect.promise(() => response.json())
+        if (!isRecord(body) || !Array.isArray(body["data"])) throw new Error("Expected a model list response")
+        expect(
+          body["data"].some((model) => isRecord(model) && model["providerID"] === "custom" && model["id"] === "chat"),
+        ).toBeTrue()
+      }),
+    (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
+  ),
+)
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value)
+}

packages/www/openapi.json

@@ -14373,6 +14373,9 @@
               },
               "agent": {
                 "type": "string"
+              },
+              "previous": {
+                "type": "string"
               }
             },
             "required": ["sessionID", "agent"],
@@ -14441,6 +14444,9 @@
               },
               "model": {
                 "$ref": "#/components/schemas/Model.Ref"
+              },
+              "previous": {
+                "$ref": "#/components/schemas/Model.Ref"
               }
             },
             "required": ["sessionID", "model"],

packages/www/public/openapi.json

@@ -14373,6 +14373,9 @@
               },
               "agent": {
                 "type": "string"
+              },
+              "previous": {
+                "type": "string"
               }
             },
             "required": ["sessionID", "agent"],
@@ -14441,6 +14444,9 @@
               },
               "model": {
                 "$ref": "#/components/schemas/Model.Ref"
+              },
+              "previous": {
+                "$ref": "#/components/schemas/Model.Ref"
               }
             },
             "required": ["sessionID", "model"],