feat(chat-runtime): tRPC router over commands + WebSocket sink adapter

packages/chat-runtime/README.md

@@ -18,6 +18,7 @@ The runtime speaks only the vocabulary in `plans/chat-protocol-v1.md`; it never
 | `sessions/` | `liveSession/` (one running session: event pump, FIFO prompt queue, cancel) and `registry/`, which builds one per harness |
 | `stream/` | `subscriptions/` — `SubscriptionHub`: replay-then-live subscribe, per-subscriber delta channels, reset frames, delta coalescing |
 | `commands/` | The client-facing verbs, each parsed with the `@superset/chat` command schemas and deduped by `commandId` |
+| `router/` | `router/` — `createChatRouter(runtime, { resolveCwd })`, the tRPC surface over `commands/` (the package owns its own `initTRPC`; procedures close over the runtime, and the host resolves `cwd` — clients never send it) — and `wsSink/`, the structural WebSocket→`Sink` adapter host-service's stream route mounts |
 | `testing/` | The cross-cutting test helpers no single module owns — `fixtures/` (protocol item factories), `testUtils/` (sinks, schedules, waits) and `testRuntime/` (the bun-sqlite runtime). Internal only: relative imports, no package export. Helpers that do have an owner stay beside it, like `harness/fake/` |
 
 ## Wiring a harness

packages/chat-runtime/package.json

