summaryrefslogtreecommitdiffhomepage
path: root/packages/transport-ws/src
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-21 16:56:42 +0900
committerAdam Malczewski <[email protected]>2026-06-21 16:56:42 +0900
commitea0e938eca3072649dc8707c999ec00cf87b986a (patch)
tree41f96ed517fe779ed612fc4cb3e8359a841ad088 /packages/transport-ws/src
parent36e950ba2cd2591e86f0dcc898f740481c59d912 (diff)
downloaddispatch-ea0e938eca3072649dc8707c999ec00cf87b986a.tar.gz
dispatch-ea0e938eca3072649dc8707c999ec00cf87b986a.zip
feat(transport): CLI endpoints + conversation.open broadcast (Wave 2)
transport-http: GET /conversations (list with ?q= prefix filter), GET /conversations/:id/last (blocks until turn settles, returns last AI text), POST /conversations/:id/open (emits conversationOpened hook), PUT /conversations/:id/title (set title). emit threaded from host.emit. extractLastAssistantText pure helper. 21 new tests (166 total). transport-ws: subscribes to conversationOpened hook, broadcasts ConversationOpenMessage to all connected WS clients. 2 new tests. session-orchestrator: conversationOpened hook descriptor (exported).
Diffstat (limited to 'packages/transport-ws/src')
-rw-r--r--packages/transport-ws/src/extension.ts31
-rw-r--r--packages/transport-ws/src/server.bun.test.ts79
2 files changed, 108 insertions, 2 deletions
diff --git a/packages/transport-ws/src/extension.ts b/packages/transport-ws/src/extension.ts
index 7a4e707..8b07a54 100644
--- a/packages/transport-ws/src/extension.ts
+++ b/packages/transport-ws/src/extension.ts
@@ -8,7 +8,7 @@
import type { Extension, HostAPI } from "@dispatch/kernel";
import type { SessionOrchestrator } from "@dispatch/session-orchestrator";
-import { sessionOrchestratorHandle } from "@dispatch/session-orchestrator";
+import { conversationOpened, sessionOrchestratorHandle } from "@dispatch/session-orchestrator";
import type { SurfaceContext, SurfaceProvider, SurfaceRegistry } from "@dispatch/surface-registry";
import { surfaceRegistryHandle } from "@dispatch/surface-registry";
import type { WsClientMessage, WsServerMessage } from "@dispatch/transport-contract";
@@ -27,6 +27,10 @@ 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> = [];
return {
manifest,
@@ -44,6 +48,13 @@ export function createTransportWsExtension(): Extension {
}
}
+ /** 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
@@ -113,6 +124,17 @@ export function createTransportWsExtension(): Extension {
}
}
+ // 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.
+ disposers.push(
+ host.on(conversationOpened, ({ conversationId }) => {
+ broadcast({ type: "conversation.open", conversationId });
+ }),
+ );
+
server = Bun.serve<ConnectionState>({
port,
fetch(req, srv) {
@@ -126,6 +148,7 @@ export function createTransportWsExtension(): Extension {
},
websocket: {
open(ws) {
+ connections.add(ws);
logger.debug("transport-ws: connection open");
send(ws, catalogMessage(registry));
},
@@ -297,6 +320,7 @@ export function createTransportWsExtension(): Extension {
},
close(ws) {
+ connections.delete(ws);
const state = ws.data;
if (state) {
// Dispose all chat subscriptions (does NOT abort turns).
@@ -318,6 +342,11 @@ export function createTransportWsExtension(): Extension {
},
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/server.bun.test.ts b/packages/transport-ws/src/server.bun.test.ts
index da3be5e..e723766 100644
--- a/packages/transport-ws/src/server.bun.test.ts
+++ b/packages/transport-ws/src/server.bun.test.ts
@@ -240,7 +240,20 @@ function startServer(
logger?: Logger,
) {
const log = logger ?? fakeLogger();
- return Bun.serve<ConnectionState>({
+ 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 = {
@@ -253,6 +266,7 @@ function startServer(
},
websocket: {
open(ws) {
+ connections.add(ws);
log.debug("transport-ws: connection open");
ws.send(JSON.stringify(catalogMessage(registry)));
},
@@ -392,6 +406,7 @@ function startServer(
},
close(ws) {
+ connections.delete(ws);
const state = ws.data;
if (state) {
for (const dispose of state.chatSubscriptions.values()) {
@@ -406,6 +421,17 @@ function startServer(
},
},
});
+
+ /**
+ * Simulate the `conversationOpened` hook firing — mirrors the
+ * `host.on(conversationOpened, ...)` subscription in extension.ts, which
+ * broadcasts a `conversation.open` WS message to every connected client.
+ */
+ return Object.assign(server, {
+ triggerConversationOpen(conversationId: string): void {
+ broadcast({ type: "conversation.open", conversationId });
+ },
+ });
}
// ── Helpers ─────────────────────────────────────────────────────────────────
@@ -938,3 +964,54 @@ describe("logging", () => {
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");
+
+ const msg = await waitForMessage(ws);
+ expect(msg).toEqual({ type: "conversation.open", conversationId: "conv-42" });
+
+ 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.
+ server.triggerConversationOpen("shared-conv");
+
+ const [msg1, msg2] = await Promise.all([waitForMessage(ws1), waitForMessage(ws2)]);
+ expect(msg1).toEqual({ type: "conversation.open", conversationId: "shared-conv" });
+ expect(msg2).toEqual({ type: "conversation.open", conversationId: "shared-conv" });
+
+ ws1.close();
+ ws2.close();
+ });
+});