feat(plugin): add experimental WebSocket handshake hook (#49131)

This commit is contained in:
Aiden Cline
2026-09-15 01:41:00 -05:00
committed by GitHub
Unverified
parent 27aaa9ce0e
commit d0a902815d
12 changed files with 296 additions and 46 deletions
-6
View File
@@ -1,4 +1,3 @@
import { Headers } from "effect/unstable/http"
import { Auth } from "../route/auth.js"
import { type AtLeastOne, type ProviderAuthOption } from "../route/auth-options.js"
import type { Route, RouteDefaultsInput, CompactionOperations } from "../route/client.js"
@@ -61,11 +60,6 @@ const responsesRoute = OpenAIResponses.route.with({
url.searchParams.delete("api-version")
return url.toString()
},
headers: (headers) => {
const apiKey = headers["api-key"]
if (!apiKey) return headers
return Headers.remove(Headers.set(headers, "authorization", `Bearer ${apiKey}`), "api-key")
},
}),
})
@@ -760,7 +760,11 @@ describe("OpenAI Responses route", () => {
)
const next = continuationDriver({
...request,
input: [...firstInput, reasoning, { type: "message", role: "user", content: [{ type: "input_text", text: "Continue" }] }],
input: [
...firstInput,
reasoning,
{ type: "message", role: "user", content: [{ type: "input_text", text: "Continue" }] },
],
})
const continued = yield* next.create(saved)
@@ -795,7 +799,10 @@ describe("OpenAI Responses route", () => {
ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_1" } }),
),
)
const appended = [...request.input, { type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] }]
const appended = [
...request.input,
{ type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] },
]
const changes = [
{ ...request, model: "gpt-5.3", input: appended },
{ ...request, instructions: "Changed", input: appended },
@@ -804,7 +811,10 @@ describe("OpenAI Responses route", () => {
{ ...request, metadata: { source: "two" }, input: appended },
{
...request,
input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "Rewritten history" }] }, appended[1]],
input: [
{ type: "message", role: "user", content: [{ type: "input_text", text: "Rewritten history" }] },
appended[1],
],
},
]
@@ -900,7 +910,10 @@ describe("OpenAI Responses route", () => {
const second = continuationDriver(
{
...firstRequest,
input: [...firstRequest.input, { type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] }],
input: [
...firstRequest.input,
{ type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] },
],
},
classifyingChannelDriver,
)
@@ -951,7 +964,10 @@ describe("OpenAI Responses route", () => {
}
const secondRequest = {
...firstRequest,
input: [...firstRequest.input, { type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] }],
input: [
...firstRequest.input,
{ type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] },
],
}
const saved = checkpoint(
yield* continuationDriver(firstRequest).observe(
@@ -1082,7 +1098,9 @@ describe("OpenAI Responses route", () => {
)
const expected = {
input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "Say \uFFFDhello \u{1F600}." }] }],
input: [
{ type: "message", role: "user", content: [{ type: "input_text", text: "Say \uFFFDhello \u{1F600}." }] },
],
metadata: { source: "overlay\uFFFD" },
}
expect(JSON.parse(yield* Ref.get(message))).toMatchObject(expected)
@@ -1124,7 +1142,9 @@ describe("OpenAI Responses route", () => {
}),
)
it.effect("builds Azure WebSocket requests with v1 URLs and bearer auth", () =>
// Azure's WebSocket upgrade accepts the same credential header as HTTP: `api-key` for keys,
// `Authorization: Bearer` for Entra tokens. Rewriting a key into a bearer is rejected.
it.effect("builds Azure WebSocket requests with v1 URLs and the route's auth header", () =>
Effect.gen(function* () {
const deps = Layer.succeed(
RequestExecutor.Service,
@@ -1133,13 +1153,13 @@ describe("OpenAI Responses route", () => {
const cases = [
{
model: Azure.configure({ resourceName: "opencode-test", apiKey: "azure-key" }).responses("deployment"),
authorization: "Bearer azure-key",
headers: { "api-key": "azure-key", authorization: undefined },
},
{
model: Azure.configure({ resourceName: "opencode-test", auth: Auth.bearer("entra-token") }).responses(
"deployment",
),
authorization: "Bearer entra-token",
headers: { "api-key": undefined, authorization: "Bearer entra-token" },
},
]
@@ -1150,8 +1170,8 @@ describe("OpenAI Responses route", () => {
Effect.gen(function* () {
expect(exchange.connect.url).toBe("wss://opencode-test.openai.azure.com/openai/v1/responses")
expect(exchange.connect.rotateAfterMs).toBe(55 * 60 * 1000)
expect(exchange.connect.headers.authorization).toBe(item.authorization)
expect(exchange.connect.headers["api-key"]).toBeUndefined()
expect(exchange.connect.headers.authorization).toBe(item.headers.authorization)
expect(exchange.connect.headers["api-key"]).toBe(item.headers["api-key"])
expect(exchange.connect.headers["openai-beta"]).toBeUndefined()
expect(JSON.parse((yield* exchange.driver.create(undefined)).message)).toMatchObject({
type: "response.create",
@@ -1850,7 +1870,11 @@ describe("OpenAI Responses route", () => {
summary: [{ type: "summary_text", text: "I inspected the previous turn." }],
},
{ role: "assistant", content: [{ type: "output_text", text: "It shows a small test image." }] },
{ type: "message", role: "user", content: [{ type: "input_text", text: "Check the weather in Paris before continuing." }] },
{
type: "message",
role: "user",
content: [{ type: "input_text", text: "Check the weather in Paris before continuing." }],
},
{ type: "function_call", call_id: "call_weather_1", name: "get_weather", arguments: '{"city":"Paris"}' },
{ type: "function_call_output", call_id: "call_weather_1", output: '{"temperature":22}' },
{ role: "assistant", content: [{ type: "output_text", text: "Paris is 22 degrees." }] },
@@ -4607,8 +4631,20 @@ describe("OpenAI Responses route", () => {
compileRequest(LLM.request({ model, messages: [response.message], providerOptions: { store } })),
)
expect(prepared.map((request) => request.body.input)).toEqual([
[{ type: "message", role: "user", content: [{ type: "input_image", image_url: "data:image/png;base64,AQID" }] }],
[{ type: "message", role: "user", content: [{ type: "input_image", image_url: "data:image/png;base64,AQID" }] }],
[
{
type: "message",
role: "user",
content: [{ type: "input_image", image_url: "data:image/png;base64,AQID" }],
},
],
[
{
type: "message",
role: "user",
content: [{ type: "input_image", image_url: "data:image/png;base64,AQID" }],
},
],
])
}),
)
+33 -12
View File
@@ -180,29 +180,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("AzurePlugin.bearer")(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 },
)
}),
})
+15 -4
View File
@@ -316,14 +316,25 @@ 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. The route decides
// whether it actually has a WebSocket channel and falls back to HTTP otherwise.
const webSocket = input.webSocket === "session" && !hasHttpHooks && model.transport === "websocket"
// HTTP hooks wrap every HTTP request, including the WebSocket fallback path. The route decides
// which transport actually carries the request, so both hook families are always offered.
const webSocket =
input.webSocket === "session" && model.transport === "websocket"
? transport.bind(session.id, (connect) =>
hooks
.trigger("session", "experimental.ws.handshake", {
...scope,
url: connect.url,
headers: connect.headers,
})
.pipe(Effect.map((event) => ({ url: event.url, headers: event.headers }))),
)
: undefined
return {
event: shaped,
request,
options: { ...(http ? { http } : {}), ...(webSocket ? { webSocket: transport.bind(session.id) } : {}) },
options: { ...(http ? { http } : {}), ...(webSocket ? { webSocket } : {}) },
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
+25 -5
View File
@@ -13,6 +13,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 +53,17 @@ interface State {
channel?: Channel
}
/** Selects the connection for one exchange. Its output feeds the affinity key, so changed headers reopen the socket. */
export interface Handshake {
readonly url: string
readonly headers: Record<string, string>
}
export interface Interface {
readonly bind: (sessionID: SessionSchema.ID) => WebSocketChannelExecutor
readonly bind: (
sessionID: SessionSchema.ID,
handshake?: (connect: Handshake) => Effect.Effect<Handshake>,
) => WebSocketChannelExecutor
readonly close: (sessionID: SessionSchema.ID) => Effect.Effect<void>
readonly closeAll: Effect.Effect<void>
}
@@ -267,7 +277,8 @@ export const makeLayer = (connector: WebSocketConnector) =>
const start = Effect.fn("SessionModelTransport.start")(function* (
owner: State,
exchange: WebSocketChannelExchange,
input: WebSocketChannelExchange,
handshake?: (connect: Handshake) => Effect.Effect<Handshake>,
) {
if (owner.closed)
return yield* transportError("Session WebSocket owner is closed", {
@@ -276,7 +287,13 @@ export const makeLayer = (connector: WebSocketConnector) =>
phase: "queue",
delivery: "not-sent",
})
if (owner.httpFallback) return fallback(exchange)
if (owner.httpFallback) return fallback(input)
const selected = handshake
? yield* handshake({ url: input.connect.url, headers: { ...input.connect.headers } })
: undefined
const exchange: WebSocketChannelExchange = selected
? { ...input, connect: { ...input.connect, url: selected.url, headers: Headers.fromInput(selected.headers) } }
: input
const key = affinity(exchange)
const now = yield* Clock.currentTimeMillis
const current = owner.channel
@@ -465,7 +482,10 @@ export const makeLayer = (connector: WebSocketConnector) =>
return { frames, complete, http: channel.connection.http }
})
const bind = (sessionID: SessionSchema.ID): WebSocketChannelExecutor => ({
const bind = (
sessionID: SessionSchema.ID,
handshake?: (connect: Handshake) => Effect.Effect<Handshake>,
): WebSocketChannelExecutor => ({
execute: (exchange) => {
const owner = state(sessionID)
let execution: WebSocketChannelExecution | undefined
@@ -475,7 +495,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, handshake)),
Effect.tap((started) =>
Effect.sync(() => {
execution = started
@@ -301,6 +301,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).not.toHaveProperty("api-key")
}),
),
)
@@ -362,9 +376,7 @@ describe("AzurePlugin", () => {
resourceName: "from-env",
baseURL: "https://from-env.cognitiveservices.azure.com/openai",
})
expect(
required(yield* models.get(Provider.ID.azure, Model.ID.make("anthropic"))).settings,
).toMatchObject({
expect(required(yield* models.get(Provider.ID.azure, Model.ID.make("anthropic"))).settings).toMatchObject({
resourceName: "model-resource",
baseURL: "https://model-resource.services.ai.azure.com/anthropic/v1",
})
@@ -469,5 +481,4 @@ describe("AzurePlugin", () => {
}),
),
)
})
@@ -11,7 +11,7 @@ import { AbsolutePath } from "@opencode/core/schema"
import { SessionModelRequest } from "@opencode/core/session/model-request"
import { SessionModelTransport } from "@opencode/core/session/model-transport"
import { SessionRunnerModel } from "@opencode/core/session/runner/model"
import { DateTime, Effect } from "effect"
import { DateTime, Effect, Stream } from "effect"
import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { testEffect } from "./lib/effect"
import { PluginTestLayer } from "./plugin/fixture"
@@ -79,4 +79,55 @@ describe("SessionModelRequest HTTP hooks", () => {
)
}).pipe(Effect.provideService(SessionModelTransport.Service, transport)),
)
it.effect("offers the WebSocket executor alongside HTTP hooks and routes the handshake hook", () =>
Effect.gen(function* () {
const hooks = yield* PluginHooks.Service
const seen: string[] = []
yield* hooks.register("session", "http.request", () => Effect.sync(() => void seen.push("http.request")))
yield* hooks.register("session", "experimental.ws.handshake", (event) =>
Effect.sync(() => {
seen.push(`handshake:${event.kind}:${event.url}`)
event.headers.authorization = "Bearer minted"
delete event.headers["api-key"]
}),
)
const bound: Array<{ url: string; headers: Record<string, string> }> = []
const websocketTransport = SessionModelTransport.Service.of({
bind: (_sessionID, handshake) => ({
execute: () =>
Effect.gen(function* () {
if (!handshake) throw new Error("Expected a handshake interceptor")
bound.push(yield* handshake({ url: "wss://example.test/v1/responses", headers: { "api-key": "k" } }))
return { frames: Stream.empty, complete: Effect.void }
}),
}),
close: () => Effect.void,
closeAll: Effect.void,
})
const requests = yield* SessionModelRequest.Service.pipe(
Effect.provide(SessionModelRequest.layer),
Effect.provideService(SessionModelTransport.Service, websocketTransport),
)
const prepared = yield* requests.primary({
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"] },
cost: [],
limit: { context: 200_000, output: 32_000 },
transport: "websocket",
}),
system: [],
messages: [],
webSocket: "session",
})
expect(prepared.options.http).toBeDefined()
expect(prepared.options.webSocket).toBeDefined()
yield* prepared.options.webSocket!.execute({} as never)
expect(bound).toEqual([{ url: "wss://example.test/v1/responses", headers: { authorization: "Bearer minted" } }])
expect(seen).toEqual(["handshake:primary:wss://example.test/v1/responses"])
}),
)
})
@@ -171,6 +171,31 @@ describe("SessionModelTransport", () => {
)
})
test("selects the connection from the handshake and reopens when it changes", async () => {
const fixture = automatic()
const tokens = ["one", "one", "two"]
await run(
fixture.connector,
Effect.gen(function* () {
const transport = yield* SessionModelTransport.Service
const executor = transport.bind(session, (connect) =>
Effect.succeed({
url: connect.url,
headers: { ...connect.headers, authorization: `Bearer ${tokens.shift()}` },
}),
)
yield* collect(executor, exchange("first", { headers: { "api-key": "k" } }))
yield* collect(executor, exchange("second", { headers: { "api-key": "k" } }))
yield* collect(executor, exchange("third", { headers: { "api-key": "k" } }))
// Same minted token reuses the socket; a rotated token changes the affinity key and reopens it.
expect(fixture.connections).toHaveLength(2)
expect(fixture.connections.map((item) => item.headers.authorization)).toEqual(["Bearer one", "Bearer two"])
expect(fixture.connections.map((item) => item.sent)).toEqual([["first", "second"], ["third"]])
}),
)
})
test("does not carry a checkpoint across physical connection rotation", async () => {
const fixture = automatic()
const checkpoints: Array<unknown> = []
+14
View File
@@ -86,6 +86,19 @@ export interface SessionHttpResponse {
response: Response
}
/**
* Connection a WebSocket-backed request opens or reuses. Runs once per model call before the
* Session's socket is selected; changing `url` or `headers` reopens the socket. Experimental.
*/
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 type SessionRetryDecision = { retry: false } | { retry: true; delay: number }
export interface SessionRetry {
@@ -106,6 +119,7 @@ export interface SessionHooks {
readonly "model.request": SessionModelRequest
readonly "http.request": SessionHttpRequest
readonly "http.response": SessionHttpResponse
readonly "experimental.ws.handshake": SessionWebSocketHandshake
readonly retry: SessionRetry
}
+14
View File
@@ -86,6 +86,19 @@ export interface SessionHttpResponse {
response: Response
}
/**
* Connection a WebSocket-backed request opens or reuses. Runs once per model call before the
* Session's socket is selected; changing `url` or `headers` reopens the socket. Experimental.
*/
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 type SessionRetryDecision = { retry: false } | { retry: true; delay: number }
export interface SessionRetry {
@@ -106,6 +119,7 @@ export interface SessionHooks {
readonly "model.request": SessionModelRequest
readonly "http.request": SessionHttpRequest
readonly "http.response": SessionHttpResponse
readonly "experimental.ws.handshake": SessionWebSocketHandshake
readonly retry: SessionRetry
}
@@ -1257,6 +1257,26 @@ effect: (ctx) =>
}),
```
Providers that stream over a WebSocket reuse one connection per session, so the HTTP hooks never see that traffic. The
experimental `experimental.ws.handshake` hook runs once per model call before the connection is selected and exposes the
connection `url` and `headers`; changing either reopens the socket. HTTP hooks still run for any request a WebSocket
route falls back to.
```ts
effect: (ctx) =>
Effect.gen(function* () {
yield* ctx.session.hook(
"experimental.ws.handshake",
(event) =>
Effect.gen(function* () {
event.headers.authorization = `Bearer ${yield* mintToken(event.url)}`
delete event.headers["api-key"]
}),
{ providerID: "azure" },
)
}),
```
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.
@@ -1389,6 +1389,29 @@ await ctx.session.hook("http.response", (event) => {
})
```
#### Native WebSocket (experimental)
Providers that stream over a WebSocket open one connection per session and reuse it across requests, so
`http.request` and `http.response` never see that traffic. The `experimental.ws.handshake` hook runs once per model
call before the connection is selected. It carries the same `sessionID`, `agent`, `model`, and `kind` as the HTTP
hooks, plus the connection `url` and `headers`. Changing `url` or `headers` changes which socket the request uses, so a
rotated bearer token reopens the connection.
HTTP hooks still run for every HTTP request, including the request a WebSocket route falls back to.
```ts
await ctx.session.hook(
"experimental.ws.handshake",
async (event) => {
event.headers.authorization = `Bearer ${await mintToken(event.url)}`
delete event.headers["api-key"]
},
{ providerID: "azure" },
)
```
This hook is experimental and its name or shape may change.
#### Retry policy
Override the retry decision for a provider failure or replace its delay in milliseconds. The hook runs after OpenCode
@@ -1434,9 +1457,19 @@ interface SessionHooks {
"model.request": SessionModelRequestHook
"http.request": SessionHttpRequestHook
"http.response": SessionHttpResponseHook
"experimental.ws.handshake": SessionWebSocketHandshakeHook
retry: SessionRetryHook
}
interface SessionWebSocketHandshakeHook {
readonly sessionID: string
readonly agent: string
readonly model: { providerID: string; id: string; variant?: string }
readonly kind: "primary" | "compaction" | "title" | "generate"
url: string
headers: Record<string, string>
}
type RetryDecision = { retry: false } | { retry: true; delay: number }
interface SessionRetryHook {