diff options
Diffstat (limited to 'packages/transport-ws')
| -rw-r--r-- | packages/transport-ws/package.json | 26 | ||||
| -rw-r--r-- | packages/transport-ws/src/extension.ts | 716 | ||||
| -rw-r--r-- | packages/transport-ws/src/index.ts | 14 | ||||
| -rw-r--r-- | packages/transport-ws/src/manifest.ts | 18 | ||||
| -rw-r--r-- | packages/transport-ws/src/router.test.ts | 1351 | ||||
| -rw-r--r-- | packages/transport-ws/src/router.ts | 467 | ||||
| -rw-r--r-- | packages/transport-ws/src/server.bun.test.ts | 2268 | ||||
| -rw-r--r-- | packages/transport-ws/tsconfig.json | 20 |
8 files changed, 2519 insertions, 2361 deletions
diff --git a/packages/transport-ws/package.json b/packages/transport-ws/package.json index b600a98..6593218 100644 --- a/packages/transport-ws/package.json +++ b/packages/transport-ws/package.json @@ -1,15 +1,15 @@ { - "name": "@dispatch/transport-ws", - "version": "0.0.0", - "type": "module", - "private": true, - "main": "dist/index.js", - "types": "dist/index.d.ts", - "dependencies": { - "@dispatch/kernel": "workspace:*", - "@dispatch/session-orchestrator": "workspace:*", - "@dispatch/surface-registry": "workspace:*", - "@dispatch/transport-contract": "workspace:*", - "@dispatch/ui-contract": "workspace:*" - } + "name": "@dispatch/transport-ws", + "version": "0.0.0", + "type": "module", + "private": true, + "main": "dist/index.js", + "types": "dist/index.d.ts", + "dependencies": { + "@dispatch/kernel": "workspace:*", + "@dispatch/session-orchestrator": "workspace:*", + "@dispatch/surface-registry": "workspace:*", + "@dispatch/transport-contract": "workspace:*", + "@dispatch/ui-contract": "workspace:*" + } } diff --git a/packages/transport-ws/src/extension.ts b/packages/transport-ws/src/extension.ts index 56bd8e2..88f721e 100644 --- a/packages/transport-ws/src/extension.ts +++ b/packages/transport-ws/src/extension.ts @@ -9,10 +9,10 @@ import type { Extension, HostAPI } from "@dispatch/kernel"; import type { SessionOrchestrator } from "@dispatch/session-orchestrator"; import { - conversationCompacted, - conversationOpened, - conversationStatusChanged, - sessionOrchestratorHandle, + conversationCompacted, + conversationOpened, + conversationStatusChanged, + sessionOrchestratorHandle, } from "@dispatch/session-orchestrator"; import type { SurfaceContext, SurfaceProvider, SurfaceRegistry } from "@dispatch/surface-registry"; import { surfaceRegistryHandle } from "@dispatch/surface-registry"; @@ -22,379 +22,397 @@ import { catalogMessage, routeClientMessage, subKey } from "./router.js"; /** Active provider subscriptions + chat subscription disposers for a single WS connection. */ interface ConnectionState { - readonly subs: Set<string>; - readonly providerDisposers: Map<string, () => void>; - /** Per-conversation chat subscription disposers (orchestrator.subscribe). */ - readonly chatSubscriptions: Map<string, () => void>; + readonly subs: Set<string>; + readonly providerDisposers: Map<string, () => void>; + /** Per-conversation chat subscription disposers (orchestrator.subscribe). */ + readonly chatSubscriptions: Map<string, () => void>; } type Ws = Bun.ServerWebSocket<ConnectionState>; export function createTransportWsExtension(): Extension { - let server: ReturnType<typeof Bun.serve<ConnectionState>> | undefined; - /** Every currently-connected WS client — used for global fan-out broadcasts. */ - const connections = new Set<Ws>(); - /** Disposers for host hook subscriptions (drained on deactivate). */ - const disposers: Array<() => void> = []; + let server: ReturnType<typeof Bun.serve<ConnectionState>> | undefined; + /** Every currently-connected WS client — used for global fan-out broadcasts. */ + const connections = new Set<Ws>(); + /** Disposers for host hook subscriptions (drained on deactivate). */ + const disposers: Array<() => void> = []; - return { - manifest, - async activate(host: HostAPI) { - const registry: SurfaceRegistry = host.getService(surfaceRegistryHandle); - const orchestrator: SessionOrchestrator = host.getService(sessionOrchestratorHandle); - const logger = host.logger; - const port = host.config.get<number>("surfaceWsPort") ?? 24205; + return { + manifest, + async activate(host: HostAPI) { + const registry: SurfaceRegistry = host.getService(surfaceRegistryHandle); + const orchestrator: SessionOrchestrator = host.getService(sessionOrchestratorHandle); + const logger = host.logger; + const port = host.config.get<number>("surfaceWsPort") ?? 24205; - function send(ws: Ws, msg: WsServerMessage): void { - try { - ws.send(JSON.stringify(msg)); - } catch { - // Connection may have been dropped; swallow. - } - } + function send(ws: Ws, msg: WsServerMessage): void { + try { + ws.send(JSON.stringify(msg)); + } catch { + // Connection may have been dropped; swallow. + } + } - /** Broadcast a message to EVERY connected WS client (global fan-out). */ - function broadcast(msg: WsServerMessage): void { - for (const ws of connections) { - send(ws, msg); - } - } + /** Broadcast a message to EVERY connected WS client (global fan-out). */ + function broadcast(msg: WsServerMessage): void { + for (const ws of connections) { + send(ws, msg); + } + } - /** - * Ensure this connection is subscribed to a conversation's chat events. - * Idempotent — no-op if already subscribed. The orchestrator replays - * buffered events to new subscribers (late-join), then streams live. - */ - function ensureChatSubscribed(ws: Ws, state: ConnectionState, conversationId: string): void { - if (state.chatSubscriptions.has(conversationId)) { - return; - } - const unsubscribe = orchestrator.subscribe(conversationId, (event) => { - send(ws, { type: "chat.delta", event }); - }); - state.chatSubscriptions.set(conversationId, unsubscribe); - } + /** + * Ensure this connection is subscribed to a conversation's chat events. + * Idempotent — no-op if already subscribed. The orchestrator replays + * buffered events to new subscribers (late-join), then streams live. + */ + function ensureChatSubscribed(ws: Ws, state: ConnectionState, conversationId: string): void { + if (state.chatSubscriptions.has(conversationId)) { + return; + } + const unsubscribe = orchestrator.subscribe(conversationId, (event) => { + send(ws, { type: "chat.delta", event }); + }); + state.chatSubscriptions.set(conversationId, unsubscribe); + } - function subscribeToProvider( - ws: Ws, - provider: SurfaceProvider, - surfaceId: string, - conversationId: string | undefined, - state: ConnectionState, - ): void { - const key = subKey(surfaceId, conversationId); - if (!provider.subscribe || state.providerDisposers.has(key)) { - return; - } - const context: SurfaceContext | undefined = - conversationId !== undefined ? { conversationId } : undefined; - const dispose = provider.subscribe(() => { - try { - const spec = provider.getSpec(context); - if (spec instanceof Promise) { - spec - .then((s) => - send(ws, { - type: "update", - update: { - surfaceId, - spec: s, - ...(conversationId !== undefined ? { conversationId } : {}), - }, - }), - ) - .catch(() => {}); - } else { - send(ws, { - type: "update", - update: { - surfaceId, - spec, - ...(conversationId !== undefined ? { conversationId } : {}), - }, - }); - } - } catch { - // Provider threw — log but don't kill the connection. - } - }); - state.providerDisposers.set(key, dispose); - } + function subscribeToProvider( + ws: Ws, + provider: SurfaceProvider, + surfaceId: string, + conversationId: string | undefined, + state: ConnectionState, + ): void { + const key = subKey(surfaceId, conversationId); + if (!provider.subscribe || state.providerDisposers.has(key)) { + return; + } + const context: SurfaceContext | undefined = + conversationId !== undefined ? { conversationId } : undefined; + const dispose = provider.subscribe(() => { + try { + const spec = provider.getSpec(context); + if (spec instanceof Promise) { + spec + .then((s) => + send(ws, { + type: "update", + update: { + surfaceId, + spec: s, + ...(conversationId !== undefined ? { conversationId } : {}), + }, + }), + ) + .catch(() => {}); + } else { + send(ws, { + type: "update", + update: { + surfaceId, + spec, + ...(conversationId !== undefined ? { conversationId } : {}), + }, + }); + } + } catch { + // Provider threw — log but don't kill the connection. + } + }); + state.providerDisposers.set(key, dispose); + } - function unsubscribeFromProvider(state: ConnectionState, key: string): void { - const dispose = state.providerDisposers.get(key); - if (dispose) { - dispose(); - state.providerDisposers.delete(key); - } - } + function unsubscribeFromProvider(state: ConnectionState, key: string): void { + const dispose = state.providerDisposers.get(key); + if (dispose) { + dispose(); + state.providerDisposers.delete(key); + } + } - // Broadcast a `conversation.open` WS message to ALL connected clients - // whenever the orchestrator signals a conversation was opened (e.g. the - // CLI `--open` flag). The frontend decides whether to open/focus a tab — - // the backend just signals. This is a GLOBAL fan-out (like the catalog), - // NOT a per-conversation chat broadcast. The payload's `workspaceId` - // is the conversation's actual persisted workspace (resolved by the - // orchestrator from the store), so a frontend opens/focuses the tab in - // the correct workspace. - disposers.push( - host.on(conversationOpened, ({ conversationId, workspaceId }) => { - broadcast({ type: "conversation.open", conversationId, workspaceId }); - }), - ); + // Broadcast a `conversation.open` WS message to ALL connected clients + // whenever the orchestrator signals a conversation was opened (e.g. the + // CLI `--open` flag). The frontend decides whether to open/focus a tab — + // the backend just signals. This is a GLOBAL fan-out (like the catalog), + // NOT a per-conversation chat broadcast. The payload's `workspaceId` + // is the conversation's actual persisted workspace (resolved by the + // orchestrator from the store), so a frontend opens/focuses the tab in + // the correct workspace. + disposers.push( + host.on(conversationOpened, ({ conversationId, workspaceId }) => { + broadcast({ type: "conversation.open", conversationId, workspaceId }); + }), + ); - // Broadcast `conversation.statusChanged` to all connected clients so - // tabs sync across devices in real time. `workspaceId` is the - // conversation's actual persisted workspace (resolved by the - // orchestrator from the store), forwarded so a frontend syncs the tab - // in the correct workspace. - disposers.push( - host.on(conversationStatusChanged, ({ conversationId, status, workspaceId }) => { - broadcast({ - type: "conversation.statusChanged", - conversationId, - status, - workspaceId, - }); - }), - ); + // Broadcast `conversation.statusChanged` to all connected clients so + // tabs sync across devices in real time. `workspaceId` is the + // conversation's actual persisted workspace (resolved by the + // orchestrator from the store), forwarded so a frontend syncs the tab + // in the correct workspace. + disposers.push( + host.on(conversationStatusChanged, ({ conversationId, status, workspaceId }) => { + broadcast({ + type: "conversation.statusChanged", + conversationId, + status, + workspaceId, + }); + }), + ); - // Broadcast `conversation.compacted` to all connected clients so - // the FE reloads the conversation history after compaction. - disposers.push( - host.on( - conversationCompacted, - ({ conversationId, newConversationId, messagesSummarized, messagesKept }) => { - broadcast({ - type: "conversation.compacted", - conversationId, - newConversationId, - messagesSummarized, - messagesKept, - }); - }, - ), - ); + // Broadcast `conversation.compacted` to all connected clients so + // the FE reloads the conversation history after compaction. + disposers.push( + host.on( + conversationCompacted, + ({ conversationId, newConversationId, messagesSummarized, messagesKept }) => { + broadcast({ + type: "conversation.compacted", + conversationId, + newConversationId, + messagesSummarized, + messagesKept, + }); + }, + ), + ); - server = Bun.serve<ConnectionState>({ - port, - fetch(req, srv) { - const initial: ConnectionState = { - subs: new Set(), - providerDisposers: new Map(), - chatSubscriptions: new Map(), - }; - if (srv.upgrade(req, { data: initial })) return; - return new Response("expected websocket", { status: 426 }); - }, - websocket: { - open(ws) { - connections.add(ws); - logger.debug("transport-ws: connection open"); - send(ws, catalogMessage(registry)); - }, + server = Bun.serve<ConnectionState>({ + port, + fetch(req, srv) { + const initial: ConnectionState = { + subs: new Set(), + providerDisposers: new Map(), + chatSubscriptions: new Map(), + }; + if (srv.upgrade(req, { data: initial })) return; + return new Response("expected websocket", { status: 426 }); + }, + websocket: { + open(ws) { + connections.add(ws); + logger.debug("transport-ws: connection open"); + send(ws, catalogMessage(registry)); + }, - message(ws, message) { - const state = ws.data; - if (!state) return; + message(ws, message) { + const state = ws.data; + if (!state) return; - let parsed: WsClientMessage; - try { - parsed = JSON.parse(String(message)) as WsClientMessage; - } catch { - send(ws, { type: "error", message: "Invalid JSON" }); - return; - } + let parsed: WsClientMessage; + try { + parsed = JSON.parse(String(message)) as WsClientMessage; + } catch { + send(ws, { type: "error", message: "Invalid JSON" }); + return; + } - const result = routeClientMessage(registry, state.subs, parsed); + const result = routeClientMessage(registry, state.subs, parsed); - switch (result.kind) { - case "surface": { - // Log surface-op errors (unknown surface or invoke failure). - for (const reply of result.replies) { - if (reply.type === "error") { - logger.warn?.("transport-ws: surface-op error", { - ...(reply.surfaceId !== undefined ? { surfaceId: reply.surfaceId } : {}), - reason: reply.message, - }); - } - } + switch (result.kind) { + case "surface": { + // Log surface-op errors (unknown surface or invoke failure). + for (const reply of result.replies) { + if (reply.type === "error") { + logger.warn?.("transport-ws: surface-op error", { + ...(reply.surfaceId !== undefined ? { surfaceId: reply.surfaceId } : {}), + reason: reply.message, + }); + } + } - // Apply sub change. - if (result.subChange) { - const key = subKey(result.subChange.surfaceId, result.subChange.conversationId); - if (result.subChange.op === "add") { - state.subs.add(key); - const provider = registry.getSurface(result.subChange.surfaceId); - if (provider) { - subscribeToProvider( - ws, - provider, - result.subChange.surfaceId, - result.subChange.conversationId, - state, - ); - } - } else { - state.subs.delete(key); - unsubscribeFromProvider(state, key); - } - } + // Apply sub change. + if (result.subChange) { + const key = subKey(result.subChange.surfaceId, result.subChange.conversationId); + if (result.subChange.op === "add") { + state.subs.add(key); + const provider = registry.getSurface(result.subChange.surfaceId); + if (provider) { + subscribeToProvider( + ws, + provider, + result.subChange.surfaceId, + result.subChange.conversationId, + state, + ); + } + } else { + state.subs.delete(key); + unsubscribeFromProvider(state, key); + } + } - // Send replies. - for (const reply of result.replies) { - send(ws, reply); - } + // Send replies. + for (const reply of result.replies) { + send(ws, reply); + } - // Perform invoke if signalled. - if (result.invoke) { - const provider = registry.getSurface(result.invoke.surfaceId); - if (provider) { - const context: SurfaceContext | undefined = - result.invoke.conversationId !== undefined - ? { conversationId: result.invoke.conversationId } - : undefined; - try { - const r = provider.invoke( - result.invoke.actionId, - result.invoke.payload, - context, - ); - if (r instanceof Promise) { - r.catch(() => {}); - } - } catch (err: unknown) { - const reason = err instanceof Error ? err.message : "invoke failed"; - logger.warn?.("transport-ws: surface-op error", { - surfaceId: result.invoke.surfaceId, - actionId: result.invoke.actionId, - reason, - }); - } - } - } - break; - } + // Perform invoke if signalled. + if (result.invoke) { + const provider = registry.getSurface(result.invoke.surfaceId); + if (provider) { + const context: SurfaceContext | undefined = + result.invoke.conversationId !== undefined + ? { conversationId: result.invoke.conversationId } + : undefined; + try { + const r = provider.invoke( + result.invoke.actionId, + result.invoke.payload, + context, + ); + if (r instanceof Promise) { + r.catch(() => {}); + } + } catch (err: unknown) { + const reason = err instanceof Error ? err.message : "invoke failed"; + logger.warn?.("transport-ws: surface-op error", { + surfaceId: result.invoke.surfaceId, + actionId: result.invoke.actionId, + reason, + }); + } + } + } + break; + } - case "chat": { - const resolvedId = result.conversationId ?? crypto.randomUUID(); - // Auto-subscribe the sender so it sees the turn's events. - ensureChatSubscribed(ws, state, resolvedId); - // Start the turn detached from this connection. - const startResult = orchestrator.startTurn({ - conversationId: resolvedId, - text: result.message, - ...(result.model !== undefined ? { modelName: result.model } : {}), - ...(result.cwd !== undefined ? { cwd: result.cwd } : {}), - ...(result.reasoningEffort !== undefined - ? { reasoningEffort: result.reasoningEffort } - : {}), - ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), - ...(result.computerId !== undefined ? { computerId: result.computerId } : {}), - }); - if (!startResult.started) { - send(ws, { - type: "chat.error", - conversationId: resolvedId, - message: "a turn is already generating for this conversation", - }); - } else { - logger.info?.("transport-ws: chat.send accepted", { - conversationId: resolvedId, - model: result.model ?? null, - }); - } - break; - } + case "chat": { + const resolvedId = result.conversationId ?? crypto.randomUUID(); + // Auto-subscribe the sender so it sees the turn's events. + ensureChatSubscribed(ws, state, resolvedId); + // Start the turn detached from this connection. + const startResult = orchestrator.startTurn({ + conversationId: resolvedId, + text: result.message, + ...(result.model !== undefined ? { modelName: result.model } : {}), + ...(result.cwd !== undefined ? { cwd: result.cwd } : {}), + ...(result.reasoningEffort !== undefined + ? { reasoningEffort: result.reasoningEffort } + : {}), + ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), + ...(result.computerId !== undefined ? { computerId: result.computerId } : {}), + ...(result.images !== undefined ? { images: result.images } : {}), + }); + if (!startResult.started) { + send(ws, { + type: "chat.error", + conversationId: resolvedId, + message: "a turn is already generating for this conversation", + }); + } else { + logger.info?.("transport-ws: chat.send accepted", { + conversationId: resolvedId, + model: result.model ?? null, + }); + } + break; + } - case "chat-subscribe": { - ensureChatSubscribed(ws, state, result.conversationId); - break; - } + case "chat-subscribe": { + ensureChatSubscribed(ws, state, result.conversationId); + break; + } - case "chat-unsubscribe": { - const dispose = state.chatSubscriptions.get(result.conversationId); - if (dispose) { - dispose(); - state.chatSubscriptions.delete(result.conversationId); - } - break; - } + case "chat-unsubscribe": { + const dispose = state.chatSubscriptions.get(result.conversationId); + if (dispose) { + dispose(); + state.chatSubscriptions.delete(result.conversationId); + } + break; + } - case "chat-queue": { - // Fire-and-forget: success is confirmed by the message-queue - // SURFACE updating (startedTurn:false) or by streaming - // chat.deltas (startedTurn:true), NOT by a reply here. On - // startedTurn:true the sender is auto-subscribed so the new - // turn's events stream to it (same as chat.send); on - // startedTurn:false (queued for steering) we emit NOTHING - // back and do not auto-subscribe. - const enqueueResult = orchestrator.enqueue({ - conversationId: result.conversationId, - text: result.text, - ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), - }); - if (enqueueResult.startedTurn) { - ensureChatSubscribed(ws, state, result.conversationId); - } - logger.info?.("transport-ws: chat.queue accepted", { - conversationId: result.conversationId, - startedTurn: enqueueResult.startedTurn, - }); - break; - } + case "chat-queue": { + // Fire-and-forget: success is confirmed by the message-queue + // SURFACE updating (startedTurn:false) or by streaming + // chat.deltas (startedTurn:true), NOT by a reply here. On + // startedTurn:true the sender is auto-subscribed so the new + // turn's events stream to it (same as chat.send); on + // startedTurn:false (queued for steering) we emit NOTHING + // back and do not auto-subscribe. + const enqueueResult = orchestrator.enqueue({ + conversationId: result.conversationId, + text: result.text, + ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), + }); + if (enqueueResult.startedTurn) { + ensureChatSubscribed(ws, state, result.conversationId); + } + logger.info?.("transport-ws: chat.queue accepted", { + conversationId: result.conversationId, + startedTurn: enqueueResult.startedTurn, + }); + break; + } - case "chat-error": { - logger.warn?.("transport-ws: malformed chat.send", { - reason: result.errorMessage, - ...(result.conversationId !== undefined - ? { conversationId: result.conversationId } - : {}), - }); - send(ws, { - type: "chat.error", - ...(result.conversationId !== undefined - ? { conversationId: result.conversationId } - : {}), - message: result.errorMessage, - }); - break; - } - } - }, + case "chat-queue-cancel": { + // Fire-and-forget: success is confirmed by the message-queue + // SURFACE updating (the cancelled message leaves the snapshot), + // NOT by a reply here. Cancelling a message that is no longer + // queued is a silent no-op (no surface update, no error). + const cancelResult = orchestrator.cancelQueuedMessage({ + conversationId: result.conversationId, + messageId: result.messageId, + }); + logger.info?.("transport-ws: chat.queue.cancel accepted", { + conversationId: result.conversationId, + messageId: result.messageId, + cancelled: cancelResult.cancelled, + }); + break; + } - close(ws) { - connections.delete(ws); - const state = ws.data; - if (state) { - // Dispose all chat subscriptions (does NOT abort turns). - for (const dispose of state.chatSubscriptions.values()) { - dispose(); - } - state.chatSubscriptions.clear(); - // Dispose surface provider subscriptions. - for (const dispose of state.providerDisposers.values()) { - dispose(); - } - } - logger.debug("transport-ws: connection close"); - }, - }, - }); + case "chat-error": { + logger.warn?.("transport-ws: malformed chat.send", { + reason: result.errorMessage, + ...(result.conversationId !== undefined + ? { conversationId: result.conversationId } + : {}), + }); + send(ws, { + type: "chat.error", + ...(result.conversationId !== undefined + ? { conversationId: result.conversationId } + : {}), + message: result.errorMessage, + }); + break; + } + } + }, - logger.info?.("transport-ws: surface WebSocket listening", { port }); - }, + close(ws) { + connections.delete(ws); + const state = ws.data; + if (state) { + // Dispose all chat subscriptions (does NOT abort turns). + for (const dispose of state.chatSubscriptions.values()) { + dispose(); + } + state.chatSubscriptions.clear(); + // Dispose surface provider subscriptions. + for (const dispose of state.providerDisposers.values()) { + dispose(); + } + } + logger.debug("transport-ws: connection close"); + }, + }, + }); - deactivate() { - for (const dispose of disposers) { - dispose(); - } - disposers.length = 0; - connections.clear(); - if (server) { - server.stop(); - server = undefined; - } - }, - }; + logger.info?.("transport-ws: surface WebSocket listening", { port }); + }, + + deactivate() { + for (const dispose of disposers) { + dispose(); + } + disposers.length = 0; + connections.clear(); + if (server) { + server.stop(); + server = undefined; + } + }, + }; } diff --git a/packages/transport-ws/src/index.ts b/packages/transport-ws/src/index.ts index e0cc66b..5a7c3f9 100644 --- a/packages/transport-ws/src/index.ts +++ b/packages/transport-ws/src/index.ts @@ -1,12 +1,12 @@ export { createTransportWsExtension } from "./extension.js"; export { manifest } from "./manifest.js"; export type { - ChatQueueRouteResult, - ChatRouteError, - ChatRouteResult, - ChatSubscribeRouteResult, - ChatUnsubscribeRouteResult, - RouteResult, - SurfaceRouteResult, + ChatQueueRouteResult, + ChatRouteError, + ChatRouteResult, + ChatSubscribeRouteResult, + ChatUnsubscribeRouteResult, + RouteResult, + SurfaceRouteResult, } from "./router.js"; export { catalogMessage, routeClientMessage, subKey } from "./router.js"; diff --git a/packages/transport-ws/src/manifest.ts b/packages/transport-ws/src/manifest.ts index 5058311..eef6f45 100644 --- a/packages/transport-ws/src/manifest.ts +++ b/packages/transport-ws/src/manifest.ts @@ -1,13 +1,13 @@ import type { Manifest } from "@dispatch/kernel"; export const manifest: Manifest = { - id: "transport-ws", - name: "Transport WebSocket", - version: "0.0.0", - apiVersion: "^0.1.0", - trust: "bundled", - dependsOn: ["surface-registry", "session-orchestrator"], - capabilities: { network: true }, - contributes: { routes: ["/ws/surfaces"] }, - activation: "eager", + id: "transport-ws", + name: "Transport WebSocket", + version: "0.0.0", + apiVersion: "^0.1.0", + trust: "bundled", + dependsOn: ["surface-registry", "session-orchestrator"], + capabilities: { network: true }, + contributes: { routes: ["/ws/surfaces"] }, + activation: "eager", }; diff --git a/packages/transport-ws/src/router.test.ts b/packages/transport-ws/src/router.test.ts index 6d01823..19b5bb5 100644 --- a/packages/transport-ws/src/router.test.ts +++ b/packages/transport-ws/src/router.test.ts @@ -7,677 +7,732 @@ import { catalogMessage, type RouteResult, routeClientMessage, subKey } from "./ // ── Fake in-memory registry (no mocks — just a plain implementation) ──────── interface FakeProviderOpts { - readonly id: string; - readonly title?: string; - readonly actions?: readonly string[]; - /** Called with the context that getSpec receives — for test assertions. */ - readonly onGetSpec?: (context: SurfaceContext | undefined) => void; - /** Called with the context that invoke receives — for test assertions. */ - readonly onInvoke?: ( - actionId: string, - payload: unknown, - context: SurfaceContext | undefined, - ) => void; + readonly id: string; + readonly title?: string; + readonly actions?: readonly string[]; + /** Called with the context that getSpec receives — for test assertions. */ + readonly onGetSpec?: (context: SurfaceContext | undefined) => void; + /** Called with the context that invoke receives — for test assertions. */ + readonly onInvoke?: ( + actionId: string, + payload: unknown, + context: SurfaceContext | undefined, + ) => void; } function fakeProvider( - idOrOpts: string | FakeProviderOpts, - title?: string, - actions?: readonly string[], + idOrOpts: string | FakeProviderOpts, + title?: string, + actions?: readonly string[], ): SurfaceProvider { - const opts: FakeProviderOpts = - typeof idOrOpts === "string" - ? { - id: idOrOpts, - ...(title !== undefined ? { title } : {}), - ...(actions !== undefined ? { actions } : {}), - } - : idOrOpts; - const catalogEntry: SurfaceCatalogEntry = { - id: opts.id, - region: "default", - title: opts.title ?? `Surface ${opts.id}`, - }; - return { - catalogEntry, - getSpec(context?: SurfaceContext): SurfaceSpec { - opts.onGetSpec?.(context); - return { - id: opts.id, - region: "default", - title: catalogEntry.title, - fields: - opts.actions?.map((a) => ({ - kind: "button" as const, - label: a, - action: { actionId: a }, - })) ?? [], - }; - }, - invoke(actionId: string, _payload?: unknown, context?: SurfaceContext) { - opts.onInvoke?.(actionId, _payload, context); - }, - }; + const opts: FakeProviderOpts = + typeof idOrOpts === "string" + ? { + id: idOrOpts, + ...(title !== undefined ? { title } : {}), + ...(actions !== undefined ? { actions } : {}), + } + : idOrOpts; + const catalogEntry: SurfaceCatalogEntry = { + id: opts.id, + region: "default", + title: opts.title ?? `Surface ${opts.id}`, + }; + return { + catalogEntry, + getSpec(context?: SurfaceContext): SurfaceSpec { + opts.onGetSpec?.(context); + return { + id: opts.id, + region: "default", + title: catalogEntry.title, + fields: + opts.actions?.map((a) => ({ + kind: "button" as const, + label: a, + action: { actionId: a }, + })) ?? [], + }; + }, + invoke(actionId: string, _payload?: unknown, context?: SurfaceContext) { + opts.onInvoke?.(actionId, _payload, context); + }, + }; } function fakeRegistry(providers: readonly SurfaceProvider[]): SurfaceRegistry { - const map = new Map(providers.map((p) => [p.catalogEntry.id, p])); - return { - register(_provider: SurfaceProvider) { - return () => {}; - }, - getCatalog() { - return [...map.values()].map((p) => p.catalogEntry); - }, - getSurface(id: string) { - return map.get(id); - }, - }; + const map = new Map(providers.map((p) => [p.catalogEntry.id, p])); + return { + register(_provider: SurfaceProvider) { + return () => {}; + }, + getCatalog() { + return [...map.values()].map((p) => p.catalogEntry); + }, + getSurface(id: string) { + return map.get(id); + }, + }; } // ── Tests ─────────────────────────────────────────────────────────────────── describe("routeClientMessage", () => { - describe("subscribe", () => { - it("replies with `surface` and tracks the subscription", () => { - const provider = fakeProvider("a", "Surface A"); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "subscribe", - surfaceId: "a", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(1); - expect(result.replies[0]).toEqual({ - type: "surface", - spec: { - id: "a", - region: "default", - title: "Surface A", - fields: [], - }, - }); - expect(result.subChange).toEqual({ op: "add", surfaceId: "a" }); - }); - - it("is idempotent — subscribing twice does not duplicate the subChange", () => { - const provider = fakeProvider("a"); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>([subKey("a")]); // already subscribed (global) - - const result = routeClientMessage(registry, connSubs, { - type: "subscribe", - surfaceId: "a", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(1); - expect(result.replies[0]?.type).toBe("surface"); - expect(result.subChange).toBeUndefined(); - }); - - it("returns `error` for an unknown surface id", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "subscribe", - surfaceId: "nonexistent", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(1); - expect(result.replies[0]).toEqual({ - type: "error", - surfaceId: "nonexistent", - message: "Unknown surface: nonexistent", - }); - expect(result.subChange).toBeUndefined(); - }); - - it("subscribe with conversationId fetches the provider spec for that conversation and tags the reply", () => { - let receivedContext: SurfaceContext | undefined; - const provider = fakeProvider({ - id: "cache-warm", - title: "Cache Warming", - onGetSpec(ctx) { - receivedContext = ctx; - }, - }); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "subscribe", - surfaceId: "cache-warm", - conversationId: "conv-42", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(receivedContext).toEqual({ conversationId: "conv-42" }); - expect(result.replies).toHaveLength(1); - const reply = result.replies[0]; - if (reply?.type !== "surface") throw new Error("expected surface reply"); - expect(reply.conversationId).toBe("conv-42"); - expect(reply.spec.id).toBe("cache-warm"); - expect(result.subChange).toEqual({ - op: "add", - surfaceId: "cache-warm", - conversationId: "conv-42", - }); - }); - - it("subscribe without conversationId behaves as before (global surface unaffected)", () => { - let receivedContext: SurfaceContext | undefined; - const provider = fakeProvider({ - id: "global-surf", - title: "Global Surface", - onGetSpec(ctx) { - receivedContext = ctx; - }, - }); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "subscribe", - surfaceId: "global-surf", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(receivedContext).toBeUndefined(); - const reply = result.replies[0]; - if (reply?.type !== "surface") throw new Error("expected surface reply"); - expect(reply.conversationId).toBeUndefined(); - expect(result.subChange).toEqual({ op: "add", surfaceId: "global-surf" }); - }); - }); - - describe("unsubscribe", () => { - it("emits a remove subChange and no replies", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>([subKey("a")]); - - const result = routeClientMessage(registry, connSubs, { - type: "unsubscribe", - surfaceId: "a", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(0); - expect(result.subChange).toEqual({ op: "remove", surfaceId: "a" }); - }); - - it("emits remove even if not currently subscribed (idempotent)", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "unsubscribe", - surfaceId: "a", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(0); - expect(result.subChange).toEqual({ op: "remove", surfaceId: "a" }); - }); - }); - - describe("invoke", () => { - it("signals the invoke effect for a known surface", () => { - const provider = fakeProvider("a", "Surface A", ["toggle"]); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "invoke", - surfaceId: "a", - actionId: "toggle", - payload: true, - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(0); - expect(result.invoke).toEqual({ - surfaceId: "a", - actionId: "toggle", - payload: true, - }); - }); - - it("returns `error` for an unknown surface id", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "invoke", - surfaceId: "nonexistent", - actionId: "toggle", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.replies).toHaveLength(1); - expect(result.replies[0]).toEqual({ - type: "error", - surfaceId: "nonexistent", - message: "Unknown surface: nonexistent", - }); - expect(result.invoke).toBeUndefined(); - }); - - it("invoke forwards the conversationId to the provider", () => { - let _receivedContext: SurfaceContext | undefined; - const provider = fakeProvider({ - id: "cache-warm", - title: "Cache Warming", - actions: ["warm"], - onInvoke(_actionId, _payload, ctx) { - _receivedContext = ctx; - }, - }); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "invoke", - surfaceId: "cache-warm", - actionId: "warm", - payload: { force: true }, - conversationId: "conv-99", - }); - - expect(result.kind).toBe("surface"); - if (result.kind !== "surface") throw new Error("expected surface"); - expect(result.invoke).toEqual({ - surfaceId: "cache-warm", - actionId: "warm", - payload: { force: true }, - conversationId: "conv-99", - }); - }); - }); - - describe("chat.send", () => { - it("classifies a chat.send message", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "hello", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - expect(result.message).toBe("hello"); - expect(result.conversationId).toBeUndefined(); - expect(result.model).toBeUndefined(); - expect(result.cwd).toBeUndefined(); - }); - - it("passes through optional fields", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - conversationId: "conv-123", - message: "follow up", - model: "gpt-4", - cwd: "/tmp", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - expect(result.conversationId).toBe("conv-123"); - expect(result.message).toBe("follow up"); - expect(result.model).toBe("gpt-4"); - expect(result.cwd).toBe("/tmp"); - }); - - it("chat.send threads workspaceId", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - conversationId: "conv-ws", - message: "hello workspace", - workspaceId: "my-workspace", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - expect(result.workspaceId).toBe("my-workspace"); - }); - - it("chat.send defaults workspaceId when omitted", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "hello no workspace", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - // workspaceId is absent (undefined) — the orchestrator receives no - // workspaceId and applies its own "default" resolution. - expect(result).not.toHaveProperty("workspaceId"); - }); - - it("chat.send threads computerId", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - conversationId: "conv-cid", - message: "hello computer", - computerId: "dev-box", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - expect(result.computerId).toBe("dev-box"); - }); - - it("chat.send omits computerId (absent/undefined) when not sent — backward compatible", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "hello no computer", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - // computerId is absent (undefined) — the orchestrator receives no - // computerId and resolves the inherited chain (conversation → - // workspace defaultComputerId → local). Mirrors workspaceId. - expect(result).not.toHaveProperty("computerId"); - }); - - it("rejects a malformed chat.send (empty message)", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "", - }); - - expect(result.kind).toBe("chat-error"); - if (result.kind !== "chat-error") throw new Error("expected chat-error"); - expect(result.errorMessage).toContain("non-empty string"); - }); - - it("rejects a malformed chat.send (missing message)", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: undefined as unknown as string, - }); - - expect(result.kind).toBe("chat-error"); - if (result.kind !== "chat-error") throw new Error("expected chat-error"); - expect(result.errorMessage).toContain("non-empty string"); - }); - - it("threads each valid reasoningEffort level through to the result", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - const levels = ["low", "medium", "high", "xhigh", "max"] as const; - - for (const level of levels) { - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "hello", - reasoningEffort: level, - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - expect(result.reasoningEffort).toBe(level); - } - }); - - it("omits reasoningEffort from result when not provided by client", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "hello", - }); - - expect(result.kind).toBe("chat"); - if (result.kind !== "chat") throw new Error("expected chat"); - expect(result).not.toHaveProperty("reasoningEffort"); - }); - - it("rejects an invalid reasoningEffort value with a chat-error", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.send", - message: "hello", - reasoningEffort: "turbo" as unknown as "low", - }); - - expect(result.kind).toBe("chat-error"); - if (result.kind !== "chat-error") throw new Error("expected chat-error"); - expect(result.errorMessage).toContain("invalid reasoningEffort"); - expect(result.errorMessage).toContain("turbo"); - }); - }); - - describe("chat.subscribe", () => { - it("routes chat.subscribe → { kind: 'chat-subscribe', conversationId }", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.subscribe", - conversationId: "conv-abc", - }); - - expect(result).toEqual({ kind: "chat-subscribe", conversationId: "conv-abc" }); - }); - }); - - describe("chat.unsubscribe", () => { - it("routes chat.unsubscribe → { kind: 'chat-unsubscribe', conversationId }", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.unsubscribe", - conversationId: "conv-abc", - }); - - expect(result).toEqual({ kind: "chat-unsubscribe", conversationId: "conv-abc" }); - }); - }); - - describe("chat.queue", () => { - it("routes a valid chat.queue → { kind: 'chat-queue', conversationId, text } (what the shell passes to orchestrator.enqueue)", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.queue", - conversationId: "conv-1", - text: "steer here", - }); - - expect(result).toEqual({ - kind: "chat-queue", - conversationId: "conv-1", - text: "steer here", - }); - }); - - it("chat.queue threads workspaceId", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.queue", - conversationId: "conv-ws", - text: "steer here", - workspaceId: "my-workspace", - }); - - expect(result.kind).toBe("chat-queue"); - if (result.kind !== "chat-queue") throw new Error("expected chat-queue"); - expect(result.workspaceId).toBe("my-workspace"); - }); - - it("rejects empty/whitespace text → chat-error (no enqueue signal)", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - for (const text of ["", " ", "\t\n"]) { - const result = routeClientMessage(registry, connSubs, { - type: "chat.queue", - conversationId: "conv-1", - text, - }); - - expect(result.kind).toBe("chat-error"); - if (result.kind !== "chat-error") throw new Error("expected chat-error"); - expect(result.conversationId).toBe("conv-1"); - expect(result.errorMessage).toContain("non-empty string"); - expect(result.errorMessage).toContain("text"); - } - }); - - it("rejects missing text → chat-error (no enqueue signal)", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - const result = routeClientMessage(registry, connSubs, { - type: "chat.queue", - conversationId: "conv-1", - text: undefined as unknown as string, - }); - - expect(result.kind).toBe("chat-error"); - if (result.kind !== "chat-error") throw new Error("expected chat-error"); - expect(result.errorMessage).toContain("non-empty string"); - }); - - it("does not trim the stored text — passes the original through to the shell", () => { - const registry = fakeRegistry([]); - const connSubs = new Set<string>(); - - // Non-empty after trim (so valid), but the value carries surrounding - // whitespace: the router passes it through unchanged (validation uses - // trim; the orchestrator receives the original text). - const result = routeClientMessage(registry, connSubs, { - type: "chat.queue", - conversationId: "conv-1", - text: " steer ", - }); - - expect(result.kind).toBe("chat-queue"); - if (result.kind !== "chat-queue") throw new Error("expected chat-queue"); - expect(result.text).toBe(" steer "); - }); - }); - - describe("exhaustive switch (regression guard for Wave-0 fan-out)", () => { - // Every WsClientMessage variant must route to a defined result with a - // known kind — no fall-through / undefined return. If the union is - // widened again, `tsc` catches the missing case (the switch is - // exhaustive); this test guards the runtime side of that contract. - it("routes every WsClientMessage variant to a defined RouteResult", () => { - const provider = fakeProvider("a", "Surface A", ["toggle"]); - const registry = fakeRegistry([provider]); - const connSubs = new Set<string>(); - - const samples: WsClientMessage[] = [ - { type: "subscribe", surfaceId: "a" }, - { type: "unsubscribe", surfaceId: "a" }, - { type: "invoke", surfaceId: "a", actionId: "toggle", payload: true }, - { type: "chat.send", message: "hi" }, - { type: "chat.subscribe", conversationId: "c1" }, - { type: "chat.unsubscribe", conversationId: "c1" }, - { type: "chat.queue", conversationId: "c1", text: "steer" }, - ]; - - const validKinds = new Set<RouteResult["kind"]>([ - "surface", - "chat", - "chat-error", - "chat-subscribe", - "chat-unsubscribe", - "chat-queue", - ]); - - for (const msg of samples) { - const result = routeClientMessage(registry, connSubs, msg); - expect(result).toBeDefined(); - expect(validKinds.has(result.kind)).toBe(true); - } - }); - }); + describe("subscribe", () => { + it("replies with `surface` and tracks the subscription", () => { + const provider = fakeProvider("a", "Surface A"); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "subscribe", + surfaceId: "a", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(1); + expect(result.replies[0]).toEqual({ + type: "surface", + spec: { + id: "a", + region: "default", + title: "Surface A", + fields: [], + }, + }); + expect(result.subChange).toEqual({ op: "add", surfaceId: "a" }); + }); + + it("is idempotent — subscribing twice does not duplicate the subChange", () => { + const provider = fakeProvider("a"); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>([subKey("a")]); // already subscribed (global) + + const result = routeClientMessage(registry, connSubs, { + type: "subscribe", + surfaceId: "a", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(1); + expect(result.replies[0]?.type).toBe("surface"); + expect(result.subChange).toBeUndefined(); + }); + + it("returns `error` for an unknown surface id", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "subscribe", + surfaceId: "nonexistent", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(1); + expect(result.replies[0]).toEqual({ + type: "error", + surfaceId: "nonexistent", + message: "Unknown surface: nonexistent", + }); + expect(result.subChange).toBeUndefined(); + }); + + it("subscribe with conversationId fetches the provider spec for that conversation and tags the reply", () => { + let receivedContext: SurfaceContext | undefined; + const provider = fakeProvider({ + id: "cache-warm", + title: "Cache Warming", + onGetSpec(ctx) { + receivedContext = ctx; + }, + }); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "subscribe", + surfaceId: "cache-warm", + conversationId: "conv-42", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(receivedContext).toEqual({ conversationId: "conv-42" }); + expect(result.replies).toHaveLength(1); + const reply = result.replies[0]; + if (reply?.type !== "surface") throw new Error("expected surface reply"); + expect(reply.conversationId).toBe("conv-42"); + expect(reply.spec.id).toBe("cache-warm"); + expect(result.subChange).toEqual({ + op: "add", + surfaceId: "cache-warm", + conversationId: "conv-42", + }); + }); + + it("subscribe without conversationId behaves as before (global surface unaffected)", () => { + let receivedContext: SurfaceContext | undefined; + const provider = fakeProvider({ + id: "global-surf", + title: "Global Surface", + onGetSpec(ctx) { + receivedContext = ctx; + }, + }); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "subscribe", + surfaceId: "global-surf", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(receivedContext).toBeUndefined(); + const reply = result.replies[0]; + if (reply?.type !== "surface") throw new Error("expected surface reply"); + expect(reply.conversationId).toBeUndefined(); + expect(result.subChange).toEqual({ op: "add", surfaceId: "global-surf" }); + }); + }); + + describe("unsubscribe", () => { + it("emits a remove subChange and no replies", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>([subKey("a")]); + + const result = routeClientMessage(registry, connSubs, { + type: "unsubscribe", + surfaceId: "a", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(0); + expect(result.subChange).toEqual({ op: "remove", surfaceId: "a" }); + }); + + it("emits remove even if not currently subscribed (idempotent)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "unsubscribe", + surfaceId: "a", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(0); + expect(result.subChange).toEqual({ op: "remove", surfaceId: "a" }); + }); + }); + + describe("invoke", () => { + it("signals the invoke effect for a known surface", () => { + const provider = fakeProvider("a", "Surface A", ["toggle"]); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "invoke", + surfaceId: "a", + actionId: "toggle", + payload: true, + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(0); + expect(result.invoke).toEqual({ + surfaceId: "a", + actionId: "toggle", + payload: true, + }); + }); + + it("returns `error` for an unknown surface id", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "invoke", + surfaceId: "nonexistent", + actionId: "toggle", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.replies).toHaveLength(1); + expect(result.replies[0]).toEqual({ + type: "error", + surfaceId: "nonexistent", + message: "Unknown surface: nonexistent", + }); + expect(result.invoke).toBeUndefined(); + }); + + it("invoke forwards the conversationId to the provider", () => { + let _receivedContext: SurfaceContext | undefined; + const provider = fakeProvider({ + id: "cache-warm", + title: "Cache Warming", + actions: ["warm"], + onInvoke(_actionId, _payload, ctx) { + _receivedContext = ctx; + }, + }); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "invoke", + surfaceId: "cache-warm", + actionId: "warm", + payload: { force: true }, + conversationId: "conv-99", + }); + + expect(result.kind).toBe("surface"); + if (result.kind !== "surface") throw new Error("expected surface"); + expect(result.invoke).toEqual({ + surfaceId: "cache-warm", + actionId: "warm", + payload: { force: true }, + conversationId: "conv-99", + }); + }); + }); + + describe("chat.send", () => { + it("classifies a chat.send message", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "hello", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + expect(result.message).toBe("hello"); + expect(result.conversationId).toBeUndefined(); + expect(result.model).toBeUndefined(); + expect(result.cwd).toBeUndefined(); + }); + + it("passes through optional fields", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + conversationId: "conv-123", + message: "follow up", + model: "gpt-4", + cwd: "/tmp", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + expect(result.conversationId).toBe("conv-123"); + expect(result.message).toBe("follow up"); + expect(result.model).toBe("gpt-4"); + expect(result.cwd).toBe("/tmp"); + }); + + it("chat.send threads workspaceId", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + conversationId: "conv-ws", + message: "hello workspace", + workspaceId: "my-workspace", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + expect(result.workspaceId).toBe("my-workspace"); + }); + + it("chat.send defaults workspaceId when omitted", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "hello no workspace", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + // workspaceId is absent (undefined) — the orchestrator receives no + // workspaceId and applies its own "default" resolution. + expect(result).not.toHaveProperty("workspaceId"); + }); + + it("chat.send threads computerId", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + conversationId: "conv-cid", + message: "hello computer", + computerId: "dev-box", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + expect(result.computerId).toBe("dev-box"); + }); + + it("chat.send omits computerId (absent/undefined) when not sent — backward compatible", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "hello no computer", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + // computerId is absent (undefined) — the orchestrator receives no + // computerId and resolves the inherited chain (conversation → + // workspace defaultComputerId → local). Mirrors workspaceId. + expect(result).not.toHaveProperty("computerId"); + }); + + it("rejects a malformed chat.send (empty message)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "", + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.errorMessage).toContain("non-empty string"); + }); + + it("rejects a malformed chat.send (missing message)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: undefined as unknown as string, + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.errorMessage).toContain("non-empty string"); + }); + + it("threads each valid reasoningEffort level through to the result", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + const levels = ["low", "medium", "high", "xhigh", "max"] as const; + + for (const level of levels) { + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "hello", + reasoningEffort: level, + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + expect(result.reasoningEffort).toBe(level); + } + }); + + it("omits reasoningEffort from result when not provided by client", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "hello", + }); + + expect(result.kind).toBe("chat"); + if (result.kind !== "chat") throw new Error("expected chat"); + expect(result).not.toHaveProperty("reasoningEffort"); + }); + + it("rejects an invalid reasoningEffort value with a chat-error", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.send", + message: "hello", + reasoningEffort: "turbo" as unknown as "low", + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.errorMessage).toContain("invalid reasoningEffort"); + expect(result.errorMessage).toContain("turbo"); + }); + }); + + describe("chat.subscribe", () => { + it("routes chat.subscribe → { kind: 'chat-subscribe', conversationId }", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.subscribe", + conversationId: "conv-abc", + }); + + expect(result).toEqual({ kind: "chat-subscribe", conversationId: "conv-abc" }); + }); + }); + + describe("chat.unsubscribe", () => { + it("routes chat.unsubscribe → { kind: 'chat-unsubscribe', conversationId }", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.unsubscribe", + conversationId: "conv-abc", + }); + + expect(result).toEqual({ kind: "chat-unsubscribe", conversationId: "conv-abc" }); + }); + }); + + describe("chat.queue", () => { + it("routes a valid chat.queue → { kind: 'chat-queue', conversationId, text } (what the shell passes to orchestrator.enqueue)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue", + conversationId: "conv-1", + text: "steer here", + }); + + expect(result).toEqual({ + kind: "chat-queue", + conversationId: "conv-1", + text: "steer here", + }); + }); + + it("chat.queue threads workspaceId", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue", + conversationId: "conv-ws", + text: "steer here", + workspaceId: "my-workspace", + }); + + expect(result.kind).toBe("chat-queue"); + if (result.kind !== "chat-queue") throw new Error("expected chat-queue"); + expect(result.workspaceId).toBe("my-workspace"); + }); + + it("rejects empty/whitespace text → chat-error (no enqueue signal)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + for (const text of ["", " ", "\t\n"]) { + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue", + conversationId: "conv-1", + text, + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.conversationId).toBe("conv-1"); + expect(result.errorMessage).toContain("non-empty string"); + expect(result.errorMessage).toContain("text"); + } + }); + + it("rejects missing text → chat-error (no enqueue signal)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue", + conversationId: "conv-1", + text: undefined as unknown as string, + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.errorMessage).toContain("non-empty string"); + }); + + it("does not trim the stored text — passes the original through to the shell", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + // Non-empty after trim (so valid), but the value carries surrounding + // whitespace: the router passes it through unchanged (validation uses + // trim; the orchestrator receives the original text). + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue", + conversationId: "conv-1", + text: " steer ", + }); + + expect(result.kind).toBe("chat-queue"); + if (result.kind !== "chat-queue") throw new Error("expected chat-queue"); + expect(result.text).toBe(" steer "); + }); + }); + + describe("chat.queue.cancel", () => { + it("routes a valid chat.queue.cancel → { kind: 'chat-queue-cancel', conversationId, messageId }", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue.cancel", + conversationId: "conv-1", + messageId: "q-42", + }); + + expect(result).toEqual({ + kind: "chat-queue-cancel", + conversationId: "conv-1", + messageId: "q-42", + }); + }); + + it("rejects empty conversationId → chat-error (no cancel signal)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue.cancel", + conversationId: "", + messageId: "q-42", + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.errorMessage).toContain("non-empty string"); + expect(result.errorMessage).toContain("conversationId"); + }); + + it("rejects empty messageId → chat-error (no cancel signal)", () => { + const registry = fakeRegistry([]); + const connSubs = new Set<string>(); + + for (const messageId of ["", undefined as unknown as string]) { + const result = routeClientMessage(registry, connSubs, { + type: "chat.queue.cancel", + conversationId: "conv-1", + messageId, + }); + + expect(result.kind).toBe("chat-error"); + if (result.kind !== "chat-error") throw new Error("expected chat-error"); + expect(result.errorMessage).toContain("non-empty string"); + expect(result.errorMessage).toContain("messageId"); + } + }); + }); + + describe("exhaustive switch (regression guard for Wave-0 fan-out)", () => { + // Every WsClientMessage variant must route to a defined result with a + // known kind — no fall-through / undefined return. If the union is + // widened again, `tsc` catches the missing case (the switch is + // exhaustive); this test guards the runtime side of that contract. + it("routes every WsClientMessage variant to a defined RouteResult", () => { + const provider = fakeProvider("a", "Surface A", ["toggle"]); + const registry = fakeRegistry([provider]); + const connSubs = new Set<string>(); + + const samples: WsClientMessage[] = [ + { type: "subscribe", surfaceId: "a" }, + { type: "unsubscribe", surfaceId: "a" }, + { type: "invoke", surfaceId: "a", actionId: "toggle", payload: true }, + { type: "chat.send", message: "hi" }, + { type: "chat.subscribe", conversationId: "c1" }, + { type: "chat.unsubscribe", conversationId: "c1" }, + { type: "chat.queue", conversationId: "c1", text: "steer" }, + { type: "chat.queue.cancel", conversationId: "c1", messageId: "m1" }, + ]; + + const validKinds = new Set<RouteResult["kind"]>([ + "surface", + "chat", + "chat-error", + "chat-subscribe", + "chat-unsubscribe", + "chat-queue", + "chat-queue-cancel", + ]); + + for (const msg of samples) { + const result = routeClientMessage(registry, connSubs, msg); + expect(result).toBeDefined(); + expect(validKinds.has(result.kind)).toBe(true); + } + }); + }); }); describe("catalogMessage", () => { - it("returns the catalog from the registry", () => { - const providerA = fakeProvider("a", "Surface A"); - const providerB = fakeProvider("b", "Surface B"); - const registry = fakeRegistry([providerA, providerB]); + it("returns the catalog from the registry", () => { + const providerA = fakeProvider("a", "Surface A"); + const providerB = fakeProvider("b", "Surface B"); + const registry = fakeRegistry([providerA, providerB]); - const msg = catalogMessage(registry); + const msg = catalogMessage(registry); - expect(msg).toEqual({ - type: "catalog", - catalog: [ - { id: "a", region: "default", title: "Surface A" }, - { id: "b", region: "default", title: "Surface B" }, - ], - }); - }); + expect(msg).toEqual({ + type: "catalog", + catalog: [ + { id: "a", region: "default", title: "Surface A" }, + { id: "b", region: "default", title: "Surface B" }, + ], + }); + }); - it("returns an empty catalog when no providers are registered", () => { - const registry = fakeRegistry([]); + it("returns an empty catalog when no providers are registered", () => { + const registry = fakeRegistry([]); - const msg = catalogMessage(registry); + const msg = catalogMessage(registry); - expect(msg).toEqual({ type: "catalog", catalog: [] }); - }); + expect(msg).toEqual({ type: "catalog", catalog: [] }); + }); }); describe("subKey", () => { - it("builds a global key when conversationId is undefined", () => { - expect(subKey("surf-a")).toBe("surf-a::"); - }); + it("builds a global key when conversationId is undefined", () => { + expect(subKey("surf-a")).toBe("surf-a::"); + }); - it("builds a conversation-scoped key when conversationId is provided", () => { - expect(subKey("surf-a", "conv-42")).toBe("surf-a::conv-42"); - }); + it("builds a conversation-scoped key when conversationId is provided", () => { + expect(subKey("surf-a", "conv-42")).toBe("surf-a::conv-42"); + }); - it("global and conversation-scoped keys are distinct", () => { - expect(subKey("surf-a")).not.toBe(subKey("surf-a", "conv-42")); - }); + it("global and conversation-scoped keys are distinct", () => { + expect(subKey("surf-a")).not.toBe(subKey("surf-a", "conv-42")); + }); }); diff --git a/packages/transport-ws/src/router.ts b/packages/transport-ws/src/router.ts index 7e9ba77..014db96 100644 --- a/packages/transport-ws/src/router.ts +++ b/packages/transport-ws/src/router.ts @@ -9,12 +9,13 @@ import type { SurfaceContext, SurfaceRegistry } from "@dispatch/surface-registry"; import type { - ChatQueueMessage, - ChatSendMessage, - ChatSubscribeMessage, - ChatUnsubscribeMessage, - ReasoningEffort, - WsClientMessage, + ChatQueueCancelMessage, + ChatQueueMessage, + ChatSendMessage, + ChatSubscribeMessage, + ChatUnsubscribeMessage, + ReasoningEffort, + WsClientMessage, } from "@dispatch/transport-contract"; import type { SurfaceServerMessage } from "@dispatch/ui-contract"; @@ -22,61 +23,67 @@ import type { SurfaceServerMessage } from "@dispatch/ui-contract"; /** The effect a surface client message should produce. */ export interface SurfaceRouteResult { - readonly kind: "surface"; - /** Server messages to send back to this connection. */ - readonly replies: readonly SurfaceServerMessage[]; - /** Whether to add or remove the surface id from connSubs. */ - readonly subChange?: { - readonly op: "add" | "remove"; - readonly surfaceId: string; - readonly conversationId?: string; - }; - /** If set, the shell must call `provider.invoke(actionId, payload, context)`. */ - readonly invoke?: { - readonly surfaceId: string; - readonly actionId: string; - readonly payload?: unknown; - readonly conversationId?: string; - }; + readonly kind: "surface"; + /** Server messages to send back to this connection. */ + readonly replies: readonly SurfaceServerMessage[]; + /** Whether to add or remove the surface id from connSubs. */ + readonly subChange?: { + readonly op: "add" | "remove"; + readonly surfaceId: string; + readonly conversationId?: string; + }; + /** If set, the shell must call `provider.invoke(actionId, payload, context)`. */ + readonly invoke?: { + readonly surfaceId: string; + readonly actionId: string; + readonly payload?: unknown; + readonly conversationId?: string; + }; } /** The effect a validated chat.send should produce. */ export interface ChatRouteResult { - readonly kind: "chat"; - readonly conversationId: string | undefined; - readonly message: string; - readonly model: string | undefined; - readonly cwd: string | undefined; - readonly reasoningEffort?: ReasoningEffort; - readonly workspaceId?: string; - /** - * The computer (SSH config alias) to run this turn's tools on — forwarded - * verbatim to the orchestrator's `startTurn` (which resolves it via - * `getEffectiveComputer`). Mirrors `cwd`/`workspaceId`: an opaque per-turn - * override, unvalidated here (validation happens at SSH connect time). - * Absent when the client omits it (the orchestrator then inherits the - * conversation → workspace → local chain). - */ - readonly computerId?: string; + readonly kind: "chat"; + readonly conversationId: string | undefined; + readonly message: string; + readonly model: string | undefined; + readonly cwd: string | undefined; + readonly reasoningEffort?: ReasoningEffort; + readonly workspaceId?: string; + /** + * The computer (SSH config alias) to run this turn's tools on — forwarded + * verbatim to the orchestrator's `startTurn` (which resolves it via + * `getEffectiveComputer`). Mirrors `cwd`/`workspaceId`: an opaque per-turn + * override, unvalidated here (validation happens at SSH connect time). + * Absent when the client omits it (the orchestrator then inherits the + * conversation → workspace → local chain). + */ + readonly computerId?: string; + /** + * Images attached to this turn (data URLs or http URLs), forwarded verbatim to + * the orchestrator. Absent when the client omits it. Each entry must have a + * non-empty string `url`; `mimeType` is optional. + */ + readonly images?: readonly { readonly url: string; readonly mimeType?: string }[]; } /** A malformed chat.send that should yield a chat.error reply. */ export interface ChatRouteError { - readonly kind: "chat-error"; - readonly conversationId: string | undefined; - readonly errorMessage: string; + readonly kind: "chat-error"; + readonly conversationId: string | undefined; + readonly errorMessage: string; } /** The effect a chat.subscribe should produce. */ export interface ChatSubscribeRouteResult { - readonly kind: "chat-subscribe"; - readonly conversationId: string; + readonly kind: "chat-subscribe"; + readonly conversationId: string; } /** The effect a chat.unsubscribe should produce. */ export interface ChatUnsubscribeRouteResult { - readonly kind: "chat-unsubscribe"; - readonly conversationId: string; + readonly kind: "chat-unsubscribe"; + readonly conversationId: string; } /** @@ -87,20 +94,35 @@ export interface ChatUnsubscribeRouteResult { * (startedTurn:true — the shell auto-subscribes the sender, same as chat.send). */ export interface ChatQueueRouteResult { - readonly kind: "chat-queue"; - readonly conversationId: string; - readonly text: string; - readonly workspaceId?: string; + readonly kind: "chat-queue"; + readonly conversationId: string; + readonly text: string; + readonly workspaceId?: string; +} + +/** + * The effect a validated chat.queue.cancel should produce. The shell calls + * `orchestrator.cancelQueuedMessage({ conversationId, messageId })` and emits + * NOTHING back (fire-and-forget): success is confirmed by the message-queue + * SURFACE updating (the cancelled message leaves the snapshot). Cancelling a + * message that is no longer queued is a silent no-op (no surface update, no + * error). Mirrors `ChatQueueRouteResult`'s fire-and-forget style. + */ +export interface ChatQueueCancelRouteResult { + readonly kind: "chat-queue-cancel"; + readonly conversationId: string; + readonly messageId: string; } /** The effect any client WS message should produce. */ export type RouteResult = - | SurfaceRouteResult - | ChatRouteResult - | ChatRouteError - | ChatSubscribeRouteResult - | ChatUnsubscribeRouteResult - | ChatQueueRouteResult; + | SurfaceRouteResult + | ChatRouteResult + | ChatRouteError + | ChatSubscribeRouteResult + | ChatUnsubscribeRouteResult + | ChatQueueRouteResult + | ChatQueueCancelRouteResult; // ── Helpers ───────────────────────────────────────────────────────────────── @@ -109,12 +131,12 @@ export type RouteResult = * The shell uses this same function so both layers agree on key format. */ export function subKey(surfaceId: string, conversationId?: string): string { - return conversationId !== undefined ? `${surfaceId}::${conversationId}` : `${surfaceId}::`; + return conversationId !== undefined ? `${surfaceId}::${conversationId}` : `${surfaceId}::`; } /** Build the catalog `SurfaceServerMessage` from the registry. */ export function catalogMessage(registry: SurfaceRegistry): SurfaceServerMessage { - return { type: "catalog", catalog: registry.getCatalog() }; + return { type: "catalog", catalog: registry.getCatalog() }; } // ── Router ────────────────────────────────────────────────────────────────── @@ -127,71 +149,104 @@ export function catalogMessage(registry: SurfaceRegistry): SurfaceServerMessage * @param msg The parsed client message (surface or chat). */ export function routeClientMessage( - registry: SurfaceRegistry, - connSubs: ReadonlySet<string>, - msg: WsClientMessage, + registry: SurfaceRegistry, + connSubs: ReadonlySet<string>, + msg: WsClientMessage, ): RouteResult { - switch (msg.type) { - case "subscribe": - return handleSubscribe(registry, connSubs, msg.surfaceId, msg.conversationId); - case "unsubscribe": - return handleUnsubscribe(msg.surfaceId, msg.conversationId); - case "invoke": - return handleInvoke(registry, msg.surfaceId, msg.actionId, msg.payload, msg.conversationId); - case "chat.send": - return handleChatSend(msg); - case "chat.subscribe": - return handleChatSubscribe(msg); - case "chat.unsubscribe": - return handleChatUnsubscribe(msg); - case "chat.queue": - return handleChatQueue(msg); - } + switch (msg.type) { + case "subscribe": + return handleSubscribe(registry, connSubs, msg.surfaceId, msg.conversationId); + case "unsubscribe": + return handleUnsubscribe(msg.surfaceId, msg.conversationId); + case "invoke": + return handleInvoke(registry, msg.surfaceId, msg.actionId, msg.payload, msg.conversationId); + case "chat.send": + return handleChatSend(msg); + case "chat.subscribe": + return handleChatSubscribe(msg); + case "chat.unsubscribe": + return handleChatUnsubscribe(msg); + case "chat.queue": + return handleChatQueue(msg); + case "chat.queue.cancel": + return handleChatQueueCancel(msg); + } } // ── Chat validation ───────────────────────────────────────────────────────── const VALID_REASONING_EFFORT: ReadonlySet<string> = new Set<ReasoningEffort>([ - "low", - "medium", - "high", - "xhigh", - "max", + "low", + "medium", + "high", + "xhigh", + "max", ]); function handleChatSend(msg: ChatSendMessage): ChatRouteResult | ChatRouteError { - if (typeof msg.message !== "string" || msg.message.length === 0) { - return { - kind: "chat-error", - conversationId: msg.conversationId, - errorMessage: "chat.send requires a non-empty string `message`", - }; - } - if (msg.reasoningEffort !== undefined && !VALID_REASONING_EFFORT.has(msg.reasoningEffort)) { - return { - kind: "chat-error", - conversationId: msg.conversationId, - errorMessage: `chat.send: invalid reasoningEffort "${msg.reasoningEffort}" — must be one of: low, medium, high, xhigh, max`, - }; - } - return { - kind: "chat", - conversationId: msg.conversationId, - message: msg.message, - model: msg.model, - cwd: msg.cwd, - ...(msg.reasoningEffort !== undefined ? { reasoningEffort: msg.reasoningEffort } : {}), - ...(msg.workspaceId !== undefined ? { workspaceId: msg.workspaceId } : {}), - ...(msg.computerId !== undefined ? { computerId: msg.computerId } : {}), - }; + if (typeof msg.message !== "string" || msg.message.length === 0) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: "chat.send requires a non-empty string `message`", + }; + } + if (msg.reasoningEffort !== undefined && !VALID_REASONING_EFFORT.has(msg.reasoningEffort)) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: `chat.send: invalid reasoningEffort "${msg.reasoningEffort}" — must be one of: low, medium, high, xhigh, max`, + }; + } + // Validate images (if present): each must be an object with a non-empty url. + let images: readonly { url: string; mimeType?: string }[] | undefined; + if (msg.images !== undefined) { + if (!Array.isArray(msg.images)) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: "chat.send: 'images' must be an array", + }; + } + const parsed: { url: string; mimeType?: string }[] = []; + for (const entry of msg.images) { + if ( + entry === null || + typeof entry !== "object" || + typeof entry.url !== "string" || + entry.url.length === 0 + ) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: "chat.send: each image must have a non-empty string 'url'", + }; + } + const p: { url: string; mimeType?: string } = { url: entry.url }; + if (entry.mimeType !== undefined) p.mimeType = entry.mimeType; + parsed.push(p); + } + if (parsed.length > 0) images = parsed; + } + return { + kind: "chat", + conversationId: msg.conversationId, + message: msg.message, + model: msg.model, + cwd: msg.cwd, + ...(msg.reasoningEffort !== undefined ? { reasoningEffort: msg.reasoningEffort } : {}), + ...(msg.workspaceId !== undefined ? { workspaceId: msg.workspaceId } : {}), + ...(msg.computerId !== undefined ? { computerId: msg.computerId } : {}), + ...(images !== undefined ? { images } : {}), + }; } function handleChatSubscribe(msg: ChatSubscribeMessage): ChatSubscribeRouteResult { - return { kind: "chat-subscribe", conversationId: msg.conversationId }; + return { kind: "chat-subscribe", conversationId: msg.conversationId }; } function handleChatUnsubscribe(msg: ChatUnsubscribeMessage): ChatUnsubscribeRouteResult { - return { kind: "chat-unsubscribe", conversationId: msg.conversationId }; + return { kind: "chat-unsubscribe", conversationId: msg.conversationId }; } /** @@ -201,105 +256,135 @@ function handleChatUnsubscribe(msg: ChatUnsubscribeMessage): ChatUnsubscribeRout * called). Valid → `chat-queue` (the shell calls `orchestrator.enqueue`). */ function handleChatQueue(msg: ChatQueueMessage): ChatQueueRouteResult | ChatRouteError { - if (typeof msg.text !== "string" || msg.text.trim().length === 0) { - return { - kind: "chat-error", - conversationId: msg.conversationId, - errorMessage: "chat.queue requires a non-empty string `text`", - }; - } - return { - kind: "chat-queue", - conversationId: msg.conversationId, - text: msg.text, - ...(msg.workspaceId !== undefined ? { workspaceId: msg.workspaceId } : {}), - }; + if (typeof msg.text !== "string" || msg.text.trim().length === 0) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: "chat.queue requires a non-empty string `text`", + }; + } + return { + kind: "chat-queue", + conversationId: msg.conversationId, + text: msg.text, + ...(msg.workspaceId !== undefined ? { workspaceId: msg.workspaceId } : {}), + }; +} + +/** + * Validate a chat.queue.cancel: both `conversationId` and `messageId` must be + * non-empty strings. Invalid → `chat-error` (the shell replies with `chat.error`, + * same style as a malformed `chat.queue`; the orchestrator is never called). + * Valid → `chat-queue-cancel` (the shell calls `orchestrator.cancelQueuedMessage`). + */ +function handleChatQueueCancel( + msg: ChatQueueCancelMessage, +): ChatQueueCancelRouteResult | ChatRouteError { + if (typeof msg.conversationId !== "string" || msg.conversationId.length === 0) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: "chat.queue.cancel requires a non-empty string `conversationId`", + }; + } + if (typeof msg.messageId !== "string" || msg.messageId.length === 0) { + return { + kind: "chat-error", + conversationId: msg.conversationId, + errorMessage: "chat.queue.cancel requires a non-empty string `messageId`", + }; + } + return { + kind: "chat-queue-cancel", + conversationId: msg.conversationId, + messageId: msg.messageId, + }; } // ── Per-message handlers ──────────────────────────────────────────────────── function handleSubscribe( - registry: SurfaceRegistry, - connSubs: ReadonlySet<string>, - surfaceId: string, - conversationId?: string, + registry: SurfaceRegistry, + connSubs: ReadonlySet<string>, + surfaceId: string, + conversationId?: string, ): SurfaceRouteResult { - const provider = registry.getSurface(surfaceId); - if (!provider) { - return { - kind: "surface", - replies: [{ type: "error", surfaceId, message: `Unknown surface: ${surfaceId}` }], - }; - } + const provider = registry.getSurface(surfaceId); + if (!provider) { + return { + kind: "surface", + replies: [{ type: "error", surfaceId, message: `Unknown surface: ${surfaceId}` }], + }; + } - const context: SurfaceContext | undefined = - conversationId !== undefined ? { conversationId } : undefined; - const spec = provider.getSpec(context); + const context: SurfaceContext | undefined = + conversationId !== undefined ? { conversationId } : undefined; + const spec = provider.getSpec(context); - // getSpec may be sync or async — the pure core treats it as a value the - // shell will resolve. We return the spec directly (it's a SurfaceSpec). - // If it's a Promise the shell awaits it; if it's sync it's already the value. - // For the pure core we just pass it through — the shell handles the resolution. - const specValue = spec as import("@dispatch/ui-contract").SurfaceSpec; + // getSpec may be sync or async — the pure core treats it as a value the + // shell will resolve. We return the spec directly (it's a SurfaceSpec). + // If it's a Promise the shell awaits it; if it's sync it's already the value. + // For the pure core we just pass it through — the shell handles the resolution. + const specValue = spec as import("@dispatch/ui-contract").SurfaceSpec; - const replies: import("@dispatch/ui-contract").SurfaceServerMessage[] = [ - { - type: "surface", - spec: specValue, - ...(conversationId !== undefined ? { conversationId } : {}), - }, - ]; + const replies: import("@dispatch/ui-contract").SurfaceServerMessage[] = [ + { + type: "surface", + spec: specValue, + ...(conversationId !== undefined ? { conversationId } : {}), + }, + ]; - // Idempotent: only emit subChange if not already subscribed. - const key = subKey(surfaceId, conversationId); - if (!connSubs.has(key)) { - return { - kind: "surface", - replies, - subChange: { - op: "add", - surfaceId, - ...(conversationId !== undefined ? { conversationId } : {}), - }, - }; - } - return { kind: "surface", replies }; + // Idempotent: only emit subChange if not already subscribed. + const key = subKey(surfaceId, conversationId); + if (!connSubs.has(key)) { + return { + kind: "surface", + replies, + subChange: { + op: "add", + surfaceId, + ...(conversationId !== undefined ? { conversationId } : {}), + }, + }; + } + return { kind: "surface", replies }; } function handleUnsubscribe(surfaceId: string, conversationId?: string): SurfaceRouteResult { - return { - kind: "surface", - replies: [], - subChange: { - op: "remove", - surfaceId, - ...(conversationId !== undefined ? { conversationId } : {}), - }, - }; + return { + kind: "surface", + replies: [], + subChange: { + op: "remove", + surfaceId, + ...(conversationId !== undefined ? { conversationId } : {}), + }, + }; } function handleInvoke( - registry: SurfaceRegistry, - surfaceId: string, - actionId: string, - payload?: unknown, - conversationId?: string, + registry: SurfaceRegistry, + surfaceId: string, + actionId: string, + payload?: unknown, + conversationId?: string, ): SurfaceRouteResult { - const provider = registry.getSurface(surfaceId); - if (!provider) { - return { - kind: "surface", - replies: [{ type: "error", surfaceId, message: `Unknown surface: ${surfaceId}` }], - }; - } - return { - kind: "surface", - replies: [], - invoke: { - surfaceId, - actionId, - payload, - ...(conversationId !== undefined ? { conversationId } : {}), - }, - }; + const provider = registry.getSurface(surfaceId); + if (!provider) { + return { + kind: "surface", + replies: [{ type: "error", surfaceId, message: `Unknown surface: ${surfaceId}` }], + }; + } + return { + kind: "surface", + replies: [], + invoke: { + surfaceId, + actionId, + payload, + ...(conversationId !== undefined ? { conversationId } : {}), + }, + }; } diff --git a/packages/transport-ws/src/server.bun.test.ts b/packages/transport-ws/src/server.bun.test.ts index e24aa6b..6b64f37 100644 --- a/packages/transport-ws/src/server.bun.test.ts +++ b/packages/transport-ws/src/server.bun.test.ts @@ -9,1202 +9,1202 @@ import { catalogMessage, routeClientMessage, subKey } from "./router.js"; // ── Fake Logger (captures records for assertions) ─────────────────────────── interface LogEntry { - readonly level: "debug" | "info" | "warn" | "error"; - readonly msg: string; - readonly attrs?: Attributes | ErrorAttributes; + readonly level: "debug" | "info" | "warn" | "error"; + readonly msg: string; + readonly attrs?: Attributes | ErrorAttributes; } function fakeLogger(): Logger & { readonly entries: readonly LogEntry[] } { - const entries: LogEntry[] = []; - return { - entries, - debug(msg, attrs) { - entries.push({ level: "debug", msg, ...(attrs !== undefined ? { attrs } : {}) }); - }, - info(msg, attrs) { - entries.push({ level: "info", msg, ...(attrs !== undefined ? { attrs } : {}) }); - }, - warn(msg, attrs) { - entries.push({ level: "warn", msg, ...(attrs !== undefined ? { attrs } : {}) }); - }, - error(msg, attrs) { - entries.push({ level: "error", msg, ...(attrs !== undefined ? { attrs } : {}) }); - }, - child() { - return fakeLogger(); - }, - span() { - return { - id: "fake-span", - log: fakeLogger(), - setAttributes() {}, - addLink() {}, - child() { - return this; - }, - end() {}, - }; - }, - }; + const entries: LogEntry[] = []; + return { + entries, + debug(msg, attrs) { + entries.push({ level: "debug", msg, ...(attrs !== undefined ? { attrs } : {}) }); + }, + info(msg, attrs) { + entries.push({ level: "info", msg, ...(attrs !== undefined ? { attrs } : {}) }); + }, + warn(msg, attrs) { + entries.push({ level: "warn", msg, ...(attrs !== undefined ? { attrs } : {}) }); + }, + error(msg, attrs) { + entries.push({ level: "error", msg, ...(attrs !== undefined ? { attrs } : {}) }); + }, + child() { + return fakeLogger(); + }, + span() { + return { + id: "fake-span", + log: fakeLogger(), + setAttributes() {}, + addLink() {}, + child() { + return this; + }, + end() {}, + }; + }, + }; } // ── Fake registry (same pattern as router.test.ts) ────────────────────────── function fakeProvider(id: string, title?: string): SurfaceProvider { - const catalogEntry: SurfaceCatalogEntry = { - id, - region: "default", - title: title ?? `Surface ${id}`, - }; - return { - catalogEntry, - getSpec(_context?: SurfaceContext): SurfaceSpec { - return { - id, - region: "default", - title: catalogEntry.title, - fields: [], - }; - }, - invoke(_actionId: string, _payload?: unknown, _context?: SurfaceContext) {}, - }; + const catalogEntry: SurfaceCatalogEntry = { + id, + region: "default", + title: title ?? `Surface ${id}`, + }; + return { + catalogEntry, + getSpec(_context?: SurfaceContext): SurfaceSpec { + return { + id, + region: "default", + title: catalogEntry.title, + fields: [], + }; + }, + invoke(_actionId: string, _payload?: unknown, _context?: SurfaceContext) {}, + }; } function fakeRegistry(providers: readonly SurfaceProvider[]): SurfaceRegistry { - const map = new Map(providers.map((p) => [p.catalogEntry.id, p])); - return { - register(_provider: SurfaceProvider) { - return () => {}; - }, - getCatalog() { - return [...map.values()].map((p) => p.catalogEntry); - }, - getSurface(id: string) { - return map.get(id); - }, - }; + const map = new Map(providers.map((p) => [p.catalogEntry.id, p])); + return { + register(_provider: SurfaceProvider) { + return () => {}; + }, + getCatalog() { + return [...map.values()].map((p) => p.catalogEntry); + }, + getSurface(id: string) { + return map.get(id); + }, + }; } // ── Fake SessionOrchestrator (DI at the transport edge, not a vi.mock) ────── interface FakeOrchestratorOpts { - /** Pre-registered listeners per conversation (for test assertions). */ - readonly listeners?: Map<string, Set<TurnEventListener>>; - /** Events to replay on subscribe (simulates buffered in-flight events). */ - readonly bufferedEvents?: Map<string, readonly AgentEvent[]>; - /** Custom startTurn impl. */ - readonly startTurn?: SessionOrchestrator["startTurn"]; - /** If true, startTurn always returns already-active. */ - readonly alreadyActive?: boolean; - /** Custom enqueue impl. */ - readonly enqueue?: SessionOrchestrator["enqueue"]; - /** If true, enqueue reports the conversation was active (startedTurn:false). */ - readonly queueActive?: boolean; + /** Pre-registered listeners per conversation (for test assertions). */ + readonly listeners?: Map<string, Set<TurnEventListener>>; + /** Events to replay on subscribe (simulates buffered in-flight events). */ + readonly bufferedEvents?: Map<string, readonly AgentEvent[]>; + /** Custom startTurn impl. */ + readonly startTurn?: SessionOrchestrator["startTurn"]; + /** If true, startTurn always returns already-active. */ + readonly alreadyActive?: boolean; + /** Custom enqueue impl. */ + readonly enqueue?: SessionOrchestrator["enqueue"]; + /** If true, enqueue reports the conversation was active (startedTurn:false). */ + readonly queueActive?: boolean; } function fakeOrchestrator(opts?: FakeOrchestratorOpts): SessionOrchestrator & { - readonly listeners: Map<string, Set<TurnEventListener>>; - readonly startCalls: readonly { conversationId: string; text: string; workspaceId?: string }[]; - readonly enqueueCalls: readonly { conversationId: string; text: string; workspaceId?: string }[]; - readonly aborted: boolean; + readonly listeners: Map<string, Set<TurnEventListener>>; + readonly startCalls: readonly { conversationId: string; text: string; workspaceId?: string }[]; + readonly enqueueCalls: readonly { conversationId: string; text: string; workspaceId?: string }[]; + readonly aborted: boolean; } { - const listeners = opts?.listeners ?? new Map<string, Set<TurnEventListener>>(); - const bufferedEvents = opts?.bufferedEvents ?? new Map<string, readonly AgentEvent[]>(); - const startCalls: { conversationId: string; text: string }[] = []; - const enqueueCalls: { conversationId: string; text: string }[] = []; - const aborted = false; - - return { - listeners, - get startCalls() { - return startCalls; - }, - get enqueueCalls() { - return enqueueCalls; - }, - get aborted() { - return aborted; - }, - startTurn(input) { - startCalls.push({ - conversationId: input.conversationId, - text: input.text, - ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), - }); - if (opts?.startTurn) { - return opts.startTurn(input); - } - if (opts?.alreadyActive) { - return { started: false, reason: "already-active" }; - } - return { started: true, turnId: "fake-turn-id" }; - }, - enqueue(input) { - enqueueCalls.push({ - conversationId: input.conversationId, - text: input.text, - ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), - }); - if (opts?.enqueue) { - return opts.enqueue(input); - } - if (opts?.queueActive) { - return { startedTurn: false, queue: [] }; - } - return { startedTurn: true, queue: [] }; - }, - subscribe(conversationId, listener) { - let set = listeners.get(conversationId); - if (!set) { - set = new Set(); - listeners.set(conversationId, set); - } - // Replay buffered events (late-join). - const buffered = bufferedEvents.get(conversationId); - if (buffered) { - for (const event of buffered) { - listener(event); - } - } - set.add(listener); - return () => { - set.delete(listener); - }; - }, - isActive(conversationId) { - return listeners.has(conversationId); - }, - closeConversation() { - return { abortedTurn: false }; - }, - stopTurn() { - return { abortedTurn: false }; - }, - async handleMessage(_input) { - // Not used by the new transport-ws, but kept for interface compat. - }, - }; + const listeners = opts?.listeners ?? new Map<string, Set<TurnEventListener>>(); + const bufferedEvents = opts?.bufferedEvents ?? new Map<string, readonly AgentEvent[]>(); + const startCalls: { conversationId: string; text: string }[] = []; + const enqueueCalls: { conversationId: string; text: string }[] = []; + const aborted = false; + + return { + listeners, + get startCalls() { + return startCalls; + }, + get enqueueCalls() { + return enqueueCalls; + }, + get aborted() { + return aborted; + }, + startTurn(input) { + startCalls.push({ + conversationId: input.conversationId, + text: input.text, + ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), + }); + if (opts?.startTurn) { + return opts.startTurn(input); + } + if (opts?.alreadyActive) { + return { started: false, reason: "already-active" }; + } + return { started: true, turnId: "fake-turn-id" }; + }, + enqueue(input) { + enqueueCalls.push({ + conversationId: input.conversationId, + text: input.text, + ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), + }); + if (opts?.enqueue) { + return opts.enqueue(input); + } + if (opts?.queueActive) { + return { startedTurn: false, queue: [] }; + } + return { startedTurn: true, queue: [] }; + }, + subscribe(conversationId, listener) { + let set = listeners.get(conversationId); + if (!set) { + set = new Set(); + listeners.set(conversationId, set); + } + // Replay buffered events (late-join). + const buffered = bufferedEvents.get(conversationId); + if (buffered) { + for (const event of buffered) { + listener(event); + } + } + set.add(listener); + return () => { + set.delete(listener); + }; + }, + isActive(conversationId) { + return listeners.has(conversationId); + }, + closeConversation() { + return { abortedTurn: false }; + }, + stopTurn() { + return { abortedTurn: false }; + }, + async handleMessage(_input) { + // Not used by the new transport-ws, but kept for interface compat. + }, + }; } /** Create a fake orchestrator that broadcasts events when `broadcast` is called. */ function fakeOrchestratorWithBroadcast(): SessionOrchestrator & { - readonly listeners: Map<string, Set<TurnEventListener>>; - readonly enqueueCalls: readonly { conversationId: string; text: string; workspaceId?: string }[]; - broadcast(conversationId: string, event: AgentEvent): void; + readonly listeners: Map<string, Set<TurnEventListener>>; + readonly enqueueCalls: readonly { conversationId: string; text: string; workspaceId?: string }[]; + broadcast(conversationId: string, event: AgentEvent): void; } { - const listeners = new Map<string, Set<TurnEventListener>>(); - const enqueueCalls: { conversationId: string; text: string }[] = []; - - return { - listeners, - enqueueCalls, - broadcast(conversationId, event) { - const set = listeners.get(conversationId); - if (set) { - for (const listener of set) { - listener(event); - } - } - }, - startTurn(_input) { - return { started: true, turnId: "fake-turn-id" }; - }, - enqueue(input) { - enqueueCalls.push({ - conversationId: input.conversationId, - text: input.text, - ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), - }); - return { startedTurn: true, queue: [] }; - }, - subscribe(conversationId, listener) { - let set = listeners.get(conversationId); - if (!set) { - set = new Set(); - listeners.set(conversationId, set); - } - set.add(listener); - return () => { - set.delete(listener); - }; - }, - isActive(conversationId) { - return listeners.has(conversationId); - }, - closeConversation() { - return { abortedTurn: false }; - }, - stopTurn() { - return { abortedTurn: false }; - }, - async handleMessage(_input) {}, - }; + const listeners = new Map<string, Set<TurnEventListener>>(); + const enqueueCalls: { conversationId: string; text: string }[] = []; + + return { + listeners, + enqueueCalls, + broadcast(conversationId, event) { + const set = listeners.get(conversationId); + if (set) { + for (const listener of set) { + listener(event); + } + } + }, + startTurn(_input) { + return { started: true, turnId: "fake-turn-id" }; + }, + enqueue(input) { + enqueueCalls.push({ + conversationId: input.conversationId, + text: input.text, + ...(input.workspaceId !== undefined ? { workspaceId: input.workspaceId } : {}), + }); + return { startedTurn: true, queue: [] }; + }, + subscribe(conversationId, listener) { + let set = listeners.get(conversationId); + if (!set) { + set = new Set(); + listeners.set(conversationId, set); + } + set.add(listener); + return () => { + set.delete(listener); + }; + }, + isActive(conversationId) { + return listeners.has(conversationId); + }, + closeConversation() { + return { abortedTurn: false }; + }, + stopTurn() { + return { abortedTurn: false }; + }, + async handleMessage(_input) {}, + }; } // ── Per-connection state (mirrors extension.ts) ───────────────────────────── interface ConnectionState { - readonly subs: Set<string>; - readonly providerDisposers: Map<string, () => void>; - readonly chatSubscriptions: Map<string, () => void>; + readonly subs: Set<string>; + readonly providerDisposers: Map<string, () => void>; + readonly chatSubscriptions: Map<string, () => void>; } // ── Server helper ─────────────────────────────────────────────────────────── function startServer( - registry: SurfaceRegistry, - orchestrator: SessionOrchestrator, - port = 0, - logger?: Logger, + registry: SurfaceRegistry, + orchestrator: SessionOrchestrator, + port = 0, + logger?: Logger, ) { - const log = logger ?? fakeLogger(); - const connections = new Set<Bun.ServerWebSocket<ConnectionState>>(); - - /** Broadcast a message to every connected client (mirrors extension.ts). */ - function broadcast(msg: WsServerMessage): void { - for (const ws of connections) { - try { - ws.send(JSON.stringify(msg)); - } catch { - // Connection may have been dropped; swallow. - } - } - } - - const server = Bun.serve<ConnectionState>({ - port, - fetch(req, srv) { - const initial: ConnectionState = { - subs: new Set(), - providerDisposers: new Map(), - chatSubscriptions: new Map(), - }; - if (srv.upgrade(req, { data: initial })) return; - return new Response("expected websocket", { status: 426 }); - }, - websocket: { - open(ws) { - connections.add(ws); - log.debug("transport-ws: connection open"); - ws.send(JSON.stringify(catalogMessage(registry))); - }, - - message(ws, raw) { - const state = ws.data; - if (!state) return; - - let parsed: SurfaceClientMessage; - try { - parsed = JSON.parse(String(raw)) as SurfaceClientMessage; - } catch { - ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" })); - return; - } - - const result = routeClientMessage(registry, state.subs, parsed); - - switch (result.kind) { - case "surface": { - for (const reply of result.replies) { - if (reply.type === "error") { - log.warn?.("transport-ws: surface-op error", { - ...(reply.surfaceId !== undefined ? { surfaceId: reply.surfaceId } : {}), - reason: reply.message, - }); - } - } - - if (result.subChange) { - const key = subKey(result.subChange.surfaceId, result.subChange.conversationId); - if (result.subChange.op === "add") { - state.subs.add(key); - } else { - state.subs.delete(key); - } - } - - for (const reply of result.replies) { - ws.send(JSON.stringify(reply)); - } - break; - } - - case "chat": { - const resolvedId = result.conversationId ?? crypto.randomUUID(); - // Auto-subscribe the sender. - if (!state.chatSubscriptions.has(resolvedId)) { - const unsubscribe = orchestrator.subscribe(resolvedId, (event) => { - ws.send(JSON.stringify({ type: "chat.delta", event })); - }); - state.chatSubscriptions.set(resolvedId, unsubscribe); - } - // Start the turn detached. - const startResult = orchestrator.startTurn({ - conversationId: resolvedId, - text: result.message, - ...(result.model !== undefined ? { modelName: result.model } : {}), - ...(result.cwd !== undefined ? { cwd: result.cwd } : {}), - ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), - }); - if (!startResult.started) { - ws.send( - JSON.stringify({ - type: "chat.error", - conversationId: resolvedId, - message: "a turn is already generating for this conversation", - }), - ); - } else { - log.info?.("transport-ws: chat.send accepted", { - conversationId: resolvedId, - model: result.model ?? null, - }); - } - break; - } - - case "chat-subscribe": { - if (!state.chatSubscriptions.has(result.conversationId)) { - const unsubscribe = orchestrator.subscribe(result.conversationId, (event) => { - ws.send(JSON.stringify({ type: "chat.delta", event })); - }); - state.chatSubscriptions.set(result.conversationId, unsubscribe); - } - break; - } - - case "chat-unsubscribe": { - const dispose = state.chatSubscriptions.get(result.conversationId); - if (dispose) { - dispose(); - state.chatSubscriptions.delete(result.conversationId); - } - break; - } - - case "chat-queue": { - // Mirror extension.ts: fire-and-forget. On startedTurn:true - // auto-subscribe the sender (deltas stream); on false emit - // nothing back. - const enqueueResult = orchestrator.enqueue({ - conversationId: result.conversationId, - text: result.text, - ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), - }); - if (enqueueResult.startedTurn) { - if (!state.chatSubscriptions.has(result.conversationId)) { - const unsubscribe = orchestrator.subscribe(result.conversationId, (event) => { - ws.send(JSON.stringify({ type: "chat.delta", event })); - }); - state.chatSubscriptions.set(result.conversationId, unsubscribe); - } - } - log.info?.("transport-ws: chat.queue accepted", { - conversationId: result.conversationId, - startedTurn: enqueueResult.startedTurn, - }); - break; - } - - case "chat-error": { - log.warn?.("transport-ws: malformed chat.send", { - reason: result.errorMessage, - ...(result.conversationId !== undefined - ? { conversationId: result.conversationId } - : {}), - }); - ws.send( - JSON.stringify({ - type: "chat.error", - conversationId: result.conversationId, - message: result.errorMessage, - }), - ); - break; - } - } - }, - - close(ws) { - connections.delete(ws); - const state = ws.data; - if (state) { - for (const dispose of state.chatSubscriptions.values()) { - dispose(); - } - state.chatSubscriptions.clear(); - for (const dispose of state.providerDisposers.values()) { - dispose(); - } - } - log.debug("transport-ws: connection close"); - }, - }, - }); - - /** - * Simulate the `conversationOpened` hook firing — mirrors the - * `host.on(conversationOpened, ...)` subscription in extension.ts, which - * broadcasts a `conversation.open` WS message (carrying the conversation's - * persisted `workspaceId`) to every connected client. - */ - return Object.assign(server, { - triggerConversationOpen(conversationId: string, workspaceId: string): void { - broadcast({ type: "conversation.open", conversationId, workspaceId }); - }, - /** - * Simulate the `conversationStatusChanged` hook firing — mirrors the - * `host.on(conversationStatusChanged, ...)` subscription in extension.ts, - * which broadcasts a `conversation.statusChanged` WS message (carrying the - * conversation's persisted `workspaceId`) to every connected client. - */ - triggerConversationStatusChanged( - conversationId: string, - status: ConversationStatus, - workspaceId: string, - ): void { - broadcast({ - type: "conversation.statusChanged", - conversationId, - status, - workspaceId, - }); - }, - }); + const log = logger ?? fakeLogger(); + const connections = new Set<Bun.ServerWebSocket<ConnectionState>>(); + + /** Broadcast a message to every connected client (mirrors extension.ts). */ + function broadcast(msg: WsServerMessage): void { + for (const ws of connections) { + try { + ws.send(JSON.stringify(msg)); + } catch { + // Connection may have been dropped; swallow. + } + } + } + + const server = Bun.serve<ConnectionState>({ + port, + fetch(req, srv) { + const initial: ConnectionState = { + subs: new Set(), + providerDisposers: new Map(), + chatSubscriptions: new Map(), + }; + if (srv.upgrade(req, { data: initial })) return; + return new Response("expected websocket", { status: 426 }); + }, + websocket: { + open(ws) { + connections.add(ws); + log.debug("transport-ws: connection open"); + ws.send(JSON.stringify(catalogMessage(registry))); + }, + + message(ws, raw) { + const state = ws.data; + if (!state) return; + + let parsed: SurfaceClientMessage; + try { + parsed = JSON.parse(String(raw)) as SurfaceClientMessage; + } catch { + ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" })); + return; + } + + const result = routeClientMessage(registry, state.subs, parsed); + + switch (result.kind) { + case "surface": { + for (const reply of result.replies) { + if (reply.type === "error") { + log.warn?.("transport-ws: surface-op error", { + ...(reply.surfaceId !== undefined ? { surfaceId: reply.surfaceId } : {}), + reason: reply.message, + }); + } + } + + if (result.subChange) { + const key = subKey(result.subChange.surfaceId, result.subChange.conversationId); + if (result.subChange.op === "add") { + state.subs.add(key); + } else { + state.subs.delete(key); + } + } + + for (const reply of result.replies) { + ws.send(JSON.stringify(reply)); + } + break; + } + + case "chat": { + const resolvedId = result.conversationId ?? crypto.randomUUID(); + // Auto-subscribe the sender. + if (!state.chatSubscriptions.has(resolvedId)) { + const unsubscribe = orchestrator.subscribe(resolvedId, (event) => { + ws.send(JSON.stringify({ type: "chat.delta", event })); + }); + state.chatSubscriptions.set(resolvedId, unsubscribe); + } + // Start the turn detached. + const startResult = orchestrator.startTurn({ + conversationId: resolvedId, + text: result.message, + ...(result.model !== undefined ? { modelName: result.model } : {}), + ...(result.cwd !== undefined ? { cwd: result.cwd } : {}), + ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), + }); + if (!startResult.started) { + ws.send( + JSON.stringify({ + type: "chat.error", + conversationId: resolvedId, + message: "a turn is already generating for this conversation", + }), + ); + } else { + log.info?.("transport-ws: chat.send accepted", { + conversationId: resolvedId, + model: result.model ?? null, + }); + } + break; + } + + case "chat-subscribe": { + if (!state.chatSubscriptions.has(result.conversationId)) { + const unsubscribe = orchestrator.subscribe(result.conversationId, (event) => { + ws.send(JSON.stringify({ type: "chat.delta", event })); + }); + state.chatSubscriptions.set(result.conversationId, unsubscribe); + } + break; + } + + case "chat-unsubscribe": { + const dispose = state.chatSubscriptions.get(result.conversationId); + if (dispose) { + dispose(); + state.chatSubscriptions.delete(result.conversationId); + } + break; + } + + case "chat-queue": { + // Mirror extension.ts: fire-and-forget. On startedTurn:true + // auto-subscribe the sender (deltas stream); on false emit + // nothing back. + const enqueueResult = orchestrator.enqueue({ + conversationId: result.conversationId, + text: result.text, + ...(result.workspaceId !== undefined ? { workspaceId: result.workspaceId } : {}), + }); + if (enqueueResult.startedTurn) { + if (!state.chatSubscriptions.has(result.conversationId)) { + const unsubscribe = orchestrator.subscribe(result.conversationId, (event) => { + ws.send(JSON.stringify({ type: "chat.delta", event })); + }); + state.chatSubscriptions.set(result.conversationId, unsubscribe); + } + } + log.info?.("transport-ws: chat.queue accepted", { + conversationId: result.conversationId, + startedTurn: enqueueResult.startedTurn, + }); + break; + } + + case "chat-error": { + log.warn?.("transport-ws: malformed chat.send", { + reason: result.errorMessage, + ...(result.conversationId !== undefined + ? { conversationId: result.conversationId } + : {}), + }); + ws.send( + JSON.stringify({ + type: "chat.error", + conversationId: result.conversationId, + message: result.errorMessage, + }), + ); + break; + } + } + }, + + close(ws) { + connections.delete(ws); + const state = ws.data; + if (state) { + for (const dispose of state.chatSubscriptions.values()) { + dispose(); + } + state.chatSubscriptions.clear(); + for (const dispose of state.providerDisposers.values()) { + dispose(); + } + } + log.debug("transport-ws: connection close"); + }, + }, + }); + + /** + * Simulate the `conversationOpened` hook firing — mirrors the + * `host.on(conversationOpened, ...)` subscription in extension.ts, which + * broadcasts a `conversation.open` WS message (carrying the conversation's + * persisted `workspaceId`) to every connected client. + */ + return Object.assign(server, { + triggerConversationOpen(conversationId: string, workspaceId: string): void { + broadcast({ type: "conversation.open", conversationId, workspaceId }); + }, + /** + * Simulate the `conversationStatusChanged` hook firing — mirrors the + * `host.on(conversationStatusChanged, ...)` subscription in extension.ts, + * which broadcasts a `conversation.statusChanged` WS message (carrying the + * conversation's persisted `workspaceId`) to every connected client. + */ + triggerConversationStatusChanged( + conversationId: string, + status: ConversationStatus, + workspaceId: string, + ): void { + broadcast({ + type: "conversation.statusChanged", + conversationId, + status, + workspaceId, + }); + }, + }); } // ── Helpers ───────────────────────────────────────────────────────────────── function waitForMessage(ws: WebSocket): Promise<WsServerMessage> { - return new Promise((resolve, reject) => { - const timeout = setTimeout(() => reject(new Error("timed out waiting for message")), 5000); - function handler(ev: MessageEvent) { - clearTimeout(timeout); - ws.removeEventListener("message", handler); - resolve(JSON.parse(ev.data as string) as WsServerMessage); - } - ws.addEventListener("message", handler); - }); + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error("timed out waiting for message")), 5000); + function handler(ev: MessageEvent) { + clearTimeout(timeout); + ws.removeEventListener("message", handler); + resolve(JSON.parse(ev.data as string) as WsServerMessage); + } + ws.addEventListener("message", handler); + }); } function waitForMessages(ws: WebSocket, count: number): Promise<WsServerMessage[]> { - return new Promise((resolve, reject) => { - const msgs: WsServerMessage[] = []; - const timeout = setTimeout( - () => reject(new Error(`timed out waiting for ${count} messages (got ${msgs.length})`)), - 5000, - ); - function handler(ev: MessageEvent) { - msgs.push(JSON.parse(ev.data as string) as WsServerMessage); - if (msgs.length === count) { - clearTimeout(timeout); - ws.removeEventListener("message", handler); - resolve(msgs); - } - } - ws.addEventListener("message", handler); - }); + return new Promise((resolve, reject) => { + const msgs: WsServerMessage[] = []; + const timeout = setTimeout( + () => reject(new Error(`timed out waiting for ${count} messages (got ${msgs.length})`)), + 5000, + ); + function handler(ev: MessageEvent) { + msgs.push(JSON.parse(ev.data as string) as WsServerMessage); + if (msgs.length === count) { + clearTimeout(timeout); + ws.removeEventListener("message", handler); + resolve(msgs); + } + } + ws.addEventListener("message", handler); + }); } // ── Tests ─────────────────────────────────────────────────────────────────── describe("Bun.serve WebSocket server", () => { - let server: ReturnType<typeof Bun.serve>; - let port: number; - const defaultOrchestrator = fakeOrchestrator(); - - beforeEach(() => { - const provider = fakeProvider("demo", "Demo Surface"); - const registry = fakeRegistry([provider]); - server = startServer(registry, defaultOrchestrator); - port = server.port as number; - }); - - afterEach(() => { - server.stop(); - }); - - test("performs WebSocket upgrade (returns 101)", async () => { - const ws = new WebSocket(`ws://localhost:${port}`); - const msg = await waitForMessage(ws); - expect(msg.type).toBe("catalog"); - ws.close(); - }); - - test("sends catalog on open", async () => { - const ws = new WebSocket(`ws://localhost:${port}`); - const msg = await waitForMessage(ws); - expect(msg).toEqual({ - type: "catalog", - catalog: [{ id: "demo", region: "default", title: "Demo Surface" }], - }); - ws.close(); - }); - - test("subscribe returns surface spec", async () => { - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "subscribe", surfaceId: "demo" })); - const msg = await waitForMessage(ws); - - expect(msg.type).toBe("surface"); - if (msg.type === "surface") { - expect(msg.spec.id).toBe("demo"); - expect(msg.spec.title).toBe("Demo Surface"); - } - ws.close(); - }); - - test("subscribe to unknown surface returns error", async () => { - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "subscribe", surfaceId: "nope" })); - const msg = await waitForMessage(ws); - - expect(msg).toEqual({ - type: "error", - surfaceId: "nope", - message: "Unknown surface: nope", - }); - ws.close(); - }); - - test("non-WebSocket request returns 426", async () => { - const res = await fetch(`http://localhost:${port}/`); - expect(res.status).toBe(426); - expect(await res.text()).toBe("expected websocket"); - }); + let server: ReturnType<typeof Bun.serve>; + let port: number; + const defaultOrchestrator = fakeOrchestrator(); + + beforeEach(() => { + const provider = fakeProvider("demo", "Demo Surface"); + const registry = fakeRegistry([provider]); + server = startServer(registry, defaultOrchestrator); + port = server.port as number; + }); + + afterEach(() => { + server.stop(); + }); + + test("performs WebSocket upgrade (returns 101)", async () => { + const ws = new WebSocket(`ws://localhost:${port}`); + const msg = await waitForMessage(ws); + expect(msg.type).toBe("catalog"); + ws.close(); + }); + + test("sends catalog on open", async () => { + const ws = new WebSocket(`ws://localhost:${port}`); + const msg = await waitForMessage(ws); + expect(msg).toEqual({ + type: "catalog", + catalog: [{ id: "demo", region: "default", title: "Demo Surface" }], + }); + ws.close(); + }); + + test("subscribe returns surface spec", async () => { + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "subscribe", surfaceId: "demo" })); + const msg = await waitForMessage(ws); + + expect(msg.type).toBe("surface"); + if (msg.type === "surface") { + expect(msg.spec.id).toBe("demo"); + expect(msg.spec.title).toBe("Demo Surface"); + } + ws.close(); + }); + + test("subscribe to unknown surface returns error", async () => { + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "subscribe", surfaceId: "nope" })); + const msg = await waitForMessage(ws); + + expect(msg).toEqual({ + type: "error", + surfaceId: "nope", + message: "Unknown surface: nope", + }); + ws.close(); + }); + + test("non-WebSocket request returns 426", async () => { + const res = await fetch(`http://localhost:${port}/`); + expect(res.status).toBe(426); + expect(await res.text()).toBe("expected websocket"); + }); }); describe("chat ops (new orchestrator API)", () => { - let server: ReturnType<typeof Bun.serve>; - let port: number; - - afterEach(() => { - server.stop(); - }); - - test("chat.send auto-subscribes the sender and delivers deltas via orchestrator.subscribe broadcast", async () => { - const orch = fakeOrchestratorWithBroadcast(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); - - // Give the message handler time to run. - await new Promise((r) => setTimeout(r, 50)); - - // The sender should be auto-subscribed. Broadcast an event. - const event = { - type: "text-delta", - conversationId: "c1", - turnId: "t1", - delta: "Hello", - } as AgentEvent; - orch.broadcast("c1", event); - - const msg = await waitForMessage(ws); - expect(msg.type).toBe("chat.delta"); - if (msg.type === "chat.delta") { - expect(msg.event).toEqual(event); - } - - ws.close(); - }); - - test("chat.send with already-active turn sends chat.error but keeps subscription", async () => { - const orch = fakeOrchestrator({ alreadyActive: true }); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); - const errMsg = await waitForMessage(ws); - - expect(errMsg.type).toBe("chat.error"); - if (errMsg.type === "chat.error") { - expect(errMsg.message).toBe("a turn is already generating for this conversation"); - expect(errMsg.conversationId).toBe("c1"); - } - - ws.close(); - }); - - test("chat.send threads workspaceId — orchestrator receives it", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send( - JSON.stringify({ - type: "chat.send", - conversationId: "c1", - message: "hello workspace", - workspaceId: "my-workspace", - }), - ); - await new Promise((r) => setTimeout(r, 50)); - - expect(orch.startCalls).toHaveLength(1); - expect(orch.startCalls[0]?.workspaceId).toBe("my-workspace"); - - ws.close(); - }); - - test("chat.send defaults workspaceId when omitted — orchestrator receives undefined", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send( - JSON.stringify({ - type: "chat.send", - conversationId: "c1", - message: "hello no workspace", - }), - ); - await new Promise((r) => setTimeout(r, 50)); - - expect(orch.startCalls).toHaveLength(1); - expect(orch.startCalls[0]).not.toHaveProperty("workspaceId"); - - ws.close(); - }); - - test("multi-client fan-out — two connections both subscribe the same conversation", async () => { - const orch = fakeOrchestratorWithBroadcast(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws1 = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws1); // drain catalog - const ws2 = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws2); // drain catalog - - // Both subscribe to the same conversation. - ws1.send(JSON.stringify({ type: "chat.subscribe", conversationId: "shared-conv" })); - ws2.send(JSON.stringify({ type: "chat.subscribe", conversationId: "shared-conv" })); - await new Promise((r) => setTimeout(r, 50)); - - // Broadcast an event. - const event = { - type: "text-delta", - conversationId: "shared-conv", - turnId: "t1", - delta: "Hi both", - } as AgentEvent; - orch.broadcast("shared-conv", event); - - const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]); - - expect(msg1.type).toBe("chat.delta"); - expect(msg2.type).toBe("chat.delta"); - if (msg1.type === "chat.delta" && msg2.type === "chat.delta") { - expect(msg1.event).toEqual(event); - expect(msg2.event).toEqual(event); - } - - ws1.close(); - ws2.close(); - }); - - test("disconnect does NOT abort the turn", async () => { - const orch = fakeOrchestratorWithBroadcast(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - // Start a turn. - ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); - await new Promise((r) => setTimeout(r, 50)); - - // Close the socket. - ws.close(); - await new Promise((r) => setTimeout(r, 50)); - - // The turn should still be "running" — the orchestrator was never told to abort. - // Broadcast a post-close event; the fake still has the listener set (real orchestrator - // would too until turn-sealed). We just verify no abort was invoked. - // The fakeOrchestratorWithBroadcast has no abort mechanism — that's the point: - // the transport never calls abort on disconnect. - expect(true).toBe(true); // If we got here, no abort was attempted. - }); - - test("late-join replay forwarded — subscribe mid-turn receives buffered events", async () => { - const bufferedEvents: AgentEvent[] = [ - { type: "turn-start", conversationId: "c1", turnId: "t1" } as AgentEvent, - { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "Hel" } as AgentEvent, - ]; - const bufferedMap = new Map<string, readonly AgentEvent[]>(); - bufferedMap.set("c1", bufferedEvents); - - const orch = fakeOrchestrator({ bufferedEvents: bufferedMap }); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - // Subscribe mid-turn — the fake orchestrator replays buffered events. - ws.send(JSON.stringify({ type: "chat.subscribe", conversationId: "c1" })); - - const msgs = await waitForMessages(ws, bufferedEvents.length); - - for (let i = 0; i < bufferedEvents.length; i++) { - const msg = msgs[i]; - const expected = bufferedEvents[i]; - if (!msg || !expected) throw new Error(`missing at index ${i}`); - expect(msg.type).toBe("chat.delta"); - if (msg.type === "chat.delta") { - expect(msg.event).toEqual(expected); - } - } - - ws.close(); - }); - - test("chat.send auto-subscribes the sender — deltas arrive without separate chat.subscribe", async () => { - const orch = fakeOrchestratorWithBroadcast(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - // Send without a separate chat.subscribe. - ws.send(JSON.stringify({ type: "chat.send", conversationId: "auto-conv", message: "go" })); - await new Promise((r) => setTimeout(r, 50)); - - // Broadcast should reach the sender. - const event = { type: "done", conversationId: "auto-conv", turnId: "t1" } as AgentEvent; - orch.broadcast("auto-conv", event); - - const msg = await waitForMessage(ws); - expect(msg.type).toBe("chat.delta"); - if (msg.type === "chat.delta") { - expect(msg.event).toEqual(event); - } - - ws.close(); - }); + let server: ReturnType<typeof Bun.serve>; + let port: number; + + afterEach(() => { + server.stop(); + }); + + test("chat.send auto-subscribes the sender and delivers deltas via orchestrator.subscribe broadcast", async () => { + const orch = fakeOrchestratorWithBroadcast(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); + + // Give the message handler time to run. + await new Promise((r) => setTimeout(r, 50)); + + // The sender should be auto-subscribed. Broadcast an event. + const event = { + type: "text-delta", + conversationId: "c1", + turnId: "t1", + delta: "Hello", + } as AgentEvent; + orch.broadcast("c1", event); + + const msg = await waitForMessage(ws); + expect(msg.type).toBe("chat.delta"); + if (msg.type === "chat.delta") { + expect(msg.event).toEqual(event); + } + + ws.close(); + }); + + test("chat.send with already-active turn sends chat.error but keeps subscription", async () => { + const orch = fakeOrchestrator({ alreadyActive: true }); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); + const errMsg = await waitForMessage(ws); + + expect(errMsg.type).toBe("chat.error"); + if (errMsg.type === "chat.error") { + expect(errMsg.message).toBe("a turn is already generating for this conversation"); + expect(errMsg.conversationId).toBe("c1"); + } + + ws.close(); + }); + + test("chat.send threads workspaceId — orchestrator receives it", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send( + JSON.stringify({ + type: "chat.send", + conversationId: "c1", + message: "hello workspace", + workspaceId: "my-workspace", + }), + ); + await new Promise((r) => setTimeout(r, 50)); + + expect(orch.startCalls).toHaveLength(1); + expect(orch.startCalls[0]?.workspaceId).toBe("my-workspace"); + + ws.close(); + }); + + test("chat.send defaults workspaceId when omitted — orchestrator receives undefined", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send( + JSON.stringify({ + type: "chat.send", + conversationId: "c1", + message: "hello no workspace", + }), + ); + await new Promise((r) => setTimeout(r, 50)); + + expect(orch.startCalls).toHaveLength(1); + expect(orch.startCalls[0]).not.toHaveProperty("workspaceId"); + + ws.close(); + }); + + test("multi-client fan-out — two connections both subscribe the same conversation", async () => { + const orch = fakeOrchestratorWithBroadcast(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws1 = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws1); // drain catalog + const ws2 = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws2); // drain catalog + + // Both subscribe to the same conversation. + ws1.send(JSON.stringify({ type: "chat.subscribe", conversationId: "shared-conv" })); + ws2.send(JSON.stringify({ type: "chat.subscribe", conversationId: "shared-conv" })); + await new Promise((r) => setTimeout(r, 50)); + + // Broadcast an event. + const event = { + type: "text-delta", + conversationId: "shared-conv", + turnId: "t1", + delta: "Hi both", + } as AgentEvent; + orch.broadcast("shared-conv", event); + + const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]); + + expect(msg1.type).toBe("chat.delta"); + expect(msg2.type).toBe("chat.delta"); + if (msg1.type === "chat.delta" && msg2.type === "chat.delta") { + expect(msg1.event).toEqual(event); + expect(msg2.event).toEqual(event); + } + + ws1.close(); + ws2.close(); + }); + + test("disconnect does NOT abort the turn", async () => { + const orch = fakeOrchestratorWithBroadcast(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + // Start a turn. + ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); + await new Promise((r) => setTimeout(r, 50)); + + // Close the socket. + ws.close(); + await new Promise((r) => setTimeout(r, 50)); + + // The turn should still be "running" — the orchestrator was never told to abort. + // Broadcast a post-close event; the fake still has the listener set (real orchestrator + // would too until turn-sealed). We just verify no abort was invoked. + // The fakeOrchestratorWithBroadcast has no abort mechanism — that's the point: + // the transport never calls abort on disconnect. + expect(true).toBe(true); // If we got here, no abort was attempted. + }); + + test("late-join replay forwarded — subscribe mid-turn receives buffered events", async () => { + const bufferedEvents: AgentEvent[] = [ + { type: "turn-start", conversationId: "c1", turnId: "t1" } as AgentEvent, + { type: "text-delta", conversationId: "c1", turnId: "t1", delta: "Hel" } as AgentEvent, + ]; + const bufferedMap = new Map<string, readonly AgentEvent[]>(); + bufferedMap.set("c1", bufferedEvents); + + const orch = fakeOrchestrator({ bufferedEvents: bufferedMap }); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + // Subscribe mid-turn — the fake orchestrator replays buffered events. + ws.send(JSON.stringify({ type: "chat.subscribe", conversationId: "c1" })); + + const msgs = await waitForMessages(ws, bufferedEvents.length); + + for (let i = 0; i < bufferedEvents.length; i++) { + const msg = msgs[i]; + const expected = bufferedEvents[i]; + if (!msg || !expected) throw new Error(`missing at index ${i}`); + expect(msg.type).toBe("chat.delta"); + if (msg.type === "chat.delta") { + expect(msg.event).toEqual(expected); + } + } + + ws.close(); + }); + + test("chat.send auto-subscribes the sender — deltas arrive without separate chat.subscribe", async () => { + const orch = fakeOrchestratorWithBroadcast(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + // Send without a separate chat.subscribe. + ws.send(JSON.stringify({ type: "chat.send", conversationId: "auto-conv", message: "go" })); + await new Promise((r) => setTimeout(r, 50)); + + // Broadcast should reach the sender. + const event = { type: "done", conversationId: "auto-conv", turnId: "t1" } as AgentEvent; + orch.broadcast("auto-conv", event); + + const msg = await waitForMessage(ws); + expect(msg.type).toBe("chat.delta"); + if (msg.type === "chat.delta") { + expect(msg.event).toEqual(event); + } + + ws.close(); + }); }); describe("chat.queue (steering enqueue)", () => { - let server: ReturnType<typeof Bun.serve>; - let port: number; - - afterEach(() => { - server.stop(); - }); - - test("chat.queue with valid text → orchestrator.enqueue called with {conversationId, text}, no reply sent", async () => { - const orch = fakeOrchestrator(); // idle → startedTurn:true - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: "steer please" })); - // Allow the message handler to run. - await new Promise((r) => setTimeout(r, 50)); - - expect(orch.enqueueCalls).toEqual([{ conversationId: "c1", text: "steer please" }]); - // chat.send-path equivalence: enqueue NOT called via startTurn. - expect(orch.startCalls).toHaveLength(0); - // Fire-and-forget: no chat.error, no ack — only the catalog was sent. - // (startedTurn:true path auto-subscribes but emits nothing itself.) - - ws.close(); - }); - - test("chat.queue threads workspaceId — orchestrator receives it", async () => { - const orch = fakeOrchestrator(); // idle → startedTurn:true - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send( - JSON.stringify({ - type: "chat.queue", - conversationId: "c1", - text: "steer here", - workspaceId: "my-workspace", - }), - ); - await new Promise((r) => setTimeout(r, 50)); - - expect(orch.enqueueCalls).toEqual([ - { conversationId: "c1", text: "steer here", workspaceId: "my-workspace" }, - ]); - - ws.close(); - }); - - test("chat.queue on startedTurn:true auto-subscribes the sender (deltas stream as chat.delta)", async () => { - const orch = fakeOrchestratorWithBroadcast(); // enqueue → startedTurn:true - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: "go" })); - await new Promise((r) => setTimeout(r, 50)); - - // The sender was auto-subscribed — a broadcast reaches it as a chat.delta. - const event = { - type: "text-delta", - conversationId: "c1", - turnId: "t1", - delta: "Hi", - } as AgentEvent; - orch.broadcast("c1", event); - - const msg = await waitForMessage(ws); - expect(msg.type).toBe("chat.delta"); - if (msg.type === "chat.delta") { - expect(msg.event).toEqual(event); - } - - ws.close(); - }); - - test("chat.queue on startedTurn:false (queued for steering) emits NOTHING back", async () => { - const orch = fakeOrchestrator({ queueActive: true }); // enqueue → startedTurn:false - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: "steer" })); - await new Promise((r) => setTimeout(r, 50)); - - expect(orch.enqueueCalls).toEqual([{ conversationId: "c1", text: "steer" }]); - // No further message should arrive within a quiet window: success is - // confirmed by the message-queue SURFACE, not a reply here. We assert - // by NOT receiving anything (a silent socket). - await expect( - Promise.race([ - waitForMessage(ws).then((m) => new Error(`unexpected reply: ${JSON.stringify(m)}`)), - new Promise((resolve) => setTimeout(() => resolve("silent"), 150)), - ]), - ).resolves.toBe("silent"); - - ws.close(); - }); - - test("chat.queue with empty text → chat.error to client, no enqueue", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: " " })); - const errMsg = await waitForMessage(ws); - - expect(errMsg.type).toBe("chat.error"); - if (errMsg.type === "chat.error") { - expect(errMsg.conversationId).toBe("c1"); - expect(errMsg.message).toContain("non-empty string"); - } - expect(orch.enqueueCalls).toHaveLength(0); - - ws.close(); - }); - - test("chat.queue with missing text → chat.error to client, no enqueue", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1" })); - const errMsg = await waitForMessage(ws); - - expect(errMsg.type).toBe("chat.error"); - if (errMsg.type === "chat.error") { - expect(errMsg.message).toContain("non-empty string"); - } - expect(orch.enqueueCalls).toHaveLength(0); - - ws.close(); - }); + let server: ReturnType<typeof Bun.serve>; + let port: number; + + afterEach(() => { + server.stop(); + }); + + test("chat.queue with valid text → orchestrator.enqueue called with {conversationId, text}, no reply sent", async () => { + const orch = fakeOrchestrator(); // idle → startedTurn:true + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: "steer please" })); + // Allow the message handler to run. + await new Promise((r) => setTimeout(r, 50)); + + expect(orch.enqueueCalls).toEqual([{ conversationId: "c1", text: "steer please" }]); + // chat.send-path equivalence: enqueue NOT called via startTurn. + expect(orch.startCalls).toHaveLength(0); + // Fire-and-forget: no chat.error, no ack — only the catalog was sent. + // (startedTurn:true path auto-subscribes but emits nothing itself.) + + ws.close(); + }); + + test("chat.queue threads workspaceId — orchestrator receives it", async () => { + const orch = fakeOrchestrator(); // idle → startedTurn:true + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send( + JSON.stringify({ + type: "chat.queue", + conversationId: "c1", + text: "steer here", + workspaceId: "my-workspace", + }), + ); + await new Promise((r) => setTimeout(r, 50)); + + expect(orch.enqueueCalls).toEqual([ + { conversationId: "c1", text: "steer here", workspaceId: "my-workspace" }, + ]); + + ws.close(); + }); + + test("chat.queue on startedTurn:true auto-subscribes the sender (deltas stream as chat.delta)", async () => { + const orch = fakeOrchestratorWithBroadcast(); // enqueue → startedTurn:true + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: "go" })); + await new Promise((r) => setTimeout(r, 50)); + + // The sender was auto-subscribed — a broadcast reaches it as a chat.delta. + const event = { + type: "text-delta", + conversationId: "c1", + turnId: "t1", + delta: "Hi", + } as AgentEvent; + orch.broadcast("c1", event); + + const msg = await waitForMessage(ws); + expect(msg.type).toBe("chat.delta"); + if (msg.type === "chat.delta") { + expect(msg.event).toEqual(event); + } + + ws.close(); + }); + + test("chat.queue on startedTurn:false (queued for steering) emits NOTHING back", async () => { + const orch = fakeOrchestrator({ queueActive: true }); // enqueue → startedTurn:false + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: "steer" })); + await new Promise((r) => setTimeout(r, 50)); + + expect(orch.enqueueCalls).toEqual([{ conversationId: "c1", text: "steer" }]); + // No further message should arrive within a quiet window: success is + // confirmed by the message-queue SURFACE, not a reply here. We assert + // by NOT receiving anything (a silent socket). + await expect( + Promise.race([ + waitForMessage(ws).then((m) => new Error(`unexpected reply: ${JSON.stringify(m)}`)), + new Promise((resolve) => setTimeout(() => resolve("silent"), 150)), + ]), + ).resolves.toBe("silent"); + + ws.close(); + }); + + test("chat.queue with empty text → chat.error to client, no enqueue", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1", text: " " })); + const errMsg = await waitForMessage(ws); + + expect(errMsg.type).toBe("chat.error"); + if (errMsg.type === "chat.error") { + expect(errMsg.conversationId).toBe("c1"); + expect(errMsg.message).toContain("non-empty string"); + } + expect(orch.enqueueCalls).toHaveLength(0); + + ws.close(); + }); + + test("chat.queue with missing text → chat.error to client, no enqueue", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.queue", conversationId: "c1" })); + const errMsg = await waitForMessage(ws); + + expect(errMsg.type).toBe("chat.error"); + if (errMsg.type === "chat.error") { + expect(errMsg.message).toContain("non-empty string"); + } + expect(orch.enqueueCalls).toHaveLength(0); + + ws.close(); + }); }); describe("logging", () => { - let server: ReturnType<typeof Bun.serve>; - let port: number; - - afterEach(() => { - server.stop(); - }); - - test("logs a warn on a surface-op error", async () => { - const logger = fakeLogger(); - const registry = fakeRegistry([]); - server = startServer(registry, fakeOrchestrator(), 0, logger); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "subscribe", surfaceId: "nonexistent" })); - await waitForMessage(ws); // drain error reply - ws.close(); - // Allow close handler to run - await new Promise((r) => setTimeout(r, 50)); - - const surfaceErrors = logger.entries.filter( - (e) => e.level === "warn" && e.msg === "transport-ws: surface-op error", - ); - expect(surfaceErrors.length).toBeGreaterThanOrEqual(1); - expect(surfaceErrors[0]?.attrs).toMatchObject({ - surfaceId: "nonexistent", - reason: "Unknown surface: nonexistent", - }); - }); - - test("logs an info when a chat.send is accepted", async () => { - const logger = fakeLogger(); - const orch = fakeOrchestrator(); - const registry = fakeRegistry([]); - server = startServer(registry, orch, 0, logger); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send( - JSON.stringify({ - type: "chat.send", - conversationId: "conv-42", - message: "hello", - model: "gpt-4", - }), - ); - // Wait for the message handler to run - await new Promise((r) => setTimeout(r, 100)); - ws.close(); - await new Promise((r) => setTimeout(r, 50)); - - const accepted = logger.entries.filter( - (e) => e.level === "info" && e.msg === "transport-ws: chat.send accepted", - ); - expect(accepted).toHaveLength(1); - expect(accepted[0]?.attrs).toMatchObject({ - conversationId: "conv-42", - model: "gpt-4", - }); - }); - - test("logs a warn on a malformed chat.send", async () => { - const logger = fakeLogger(); - const registry = fakeRegistry([]); - server = startServer(registry, fakeOrchestrator(), 0, logger); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.send", message: "" })); - await waitForMessage(ws); // drain chat.error reply - ws.close(); - await new Promise((r) => setTimeout(r, 50)); - - const malformed = logger.entries.filter( - (e) => e.level === "warn" && e.msg === "transport-ws: malformed chat.send", - ); - expect(malformed).toHaveLength(1); - expect(malformed[0]?.attrs).toMatchObject({ - reason: "chat.send requires a non-empty string `message`", - }); - }); - - test("does not log 'in-flight turn aborted' on close", async () => { - const logger = fakeLogger(); - const orch = fakeOrchestratorWithBroadcast(); - const registry = fakeRegistry([]); - server = startServer(registry, orch, 0, logger); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); - await new Promise((r) => setTimeout(r, 50)); - ws.close(); - await new Promise((r) => setTimeout(r, 50)); - - const abortLogs = logger.entries.filter( - (e) => e.msg.includes("aborted") || e.msg.includes("abort"), - ); - expect(abortLogs).toHaveLength(0); - }); + let server: ReturnType<typeof Bun.serve>; + let port: number; + + afterEach(() => { + server.stop(); + }); + + test("logs a warn on a surface-op error", async () => { + const logger = fakeLogger(); + const registry = fakeRegistry([]); + server = startServer(registry, fakeOrchestrator(), 0, logger); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "subscribe", surfaceId: "nonexistent" })); + await waitForMessage(ws); // drain error reply + ws.close(); + // Allow close handler to run + await new Promise((r) => setTimeout(r, 50)); + + const surfaceErrors = logger.entries.filter( + (e) => e.level === "warn" && e.msg === "transport-ws: surface-op error", + ); + expect(surfaceErrors.length).toBeGreaterThanOrEqual(1); + expect(surfaceErrors[0]?.attrs).toMatchObject({ + surfaceId: "nonexistent", + reason: "Unknown surface: nonexistent", + }); + }); + + test("logs an info when a chat.send is accepted", async () => { + const logger = fakeLogger(); + const orch = fakeOrchestrator(); + const registry = fakeRegistry([]); + server = startServer(registry, orch, 0, logger); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send( + JSON.stringify({ + type: "chat.send", + conversationId: "conv-42", + message: "hello", + model: "gpt-4", + }), + ); + // Wait for the message handler to run + await new Promise((r) => setTimeout(r, 100)); + ws.close(); + await new Promise((r) => setTimeout(r, 50)); + + const accepted = logger.entries.filter( + (e) => e.level === "info" && e.msg === "transport-ws: chat.send accepted", + ); + expect(accepted).toHaveLength(1); + expect(accepted[0]?.attrs).toMatchObject({ + conversationId: "conv-42", + model: "gpt-4", + }); + }); + + test("logs a warn on a malformed chat.send", async () => { + const logger = fakeLogger(); + const registry = fakeRegistry([]); + server = startServer(registry, fakeOrchestrator(), 0, logger); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.send", message: "" })); + await waitForMessage(ws); // drain chat.error reply + ws.close(); + await new Promise((r) => setTimeout(r, 50)); + + const malformed = logger.entries.filter( + (e) => e.level === "warn" && e.msg === "transport-ws: malformed chat.send", + ); + expect(malformed).toHaveLength(1); + expect(malformed[0]?.attrs).toMatchObject({ + reason: "chat.send requires a non-empty string `message`", + }); + }); + + test("does not log 'in-flight turn aborted' on close", async () => { + const logger = fakeLogger(); + const orch = fakeOrchestratorWithBroadcast(); + const registry = fakeRegistry([]); + server = startServer(registry, orch, 0, logger); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + ws.send(JSON.stringify({ type: "chat.send", conversationId: "c1", message: "hi" })); + await new Promise((r) => setTimeout(r, 50)); + ws.close(); + await new Promise((r) => setTimeout(r, 50)); + + const abortLogs = logger.entries.filter( + (e) => e.msg.includes("aborted") || e.msg.includes("abort"), + ); + expect(abortLogs).toHaveLength(0); + }); }); describe("conversation.open broadcast (conversationOpened hook)", () => { - let server: ReturnType<typeof startServer>; - let port: number; - - afterEach(() => { - server.stop(); - }); - - test("conversation.open broadcast on conversationOpened hook", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - // Simulate the conversationOpened hook firing (extension.ts's - // `host.on(conversationOpened, ...)` handler runs and broadcasts). - server.triggerConversationOpen("conv-42", "ws-7"); - - const msg = await waitForMessage(ws); - expect(msg).toEqual({ - type: "conversation.open", - conversationId: "conv-42", - workspaceId: "ws-7", - }); - - ws.close(); - }); - - test("conversation.open sent to all connected clients", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); - server = startServer(registry, orch); - port = server.port as number; - - const ws1 = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws1); // drain catalog - const ws2 = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws2); // drain catalog - - // Global fan-out: BOTH connected clients receive the broadcast, - // regardless of any per-conversation subscription state. The forwarded - // `workspaceId` is identical on both. - server.triggerConversationOpen("shared-conv", "ws-shared"); - - const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]); - expect(msg1).toEqual({ - type: "conversation.open", - conversationId: "shared-conv", - workspaceId: "ws-shared", - }); - expect(msg2).toEqual({ - type: "conversation.open", - conversationId: "shared-conv", - workspaceId: "ws-shared", - }); - - ws1.close(); - ws2.close(); - }); + let server: ReturnType<typeof startServer>; + let port: number; + + afterEach(() => { + server.stop(); + }); + + test("conversation.open broadcast on conversationOpened hook", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + // Simulate the conversationOpened hook firing (extension.ts's + // `host.on(conversationOpened, ...)` handler runs and broadcasts). + server.triggerConversationOpen("conv-42", "ws-7"); + + const msg = await waitForMessage(ws); + expect(msg).toEqual({ + type: "conversation.open", + conversationId: "conv-42", + workspaceId: "ws-7", + }); + + ws.close(); + }); + + test("conversation.open sent to all connected clients", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); + server = startServer(registry, orch); + port = server.port as number; + + const ws1 = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws1); // drain catalog + const ws2 = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws2); // drain catalog + + // Global fan-out: BOTH connected clients receive the broadcast, + // regardless of any per-conversation subscription state. The forwarded + // `workspaceId` is identical on both. + server.triggerConversationOpen("shared-conv", "ws-shared"); + + const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]); + expect(msg1).toEqual({ + type: "conversation.open", + conversationId: "shared-conv", + workspaceId: "ws-shared", + }); + expect(msg2).toEqual({ + type: "conversation.open", + conversationId: "shared-conv", + workspaceId: "ws-shared", + }); + + ws1.close(); + ws2.close(); + }); }); describe("conversation.statusChanged broadcast (conversationStatusChanged hook)", () => { - let server: ReturnType<typeof startServer>; - let port: number; - - afterEach(() => { - server.stop(); - }); - - test("conversation.statusChanged broadcast forwards workspaceId", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); - server = startServer(registry, orch); - port = server.port as number; - - const ws = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws); // drain catalog - - // Simulate the conversationStatusChanged hook firing (extension.ts's - // `host.on(conversationStatusChanged, ...)` handler runs and broadcasts). - server.triggerConversationStatusChanged("conv-9", "active", "ws-9"); - - const msg = await waitForMessage(ws); - expect(msg).toEqual({ - type: "conversation.statusChanged", - conversationId: "conv-9", - status: "active", - workspaceId: "ws-9", - }); - - ws.close(); - }); - - test("conversation.statusChanged sent to all connected clients with the same workspaceId", async () => { - const orch = fakeOrchestrator(); - const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); - server = startServer(registry, orch); - port = server.port as number; - - const ws1 = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws1); // drain catalog - const ws2 = new WebSocket(`ws://localhost:${port}`); - await waitForMessage(ws2); // drain catalog - - // Global fan-out: BOTH connected clients receive the broadcast with the - // conversation's persisted workspaceId forwarded unchanged. - server.triggerConversationStatusChanged("shared-conv", "idle", "ws-shared"); - - const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]); - expect(msg1).toEqual({ - type: "conversation.statusChanged", - conversationId: "shared-conv", - status: "idle", - workspaceId: "ws-shared", - }); - expect(msg2).toEqual({ - type: "conversation.statusChanged", - conversationId: "shared-conv", - status: "idle", - workspaceId: "ws-shared", - }); - - ws1.close(); - ws2.close(); - }); + let server: ReturnType<typeof startServer>; + let port: number; + + afterEach(() => { + server.stop(); + }); + + test("conversation.statusChanged broadcast forwards workspaceId", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); + server = startServer(registry, orch); + port = server.port as number; + + const ws = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws); // drain catalog + + // Simulate the conversationStatusChanged hook firing (extension.ts's + // `host.on(conversationStatusChanged, ...)` handler runs and broadcasts). + server.triggerConversationStatusChanged("conv-9", "active", "ws-9"); + + const msg = await waitForMessage(ws); + expect(msg).toEqual({ + type: "conversation.statusChanged", + conversationId: "conv-9", + status: "active", + workspaceId: "ws-9", + }); + + ws.close(); + }); + + test("conversation.statusChanged sent to all connected clients with the same workspaceId", async () => { + const orch = fakeOrchestrator(); + const registry = fakeRegistry([fakeProvider("demo", "Demo Surface")]); + server = startServer(registry, orch); + port = server.port as number; + + const ws1 = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws1); // drain catalog + const ws2 = new WebSocket(`ws://localhost:${port}`); + await waitForMessage(ws2); // drain catalog + + // Global fan-out: BOTH connected clients receive the broadcast with the + // conversation's persisted workspaceId forwarded unchanged. + server.triggerConversationStatusChanged("shared-conv", "idle", "ws-shared"); + + const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]); + expect(msg1).toEqual({ + type: "conversation.statusChanged", + conversationId: "shared-conv", + status: "idle", + workspaceId: "ws-shared", + }); + expect(msg2).toEqual({ + type: "conversation.statusChanged", + conversationId: "shared-conv", + status: "idle", + workspaceId: "ws-shared", + }); + + ws1.close(); + ws2.close(); + }); }); diff --git a/packages/transport-ws/tsconfig.json b/packages/transport-ws/tsconfig.json index 2a1d7ab..c861681 100644 --- a/packages/transport-ws/tsconfig.json +++ b/packages/transport-ws/tsconfig.json @@ -1,12 +1,12 @@ { - "extends": "../../tsconfig.base.json", - "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true }, - "include": ["src/**/*.ts"], - "references": [ - { "path": "../kernel" }, - { "path": "../session-orchestrator" }, - { "path": "../surface-registry" }, - { "path": "../transport-contract" }, - { "path": "../ui-contract" } - ] + "extends": "../../tsconfig.base.json", + "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true }, + "include": ["src/**/*.ts"], + "references": [ + { "path": "../kernel" }, + { "path": "../session-orchestrator" }, + { "path": "../surface-registry" }, + { "path": "../transport-contract" }, + { "path": "../ui-contract" } + ] } |