feat(plugin): add experimental WebSocket session hooks

packages/core/src/plugin/provider/azure.ts

@@ -178,29 +178,50 @@ export const AzurePlugin = define({
       Effect.forkScoped({ startImmediately: true }),
     )
 
+    // Entra bearer tokens are minted per request from the target URL's scope, so they are injected at the
+    // transport hooks rather than stored as a credential.
+    const bearer = Effect.fn(function* (url: string) {
+      const connection = yield* ctx.integration.connection.active(Provider.ID.azure)
+      const credential = connection
+        ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined))
+        : undefined
+      if (credential?.type !== "oauth" || credential.methodID !== methodID) return
+      const target = new URL(url)
+      const scope =
+        target.hostname.endsWith(".services.ai.azure.com") && !target.pathname.startsWith("/models")
+          ? foundryScope
+          : cognitiveScope
+      const current = yield* token(scope).pipe(Effect.orDie)
+      return `Bearer ${current.access}`
+    })
     yield* ctx.session.hook(
       "http.request",
       (evt) =>
         Effect.gen(function* () {
           if (evt.model.providerID !== Provider.ID.azure) return
-          const connection = yield* ctx.integration.connection.active(Provider.ID.azure)
-          const credential = connection
-            ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined))
-            : undefined
-          if (credential?.type !== "oauth" || credential.methodID !== methodID) return
-          const url = new URL(evt.request.url)
-          const scope =
-            url.hostname.endsWith(".services.ai.azure.com") && !url.pathname.startsWith("/models")
-              ? foundryScope
-              : cognitiveScope
-          const current = yield* token(scope).pipe(Effect.orDie)
+          const authorization = yield* bearer(evt.request.url)
+          if (!authorization) return
           evt.request.headers.delete("api-key")
           evt.request.headers.delete("x-api-key")
-          evt.request.headers.set("authorization", `Bearer ${current.access}`)
+          evt.request.headers.set("authorization", authorization)
           evt.request.headers.set("user-agent", App.useragent(ctx.app))
         }),
       { providerID: Provider.ID.azure },
     )
+    yield* ctx.session.hook(
+      "experimental.ws.handshake",
+      (evt) =>
+        Effect.gen(function* () {
+          if (evt.model.providerID !== Provider.ID.azure) return
+          const authorization = yield* bearer(evt.url)
+          if (!authorization) return
+          delete evt.headers["api-key"]
+          delete evt.headers["x-api-key"]
+          evt.headers.authorization = authorization
+          evt.headers["user-agent"] = App.useragent(ctx.app)
+        }),
+      { providerID: Provider.ID.azure },
+    )
 
     yield* ctx.aisdk.hook(
       "sdk",

packages/core/src/session/model-request.ts

@@ -316,17 +316,28 @@ export const layer = Layer.effect(
               return HttpClientResponse.fromWeb(sent, after.response)
             }).pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause)))))
         : undefined
-      // HTTP hooks must observe every request, so they keep the provider on HTTP.
       const webSocket =
