diff --git a/packages/control-plane/src/cloudflare/durable-object-session-connections.test.ts b/packages/control-plane/src/cloudflare/durable-object-session-connections.test.ts new file mode 100644 index 000000000..0360b68f8 --- /dev/null +++ b/packages/control-plane/src/cloudflare/durable-object-session-connections.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, it, vi } from "vitest"; +import { + DurableObjectSessionConnections, + type DurableObjectSessionConnectionSockets, +} from "./durable-object-session-connections"; +import type { SandboxDeliveryUnavailableError } from "../session/connections"; + +function harness() { + const browser = { readyState: WebSocket.OPEN } as WebSocket; + const sandbox = { readyState: WebSocket.OPEN } as WebSocket; + const manager: DurableObjectSessionConnectionSockets = { + forEachClientSocket: vi.fn( + (_mode: "all_clients" | "authenticated_only", fn: (ws: WebSocket) => void) => fn(browser) + ), + getSandboxSocket: vi.fn(() => sandbox), + send: vi.fn(() => true), + configureAutoPing: vi.fn(), + createUpgradeSockets: vi.fn(), + }; + return { + connections: new DurableObjectSessionConnections(manager), + manager, + browser, + sandbox, + }; +} + +describe("DurableObjectSessionConnections", () => { + it("owns Cloudflare auto-response configuration", () => { + const { manager } = harness(); + + expect(manager.configureAutoPing).toHaveBeenCalledTimes(1); + }); + + it("broadcasts to authenticated browser sockets", async () => { + const { connections, manager, browser } = harness(); + const message = { type: "sandbox_status", status: "ready" } as const; + + await connections.broadcastToBrowsers(message); + + expect(manager.forEachClientSocket).toHaveBeenCalledWith( + "authenticated_only", + expect.any(Function) + ); + expect(manager.send).toHaveBeenCalledWith(browser, message); + }); + + it("sends typed commands to the active sandbox", async () => { + const { connections, manager, sandbox } = harness(); + + await connections.sendToSandbox({ type: "snapshot" }); + + expect(manager.send).toHaveBeenCalledWith(sandbox, { type: "snapshot" }); + }); + + it("distinguishes missing sockets from failed delivery", async () => { + const { connections, manager } = harness(); + vi.mocked(manager.getSandboxSocket).mockReturnValueOnce(null); + + await expect(connections.sendToSandbox({ type: "snapshot" })).rejects.toMatchObject({ + reason: "not_connected", + } satisfies Partial); + + vi.mocked(manager.send).mockReturnValueOnce(false); + await expect(connections.sendToSandbox({ type: "snapshot" })).rejects.toMatchObject({ + reason: "send_failed", + } satisfies Partial); + }); +}); diff --git a/packages/control-plane/src/cloudflare/durable-object-session-connections.ts b/packages/control-plane/src/cloudflare/durable-object-session-connections.ts new file mode 100644 index 000000000..561e12faa --- /dev/null +++ b/packages/control-plane/src/cloudflare/durable-object-session-connections.ts @@ -0,0 +1,44 @@ +import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; +import type { SessionConnections } from "../session/connections"; +import { SandboxDeliveryUnavailableError } from "../session/connections"; +import type { SandboxCommand } from "../session/types"; + +export interface DurableObjectSessionConnectionSockets { + configureAutoPing(request: string, response: string): void; + createUpgradeSockets(): { client: WebSocket; server: WebSocket }; + forEachClientSocket( + mode: "all_clients" | "authenticated_only", + fn: (ws: WebSocket) => void + ): void; + getSandboxSocket(): WebSocket | null; + send(ws: WebSocket, message: ServerMessage | SandboxCommand): boolean; +} + +/** Cloudflare Durable Object implementation of the session connection port. */ +export class DurableObjectSessionConnections implements SessionConnections { + constructor(private readonly sockets: DurableObjectSessionConnectionSockets) { + this.sockets.configureAutoPing( + JSON.stringify({ type: "ping" }), + JSON.stringify({ type: "pong", timestamp: Date.now() }) + ); + } + + createUpgradeSockets(): { client: WebSocket; server: WebSocket } { + return this.sockets.createUpgradeSockets(); + } + + sendToSandbox(message: SandboxCommand): Promise { + const ws = this.sockets.getSandboxSocket(); + if (!ws) return Promise.reject(new SandboxDeliveryUnavailableError()); + return this.sockets.send(ws, message) + ? Promise.resolve() + : Promise.reject(new SandboxDeliveryUnavailableError("send_failed")); + } + + broadcastToBrowsers(message: ServerMessage): Promise { + this.sockets.forEachClientSocket("authenticated_only", (ws) => { + this.sockets.send(ws, message); + }); + return Promise.resolve(); + } +} diff --git a/packages/control-plane/src/session/websocket-manager.test.ts b/packages/control-plane/src/cloudflare/durable-object-socket-registry.test.ts similarity index 91% rename from packages/control-plane/src/session/websocket-manager.test.ts rename to packages/control-plane/src/cloudflare/durable-object-socket-registry.test.ts index 7fd540cd9..8bde2bd5b 100644 --- a/packages/control-plane/src/session/websocket-manager.test.ts +++ b/packages/control-plane/src/cloudflare/durable-object-socket-registry.test.ts @@ -1,21 +1,23 @@ /** - * Unit tests for SessionWebSocketManagerImpl. + * Unit tests for DurableObjectSocketRegistry. * * Uses fake DurableObjectState and mock repositories to test * all WebSocket mechanics in isolation from the full DO. */ import { describe, it, expect, vi } from "vitest"; -import { SessionWebSocketManagerImpl } from "./websocket-manager"; -import type { WebSocketManagerConfig } from "./websocket-manager"; +import { + DurableObjectSocketRegistry, + type SocketRegistryConfig, +} from "./durable-object-socket-registry"; import type { Logger } from "../logger"; import type { ClientInfo } from "../types"; -import type { SandboxRepository } from "./sandbox-repository"; +import type { SandboxRepository } from "../session/sandbox-repository"; import type { WsClientMappingRepository, WsClientMappingResult, -} from "./ws-client-mapping-repository"; -import type { SandboxRow } from "./types"; +} from "../session/ws-client-mapping-repository"; +import type { SandboxRow } from "../session/types"; // --------------------------------------------------------------------------- // Fakes & Helpers @@ -70,8 +72,10 @@ function createFakeCtx(): FakeCtx { getTags(ws: WebSocket): string[] { return sockets.get(ws) ?? []; }, - getWebSockets(): WebSocket[] { - return Array.from(sockets.keys()); + getWebSockets(tag?: string): WebSocket[] { + return Array.from(sockets, ([ws, tags]) => ({ ws, tags })) + .filter(({ tags }) => !tag || tags.includes(tag)) + .map(({ ws }) => ws); }, setWebSocketAutoResponse: vi.fn(), storage: { setAlarm: vi.fn() }, @@ -180,7 +184,7 @@ function createSandboxRow(modalSandboxId: string): SandboxRow { }; } -const TEST_CONFIG: WebSocketManagerConfig = { authTimeoutMs: 100 }; +const TEST_CONFIG: SocketRegistryConfig = { authTimeoutMs: 100 }; /** Create a fresh manager with all dependencies. */ function createManager() { @@ -188,22 +192,38 @@ function createManager() { const mockRepo = createMockRepository(); const log = createMockLogger(); - const manager = new SessionWebSocketManagerImpl( + const createRecoveredClient = vi.fn((ws: WebSocket, mapping: WsClientMappingResult) => + createClientInfo({ + ws, + participantId: mapping.participant_id, + userId: mapping.user_id, + clientId: mapping.client_id, + }) + ); + const manager = new DurableObjectSocketRegistry( fakeCtx.state, mockRepo.repo, mockRepo.repo as unknown as WsClientMappingRepository, - log, + () => log, + createRecoveredClient, TEST_CONFIG ); - return { manager, sockets: fakeCtx.sockets, state: fakeCtx.state, mockRepo, log }; + return { + manager, + sockets: fakeCtx.sockets, + state: fakeCtx.state, + mockRepo, + log, + createRecoveredClient, + }; } // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- -describe("SessionWebSocketManagerImpl", () => { +describe("DurableObjectSocketRegistry", () => { describe("classify", () => { it("classifies sandbox socket with sandbox ID", () => { const { manager, sockets } = createManager(); @@ -439,13 +459,13 @@ describe("SessionWebSocketManagerImpl", () => { }); }); - describe("clearSandboxSocketIfMatch", () => { + describe("clearSandboxIfMatch", () => { it("clears and returns true when ws matches", () => { const { manager } = createManager(); const ws = createFakeWebSocket(); manager.acceptAndSetSandboxSocket(ws, "sb-1"); - const result = manager.clearSandboxSocketIfMatch(ws); + const result = manager.clearSandboxIfMatch(ws); expect(result).toBe(true); // Verify it was actually cleared @@ -462,7 +482,7 @@ describe("SessionWebSocketManagerImpl", () => { manager.acceptAndSetSandboxSocket(newWs, "sb-2"); // Try to clear with old socket — should not affect new socket - const result = manager.clearSandboxSocketIfMatch(oldWs); + const result = manager.clearSandboxIfMatch(oldWs); expect(result).toBe(false); expect(manager.getSandboxSocket()).toBe(newWs); @@ -474,7 +494,7 @@ describe("SessionWebSocketManagerImpl", () => { // When sandboxWs is null (e.g., post-hibernation), the closing socket // is treated as active since there's no replacement to compare against. - expect(manager.clearSandboxSocketIfMatch(ws)).toBe(true); + expect(manager.clearSandboxIfMatch(ws)).toBe(true); }); }); @@ -496,6 +516,37 @@ describe("SessionWebSocketManagerImpl", () => { expect(manager.getClient(ws)).toBeNull(); }); + it("reconstructs and caches client identity after hibernation", () => { + const { manager, sockets, mockRepo, createRecoveredClient } = createManager(); + const ws = createFakeWebSocket(); + const mapping: WsClientMappingResult = { + participant_id: "part-recovered", + client_id: "client-recovered", + user_id: "user-recovered", + scm_name: null, + auth_name: null, + scm_login: null, + }; + sockets.set(ws, ["wsid:ws-recovered"]); + mockRepo.addMapping("ws-recovered", mapping); + + const recovered = manager.getClient(ws); + + expect(createRecoveredClient).toHaveBeenCalledWith(ws, mapping); + expect(recovered).toMatchObject({ participantId: "part-recovered" }); + expect(manager.getClient(ws)).toBe(recovered); + expect(createRecoveredClient).toHaveBeenCalledTimes(1); + }); + + it("reports whether an authenticated participant is connected", () => { + const { manager } = createManager(); + const ws = createFakeWebSocket(); + manager.setClient(ws, createClientInfo({ ws, participantId: "part-1" })); + + expect(manager.hasParticipant("part-1")).toBe(true); + expect(manager.hasParticipant("part-2")).toBe(false); + }); + it("removeClient returns and removes the client", () => { const { manager } = createManager(); const ws = createFakeWebSocket(); diff --git a/packages/control-plane/src/session/websocket-manager.ts b/packages/control-plane/src/cloudflare/durable-object-socket-registry.ts similarity index 72% rename from packages/control-plane/src/session/websocket-manager.ts rename to packages/control-plane/src/cloudflare/durable-object-socket-registry.ts index 389c8657d..a4a9ea3d0 100644 --- a/packages/control-plane/src/session/websocket-manager.ts +++ b/packages/control-plane/src/cloudflare/durable-object-socket-registry.ts @@ -1,6 +1,5 @@ /** - * SessionWebSocketManager — centralizes all Cloudflare WebSocket API usage - * into a single, testable module. + * DurableObjectSocketRegistry - the Cloudflare implementation of the session socket port. * * The manager is a registry for ClientInfo, not a factory. The DO builds * ClientInfo and stores it here via setClient/getClient. @@ -8,85 +7,22 @@ import type { Logger } from "../logger"; import type { ClientInfo } from "../types"; -import type { ConnectionClassification } from "./ports"; -import type { SandboxRepository } from "./sandbox-repository"; +import type { ConnectionClassification, SocketRegistry } from "../session/ports"; +import type { SandboxRepository } from "../session/sandbox-repository"; import type { WsClientMappingRepository, WsClientMappingResult, -} from "./ws-client-mapping-repository"; +} from "../session/ws-client-mapping-repository"; -/** Configuration for the WebSocket manager. */ -export interface WebSocketManagerConfig { +export interface SocketRegistryConfig { authTimeoutMs: number; } -// --------------------------------------------------------------------------- -// Interface -// --------------------------------------------------------------------------- - -export interface SessionWebSocketManager { - /** Accept a client WebSocket with a wsId tag for hibernation recovery. */ - acceptClientSocket(ws: WebSocket, wsId: string): void; - - /** - * Accept a sandbox WebSocket, close any existing sandbox socket, and set - * as the active sandbox connection. - */ - acceptAndSetSandboxSocket(ws: WebSocket, sandboxId?: string): { replaced: boolean }; - - /** Parse a WebSocket's tags to determine its kind and identity. */ - classify(ws: WebSocket): ConnectionClassification; - - /** - * Get the active sandbox socket, recovering from hibernation if needed. - * Validates sandbox ID against the repository during hibernation recovery. - */ - getSandboxSocket(): WebSocket | null; - - /** Clear the in-memory sandbox socket reference. */ - clearSandboxSocket(): void; - - /** Clear and close all active sandbox sockets without consulting persisted dispatch status. */ - detachSandboxSocket(code: number, reason: string): void; - - /** Clear sandbox socket only if ws matches current reference. Returns true if it was the active socket. */ - clearSandboxSocketIfMatch(ws: WebSocket): boolean; - - setClient(ws: WebSocket, info: ClientInfo): void; - getClient(ws: WebSocket): ClientInfo | null; - removeClient(ws: WebSocket): ClientInfo | null; - - /** Returns raw DB mapping for hibernation recovery. The DO builds ClientInfo from this. */ - recoverClientMapping(ws: WebSocket): WsClientMappingResult | null; - - /** Persist ws-to-participant mapping for hibernation survival. */ - persistClientMapping(wsId: string, participantId: string, clientId: string): void; - - setClientSynchronizing(ws: WebSocket, synchronizing: boolean): void; - isClientSynchronizing(ws: WebSocket): boolean; - isClientAuthenticated(ws: WebSocket): boolean; - - /** Check if a wsId has a persisted mapping (used by auth timeout). */ - hasPersistedMapping(wsId: string): boolean; - - send(ws: WebSocket, message: string | object): boolean; - close(ws: WebSocket, code: number, reason: string): void; - - forEachClientSocket( - mode: "all_clients" | "authenticated_only", - fn: (ws: WebSocket) => void - ): void; - - enforceAuthTimeout(ws: WebSocket, wsId: string): Promise; - getAuthenticatedClients(): IterableIterator; - getConnectedClientCount(): number; -} - // --------------------------------------------------------------------------- // Implementation // --------------------------------------------------------------------------- -export class SessionWebSocketManagerImpl implements SessionWebSocketManager { +export class DurableObjectSocketRegistry implements SocketRegistry { private clients = new Map(); private synchronizingClients = new Set(); private sandboxWs: WebSocket | null = null; @@ -95,10 +31,28 @@ export class SessionWebSocketManagerImpl implements SessionWebSocketManager { private readonly ctx: DurableObjectState, private readonly sandboxRepository: SandboxRepository, private readonly wsClientMappingRepository: WsClientMappingRepository, - private readonly log: Logger, - private readonly config: WebSocketManagerConfig + private readonly getLog: () => Logger, + private readonly createRecoveredClient: ( + ws: WebSocket, + mapping: WsClientMappingResult + ) => ClientInfo, + private readonly config: SocketRegistryConfig ) {} + private get log(): Logger { + return this.getLog(); + } + + createUpgradeSockets(): { client: WebSocket; server: WebSocket } { + const pair = new WebSocketPair(); + const [client, server] = Object.values(pair); + return { client, server }; + } + + configureAutoPing(request: string, response: string): void { + this.ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair(request, response)); + } + // ------------------------------------------------------------------------- // Accept // ------------------------------------------------------------------------- @@ -157,11 +111,8 @@ export class SessionWebSocketManagerImpl implements SessionWebSocketManager { if (sandbox && terminalStatuses.includes(sandbox.status)) { this.sandboxWs = null; // Close any lingering sandbox WebSockets so they don't persist - for (const ws of this.ctx.getWebSockets()) { - const parsed = this.classify(ws); - if (parsed.kind === "sandbox") { - this.close(ws, 1000, "Sandbox terminated"); - } + for (const ws of this.ctx.getWebSockets("sandbox")) { + this.close(ws, 1000, "Sandbox terminated"); } return null; } @@ -172,7 +123,7 @@ export class SessionWebSocketManagerImpl implements SessionWebSocketManager { // Hibernation recovery: scan all WebSockets, validate sandbox identity - for (const ws of this.ctx.getWebSockets()) { + for (const ws of this.ctx.getWebSockets("sandbox")) { const parsed = this.classify(ws); if (parsed.kind !== "sandbox" || ws.readyState !== WebSocket.OPEN) continue; @@ -199,14 +150,12 @@ export class SessionWebSocketManagerImpl implements SessionWebSocketManager { detachSandboxSocket(code: number, reason: string): void { const sockets = new Set(); if (this.sandboxWs) sockets.add(this.sandboxWs); - for (const ws of this.ctx.getWebSockets()) { - if (this.classify(ws).kind === "sandbox") sockets.add(ws); - } + for (const ws of this.ctx.getWebSockets("sandbox")) sockets.add(ws); this.sandboxWs = null; for (const ws of sockets) this.close(ws, code, reason); } - clearSandboxSocketIfMatch(ws: WebSocket): boolean { + clearSandboxIfMatch(ws: WebSocket): boolean { if (this.sandboxWs === ws) { this.sandboxWs = null; return true; @@ -225,7 +174,20 @@ export class SessionWebSocketManagerImpl implements SessionWebSocketManager { } getClient(ws: WebSocket): ClientInfo | null { - return this.clients.get(ws) ?? null; + const cached = this.clients.get(ws); + if (cached) return cached; + + const mapping = this.recoverClientMapping(ws); + if (!mapping) { + this.log.warn("No client mapping found after hibernation, closing WebSocket"); + this.close(ws, 4002, "Session expired, please reconnect"); + return null; + } + + this.log.info("Recovered client info from DB", { user_id: mapping.user_id }); + const client = this.createRecoveredClient(ws, mapping); + this.clients.set(ws, client); + return client; } removeClient(ws: WebSocket): ClientInfo | null { @@ -355,6 +317,13 @@ export class SessionWebSocketManagerImpl implements SessionWebSocketManager { return this.clients.values(); } + hasParticipant(participantId: string): boolean { + for (const client of this.clients.values()) { + if (client.participantId === participantId) return true; + } + return false; + } + getConnectedClientCount(): number { let count = 0; for (const ws of this.ctx.getWebSockets()) { diff --git a/packages/control-plane/src/session/connections.ts b/packages/control-plane/src/session/connections.ts index 25a8f6670..92bba921d 100644 --- a/packages/control-plane/src/session/connections.ts +++ b/packages/control-plane/src/session/connections.ts @@ -1,5 +1,6 @@ import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; import type { SandboxCommand } from "./types"; +import type { SandboxCommandSender } from "./ports"; export interface ConnectedParticipant { participantId: string; @@ -27,7 +28,10 @@ export interface DisconnectReason { } export class SandboxDeliveryUnavailableError extends Error { - constructor(message = "No sandbox connected") { + constructor( + readonly reason: "not_connected" | "send_failed" = "not_connected", + message = reason === "not_connected" ? "No sandbox connected" : "Failed to send to sandbox" + ) { super(message); this.name = "SandboxDeliveryUnavailableError"; } @@ -58,13 +62,8 @@ export function projectConnectedParticipants( } /** Platform-neutral connection and fan-out boundary consumed by the session engine. */ -export interface SessionConnections { - registerBrowser(input: BrowserConnection): Promise; - registerSandbox(input: SandboxConnection): Promise; - sendToSandbox(message: SandboxCommand): Promise; +export interface SessionConnections extends SandboxCommandSender { broadcastToBrowsers(message: ServerMessage): Promise; - disconnectSandbox(reason: DisconnectReason): Promise; - listParticipants(): Promise; } /** Deterministic connection adapter for application tests and non-WebSocket runtimes. */ diff --git a/packages/control-plane/src/session/durable-object-session-connections.test.ts b/packages/control-plane/src/session/durable-object-session-connections.test.ts deleted file mode 100644 index dd2cd05a8..000000000 --- a/packages/control-plane/src/session/durable-object-session-connections.test.ts +++ /dev/null @@ -1,119 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; -import { DurableObjectSessionConnections } from "./durable-object-session-connections"; -import type { SessionWebSocketManager } from "./websocket-manager"; - -vi.stubGlobal( - "WebSocketRequestResponsePair", - class WebSocketRequestResponsePair { - constructor( - readonly request: string, - readonly response: string - ) {} - } -); - -function harness() { - const browser = { readyState: WebSocket.OPEN } as WebSocket; - const sandbox = { readyState: WebSocket.OPEN } as WebSocket; - const clientInfo = { - participantId: "part-1", - userId: "user-1", - name: "Test User", - status: "active" as const, - lastSeen: 123, - clientId: "client-1", - ws: browser, - }; - const manager = { - classify: vi.fn((ws: WebSocket) => - ws === sandbox - ? ({ kind: "sandbox" as const, sandboxId: "sb-1" } as const) - : ({ kind: "client" as const, wsId: "ws-1" } as const) - ), - forEachClientSocket: vi.fn( - (_mode: "all_clients" | "authenticated_only", fn: (ws: WebSocket) => void) => fn(browser) - ), - setClient: vi.fn(), - persistClientMapping: vi.fn(), - getSandboxSocket: vi.fn(() => sandbox), - send: vi.fn(() => true), - detachSandboxSocket: vi.fn(), - getAuthenticatedClients: vi.fn(() => [clientInfo].values()), - recoverClientMapping: vi.fn(() => null), - } as unknown as SessionWebSocketManager; - const state = { setWebSocketAutoResponse: vi.fn() } as unknown as DurableObjectState; - return { - connections: new DurableObjectSessionConnections(state, manager), - manager, - state, - browser, - sandbox, - }; -} - -describe("DurableObjectSessionConnections", () => { - it("owns Cloudflare auto-response configuration", () => { - const { state } = harness(); - - expect(state.setWebSocketAutoResponse).toHaveBeenCalledTimes(1); - }); - - it("registers and broadcasts to a browser through the WebSocket registry", async () => { - const { connections, manager, browser } = harness(); - const input = { - connectionId: "ws-1", - clientId: "client-1", - participant: { - participantId: "part-1", - userId: "user-1", - name: "Test User", - status: "active" as const, - lastSeen: 123, - }, - }; - - await connections.registerBrowser(input); - const message = { type: "sandbox_status", status: "ready" } as const; - await connections.broadcastToBrowsers(message); - - expect(manager.setClient).toHaveBeenCalledWith( - browser, - expect.objectContaining(input.participant) - ); - expect(manager.persistClientMapping).toHaveBeenCalledWith("ws-1", "part-1", "client-1"); - expect(manager.send).toHaveBeenCalledWith(browser, message); - }); - - it("sends to and disconnects the active sandbox", async () => { - const { connections, manager, sandbox } = harness(); - - await connections.registerSandbox({ connectionId: "sandbox-1", sandboxId: "sb-1" }); - await connections.sendToSandbox({ type: "snapshot" }); - await connections.disconnectSandbox({ code: 1011, reason: "Unresponsive sandbox" }); - - expect(manager.send).toHaveBeenCalledWith(sandbox, { type: "snapshot" }); - expect(manager.detachSandboxSocket).toHaveBeenCalledWith(1011, "Unresponsive sandbox"); - }); - - it("rejects registration for a different sandbox identity", async () => { - const { connections } = harness(); - - await expect( - connections.registerSandbox({ connectionId: "sandbox-2", sandboxId: "sb-2" }) - ).rejects.toThrow("does not match"); - }); - - it("lists participants without exposing WebSocket details", async () => { - const { connections } = harness(); - - await expect(connections.listParticipants()).resolves.toEqual([ - { - participantId: "part-1", - userId: "user-1", - name: "Test User", - status: "active", - lastSeen: 123, - }, - ]); - }); -}); diff --git a/packages/control-plane/src/session/durable-object-session-connections.ts b/packages/control-plane/src/session/durable-object-session-connections.ts deleted file mode 100644 index c403e6166..000000000 --- a/packages/control-plane/src/session/durable-object-session-connections.ts +++ /dev/null @@ -1,94 +0,0 @@ -import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; -import type { - BrowserConnection, - ConnectedParticipant, - DisconnectReason, - SandboxConnection, - SessionConnections, -} from "./connections"; -import { projectConnectedParticipants, SandboxDeliveryUnavailableError } from "./connections"; -import type { SandboxCommand } from "./types"; -import type { SessionWebSocketManager } from "./websocket-manager"; - -/** Cloudflare Durable Object implementation of the session connection port. */ -export class DurableObjectSessionConnections implements SessionConnections { - constructor( - private readonly ctx: DurableObjectState, - private readonly wsManager: SessionWebSocketManager - ) { - this.ctx.setWebSocketAutoResponse( - new WebSocketRequestResponsePair( - JSON.stringify({ type: "ping" }), - JSON.stringify({ type: "pong", timestamp: Date.now() }) - ) - ); - } - - createUpgradeSockets(): { client: WebSocket; server: WebSocket } { - const pair = new WebSocketPair(); - const [client, server] = Object.values(pair); - return { client, server }; - } - - registerBrowser(input: BrowserConnection): Promise { - let matchedSocket: WebSocket | null = null; - this.wsManager.forEachClientSocket("all_clients", (ws) => { - const connection = this.wsManager.classify(ws); - if (connection.kind === "client" && connection.wsId === input.connectionId) { - matchedSocket = ws; - } - }); - if (!matchedSocket) { - return Promise.reject(new Error(`Browser connection ${input.connectionId} not found`)); - } - - this.wsManager.setClient(matchedSocket, { - ...input.participant, - clientId: input.clientId, - ws: matchedSocket, - }); - this.wsManager.persistClientMapping( - input.connectionId, - input.participant.participantId, - input.clientId - ); - return Promise.resolve(); - } - - registerSandbox(input: SandboxConnection): Promise { - const ws = this.wsManager.getSandboxSocket(); - if (!ws) return Promise.reject(new Error(`Sandbox connection ${input.connectionId} not found`)); - const connection = this.wsManager.classify(ws); - if ( - connection.kind !== "sandbox" || - (input.sandboxId !== undefined && connection.sandboxId !== input.sandboxId) - ) { - return Promise.reject(new Error(`Sandbox connection ${input.connectionId} does not match`)); - } - return Promise.resolve(); - } - - sendToSandbox(message: SandboxCommand): Promise { - const ws = this.wsManager.getSandboxSocket(); - if (!ws) return Promise.reject(new SandboxDeliveryUnavailableError()); - return this.wsManager.send(ws, message) - ? Promise.resolve() - : Promise.reject(new SandboxDeliveryUnavailableError("Failed to send message to sandbox")); - } - - broadcastToBrowsers(message: ServerMessage): Promise { - this.wsManager.forEachClientSocket("authenticated_only", (ws) => { - this.wsManager.send(ws, message); - }); - return Promise.resolve(); - } - - disconnectSandbox(reason: DisconnectReason): Promise { - this.wsManager.detachSandboxSocket(reason.code, reason.reason); - return Promise.resolve(); - } - - listParticipants(): Promise { - return Promise.resolve(projectConnectedParticipants(this.wsManager.getAuthenticatedClients())); - } -} diff --git a/packages/control-plane/src/session/durable-object.ts b/packages/control-plane/src/session/durable-object.ts index de79d4070..590ffe5a4 100644 --- a/packages/control-plane/src/session/durable-object.ts +++ b/packages/control-plane/src/session/durable-object.ts @@ -68,8 +68,8 @@ import { WsClientMappingRepository } from "./ws-client-mapping-repository"; import { resolveParticipantName } from "./participant-name"; import { validateReasoningEffort } from "./reasoning-effort"; import { parseTunnelUrls } from "./tunnel-urls"; -import { SessionWebSocketManagerImpl, type SessionWebSocketManager } from "./websocket-manager"; -import { DurableObjectSessionConnections } from "./durable-object-session-connections"; +import { DurableObjectSocketRegistry } from "../cloudflare/durable-object-socket-registry"; +import { DurableObjectSessionConnections } from "../cloudflare/durable-object-session-connections"; import { SessionPullRequestStore } from "../db/session-pull-request-store"; import { PullRequestCreationClaims, SessionPullRequestService } from "./pull-request-service"; import { refreshSessionPullRequests } from "./pull-request-refresh"; @@ -144,7 +144,7 @@ import { SessionServer } from "./server"; import { SessionHttpDispatcher } from "./http/dispatcher"; import { SessionMessageRouter, type SessionClientCommands } from "./message-router"; import { SessionDisconnectHandler } from "./disconnect-handler"; -import type { Clock, SandboxDisconnectMonitor, SessionBroadcaster, SocketRegistry } from "./ports"; +import type { Clock, SocketRegistry, SandboxDisconnectMonitor, SessionBroadcaster } from "./ports"; /** * Timeout for WebSocket authentication (in milliseconds). @@ -192,7 +192,7 @@ export class SessionDO extends DurableObject { // (with trace_id / request_id) threaded explicitly from fetch(). private log: Logger; // WebSocket manager (lazily initialized like lifecycleManager) - private _wsManager: SessionWebSocketManager | null = null; + private _wsManager: DurableObjectSocketRegistry | null = null; private _connections: DurableObjectSessionConnections | null = null; // Session messenger (constructed in ensureInitialized once the session logger exists) private messenger!: SessionMessenger; @@ -211,7 +211,7 @@ export class SessionDO extends DurableObject { // Presence service (lazily initialized) private _presenceService: PresenceService | null = null; // Message queue service (lazily initialized) - private _messageQueue: SessionMessageQueue | null = null; + private _messageQueue: SessionMessageQueue | null = null; // Message service (lazily initialized) private _messageService: MessageService | null = null; private _eventStream: SessionEventStream | null = null; @@ -322,18 +322,7 @@ export class SessionDO extends DurableObject { nowMs: () => Date.now(), monotonicNowMs: () => performance.now(), }; - const sockets: SocketRegistry = { - classify: (ws) => this.wsManager.classify(ws), - send: (ws, message) => this.safeSend(ws, message), - getClient: (ws) => this.getClientInfo(ws), - close: (ws, code, reason) => this.wsManager.close(ws, code, reason), - clearSandboxIfMatch: (ws) => this.wsManager.clearSandboxSocketIfMatch(ws), - removeClient: (ws) => this.wsManager.removeClient(ws), - hasParticipant: (participantId) => - Array.from(this.wsManager.getAuthenticatedClients()).some( - (client) => client.participantId === participantId - ), - }; + const sockets: SocketRegistry = this.wsManager; const clientCommands: SessionClientCommands = { subscribe: (ws, message) => this.handleSubscribe(ws, message), submitPrompt: (ws, client, message) => this.handlePromptMessage(ws, client, message), @@ -472,18 +461,24 @@ export class SessionDO extends DurableObject { return this._presenceService; } - /** - * Get the WebSocket manager, creating it lazily if needed. - * Lazy initialization ensures the logger has session_id context - * (set by ensureInitialized()) by the time the manager is created. - */ - private get wsManager(): SessionWebSocketManager { + /** Get the WebSocket manager while resolving its current request-scoped logger on demand. */ + private get wsManager(): DurableObjectSocketRegistry { if (!this._wsManager) { - this._wsManager = new SessionWebSocketManagerImpl( + this._wsManager = new DurableObjectSocketRegistry( this.ctx, this.sandboxRepository, this.wsClientMappingRepository, - this.log, + () => this.log, + (ws, mapping) => ({ + participantId: mapping.participant_id, + userId: mapping.canonical_user_id ?? mapping.user_id, + name: resolveParticipantName(mapping), + avatar: getAvatarUrl(mapping.scm_login, resolveScmProviderFromEnv(this.env.SCM_PROVIDER)), + status: "active", + lastSeen: Date.now(), + clientId: mapping.client_id || `client-${Date.now()}`, + ws, + }), { authTimeoutMs: WS_AUTH_TIMEOUT_MS } ); } @@ -492,7 +487,7 @@ export class SessionDO extends DurableObject { private get connections(): DurableObjectSessionConnections { if (!this._connections) { - this._connections = new DurableObjectSessionConnections(this.ctx, this.wsManager); + this._connections = new DurableObjectSessionConnections(this.wsManager); } return this._connections; } @@ -521,7 +516,7 @@ export class SessionDO extends DurableObject { return this._alarmScheduler; } - private get messageQueue(): SessionMessageQueue { + private get messageQueue(): SessionMessageQueue { if (!this._messageQueue) { this._messageQueue = new SessionMessageQueue( this.backgroundTasks, @@ -836,7 +831,6 @@ export class SessionDO extends DurableObject { this.eventRepository, this.artifactRepository, this.callbackService, - this.wsManager, this.messenger, this.diffService, (title, options) => this.applySessionTitleUpdate(title, options), @@ -1425,40 +1419,6 @@ export class SessionDO extends DurableObject { return true; } - /** - * Get client info for a WebSocket, reconstructing from storage if needed after hibernation. - */ - private getClientInfo(ws: WebSocket): ClientInfo | null { - // 1. In-memory cache (manager) - const cached = this.wsManager.getClient(ws); - if (cached) return cached; - - // 2. DB recovery (manager handles tag parsing + DB lookup) - const mapping = this.wsManager.recoverClientMapping(ws); - if (!mapping) { - this.log.warn("No client mapping found after hibernation, closing WebSocket"); - this.wsManager.close(ws, 4002, "Session expired, please reconnect"); - return null; - } - - // 3. Build ClientInfo (DO owns domain logic) - this.log.info("Recovered client info from DB", { user_id: mapping.user_id }); - const clientInfo: ClientInfo = { - participantId: mapping.participant_id, - userId: mapping.canonical_user_id ?? mapping.user_id, - name: resolveParticipantName(mapping), - avatar: getAvatarUrl(mapping.scm_login, resolveScmProviderFromEnv(this.env.SCM_PROVIDER)), - status: "active", - lastSeen: Date.now(), - clientId: mapping.client_id || `client-${Date.now()}`, - ws, - }; - - // 4. Re-cache - this.wsManager.setClient(ws, clientInfo); - return clientInfo; - } - /** * Handle prompt message from client. */ diff --git a/packages/control-plane/src/session/message-queue.test.ts b/packages/control-plane/src/session/message-queue.test.ts index e90585429..5189bd8bf 100644 --- a/packages/control-plane/src/session/message-queue.test.ts +++ b/packages/control-plane/src/session/message-queue.test.ts @@ -5,15 +5,21 @@ import type { SessionAttachmentRepository } from "./session-attachment-repositor import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; import { MAX_UNFINISHED_PROMPTS } from "@open-inspect/shared/types/prompts"; import type { ClientInfo } from "../types"; -import type { MessageRow, ParticipantRow, SessionRow, SessionAttachmentRow } from "./types"; +import type { + MessageRow, + ParticipantRow, + SandboxCommand, + SessionRow, + SessionAttachmentRow, +} from "./types"; import type { SessionCoreRepository } from "./session-core-repository"; import type { ParticipantRepository } from "./participant-repository"; import type { MessageRepository } from "./message-repository"; -import type { SessionWebSocketManager } from "./websocket-manager"; import type { ParticipantService } from "./participant-service"; import type { CallbackNotificationService } from "./callback-notification-service"; import { createEarliestAlarmScheduler } from "./alarm/scheduler"; import type { SessionStatusService } from "./session-status-service"; +import { SandboxDeliveryUnavailableError } from "./connections"; function createParticipant(overrides: Partial = {}): ParticipantRow { return { @@ -167,7 +173,6 @@ function buildQueue() { }; const wsManager = { - getSandboxSocket: vi.fn(() => null as WebSocket | null), send: vi.fn((_ws: WebSocket, _message: ServerMessage) => true), }; @@ -182,7 +187,12 @@ function buildQueue() { }; const broadcast = vi.fn((_message: ServerMessage) => {}); - const messenger = { broadcast, sendToSandbox: vi.fn(async () => {}) }; + const messenger = { + broadcast, + sendToSandbox: vi.fn<(command: SandboxCommand) => Promise>(async () => { + throw new SandboxDeliveryUnavailableError(); + }), + }; const sessionStatus = { transition: vi.fn(async (_status: string) => true), reconcileAfterExecution: vi.fn(async (_success: boolean) => {}), @@ -208,7 +218,7 @@ function buildQueue() { repository as unknown as MessageRepository, repository as unknown as ParticipantRepository, attachmentRepository as unknown as SessionAttachmentRepository, - wsManager as unknown as SessionWebSocketManager, + wsManager, messenger, participantService as unknown as ParticipantService, callbackService as unknown as CallbackNotificationService, @@ -238,6 +248,7 @@ function buildQueue() { repository, attachmentRepository, wsManager, + messenger, participantService, broadcast, sessionStatus, @@ -311,7 +322,8 @@ describe("SessionMessageQueue", () => { expect(h.broadcast).toHaveBeenCalledWith({ type: "sandbox_spawning" }); expect(h.sandboxLifecycle.spawnSandbox).toHaveBeenCalledTimes(1); expect(h.repository.updateMessageToProcessing).not.toHaveBeenCalled(); - expect(h.repository.startMessageProcessing).not.toHaveBeenCalled(); + expect(h.repository.startMessageProcessing).toHaveBeenCalled(); + expect(h.repository.updateMessageToPending).toHaveBeenCalled(); expect(h.callbackService.notifyStarted).not.toHaveBeenCalled(); }); @@ -643,9 +655,8 @@ describe("SessionMessageQueue", () => { it("materializes the user_message at processing start", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; h.repository.getNextPendingMessage.mockReturnValue(createMessage()); - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); @@ -677,7 +688,7 @@ describe("SessionMessageQueue", () => { h.repository.getNextPendingMessage.mockReturnValue( createMessage({ author_id: participant.id, source: "slack" }) ); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); @@ -693,9 +704,8 @@ describe("SessionMessageQueue", () => { it("dispatches prompt command when sandbox socket exists", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-42" })); - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); @@ -704,8 +714,7 @@ describe("SessionMessageQueue", () => { expect.any(Number), expect.objectContaining({ type: "user_message", messageId: "msg-42" }) ); - expect(h.wsManager.send).toHaveBeenCalledWith( - sandboxWs, + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith( expect.objectContaining({ type: "prompt", messageId: "msg-42" }) ); expect(h.broadcast).toHaveBeenCalledWith({ type: "processing_status", isProcessing: true }); @@ -718,8 +727,7 @@ describe("SessionMessageQueue", () => { it("leaves the prompt pending and timeline untouched when sandbox send fails", async () => { const h = buildQueue(); h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-unsent" })); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); - h.wsManager.send.mockReturnValue(false); + h.messenger.sendToSandbox.mockRejectedValue(new SandboxDeliveryUnavailableError("send_failed")); await h.queue.processMessageQueue(); @@ -744,11 +752,11 @@ describe("SessionMessageQueue", () => { const h = buildQueue(); h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-lost" })); h.repository.startMessageProcessing.mockReturnValue(false); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); - expect(h.wsManager.send).not.toHaveBeenCalled(); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalled(); expect(h.broadcast).not.toHaveBeenCalledWith( expect.objectContaining({ type: "processing_status" }) ); @@ -769,17 +777,15 @@ describe("SessionMessageQueue", () => { it("drops a persisted reasoning effort that the session model does not support", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; h.repository.getNextPendingMessage.mockReturnValue(createMessage()); h.repository.getSession.mockReturnValue( createSession({ model: "xai/grok-build-0.1", reasoning_effort: "high" }) ); - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); - expect(h.wsManager.send).toHaveBeenCalledWith( - sandboxWs, + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith( expect.objectContaining({ model: "xai/grok-build-0.1", reasoningEffort: undefined, @@ -789,7 +795,6 @@ describe("SessionMessageQueue", () => { it("falls back atomically when GitHub author mapping is incomplete", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-agent-only" })); h.repository.getParticipantById.mockReturnValue( createParticipant({ @@ -799,12 +804,11 @@ describe("SessionMessageQueue", () => { scm_email: "private@example.com", }) ); - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); - expect(h.wsManager.send).toHaveBeenCalledWith( - sandboxWs, + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith( expect.objectContaining({ author: { userId: "user-1", @@ -816,8 +820,7 @@ describe("SessionMessageQueue", () => { it("resolves each dispatched prompt's Git author from its current participant", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); h.repository.getNextPendingMessage .mockReturnValueOnce(createMessage({ id: "msg-ada", author_id: "part-ada" })) .mockReturnValueOnce(createMessage({ id: "msg-grace", author_id: "part-grace" })); @@ -844,7 +847,7 @@ describe("SessionMessageQueue", () => { await h.queue.processMessageQueue(); await h.queue.processMessageQueue(); - expect(h.wsManager.send.mock.calls.map(([, command]) => command)).toEqual([ + expect(h.messenger.sendToSandbox.mock.calls.map(([command]) => command)).toEqual([ expect.objectContaining({ author: { userId: "user-ada", @@ -870,9 +873,8 @@ describe("SessionMessageQueue", () => { it("notifies the integration after a prompt is dispatched to the sandbox", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-linear" })); - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); @@ -883,8 +885,7 @@ describe("SessionMessageQueue", () => { it("does not notify the integration when sandbox dispatch fails", async () => { const h = buildQueue(); h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-failed" })); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); - h.wsManager.send.mockReturnValue(false); + h.messenger.sendToSandbox.mockRejectedValue(new SandboxDeliveryUnavailableError("send_failed")); await h.queue.processMessageQueue(); @@ -895,7 +896,7 @@ describe("SessionMessageQueue", () => { describe("execution timeout scheduling", () => { function dispatchPrompt(h: ReturnType) { h.repository.getNextPendingMessage.mockReturnValue(createMessage()); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); + h.messenger.sendToSandbox.mockResolvedValue(undefined); return h.queue.processMessageQueue(); } @@ -946,8 +947,7 @@ describe("SessionMessageQueue", () => { it("delegates stop finalization before broadcasting idle and stopping the sandbox", async () => { const h = buildQueue(); - const sandboxWs = { readyState: 1 } as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); + h.messenger.sendToSandbox.mockResolvedValue(undefined); h.repository.getProcessingMessageWithCreatedAt.mockReturnValue({ id: "msg-9", created_at: 900, @@ -970,14 +970,14 @@ describe("SessionMessageQueue", () => { expect.any(Number) ); expect(h.broadcast).toHaveBeenCalledWith({ type: "processing_status", isProcessing: false }); - expect(h.wsManager.send).toHaveBeenCalledWith(sandboxWs, { type: "stop" }); + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "stop" }); expect(h.repository.recordMessageCompletion.mock.invocationCallOrder[0]).toBeLessThan( h.repository.markMessageAwaitingStopConfirmation.mock.invocationCallOrder[0] ); expect(h.projectTerminalMessage).toHaveBeenCalledWith("msg-9", 1000, expect.any(Number)); expect( h.repository.markMessageAwaitingStopConfirmation.mock.invocationCallOrder[0] - ).toBeLessThan(h.wsManager.send.mock.invocationCallOrder[0]); + ).toBeLessThan(h.messenger.sendToSandbox.mock.invocationCallOrder[0]); }); it("projects terminal unread state before broadcasting synthetic completion", async () => { @@ -1013,7 +1013,7 @@ describe("SessionMessageQueue", () => { created_at: 900, }); h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-next" })); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.stopExecution(); @@ -1021,9 +1021,8 @@ describe("SessionMessageQueue", () => { "msg-next", expect.any(Number) ); - expect(h.wsManager.send).toHaveBeenCalledWith(expect.anything(), { type: "stop" }); - expect(h.wsManager.send).not.toHaveBeenCalledWith( - expect.anything(), + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "stop" }); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalledWith( expect.objectContaining({ type: "prompt", messageId: "msg-next" }) ); expect(h.setAlarm).toHaveBeenCalledOnce(); @@ -1036,7 +1035,7 @@ describe("SessionMessageQueue", () => { created_at: 900, }); h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-next" })); - h.wsManager.getSandboxSocket.mockReturnValue(null); + h.messenger.sendToSandbox.mockRejectedValue(new SandboxDeliveryUnavailableError()); await h.queue.stopExecution(); @@ -1052,8 +1051,7 @@ describe("SessionMessageQueue", () => { id: "msg-running", created_at: 900, }); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); - h.wsManager.send.mockReturnValue(false); + h.messenger.sendToSandbox.mockRejectedValue(new SandboxDeliveryUnavailableError("send_failed")); await h.queue.stopExecution(); @@ -1095,12 +1093,12 @@ describe("SessionMessageQueue", () => { deadline: Date.now() + 10_000, }); h.repository.getNextPendingMessage.mockReturnValue(createMessage({ id: "msg-next" })); - h.wsManager.getSandboxSocket.mockReturnValue({ readyState: 1 } as WebSocket); + h.messenger.sendToSandbox.mockResolvedValue(undefined); await h.queue.processMessageQueue(); expect(h.repository.updateMessageToProcessing).not.toHaveBeenCalled(); - expect(h.wsManager.send).not.toHaveBeenCalled(); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalled(); }); it("suppresses session status reconcile when stopExecution is called with suppress flag", async () => { @@ -1121,7 +1119,7 @@ describe("SessionMessageQueue", () => { await h.queue.failStuckProcessingMessage(); expect(h.repository.recordMessageCompletion).not.toHaveBeenCalled(); - expect(h.wsManager.send).not.toHaveBeenCalledWith(expect.anything(), { type: "stop" }); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalledWith({ type: "stop" }); expect(h.sessionStatus.reconcileAfterExecution).not.toHaveBeenCalled(); }); diff --git a/packages/control-plane/src/session/message-queue.ts b/packages/control-plane/src/session/message-queue.ts index 133a6f2cd..606d6a507 100644 --- a/packages/control-plane/src/session/message-queue.ts +++ b/packages/control-plane/src/session/message-queue.ts @@ -26,7 +26,6 @@ import { type SessionAttachmentRepository, } from "./session-attachment-repository"; import type { SessionMessenger } from "./messenger"; -import type { SessionWebSocketManager } from "./websocket-manager"; import type { ParticipantService } from "./participant-service"; import type { CallbackNotificationService } from "./callback-notification-service"; import type { SessionStatusService } from "./session-status-service"; @@ -34,6 +33,8 @@ import type { EnqueuePromptRequest } from "./enqueue-prompt-contract"; import { getAvatarUrl } from "./participant-service"; import { resolveParticipantName } from "./participant-name"; import type { AlarmScheduler, BackgroundTasks } from "../platform-ports"; +import { SandboxDeliveryUnavailableError } from "./connections"; +import type { ClientResponder } from "./ports"; import { resolveGitAuthorIdentity } from "./identity"; import { validateReasoningEffort } from "./reasoning-effort"; import { @@ -139,7 +140,7 @@ function resolveParticipantGitIdentity( : { mode: "agent-only" }; } -export class SessionMessageQueue { +export class SessionMessageQueue { constructor( private readonly backgroundTasks: BackgroundTasks, private readonly log: Logger, @@ -147,7 +148,7 @@ export class SessionMessageQueue { private readonly messageRepository: MessageRepository, private readonly participantRepository: ParticipantRepository, private readonly attachmentRepository: SessionAttachmentRepository, - private readonly wsManager: SessionWebSocketManager, + private readonly clientResponder: ClientResponder, private readonly messenger: SessionMessenger, private readonly participantService: ParticipantService, private readonly callbackService: CallbackNotificationService, @@ -165,7 +166,7 @@ export class SessionMessageQueue { ) {} async handlePromptMessage( - ws: WebSocket, + ws: Connection, client: ClientInfo, data: PromptMessageData ): Promise { @@ -190,7 +191,7 @@ export class SessionMessageQueue { }); } catch (error) { if (error instanceof SessionAttachmentError) { - this.wsManager.send(ws, { + this.clientResponder.send(ws, { type: "error", code: "INVALID_ATTACHMENTS", message: error.message, @@ -199,7 +200,7 @@ export class SessionMessageQueue { return; } if (error instanceof SessionNotPromptableError) { - this.wsManager.send(ws, { + this.clientResponder.send(ws, { type: "error", code: "SESSION_NOT_PROMPTABLE", message: error.message, @@ -208,7 +209,7 @@ export class SessionMessageQueue { return; } if (error instanceof PromptQueueFullError) { - this.wsManager.send(ws, { + this.clientResponder.send(ws, { type: "error", code: "PROMPT_QUEUE_FULL", message: error.message, @@ -217,7 +218,7 @@ export class SessionMessageQueue { return; } if (error instanceof PromptRequestConflictError) { - this.wsManager.send(ws, { + this.clientResponder.send(ws, { type: "error", code: "PROMPT_REQUEST_CONFLICT", message: error.message, @@ -240,22 +241,24 @@ export class SessionMessageQueue { } } - this.wsManager.send(ws, { - type: "prompt_queued", - clientRequestId: data.clientRequestId, - messageId: enqueued.messageId, - position: enqueued.position, - }); + if (data.clientRequestId) { + this.clientResponder.send(ws, { + type: "prompt_queued", + clientRequestId: data.clientRequestId, + messageId: enqueued.messageId, + position: enqueued.position, + }); + } await this.processMessageQueue(); } async cancelQueuedPrompt( - ws: WebSocket, + ws: Connection, data: { messageId: string; clientRequestId: string } ): Promise { if (!this.messageRepository.cancelPendingMessage(data.messageId)) { - this.wsManager.send(ws, { + this.clientResponder.send(ws, { type: "error", code: "PROMPT_NOT_CANCELLABLE", message: "This prompt is no longer pending and cannot be removed", @@ -264,7 +267,7 @@ export class SessionMessageQueue { return; } - this.wsManager.send(ws, { + this.clientResponder.send(ws, { type: "prompt_cancelled", clientRequestId: data.clientRequestId, messageId: data.messageId, @@ -304,39 +307,6 @@ export class SessionMessageQueue { } const now = Date.now(); - const sandboxWs = this.wsManager.getSandboxSocket(); - if (!sandboxWs) { - this.log.info("prompt.dispatch", { - event: "prompt.dispatch", - message_id: message.id, - outcome: "deferred", - reason: "no_sandbox", - }); - this.messenger.broadcast({ type: "sandbox_spawning" }); - // Spawn in the background: a snapshot restore can take tens of seconds, - // and awaiting it here holds the prompt HTTP response open past bot - // callers' request timeouts. The message is already persisted as - // pending and dispatches when the sandbox WebSocket connects. - this.backgroundTasks.submit( - this.sandboxLifecycle.spawnSandbox().catch((error) => { - // Expected provider failures broadcast sandbox_error inside the - // lifecycle manager; this catch only sees throws from before those - // handlers. Surface them the same way so clients aren't left - // watching a silent "sandbox_spawning" forever. - this.messenger.broadcast({ - type: "sandbox_error", - error: error instanceof Error ? error.message : "Failed to spawn sandbox", - }); - throw error; - }), - { - name: "sandbox.spawn", - context: { message_id: message.id }, - } - ); - return; - } - const author = this.participantRepository.getParticipantById(message.author_id); if (!author) { throw new Error(`Missing prompt author ${message.author_id}`); @@ -385,12 +355,20 @@ export class SessionMessageQueue { return; } - const sent = this.wsManager.send(sandboxWs, command); - - if (!sent) { + let sent = false; + try { + await this.messenger.sendToSandbox(command); + sent = true; + } catch (error) { this.messageRepository.updateMessageToPending(message.id); + if (error instanceof SandboxDeliveryUnavailableError && error.reason === "not_connected") { + this.deferUntilSandboxConnects(message.id); + return; + } await this.sandboxLifecycle.terminateUnresponsiveSandbox("prompt_dispatch_send_failed"); - } else { + } + + if (sent) { this.messenger.broadcast({ type: "sandbox_event", event: userMessageEvent }); this.messenger.broadcast({ type: "processing_status", isProcessing: true }); this.broadcastPromptQueue(); @@ -416,7 +394,6 @@ export class SessionMessageQueue { user_id: author?.user_id ?? "unknown", source: message.source, has_sandbox_ws: true, - sandbox_ready_state: sandboxWs.readyState, queue_wait_ms: now - message.created_at, has_attachments: !!message.attachments, }); @@ -450,9 +427,12 @@ export class SessionMessageQueue { this.messenger.broadcast({ type: "processing_status", isProcessing: false }); - const sandboxWs = this.wsManager.getSandboxSocket(); - if (stoppedMessageId && (!sandboxWs || !this.wsManager.send(sandboxWs, { type: "stop" }))) { - await this.sandboxLifecycle.terminateUnresponsiveSandbox("stop_send_failed"); + if (stoppedMessageId) { + try { + await this.messenger.sendToSandbox({ type: "stop" }); + } catch { + await this.sandboxLifecycle.terminateUnresponsiveSandbox("stop_send_failed"); + } } } @@ -488,8 +468,33 @@ export class SessionMessageQueue { this.messenger.broadcast({ type: "processing_status", isProcessing: false }); this.broadcastPromptQueue(); - const sandboxWs = this.wsManager.getSandboxSocket(); - if (sandboxWs) this.wsManager.send(sandboxWs, { type: "stop" }); + void this.messenger.sendToSandbox({ type: "stop" }).catch(() => { + // Cancellation is synchronous and sandbox termination owns eventual cleanup. + }); + } + + private deferUntilSandboxConnects(messageId: string): void { + this.log.info("prompt.dispatch", { + event: "prompt.dispatch", + message_id: messageId, + outcome: "deferred", + reason: "no_sandbox", + }); + this.messenger.broadcast({ type: "sandbox_spawning" }); + // Keep the caller responsive while a snapshot restore runs in the background. + this.backgroundTasks.submit( + this.sandboxLifecycle.spawnSandbox().catch((error) => { + this.messenger.broadcast({ + type: "sandbox_error", + error: error instanceof Error ? error.message : "Failed to spawn sandbox", + }); + throw error; + }), + { + name: "sandbox.spawn", + context: { message_id: messageId }, + } + ); } /** diff --git a/packages/control-plane/src/session/messenger.test.ts b/packages/control-plane/src/session/messenger.test.ts index bae8df108..b4a491705 100644 --- a/packages/control-plane/src/session/messenger.test.ts +++ b/packages/control-plane/src/session/messenger.test.ts @@ -4,12 +4,8 @@ import type { SessionConnections } from "./connections"; function harness() { const connections = { - registerBrowser: vi.fn(async () => {}), - registerSandbox: vi.fn(async () => {}), sendToSandbox: vi.fn(async () => {}), broadcastToBrowsers: vi.fn(async () => {}), - disconnectSandbox: vi.fn(async () => {}), - listParticipants: vi.fn(async () => []), } satisfies SessionConnections; return { messenger: new SessionMessengerImpl(connections), connections }; } diff --git a/packages/control-plane/src/session/messenger.ts b/packages/control-plane/src/session/messenger.ts index e27ece1f8..053c0c649 100644 --- a/packages/control-plane/src/session/messenger.ts +++ b/packages/control-plane/src/session/messenger.ts @@ -6,8 +6,9 @@ import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; import type { SandboxCommand } from "./types"; import type { SessionConnections } from "./connections"; +import type { SandboxCommandSender } from "./ports"; -export interface SessionMessenger { +export interface SessionMessenger extends SandboxCommandSender { /** Broadcast a message to all authenticated client sockets. */ broadcast(message: ServerMessage): void; diff --git a/packages/control-plane/src/session/ports.ts b/packages/control-plane/src/session/ports.ts index 4cc47b9c8..a1c5da58d 100644 --- a/packages/control-plane/src/session/ports.ts +++ b/packages/control-plane/src/session/ports.ts @@ -1,3 +1,7 @@ +import type { SandboxStatus } from "@open-inspect/shared/types/sessions"; +import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; +import type { SandboxCommand } from "./types"; + /** Mutable state associated with one authenticated browser connection. */ export interface ConnectedClient { participantId: string; @@ -27,6 +31,16 @@ export interface SocketRegistry { hasParticipant(participantId: string): boolean; } +/** Typed command delivery to the active sandbox. */ +export interface SandboxCommandSender { + sendToSandbox(command: SandboxCommand): Promise; +} + +/** Typed replies to one browser connection. */ +export interface ClientResponder { + send(connection: Connection, message: ServerMessage): boolean; +} + /** Participant-facing notifications emitted by disconnect policy. */ export interface SessionBroadcaster { broadcast(message: ServerMessage): void; @@ -38,5 +52,3 @@ export interface SandboxDisconnectMonitor { getStatus(): SandboxStatus | undefined; scheduleCheck(): Promise; } -import type { SandboxStatus } from "@open-inspect/shared/types/sessions"; -import type { ServerMessage } from "@open-inspect/shared/types/server-messages"; diff --git a/packages/control-plane/src/session/sandbox-events.test.ts b/packages/control-plane/src/session/sandbox-events.test.ts index bd92c01f7..ae55b9c4f 100644 --- a/packages/control-plane/src/session/sandbox-events.test.ts +++ b/packages/control-plane/src/session/sandbox-events.test.ts @@ -11,7 +11,7 @@ import type { ArtifactRepository } from "./artifact-repository"; import type { EventRepository } from "./event-repository"; import type { MessageRepository } from "./message-repository"; import type { SessionStatusService } from "./session-status-service"; -import type { SessionWebSocketManager } from "./websocket-manager"; +import { SandboxDeliveryUnavailableError } from "./connections"; function createPushSpec(repoOwner: string, repoName: string, targetBranch: string): GitPushSpec { return { @@ -58,11 +58,6 @@ function createProcessor() { notifyComplete: vi.fn(async () => {}), }; - const wsManager = { - getSandboxSocket: vi.fn(() => null as WebSocket | null), - send: vi.fn(() => true), - }; - const broadcast = vi.fn((_message: ServerMessage) => {}); const messenger = { broadcast, sendToSandbox: vi.fn(async () => {}) }; const diffService = { pinBaselines: vi.fn() }; @@ -95,7 +90,6 @@ function createProcessor() { eventRepository, artifactRepository, callbackService as unknown as CallbackNotificationService, - wsManager as unknown as SessionWebSocketManager, messenger, diffService as unknown as SessionDiffService, applySessionTitleUpdate, @@ -113,7 +107,7 @@ function createProcessor() { artifactRepository, repository, eventRepository, - wsManager, + messenger, callbackService, broadcast, diffService, @@ -442,9 +436,7 @@ describe("SessionSandboxEventProcessor", () => { it("waits for terminal projection before snapshot, queue drain, and acknowledgement", async () => { const h = createProcessor(); - const sandboxWs = { readyState: WebSocket.OPEN } as WebSocket; h.repository.getProcessingMessage.mockReturnValue({ id: "msg-1" }); - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); let resolveCompletion!: () => void; h.projectTerminalMessage.mockReturnValue( new Promise((resolve) => { @@ -463,14 +455,14 @@ describe("SessionSandboxEventProcessor", () => { expect(h.triggerSnapshot).not.toHaveBeenCalled(); expect(h.processMessageQueue).not.toHaveBeenCalled(); - expect(h.wsManager.send).not.toHaveBeenCalled(); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalled(); resolveCompletion(); await processing; expect(h.triggerSnapshot).toHaveBeenCalledWith("execution_complete"); expect(h.processMessageQueue).toHaveBeenCalledOnce(); - expect(h.wsManager.send).toHaveBeenCalledWith(sandboxWs, { type: "ack", ackId: "ack-1" }); + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "ack", ackId: "ack-1" }); }); it("delegates a late terminal event with no processing owner", async () => { @@ -511,9 +503,6 @@ describe("SessionSandboxEventProcessor", () => { it("resolves pending push when push_complete event arrives", async () => { const h = createProcessor(); - const sandboxWs = { readyState: WebSocket.OPEN } as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); - const pushPromise = h.processor.pushBranchToRemote( createPushSpec("acme", "web", "feature/test") ); @@ -527,17 +516,40 @@ describe("SessionSandboxEventProcessor", () => { }); await expect(pushPromise).resolves.toEqual({ success: true }); - expect(h.wsManager.send).toHaveBeenCalledWith( - sandboxWs, + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith( expect.objectContaining({ type: "push" }) ); }); + it("treats a missing sandbox as an already-manual push", async () => { + const h = createProcessor(); + h.messenger.sendToSandbox.mockRejectedValue(new SandboxDeliveryUnavailableError()); + const legacyPushSpec = { + ...createPushSpec("acme", "web", "feature/test"), + repoOwner: undefined, + repoName: undefined, + } as unknown as GitPushSpec; + + await expect(h.processor.pushBranchToRemote(legacyPushSpec)).resolves.toEqual({ + success: true, + }); + }); + + it("reports a sandbox transport send failure", async () => { + const h = createProcessor(); + h.messenger.sendToSandbox.mockRejectedValue(new SandboxDeliveryUnavailableError("send_failed")); + + await expect( + h.processor.pushBranchToRemote(createPushSpec("acme", "web", "feature/test")) + ).resolves.toEqual({ + success: false, + error: expect.stringContaining("Failed to send to sandbox"), + }); + }); + describe("push resolver keying", () => { function connectSandbox(h: ReturnType) { - const sandboxWs = { readyState: WebSocket.OPEN } as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); - return sandboxWs; + h.messenger.sendToSandbox.mockResolvedValue(undefined); } it("settles the matching push when two repos push the same branch name", async () => { @@ -780,8 +792,6 @@ describe("SessionSandboxEventProcessor", () => { describe("ACK mechanism", () => { it("sends ACK after execution_complete when ackId is present", async () => { const h = createProcessor(); - const sandboxWs = {} as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); h.repository.getProcessingMessage.mockReturnValue({ id: "msg-1" }); const event = { @@ -795,7 +805,7 @@ describe("SessionSandboxEventProcessor", () => { await h.processor.processSandboxEvent(event); - expect(h.wsManager.send).toHaveBeenCalledWith(sandboxWs, { + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "ack", ackId: "execution_complete:msg-1", }); @@ -803,8 +813,6 @@ describe("SessionSandboxEventProcessor", () => { it("sends ACK for push_complete when ackId is present", async () => { const h = createProcessor(); - const sandboxWs = {} as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); const event = { type: "push_complete", @@ -815,7 +823,7 @@ describe("SessionSandboxEventProcessor", () => { await h.processor.processSandboxEvent(event); - expect(h.wsManager.send).toHaveBeenCalledWith(sandboxWs, { + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "ack", ackId: "push_complete:msg-2", }); @@ -823,8 +831,6 @@ describe("SessionSandboxEventProcessor", () => { it("sends ACK for error events when ackId is present", async () => { const h = createProcessor(); - const sandboxWs = {} as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); const event = { type: "error", @@ -837,7 +843,7 @@ describe("SessionSandboxEventProcessor", () => { await h.processor.processSandboxEvent(event); - expect(h.wsManager.send).toHaveBeenCalledWith(sandboxWs, { + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "ack", ackId: "error:msg-3", }); @@ -845,8 +851,6 @@ describe("SessionSandboxEventProcessor", () => { it("does not send ACK when ackId is absent (backward compatibility)", async () => { const h = createProcessor(); - const sandboxWs = {} as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); h.repository.getProcessingMessage.mockReturnValue({ id: "msg-1" }); const event: SandboxEvent = { @@ -859,13 +863,11 @@ describe("SessionSandboxEventProcessor", () => { await h.processor.processSandboxEvent(event); - expect(h.wsManager.send).not.toHaveBeenCalled(); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalled(); }); it("sends ACK on already_stopped path for execution_complete", async () => { const h = createProcessor(); - const sandboxWs = {} as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); // No processing message — triggers the "already_stopped" branch h.repository.getProcessingMessage.mockReturnValue(null); @@ -880,7 +882,7 @@ describe("SessionSandboxEventProcessor", () => { await h.processor.processSandboxEvent(event); - expect(h.wsManager.send).toHaveBeenCalledWith(sandboxWs, { + expect(h.messenger.sendToSandbox).toHaveBeenCalledWith({ type: "ack", ackId: "execution_complete:msg-1", }); @@ -889,8 +891,6 @@ describe("SessionSandboxEventProcessor", () => { it("does not send ACK for non-critical events even with ackId", async () => { const h = createProcessor(); - const sandboxWs = {} as WebSocket; - h.wsManager.getSandboxSocket.mockReturnValue(sandboxWs); const event = { type: "token", @@ -904,7 +904,30 @@ describe("SessionSandboxEventProcessor", () => { await h.processor.processSandboxEvent(event); // Token events return early before ACK logic - expect(h.wsManager.send).not.toHaveBeenCalled(); + expect(h.messenger.sendToSandbox).not.toHaveBeenCalled(); + }); + + it("logs failed ACK delivery without failing event processing", async () => { + const h = createProcessor(); + h.messenger.sendToSandbox.mockRejectedValue( + new SandboxDeliveryUnavailableError("send_failed") + ); + + await expect( + h.processor.processSandboxEvent({ + type: "error", + error: "failed", + sandboxId: "sb-1", + timestamp: 3000, + ackId: "error:msg-1", + } as SandboxEvent & { ackId: string }) + ).resolves.toBeUndefined(); + await Promise.resolve(); + + expect(h.log.warn).toHaveBeenCalledWith( + "Failed to send sandbox event ACK", + expect.objectContaining({ ack_id: "error:msg-1" }) + ); }); }); }); diff --git a/packages/control-plane/src/session/sandbox-events.ts b/packages/control-plane/src/session/sandbox-events.ts index f237ca6e2..dd4610497 100644 --- a/packages/control-plane/src/session/sandbox-events.ts +++ b/packages/control-plane/src/session/sandbox-events.ts @@ -13,9 +13,9 @@ import type { CallbackNotificationService } from "./callback-notification-servic import type { SessionDiffService } from "./diffs/service"; import type { SessionMessenger } from "./messenger"; import type { SessionStatusService } from "./session-status-service"; -import type { SessionWebSocketManager } from "./websocket-manager"; import type { SessionTitleUpdateOptions, SessionTitleUpdateResult } from "./title"; import type { BackgroundTasks } from "../platform-ports"; +import { SandboxDeliveryUnavailableError } from "./connections"; type PushResolver = { resolve: () => void; reject: (err: Error) => void }; type SandboxEventWithAck = SandboxEvent & { ackId?: string }; @@ -48,7 +48,6 @@ export class SessionSandboxEventProcessor { private readonly eventRepository: EventRepository, private readonly artifactRepository: ArtifactRepository, private readonly callbackService: CallbackNotificationService, - private readonly wsManager: SessionWebSocketManager, private readonly messenger: SessionMessenger, private readonly diffService: SessionDiffService, private readonly applySessionTitleUpdate: ( @@ -298,13 +297,6 @@ export class SessionSandboxEventProcessor { async pushBranchToRemote( pushSpec: GitPushSpec ): Promise<{ success: true } | { success: false; error: string }> { - const sandboxWs = this.wsManager.getSandboxSocket(); - - if (!sandboxWs) { - this.log.info("No sandbox connected, assuming branch was pushed manually"); - return { success: true }; - } - const resolverKey = this.pushResolverKey( pushSpec.repoOwner, pushSpec.repoName, @@ -328,10 +320,21 @@ export class SessionSandboxEventProcessor { repo_owner: pushSpec.repoOwner, repo_name: pushSpec.repoName, }); - this.wsManager.send(sandboxWs, { - type: "push", - pushSpec, - }); + try { + await this.messenger.sendToSandbox({ type: "push", pushSpec }); + } catch (error) { + this.pendingPushResolvers.delete(resolverKey); + if (timeoutId) clearTimeout(timeoutId); + if (error instanceof SandboxDeliveryUnavailableError && error.reason === "not_connected") { + this.log.info("No sandbox connected, assuming branch was pushed manually"); + return { success: true }; + } + this.log.error("Failed to send push command", { + branch_name: pushSpec.targetBranch, + error: error instanceof Error ? error : String(error), + }); + return { success: false, error: `Failed to push branch: ${error}` }; + } try { await pushPromise; @@ -406,15 +409,23 @@ export class SessionSandboxEventProcessor { private sendAck(ackId: string | undefined): void { if (!ackId) return; - const sandboxWs = this.wsManager.getSandboxSocket(); - if (sandboxWs) { - this.wsManager.send(sandboxWs, { type: "ack", ackId }); - } else { - this.log.debug("Cannot send ACK: no sandbox socket", { ack_id: ackId }); - } + void this.messenger.sendToSandbox({ type: "ack", ackId }).catch((error) => { + if (error instanceof SandboxDeliveryUnavailableError && error.reason === "not_connected") { + this.log.debug("Cannot send ACK: no sandbox socket", { ack_id: ackId }); + return; + } + this.log.warn("Failed to send sandbox event ACK", { + ack_id: ackId, + error: error instanceof Error ? error : String(error), + }); + }); } - private pushResolverKey(repoOwner: string, repoName: string, branchName: string): string { - return `${repoOwner.toLowerCase()}/${repoName.toLowerCase()}::${branchName.trim().toLowerCase()}`; + private pushResolverKey( + repoOwner: string | undefined, + repoName: string | undefined, + branchName: string + ): string { + return `${repoOwner?.toLowerCase() ?? ""}/${repoName?.toLowerCase() ?? ""}::${branchName.trim().toLowerCase()}`; } } diff --git a/packages/control-plane/src/session/server.test.ts b/packages/control-plane/src/session/server.test.ts index ae5d68d25..11de7f6a6 100644 --- a/packages/control-plane/src/session/server.test.ts +++ b/packages/control-plane/src/session/server.test.ts @@ -8,7 +8,7 @@ import { type SessionClientCommands, type SessionMessageRouterDeps, } from "./message-router"; -import type { Clock, SandboxDisconnectMonitor, SessionBroadcaster, SocketRegistry } from "./ports"; +import type { Clock, SocketRegistry, SandboxDisconnectMonitor, SessionBroadcaster } from "./ports"; import { SessionServer } from "./server"; interface TestClient {