feat(streams): vendor durable-session-proxy into apps/streams

apps/streams/package.json

@@ -13,10 +13,15 @@
 		"test:conformance": "npx @durable-streams/server-conformance-tests --run http://localhost:8080"
 	},
 	"dependencies": {
-		"@durable-streams/server": "^0.2.0"
+		"@durable-streams/client": "^0.2.0",
+		"@durable-streams/server": "^0.2.0",
+		"@superset/durable-session": "workspace:*",
+		"@tanstack/db": "^0.5.22",
+		"@hono/node-server": "^1.13.0",
+		"hono": "^4.4.0",
+		"zod": "^4.1.12"
 	},
 	"devDependencies": {
-		"@durable-streams/client": "^0.2.0",
 		"@durable-streams/server-conformance-tests": "^0.2.0",
 		"@superset/typescript": "workspace:*",
 		"@types/node": "^24.9.1",

apps/streams/src/handlers/index.ts

@@ -0,0 +1,7 @@
+export {
+	handleInvokeAgent,
+	handleRegisterAgents,
+	handleUnregisterAgent,
+} from "./invoke-agent";
+export { handleSendMessage } from "./send-message";
+export { createStreamWriter, StreamWriter } from "./stream-writer";

apps/streams/src/handlers/invoke-agent.ts

@@ -0,0 +1,101 @@
+import type { Context } from "hono";
+import { z } from "zod";
+import type { AIDBSessionProtocol } from "../protocol";
+import { type AgentSpec, agentSpecSchema } from "../types";
+
+const invokeAgentRequestSchema = z.object({
+	agent: agentSpecSchema,
+	messages: z.array(
+		z.object({
+			role: z.string(),
+			content: z.string(),
+		}),
+	),
+});
+
+type InvokeAgentRequest = z.infer<typeof invokeAgentRequestSchema>;
+
+export async function handleInvokeAgent(
+	c: Context,
+	protocol: AIDBSessionProtocol,
+): Promise<Response> {
+	const sessionId = c.req.param("sessionId");
+
+	let body: InvokeAgentRequest;
+	try {
+		const rawBody = await c.req.json();
+		body = invokeAgentRequestSchema.parse(rawBody);
+	} catch (error) {
+		return c.json(
+			{ error: "Invalid request body", details: (error as Error).message },
+			400,
+		);
+	}
+
+	try {
+		const stream = await protocol.getOrCreateSession(sessionId);
+		await protocol.invokeAgent(stream, sessionId, body.agent, body.messages);
+		return c.json({ success: true }, 200);
+	} catch (error) {
+		console.error("Failed to invoke agent:", error);
+		return c.json(
+			{ error: "Failed to invoke agent", details: (error as Error).message },
+			500,
+		);
+	}
+}
+
+export async function handleRegisterAgents(
+	c: Context,
+	protocol: AIDBSessionProtocol,
+): Promise<Response> {
+	const sessionId = c.req.param("sessionId");
+
+	let agents: AgentSpec[];
+	try {
+		const rawBody = await c.req.json();
+		const parsed = z
+			.object({ agents: z.array(agentSpecSchema) })
+			.parse(rawBody);
+		agents = parsed.agents;
+	} catch (error) {
+		return c.json(
+			{ error: "Invalid request body", details: (error as Error).message },
+			400,
+		);
+	}
+
+	try {
+		await protocol.getOrCreateSession(sessionId);
+		await protocol.registerAgents(sessionId, agents);
+		return c.json({ success: true }, 200);
+	} catch (error) {
+		console.error("Failed to register agents:", error);
+		return c.json(
+			{ error: "Failed to register agents", details: (error as Error).message },
+			500,
+		);
+	}
+}
+
+export async function handleUnregisterAgent(
+	c: Context,
+	protocol: AIDBSessionProtocol,
+): Promise<Response> {
+	const sessionId = c.req.param("sessionId");
+	const agentId = c.req.param("agentId");
+
+	try {
+		await protocol.unregisterAgent(sessionId, agentId);
+		return new Response(null, { status: 204 });
+	} catch (error) {
+		console.error("Failed to unregister agent:", error);
+		return c.json(
+			{
+				error: "Failed to unregister agent",
+				details: (error as Error).message,
+			},
+			500,
+		);
+	}
+}

apps/streams/src/handlers/send-message.ts

@@ -0,0 +1,61 @@
+import type { Context } from "hono";
+import type { AIDBSessionProtocol } from "../protocol";
+import {
+	type SendMessageRequest,
+	type SendMessageResponse,
+	sendMessageRequestSchema,
+} from "../types";
+
+export async function handleSendMessage(
+	c: Context,
+	protocol: AIDBSessionProtocol,
+): Promise<Response> {
+	const sessionId = c.req.param("sessionId");
+
+	let body: SendMessageRequest;
+	try {
+		const rawBody = await c.req.json();
+		body = sendMessageRequestSchema.parse(rawBody);
+	} catch (error) {
+		return c.json(
+			{ error: "Invalid request body", details: (error as Error).message },
+			400,
+		);
+	}
+
+	const actorId =
+		body.actorId ?? c.req.header("X-Actor-Id") ?? crypto.randomUUID();
+
+	const messageId = body.messageId ?? crypto.randomUUID();
+
+	try {
+		const stream = await protocol.getOrCreateSession(sessionId);
+
+		await protocol.writeUserMessage(
+			stream,
+			sessionId,
+			messageId,
+			actorId,
+			body.content,
+			body.txid,
+		);
+
+		if (body.agent) {
+			const messageHistory = await protocol.getMessageHistory(sessionId);
+			protocol
+				.invokeAgent(stream, sessionId, body.agent, messageHistory)
+				.catch((err) => {
+					console.error("[streams/send-message] Agent invocation failed:", err);
+				});
+		}
+
+		const response: SendMessageResponse = { messageId };
+		return c.json(response, 200);
+	} catch (error) {
+		console.error("Failed to send message:", error);
+		return c.json(
+			{ error: "Failed to send message", details: (error as Error).message },
+			500,
+		);
+	}
+}

apps/streams/src/handlers/stream-writer.ts

@@ -0,0 +1,91 @@
+import type { DurableStream } from "@durable-streams/client";
+import type { AIDBSessionProtocol } from "../protocol";
+import type { StreamChunk } from "../types";
+
+type MessageRole = "user" | "assistant" | "system";
+
+export class StreamWriter {
+	constructor(
+		private readonly protocol: AIDBSessionProtocol,
+		private readonly stream: DurableStream,
+		private readonly sessionId: string,
+	) {}
+
+	async writeUserMessage(
+		messageId: string,
+		actorId: string,
+		content: string,
+		txid?: string,
+	): Promise<void> {
+		await this.protocol.writeUserMessage(
+			this.stream,
+			this.sessionId,
+			messageId,
+			actorId,
+			content,
+			txid,
+		);
+	}
+
+	async writeChunk(
+		messageId: string,
+		actorId: string,
+		role: MessageRole,
+		chunk: StreamChunk,
+		txid?: string,
+	): Promise<void> {
+		await this.protocol.writeChunk(
+			this.stream,
+			this.sessionId,
+			messageId,
+			actorId,
+			role,
+			chunk,
+			txid,
+		);
+	}
+
+	async writeToolResult(
+		messageId: string,
+		actorId: string,
+		toolCallId: string,
+		output: unknown,
+		error: string | null,
+		txid?: string,
+	): Promise<void> {
+		await this.protocol.writeToolResult(
+			this.stream,
+			this.sessionId,
+			messageId,
+			actorId,
+			toolCallId,
+			output,
+			error,
+			txid,
+		);
+	}
+
+	async writeApprovalResponse(
+		actorId: string,
+		approvalId: string,
+		approved: boolean,
+		txid?: string,
+	): Promise<void> {
+		await this.protocol.writeApprovalResponse(
+			this.stream,
+			this.sessionId,
+			actorId,
+			approvalId,
+			approved,
+			txid,
+		);
+	}
+}
+
+export function createStreamWriter(
+	protocol: AIDBSessionProtocol,
+	stream: DurableStream,
+	sessionId: string,
+): StreamWriter {
+	return new StreamWriter(protocol, stream, sessionId);
+}