-        input.webSocket === "session" &&
-        !hasHttpHooks &&
-        model.capabilities.responsesWebsockets === true &&
-        model.websocket
+        input.webSocket === "session" && model.capabilities.responsesWebsockets === true && model.websocket
+      const interceptor: SessionModelTransport.Interceptor = {
+        handshake: (connect) =>
+          hooks.trigger("session", "experimental.ws.handshake", {
+            ...scope,
+            url: connect.url,
+            headers: connect.headers,
+          }),
+        send: (frame, mode) =>
+          hooks.trigger("session", "experimental.ws.send", { ...scope, mode, frame }).pipe(Effect.map((e) => e.frame)),
+        receive: (frame) =>
+          hooks.trigger("session", "experimental.ws.receive", { ...scope, frame }).pipe(Effect.map((e) => e.frame)),
+      }
 
       return {
         event: shaped,
         request,
-        options: { ...(http ? { http } : {}), ...(webSocket ? { webSocket: transport.bind(session.id) } : {}) },
+        options: {
+          ...(http ? { http } : {}),
+          ...(webSocket ? { webSocket: transport.bind(session.id, interceptor) } : {}),
+        },
         retry: (event: Parameters<Prepared["retry"]>[0]) =>
           hooks.trigger("session", "retry", event).pipe(Effect.asVoid),
         // Permission.assert and the question tool throw declines as defects so tools cannot

packages/core/src/session/model-transport.ts

@@ -2,6 +2,7 @@ export * as SessionModelTransport from "./model-transport.js"
 
 import {
   WebSocketTransport,
+  type ChannelCreate,
   type ChannelObservation,
   type ChannelCheckpoint,
   type WebSocketChannelExchange,
@@ -13,6 +14,7 @@ import {
 import { AIError, AIErrorReason, TransportError, type TransportOperation } from "@opencode/ai"
 import { Hash } from "@opencode/util/hash"
 import { Cause, Clock, Context, Effect, Fiber, Layer, Metric, Queue, Scope, Semaphore, Stream } from "effect"
+import { Headers } from "effect/unstable/http"
 import { Socket } from "effect/unstable/socket"
 import { makeGlobalNode } from "@opencode/util/effect/app-node"
 import { SessionSchema } from "./schema.js"
@@ -52,8 +54,18 @@ interface State {
   channel?: Channel
 }
 
+/** Per-exchange plugin hooks. `handshake` output selects the connection; frames are what crosses the wire. */
+export interface Interceptor {
+  readonly handshake: (connect: {
+    readonly url: string
+    readonly headers: Record<string, string>
+  }) => Effect.Effect<{ readonly url: string; readonly headers: Record<string, string> }>
+  readonly send: (frame: string, mode: ChannelCreate["mode"]) => Effect.Effect<string>
+  readonly receive: (frame: string) => Effect.Effect<string>
+}
+
 export interface Interface {
-  readonly bind: (sessionID: SessionSchema.ID) => WebSocketChannelExecutor
+  readonly bind: (sessionID: SessionSchema.ID, interceptor?: Interceptor) => WebSocketChannelExecutor
   readonly close: (sessionID: SessionSchema.ID) => Effect.Effect<void>
   readonly closeAll: Effect.Effect<void>
 }
@@ -267,7 +279,8 @@ export const makeLayer = (connector: WebSocketConnector) =>
 
       const start = Effect.fn("SessionModelTransport.start")(function* (
         owner: State,
-        exchange: WebSocketChannelExchange,
+        input: WebSocketChannelExchange,
+        interceptor?: Interceptor,
       ) {
         if (owner.closed)
           return yield* transportError("Session WebSocket owner is closed", {
@@ -276,7 +289,16 @@ export const makeLayer = (connector: WebSocketConnector) =>
             phase: "queue",
             delivery: "not-sent",
           })
-        if (owner.httpFallback) return fallback(exchange)
+        if (owner.httpFallback) return fallback(input)
+        const handshake = interceptor
+          ? yield* interceptor.handshake({ url: input.connect.url, headers: { ...input.connect.headers } })
+          : undefined
+        const exchange: WebSocketChannelExchange = handshake
+          ? {
+              ...input,
+              connect: { ...input.connect, url: handshake.url, headers: Headers.fromInput(handshake.headers) },
+            }
+          : input
         const key = affinity(exchange)
         const now = yield* Clock.currentTimeMillis
         const current = owner.channel
@@ -337,6 +359,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
           Effect.onInterrupt(() => closeChannel(owner, channel)),
         )
         if (create.mode === "full") channel.checkpoint = undefined
+        const message = interceptor ? yield* interceptor.send(create.message, create.mode) : create.message
         yield* Effect.logDebug("session websocket sending", {
           sessionTransport: "websocket",
           phase: "send",
@@ -347,7 +370,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
           delivery: "send-attempted",
         }
         channel.active = active
-        const sent = yield* channel.connection.sendText(create.message).pipe(
+        const sent = yield* channel.connection.sendText(message).pipe(
           Effect.withSpan("SessionModelTransport.send"),
           Effect.onInterrupt(() => closeChannel(owner, channel)),
           Effect.result,
@@ -388,6 +411,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
                 }),
               ),
           }),
+          Stream.mapEffect((frame) => (interceptor ? interceptor.receive(frame) : Effect.succeed(frame))),
           Stream.mapEffect((frame) => exchange.driver.observe(create, frame)),
           Stream.tap((observation) =>
             Effect.sync(() => {
@@ -465,7 +489,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
         return { frames, complete, http: channel.connection.http }
       })
 
-      const bind = (sessionID: SessionSchema.ID): WebSocketChannelExecutor => ({
+      const bind = (sessionID: SessionSchema.ID, interceptor?: Interceptor): WebSocketChannelExecutor => ({
         execute: (exchange) => {
           const owner = state(sessionID)
           let execution: WebSocketChannelExecution | undefined
@@ -475,7 +499,7 @@ export const makeLayer = (connector: WebSocketConnector) =>
             },
             frames: Stream.unwrap(
               Effect.acquireRelease(owner.lock.take(1), () => owner.lock.release(1), { interruptible: true }).pipe(
-                Effect.andThen(start(owner, exchange)),
+                Effect.andThen(start(owner, exchange, interceptor)),
                 Effect.tap((started) =>
                   Effect.sync(() => {
                     execution = started

packages/core/test/plugin/provider-azure.test.ts

@@ -303,6 +303,20 @@ describe("AzurePlugin", () => {
           })
           expect(foundry.request.headers.get("authorization")).toBe("Bearer https://ai.azure.com/.default-token")
           expect(foundry.request.headers.has("x-api-key")).toBe(false)
+
+          const handshake = yield* hooks.trigger("session", "experimental.ws.handshake", {
+            sessionID: Session.ID.make("ses_azure_ws"),
+            agent: Agent.ID.make("build"),
+            model,
+            kind: "primary",
+            url: "wss://test-resource.openai.azure.com/openai/v1/responses",
+            headers: { "api-key": "stored-token", "x-keep": "yes" },
+          })
+          expect(handshake.headers).toMatchObject({
+            authorization: "Bearer https://cognitiveservices.azure.com/.default-token",
+            "x-keep": "yes",
+          })
+          expect(handshake.headers["api-key"]).toBeUndefined()
         }),
     ),
   )

packages/core/test/session-model-request-hooks.test.ts

@@ -79,4 +79,68 @@ describe("SessionModelRequest HTTP hooks", () => {
       )
     }).pipe(Effect.provideService(SessionModelTransport.Service, transport)),
   )
+
+  it.effect("runs experimental.ws hooks through the transport interceptor alongside http hooks", () =>
+    Effect.gen(function* () {
+      const hooks = yield* PluginHooks.Service
+      const seen: Array<string> = []
+      yield* hooks.register("session", "http.request", () => Effect.void)
+      yield* hooks.register("session", "experimental.ws.handshake", (event) =>
+        Effect.sync(() => {
+          seen.push(`handshake:${event.kind}:${event.agent}`)
+          event.url = `${event.url}?hooked`
+          event.headers.authorization = "Bearer hooked"
+        }),
+      )
+      yield* hooks.register("session", "experimental.ws.send", (event) =>
+        Effect.sync(() => {
+          seen.push(`send:${event.kind}:${event.mode}`)
+          event.frame = `${event.frame}:sent`
+        }),
+      )
+      yield* hooks.register("session", "experimental.ws.receive", (event) =>
+        Effect.sync(() => {
+          seen.push(`receive:${event.kind}`)
+          event.frame = `${event.frame}:received`
+        }),
+      )
+      let interceptor: SessionModelTransport.Interceptor | undefined
+      const capturing = SessionModelTransport.Service.of({
+        bind: (_sessionID, bound) => {
+          interceptor = bound
+          return { execute: () => Effect.die("unused WebSocket execution") }
+        },
+        close: () => Effect.void,
+        closeAll: Effect.void,
+      })
+      const requests = yield* SessionModelRequest.Service.pipe(
+        Effect.provide(SessionModelRequest.layer),
+        Effect.provideService(SessionModelTransport.Service, capturing),
+      )
+      const prepared = yield* requests.compaction({
+        session,
+        agent: Agent.ID.make("build"),
+        model: SessionRunnerModel.resolved(OpenAIChat.route.model({ id: "gpt-5.5", provider: "test" }), {
+          capabilities: { tools: true, input: ["text"], output: ["text"], responsesWebsockets: true },
+          cost: [],
+          limit: { context: 200_000, output: 32_000 },
+          websocket: true,
+        }),
+        system: [],
+        messages: [],
+        webSocket: "session",
+      })
+      expect(prepared.options.http).toBeDefined()
+      expect(prepared.options.webSocket).toBeDefined()
+      if (!interceptor) throw new Error("Expected the transport to receive an interceptor")
+
+      expect(yield* interceptor.handshake({ url: "wss://example.test/v1/responses", headers: {} })).toMatchObject({
+        url: "wss://example.test/v1/responses?hooked",
+        headers: { authorization: "Bearer hooked" },
+      })
+      expect(yield* interceptor.send("frame", "incremental")).toBe("frame:sent")
+      expect(yield* interceptor.receive("frame")).toBe("frame:received")
+      expect(seen).toEqual(["handshake:compaction:build", "send:compaction:incremental", "receive:compaction"])
+    }),
+  )
 })

packages/core/test/session-model-transport.test.ts

@@ -822,6 +822,36 @@ describe("SessionModelTransport", () => {
     )
   })
 
+  test("runs interceptors on the handshake and both frame directions", async () => {
+    const fixture = automatic()
+    const seen: Array<string> = []
+    let authorization = "one"
+
+    await run(
+      fixture.connector,
+      Effect.gen(function* () {
+        const transport = yield* SessionModelTransport.Service
+        const executor = transport.bind(session, {
+          handshake: (connect) =>
+            Effect.succeed({ url: `${connect.url}?hooked`, headers: { ...connect.headers, authorization } }),
+          send: (frame, mode) => Effect.succeed(`${frame}:${mode}`),
+          receive: (frame) =>
+            Effect.sync(() => {
+              seen.push(frame)
+              return frame.toUpperCase()
+            }),
+        })
+        expect(yield* collect(executor, exchange("first"))).toEqual(["COMPLETED:FIRST:FULL"])
+        authorization = "two"
+        expect(yield* collect(executor, exchange("second"))).toEqual(["COMPLETED:SECOND:FULL"])
+        expect(seen).toEqual(["completed:first:full", "completed:second:full"])
+        expect(fixture.connections).toHaveLength(2)
+        expect(fixture.connections.map((item) => item.headers.authorization)).toEqual(["one", "two"])
+        expect(fixture.connections.map((item) => item.sent)).toEqual([["first:full"], ["second:full"]])
+      }),
+    )
+  })
+
   test("rotates when the connection exceeds its requested age limit", async () => {
     const fixture = automatic()
 

packages/plugin/src/effect/session.ts

@@ -86,6 +86,34 @@ export interface SessionHttpResponse {
   response: Response
 }
 
+/** Connection a WebSocket request needs. Changing `url` or `headers` reopens the Session's socket. */
+export interface SessionWebSocketHandshake {
+  readonly sessionID: Session.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly kind: SessionRequestKind
+  url: string
+  headers: Record<string, string>
+}
+
+export interface SessionWebSocketSend {
+  readonly sessionID: Session.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly kind: SessionRequestKind
+  /** Incremental frames carry only what changed since the provider's last checkpoint. */
+  readonly mode: "full" | "incremental"
+  frame: string
+}
+
+export interface SessionWebSocketReceive {
+  readonly sessionID: Session.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly kind: SessionRequestKind
+  frame: string
+}
+
 export type SessionRetryDecision = { retry: false } | { retry: true; delay: number }
 
 export interface SessionRetry {
@@ -106,6 +134,9 @@ export interface SessionHooks {
   readonly "model.request": SessionModelRequest
   readonly "http.request": SessionHttpRequest
   readonly "http.response": SessionHttpResponse
+  readonly "experimental.ws.handshake": SessionWebSocketHandshake
+  readonly "experimental.ws.send": SessionWebSocketSend
+  readonly "experimental.ws.receive": SessionWebSocketReceive
   readonly retry: SessionRetry
 }
 

packages/plugin/src/promise/session.ts

@@ -86,6 +86,34 @@ export interface SessionHttpResponse {
   response: Response
 }
 
+/** Connection a WebSocket request needs. Changing `url` or `headers` reopens the Session's socket. */
+export interface SessionWebSocketHandshake {
+  readonly sessionID: Session.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly kind: SessionRequestKind
+  url: string
+  headers: Record<string, string>
+}
+
+export interface SessionWebSocketSend {
+  readonly sessionID: Session.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly kind: SessionRequestKind
+  /** Incremental frames carry only what changed since the provider's last checkpoint. */
+  readonly mode: "full" | "incremental"
+  frame: string
+}
+
+export interface SessionWebSocketReceive {
+  readonly sessionID: Session.ID
+  readonly agent: Agent.ID
+  readonly model: Model.Ref
+  readonly kind: SessionRequestKind
+  frame: string
+}
+
 export type SessionRetryDecision = { retry: false } | { retry: true; delay: number }
 
 export interface SessionRetry {
@@ -106,6 +134,9 @@ export interface SessionHooks {
   readonly "model.request": SessionModelRequest
   readonly "http.request": SessionHttpRequest
   readonly "http.response": SessionHttpResponse
+  readonly "experimental.ws.handshake": SessionWebSocketHandshake
+  readonly "experimental.ws.send": SessionWebSocketSend
+  readonly "experimental.ws.receive": SessionWebSocketReceive
   readonly retry: SessionRetry
 }
 

services/www/src/docs/content/build/plugins/effect.mdx

@@ -1151,6 +1151,24 @@ effect: (ctx) =>
   }),
 ```
 
+WebSocket providers do not issue one HTTP request per model call, so the HTTP hooks never see that traffic. Three
+experimental hooks cover it: `experimental.ws.handshake` runs once per model call with the URL and headers the connection
+needs (changing either reopens the session's socket), `experimental.ws.send` runs on the outbound frame, and
+`experimental.ws.receive` on every inbound frame. Incremental `send` frames carry only what changed since the provider's
+last checkpoint; rewriting them changes what the provider sees without changing what OpenCode believes it sent.
+
+```ts
+effect: (ctx) =>
+  Effect.gen(function* () {
+    yield* ctx.session.hook("experimental.ws.handshake", (event) =>
+      Effect.sync(() => {
+        event.headers.authorization = `Bearer ${token}`
+      }),
+    )
+    yield* ctx.session.hook("experimental.ws.receive", (event) => Effect.log(event.frame))
+  }),
+```
+
 Override the retry decision for a provider failure or replace its delay in milliseconds. The hook runs after OpenCode
 classifies the failure and proposes its policy, but before any retry is scheduled. It does not expose how OpenCode
 internally performs the next attempt.
@@ -1191,6 +1209,9 @@ interface SessionHooks {
   readonly "model.request": SessionModelRequest
   readonly "http.request": SessionHttpRequest
   readonly "http.response": SessionHttpResponse
+  readonly "experimental.ws.handshake": SessionWebSocketHandshake
+  readonly "experimental.ws.send": SessionWebSocketSend
+  readonly "experimental.ws.receive": SessionWebSocketReceive
   readonly retry: SessionRetry
 }
 

services/www/src/docs/content/build/plugins/index.mdx

@@ -1294,6 +1294,32 @@ await ctx.session.hook("http.response", (event) => {
 })
 ```
 
+#### WebSocket (experimental)
+
+Providers that stream over a WebSocket do not issue one HTTP request per model call, so `http.request` and
+`http.response` never see that traffic. Three experimental hooks cover it instead. `experimental.ws.handshake` runs
+once per model call with the URL and headers the connection needs; changing either reopens the session's socket.
+`experimental.ws.send` runs on the outbound frame, and `experimental.ws.receive` on every inbound frame. All three carry
+the same `sessionID`, `agent`, `model`, and `kind` as the HTTP hooks.
+
+```ts
+await ctx.session.hook("experimental.ws.handshake", (event) => {
+  event.headers.authorization = `Bearer ${token}`
+})
+
+await ctx.session.hook("experimental.ws.send", (event) => {
+  if (event.mode === "full") event.frame = redact(event.frame)
+})
+
+await ctx.session.hook("experimental.ws.receive", (event) => {
+  log(event.frame)
+})
+```
+
+`send` frames in `"incremental"` mode carry only what changed since the provider's last checkpoint. Rewriting them
+changes what the provider sees without changing what OpenCode believes it sent, so treat them as read-only unless you
+also handle the resulting drift.
+
 #### Retry policy
 
 Override the retry decision for a provider failure or replace its delay in milliseconds. The hook runs after OpenCode
@@ -1339,6 +1365,9 @@ interface SessionHooks {
   "model.request": SessionModelRequestHook
   "http.request": SessionHttpRequestHook
   "http.response": SessionHttpResponseHook
+  "experimental.ws.handshake": SessionWebSocketHandshakeHook
+  "experimental.ws.send": SessionWebSocketSendHook
+  "experimental.ws.receive": SessionWebSocketReceiveHook
   retry: SessionRetryHook
 }