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)) {
+  const cause =
-    return transportError({ message: error.message, kind: "Timeout" })
+    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",
+      message: message ?? error.reason.description ?? "HTTP transport failed",
-      kind: error.reason._tag,
+      kind: code ?? error.reason._tag,
       request,
     })
   }
   return transportError({
-    message: `HTTP transport failed: ${error.reason._tag}`,
+    message: message ?? `HTTP transport failed: ${error.reason._tag}`,
-    kind: 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
+          return yield* http.execute(request).pipe(
-            .execute(request)
+            Effect.mapError((error) => mapHttpError(error, redactedNames)),
-            .pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(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) => 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(
+            Stream.unwrap(
-              response.stream.pipe(
+              Effect.map(Headers.CurrentRedactedNames, (redactedNames) =>
-                Stream.mapError((error) =>
+                prepared.framing.frame(
-                  ProviderShared.eventError(
+                  response.stream.pipe(Stream.mapError((error) => mapHttpError(error, redactedNames))),
-                    `${request.model.provider}/${request.model.route.id}`,
-                    `Failed to read ${request.model.provider}/${request.model.route.id} stream`,
-                    ProviderShared.errorText(error),
-                  ),
                 ),
               ),
             ),

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([
+      const layer = truncatedStream(
-        `data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`,
+        [`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/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(
+          previous:
-            (item): item is Extract<SessionMessageInfo, { type: "model-switched" | "assistant" }> =>
+            event.data.previous ??
-              item.type === "model-switched" || item.type === "assistant",
+            source.findLast(
-          )?.model,
+              (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/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/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/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/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"],