@@ -18,6 +18,7 @@
 	},
 	"dependencies": {
 		"@superset/chat": "workspace:*",
+		"@trpc/server": "11.16.0",
 		"better-sqlite3": "12.6.2",
 		"drizzle-orm": "0.45.2",
 		"zod": "4.4.3"

packages/chat-runtime/src/index.ts

@@ -15,6 +15,7 @@ export * from "./harness";
 export * from "./journal";
 export * from "./projection";
 export * from "./replay";
+export * from "./router";
 export * from "./sessions";
 export * from "./stream";
 

packages/chat-runtime/src/router/index.ts

@@ -0,0 +1,4 @@
+export type { ChatRouter, ChatRouterOptions } from "./router";
+export { createChatCallerFactory, createChatRouter } from "./router";
+export type { WsSinkSocket } from "./wsSink";
+export { createWsSink } from "./wsSink";

packages/chat-runtime/src/router/router/index.ts

@@ -0,0 +1,2 @@
+export type { ChatRouter, ChatRouterOptions } from "./router";
+export { createChatCallerFactory, createChatRouter } from "./router";

packages/chat-runtime/src/router/router/router.test.ts

@@ -0,0 +1,222 @@
+import { describe, expect, test } from "bun:test";
+import { randomUUID } from "node:crypto";
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import type { FakeHarnessScript } from "../../harness/fake";
+import type { ChatRuntime } from "../../index";
+import { agentMessage, turn } from "../../testing/fixtures";
+import { createTestRuntime } from "../../testing/testRuntime";
+import {
+	FAKE_HARNESS,
+	fakeHarnessRegistry,
+	journalEnvelopes,
+	waitFor,
+} from "../../testing/testUtils";
+import { createChatCallerFactory, createChatRouter } from "./router";
+
+const SCRIPT: FakeHarnessScript = {
+	turns: [
+		[
+			{ kind: "turn", turn: turn("t1") },
+			{ kind: "item", item: agentMessage("a1", "done"), turnId: "t1" },
+			{
+				kind: "turn",
+				turn: turn("t1", { status: "completed", completedAtMs: 2 }),
+			},
+			{ kind: "session", session: { status: "idle" } },
+		],
+	],
+};
+
+function newCaller() {
+	const { harnesses, adapters } = fakeHarnessRegistry(SCRIPT);
+	const runtime = createTestRuntime({ harnesses });
+	const cwd = mkdtempSync(join(tmpdir(), "chat-router-cwd-"));
+	const resolvedWorkspaceIds: string[] = [];
+	const router = createChatRouter(runtime, {
+		resolveCwd: (workspaceId) => {
+			resolvedWorkspaceIds.push(workspaceId);
+			return cwd;
+		},
+	});
+	const caller = createChatCallerFactory(router)({});
+	return {
+		runtime,
+		caller,
+		resolvedWorkspaceIds,
+		adapterCount: () => adapters.length,
+	};
+}
+
+function createSessionInput(workspaceId = "workspace-1") {
+	return {
+		commandId: randomUUID(),
+		workspaceId,
+		harness: FAKE_HARNESS,
+	};
+}
+
+async function promptIdle(
+	caller: ReturnType<typeof newCaller>["caller"],
+	runtime: ChatRuntime,
+	sessionId: string,
+): Promise<void> {
+	await caller.prompt({
+		commandId: randomUUID(),
+		sessionId,
+		clientId: "client-1",
+		content: [{ type: "text", text: "hello" }],
+	});
+	await waitFor(() => runtime.sessions.get(sessionId)?.status === "idle");
+}
+
+describe("createChatRouter", () => {
+	test("create, prompt and getItems round trip through the router", async () => {
+		const { runtime, caller, resolvedWorkspaceIds } = newCaller();
+		const created = await caller.createSession(createSessionInput());
+		expect(created.sessionId).toMatch(/^[0-9a-f-]{36}$/);
+		expect(resolvedWorkspaceIds).toEqual(["workspace-1"]);
+
+		await promptIdle(caller, runtime, created.sessionId);
+
+		const page = await caller.getItems({ sessionId: created.sessionId });
+		expect(page.ok).toBe(true);
+		if (!page.ok) return;
+		const itemKinds = page.envelopes
+			.filter((envelope) => envelope.event.type === "item")
+			.map((envelope) =>
+				envelope.event.type === "item" ? envelope.event.item.kind : "",
+			);
+		expect(itemKinds).toContain("user_message");
+		expect(itemKinds).toContain("agent_message");
+
+		const session = await caller.getSession({ sessionId: created.sessionId });
+		expect(session.session).toMatchObject({ sessionId: created.sessionId });
+		expect(session.cursor).toEqual({
+			epoch: created.epoch,
+			seq: runtime.journal.cursor(created.sessionId).seq,
+		});
+		await runtime.dispose();
+	});
+
+	test("invalid input is rejected by the zod schemas", async () => {
+		const { runtime, caller } = newCaller();
+		const created = await caller.createSession(createSessionInput());
+
+		await expect(
+			caller.createSession({
+				commandId: randomUUID(),
+				workspaceId: "",
+				harness: FAKE_HARNESS,
+			}),
+		).rejects.toMatchObject({ code: "BAD_REQUEST" });
+
+		await expect(
+			caller.prompt({
+				commandId: "not-a-uuid",
+				sessionId: created.sessionId,
+				clientId: "client-1",
+				content: [{ type: "text", text: "hi" }],
+			}),
+		).rejects.toMatchObject({ code: "BAD_REQUEST" });
+
+		await expect(
+			caller.prompt({
+				commandId: randomUUID(),
+				sessionId: created.sessionId,
+				clientId: "client-1",
+				content: [],
+			}),
+		).rejects.toMatchObject({ code: "BAD_REQUEST" });
+
+		await expect(
+			caller.getItems({ sessionId: created.sessionId, limit: 0 }),
+		).rejects.toMatchObject({ code: "BAD_REQUEST" });
+		await runtime.dispose();
+	});
+
+	test("command errors surface as typed tRPC errors", async () => {
+		const { runtime, caller } = newCaller();
+
+		await expect(
+			caller.createSession({
+				commandId: randomUUID(),
+				workspaceId: "workspace-1",
+				harness: "nope",
+			}),
+		).rejects.toMatchObject({ code: "BAD_REQUEST" });
+
+		await expect(
+			caller.prompt({
+				commandId: randomUUID(),
+				sessionId: "missing",
+				clientId: "client-1",
+				content: [{ type: "text", text: "hi" }],
+			}),
+		).rejects.toMatchObject({ code: "NOT_FOUND" });
+
+		const created = await caller.createSession(createSessionInput());
+		await runtime.live.dispose(created.sessionId);
+		await expect(
+			caller.prompt({
+				commandId: randomUUID(),
+				sessionId: created.sessionId,
+				clientId: "client-1",
+				content: [{ type: "text", text: "hi" }],
+			}),
+		).rejects.toMatchObject({ code: "CONFLICT" });
+		await runtime.dispose();
+	});
+
+	test("a retried commandId dedupes through the router", async () => {
+		const { runtime, caller, adapterCount } = newCaller();
+		const input = createSessionInput();
+		const first = await caller.createSession(input);
+		const second = await caller.createSession(input);
+
+		expect(second).toEqual(first);
+		expect(adapterCount()).toBe(1);
+		expect(await caller.listSessions({})).toHaveLength(1);
+
+		const promptInput = {
+			commandId: randomUUID(),
+			sessionId: first.sessionId,
+			clientId: "client-1",
+			content: [{ type: "text" as const, text: "hello" }],
+		};
+		const promptResult = await caller.prompt(promptInput);
+		const retried = await caller.prompt(promptInput);
+		expect(retried).toEqual(promptResult);
+
+		await waitFor(
+			() => runtime.sessions.get(first.sessionId)?.status === "idle",
+		);
+		const userMessages = journalEnvelopes(runtime, first.sessionId).filter(
+			(envelope) =>
+				envelope.event.type === "item" &&
+				envelope.event.item.kind === "user_message",
+		);
+		expect(
+			new Set(
+				userMessages.map((envelope) =>
+					envelope.event.type === "item" ? envelope.event.item.id : "",
+				),
+			).size,
+		).toBe(1);
+		await runtime.dispose();
+	});
+
+	test("listSessions filters by workspace and honours the limit", async () => {
+		const { runtime, caller } = newCaller();
+		await caller.createSession(createSessionInput("workspace-1"));
+		await caller.createSession(createSessionInput("workspace-2"));
+
+		expect(await caller.listSessions({})).toHaveLength(2);
+		expect(
+			await caller.listSessions({ workspaceId: "workspace-2" }),
+		).toHaveLength(1);
+		expect(await caller.listSessions({ limit: 1 })).toHaveLength(1);
+		await runtime.dispose();
+	});
+});

packages/chat-runtime/src/router/router/router.ts

@@ -0,0 +1,100 @@
+import {
+	cancelTurnInputSchema,
+	createSessionInputSchema,
+	getItemsInputSchema,
+	getSessionInputSchema,
+	listSessionsInputSchema,
+	promptInputSchema,
+	respondToApprovalInputSchema,
+	setModeInputSchema,
+} from "@superset/chat/protocol";
+import { initTRPC, TRPCError } from "@trpc/server";
+import type { ChatRuntime } from "../../index";
+
+const t = initTRPC.create();
+
+export const createChatCallerFactory = t.createCallerFactory;
+
+export type ChatRouterOptions = {
+	resolveCwd(workspaceId: string): string | Promise<string>;
+};
+
+const UNKNOWN_HARNESS = /^unknown harness /;
+const NOT_RUNNING = /^chat session (.+) is not running$/;
+
+function mapCommandError(runtime: ChatRuntime, error: unknown): unknown {
+	if (error instanceof TRPCError) return error;
+	if (!(error instanceof Error)) return error;
+	if (UNKNOWN_HARNESS.test(error.message)) {
+		return new TRPCError({
+			code: "BAD_REQUEST",
+			message: error.message,
+			cause: error,
+		});
+	}
+	const sessionId = NOT_RUNNING.exec(error.message)?.[1];
+	if (sessionId) {
+		return new TRPCError({
+			code: runtime.sessions.get(sessionId) ? "CONFLICT" : "NOT_FOUND",
+			message: error.message,
+			cause: error,
+		});
+	}
+	return error;
+}
+
+export function createChatRouter(
+	runtime: ChatRuntime,
+	options: ChatRouterOptions,
+) {
+	function guarded<T>(execute: () => T): T {
+		try {
+			return execute();
+		} catch (error) {
+			throw mapCommandError(runtime, error);
+		}
+	}
+
+	return t.router({
+		createSession: t.procedure
+			.input(createSessionInputSchema)
+			.mutation(async ({ input }) => {
+				const cwd = await options.resolveCwd(input.workspaceId);
+				return guarded(() => runtime.commands.createSession({ ...input, cwd }));
+			}),
+
+		prompt: t.procedure
+			.input(promptInputSchema)
+			.mutation(({ input }) => guarded(() => runtime.commands.prompt(input))),
+
+		cancelTurn: t.procedure
+			.input(cancelTurnInputSchema)
+			.mutation(({ input }) =>
+				guarded(() => runtime.commands.cancelTurn(input)),
+			),
+
+		respondToApproval: t.procedure
+			.input(respondToApprovalInputSchema)
+			.mutation(({ input }) =>
+				guarded(() => runtime.commands.respondToApproval(input)),
+			),
+
+		setMode: t.procedure
+			.input(setModeInputSchema)
+			.mutation(({ input }) => guarded(() => runtime.commands.setMode(input))),
+
+		getSession: t.procedure
+			.input(getSessionInputSchema)
+			.query(({ input }) => runtime.commands.getSession(input)),
+
+		listSessions: t.procedure
+			.input(listSessionsInputSchema)
+			.query(({ input }) => runtime.commands.listSessions(input)),
+
+		getItems: t.procedure
+			.input(getItemsInputSchema)
+			.query(({ input }) => runtime.commands.getItems(input)),
+	});
+}
+
+export type ChatRouter = ReturnType<typeof createChatRouter>;

packages/chat-runtime/src/router/wsSink/index.ts

@@ -0,0 +1,2 @@
+export type { WsSinkSocket } from "./wsSink";
+export { createWsSink } from "./wsSink";