From 2ffff8531348e09376b03d6f67eaf0f0f61cdea4 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Thu, 7 May 2026 13:12:48 +0200 Subject: [PATCH] Split session daemon from web server --- README.md | 21 ++++-- package-lock.json | 3 +- package.json | 12 ++- src/server/index.ts | 85 +--------------------- src/server/sessiond.ts | 28 +++++++ src/server/sessiond/config.ts | 10 +++ src/server/sessiond/sessionDaemonClient.ts | 65 +++++++++++++++++ src/server/sessiond/sessionProxyRoutes.ts | 56 ++++++++++++++ src/server/sessions/sessionRoutes.ts | 85 ++++++++++++++++++++++ 9 files changed, 272 insertions(+), 93 deletions(-) create mode 100644 src/server/sessiond.ts create mode 100644 src/server/sessiond/config.ts create mode 100644 src/server/sessiond/sessionDaemonClient.ts create mode 100644 src/server/sessiond/sessionProxyRoutes.ts create mode 100644 src/server/sessions/sessionRoutes.ts diff --git a/README.md b/README.md index da400a0..61a9f78 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Small web wrapper around `@mariozechner/pi-coding-agent`. - Discover workspaces from `git worktree list --porcelain`. - For non-git projects, show the project folder as the only workspace. - List Pi sessions for a workspace using Pi's default session storage. -- Start Pi sessions, chat over WebSocket events, and close active runtimes. +- Start Pi sessions, chat over WebSocket events, and stop individual session runtimes. ## State @@ -17,7 +17,7 @@ This POC intentionally keeps state minimal: - Projects: `~/.pi-web/projects.json` - Workspaces: discovered from git, not stored - Sessions/chat history: Pi default JSONL session storage -- Active sessions/WebSockets: memory only +- Active session runtimes/WebSockets: memory only in `pi-web-sessiond` ## Run @@ -28,19 +28,30 @@ npm run dev Open the Vite URL, usually . -For a single-process/proxied deployment: +The session runtime owner is split into a tiny long-lived daemon. To iterate on only the web/API/UI process while keeping active Pi sessions alive, run these in separate terminals: + +```bash +npm run dev:sessiond +npm run dev:web +npm run dev:client +``` + +Then restart `dev:web` or `dev:client` freely; active Pi sessions continue in `dev:sessiond`. + +For deployment: ```bash npm run build +npm run start:sessiond PI_WEB_PORT=3000 npm start ``` Then proxy Traefik to `http://127.0.0.1:3000`. -The server defaults to `127.0.0.1:3000`. Use `PI_WEB_HOST=0.0.0.0` only if you want to bind directly on all interfaces. +The web server defaults to `127.0.0.1:3000`. Use `PI_WEB_HOST=0.0.0.0` only if you want to bind directly on all interfaces. The session daemon defaults to a private Unix socket at `~/.pi-web/sessiond.sock`; override with `PI_WEB_SESSIOND_SOCKET` or use TCP with `PI_WEB_SESSIOND_PORT` plus `PI_WEB_SESSIOND_URL` for the web process. ## Notes - The backend uses your normal Pi auth/model settings from `~/.pi/agent`. - Slash commands that belong to Pi's interactive TUI, such as `/model`, are not implemented in this POC UI yet. Plain prompts and extension/prompt-template handling go through the SDK path. -- `Close` currently only disposes the in-memory runtime. It does not hide or delete sessions. +- Browser disconnects and web-server restarts do not stop active Pi sessions. Only the explicit `Stop session` action aborts/disposes that one session runtime. diff --git a/package-lock.json b/package-lock.json index 8498e96..a2e7024 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,7 +13,8 @@ "@mariozechner/pi-coding-agent": "^0.73.0", "fastify": "^5.6.1", "lit": "^3.3.1", - "marked": "^18.0.3" + "marked": "^18.0.3", + "ws": "^8.18.3" }, "devDependencies": { "@types/node": "^24.10.1", diff --git a/package.json b/package.json index 79cace1..0aa2898 100644 --- a/package.json +++ b/package.json @@ -4,11 +4,14 @@ "private": true, "type": "module", "scripts": { - "dev": "bash -c 'trap \"kill 0\" EXIT; npm run dev:server & npm run dev:client & wait'", - "dev:server": "tsx watch src/server/index.ts", + "dev": "bash -c 'trap \"kill 0\" EXIT; npm run dev:sessiond & npm run dev:web & npm run dev:client & wait'", + "dev:sessiond": "tsx watch src/server/sessiond.ts", + "dev:web": "tsx watch src/server/index.ts", + "dev:server": "npm run dev:web", "dev:client": "vite --host 0.0.0.0", "build": "tsc && vite build", - "start": "tsx src/server/index.ts" + "start": "tsx src/server/index.ts", + "start:sessiond": "tsx src/server/sessiond.ts" }, "dependencies": { "@fastify/static": "^8.3.0", @@ -16,7 +19,8 @@ "@mariozechner/pi-coding-agent": "^0.73.0", "fastify": "^5.6.1", "lit": "^3.3.1", - "marked": "^18.0.3" + "marked": "^18.0.3", + "ws": "^8.18.3" }, "devDependencies": { "@types/node": "^24.10.1", diff --git a/src/server/index.ts b/src/server/index.ts index 54f1e2a..aaa4185 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -6,17 +6,14 @@ import fastifyWebsocket from "@fastify/websocket"; import { ProjectStore } from "./storage/projectStore.js"; import { ProjectService } from "./projects/projectService.js"; import { WorkspaceService } from "./workspaces/workspaceService.js"; -import { SessionEventHub } from "./realtime/sessionEventHub.js"; -import { PiSessionService } from "./sessions/piSessionService.js"; import { listFileSuggestions } from "./workspaces/fileSuggestions.js"; +import { registerSessionProxyRoutes } from "./sessiond/sessionProxyRoutes.js"; const app = Fastify({ logger: true }); await app.register(fastifyWebsocket); const projects = new ProjectService(new ProjectStore()); const workspaces = new WorkspaceService(); -const eventHub = new SessionEventHub(); -const sessions = new PiSessionService(eventHub); app.get("/api/projects", async () => projects.list()); @@ -37,85 +34,7 @@ app.get<{ Params: { projectId: string } }>("/api/projects/:projectId/workspaces" } }); -app.get<{ Querystring: { cwd?: string } }>("/api/sessions", async (request, reply) => { - if (!request.query.cwd) return reply.code(400).send({ error: "cwd query parameter is required" }); - return sessions.list(request.query.cwd); -}); - -app.post<{ Body: { cwd: string } }>("/api/sessions", async (request, reply) => { - try { - return await sessions.start(request.body.cwd); - } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/messages", async (request, reply) => { - try { - return await sessions.messages(request.params.sessionId); - } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/status", async (request, reply) => { - try { - return await sessions.status(request.params.sessionId); - } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/commands", async (request, reply) => { - try { - return await sessions.commands(request.params.sessionId); - } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.post<{ Params: { sessionId: string }; Body: { text: string } }>("/api/sessions/:sessionId/prompt", async (request, reply) => { - try { - await sessions.prompt(request.params.sessionId, request.body.text); - return { accepted: true }; - } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.post<{ Params: { sessionId: string }; Body: { text: string } }>("/api/sessions/:sessionId/commands/run", async (request, reply) => { - try { - return await sessions.runCommand(request.params.sessionId, request.body.text); - } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.post<{ Params: { sessionId: string }; Body: { requestId: string; value: string } }>("/api/sessions/:sessionId/commands/respond", async (request, reply) => { - try { - return await sessions.respondToCommand(request.params.sessionId, request.body.requestId, request.body.value); - } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); - } -}); - -app.post<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/abort", async (request) => { - await sessions.abort(request.params.sessionId); - return { aborted: true }; -}); - -app.post<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/stop", async (request) => { - sessions.stop(request.params.sessionId); - return { stopped: true }; -}); - -app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/events", { websocket: true }, (socket, request) => { - eventHub.add(request.params.sessionId, socket); -}); - -app.get("/api/sessions/events", { websocket: true }, (socket) => { - eventHub.addGlobal(socket); -}); +await registerSessionProxyRoutes(app); app.get<{ Querystring: { cwd?: string; q?: string; kind?: "tracked" | "untracked" | "other" } }>("/api/files", async (request, reply) => { if (!request.query.cwd) return reply.code(400).send({ error: "cwd query parameter is required" }); diff --git a/src/server/sessiond.ts b/src/server/sessiond.ts new file mode 100644 index 0000000..dd8b7b0 --- /dev/null +++ b/src/server/sessiond.ts @@ -0,0 +1,28 @@ +import { mkdir, rm } from "node:fs/promises"; +import { dirname } from "node:path"; +import Fastify from "fastify"; +import fastifyWebsocket from "@fastify/websocket"; +import { SessionEventHub } from "./realtime/sessionEventHub.js"; +import { PiSessionService } from "./sessions/piSessionService.js"; +import { registerSessionRoutes } from "./sessions/sessionRoutes.js"; +import { sessiondSocketPath } from "./sessiond/config.js"; + +const app = Fastify({ logger: true }); +await app.register(fastifyWebsocket); + +const eventHub = new SessionEventHub(); +const sessions = new PiSessionService(eventHub); +await registerSessionRoutes(app, sessions, eventHub); + +const port = process.env.PI_WEB_SESSIOND_PORT ? Number(process.env.PI_WEB_SESSIOND_PORT) : undefined; +const host = process.env.PI_WEB_SESSIOND_HOST ?? "127.0.0.1"; + +if (port) { + await app.listen({ port, host }); +} else { + const path = sessiondSocketPath(); + await mkdir(dirname(path), { recursive: true }); + await rm(path, { force: true }); + await app.listen({ path }); + process.on("exit", () => void rm(path, { force: true })); +} diff --git a/src/server/sessiond/config.ts b/src/server/sessiond/config.ts new file mode 100644 index 0000000..10006b8 --- /dev/null +++ b/src/server/sessiond/config.ts @@ -0,0 +1,10 @@ +import { homedir } from "node:os"; +import { join } from "node:path"; + +export function sessiondSocketPath(): string { + return process.env.PI_WEB_SESSIOND_SOCKET ?? join(homedir(), ".pi-web", "sessiond.sock"); +} + +export function sessiondHttpUrl(): string | undefined { + return process.env.PI_WEB_SESSIOND_URL; +} diff --git a/src/server/sessiond/sessionDaemonClient.ts b/src/server/sessiond/sessionDaemonClient.ts new file mode 100644 index 0000000..8b0d5e2 --- /dev/null +++ b/src/server/sessiond/sessionDaemonClient.ts @@ -0,0 +1,65 @@ +import http from "node:http"; +import { WebSocket } from "ws"; +import { sessiondHttpUrl, sessiondSocketPath } from "./config.js"; + +export class SessionDaemonClient { + private readonly baseUrl = sessiondHttpUrl(); + private readonly socketPath = sessiondSocketPath(); + + async request(method: string, path: string, body?: unknown): Promise<{ statusCode: number; headers: Record; body: string }> { + const payload = body === undefined ? undefined : JSON.stringify(body); + if (this.baseUrl) return this.requestUrl(method, path, payload); + return this.requestSocket(method, path, payload); + } + + connectWebSocket(path: string): WebSocket { + if (this.baseUrl) { + const url = new URL(path, this.baseUrl); + url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; + return new WebSocket(url); + } + return new WebSocket(`ws+unix:${this.socketPath}:${path}`); + } + + private async requestUrl(method: string, path: string, payload?: string) { + const response = await fetch(new URL(path, this.baseUrl), { + method, + headers: payload ? { "content-type": "application/json" } : undefined, + body: payload, + }); + return { + statusCode: response.status, + headers: Object.fromEntries(response.headers.entries()), + body: await response.text(), + }; + } + + private requestSocket(method: string, path: string, payload?: string): Promise<{ statusCode: number; headers: Record; body: string }> { + return new Promise((resolve, reject) => { + const request = http.request( + { + socketPath: this.socketPath, + path, + method, + headers: payload + ? { "content-type": "application/json", "content-length": Buffer.byteLength(payload) } + : undefined, + }, + (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))); + response.on("end", () => { + resolve({ + statusCode: response.statusCode ?? 500, + headers: Object.fromEntries(Object.entries(response.headers).map(([key, value]) => [key, Array.isArray(value) ? value.join(", ") : value ?? ""])), + body: Buffer.concat(chunks).toString("utf8"), + }); + }); + }, + ); + request.on("error", reject); + if (payload) request.write(payload); + request.end(); + }); + } +} diff --git a/src/server/sessiond/sessionProxyRoutes.ts b/src/server/sessiond/sessionProxyRoutes.ts new file mode 100644 index 0000000..5b4b611 --- /dev/null +++ b/src/server/sessiond/sessionProxyRoutes.ts @@ -0,0 +1,56 @@ +import type { FastifyInstance, FastifyReply } from "fastify"; +import { WebSocket, type RawData } from "ws"; +import { SessionDaemonClient } from "./sessionDaemonClient.js"; + +export async function registerSessionProxyRoutes(app: FastifyInstance, daemon = new SessionDaemonClient()): Promise { + const proxy = async (request: { method: string; url: string; body?: unknown }, reply: FastifyReply) => { + try { + const upstream = await daemon.request(request.method, stripApiPrefix(request.url), request.body); + reply.code(upstream.statusCode); + if (upstream.headers["content-type"]) reply.header("content-type", upstream.headers["content-type"]); + return upstream.body ? JSON.parse(upstream.body) : undefined; + } catch (error) { + requestFailed(reply, error); + } + }; + + app.get<{ Querystring: { cwd?: string } }>("/api/sessions", (request, reply) => proxy(request, reply)); + app.post<{ Body: { cwd: string } }>("/api/sessions", (request, reply) => proxy(request, reply)); + app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/messages", (request, reply) => proxy(request, reply)); + app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/status", (request, reply) => proxy(request, reply)); + app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/commands", (request, reply) => proxy(request, reply)); + app.post<{ Params: { sessionId: string }; Body: { text: string } }>("/api/sessions/:sessionId/prompt", (request, reply) => proxy(request, reply)); + app.post<{ Params: { sessionId: string }; Body: { text: string } }>("/api/sessions/:sessionId/commands/run", (request, reply) => proxy(request, reply)); + app.post<{ Params: { sessionId: string }; Body: { requestId: string; value: string } }>("/api/sessions/:sessionId/commands/respond", (request, reply) => proxy(request, reply)); + app.post<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/abort", (request, reply) => proxy(request, reply)); + app.post<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/stop", (request, reply) => proxy(request, reply)); + + app.get<{ Params: { sessionId: string } }>("/api/sessions/:sessionId/events", { websocket: true }, (socket, request) => { + bridgeSockets(socket, daemon.connectWebSocket(`/sessions/${request.params.sessionId}/events`)); + }); + + app.get("/api/sessions/events", { websocket: true }, (socket) => { + bridgeSockets(socket, daemon.connectWebSocket("/sessions/events")); + }); +} + +function stripApiPrefix(url: string): string { + return url.startsWith("/api") ? url.slice(4) || "/" : url; +} + +function requestFailed(reply: FastifyReply, error: unknown) { + reply.code(502).send({ error: `Session daemon unavailable: ${error instanceof Error ? error.message : String(error)}` }); +} + +function bridgeSockets(client: WebSocket, upstream: WebSocket): void { + client.on("message", (data) => sendIfOpen(upstream, data)); + upstream.on("message", (data) => sendIfOpen(client, data)); + client.on("close", () => upstream.close()); + upstream.on("close", () => client.close()); + upstream.on("error", () => client.close()); + client.on("error", () => upstream.close()); +} + +function sendIfOpen(socket: WebSocket, data: RawData): void { + if (socket.readyState === WebSocket.OPEN) socket.send(data); +} diff --git a/src/server/sessions/sessionRoutes.ts b/src/server/sessions/sessionRoutes.ts new file mode 100644 index 0000000..132bc8e --- /dev/null +++ b/src/server/sessions/sessionRoutes.ts @@ -0,0 +1,85 @@ +import type { FastifyInstance } from "fastify"; +import type { SessionEventHub } from "../realtime/sessionEventHub.js"; +import type { PiSessionService } from "./piSessionService.js"; + +export async function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionService, eventHub: SessionEventHub, prefix = ""): Promise { + app.get<{ Querystring: { cwd?: string } }>(`${prefix}/sessions`, async (request, reply) => { + if (!request.query.cwd) return reply.code(400).send({ error: "cwd query parameter is required" }); + return sessions.list(request.query.cwd); + }); + + app.post<{ Body: { cwd: string } }>(`${prefix}/sessions`, async (request, reply) => { + try { + return await sessions.start(request.body.cwd); + } catch (error) { + return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => { + try { + return await sessions.messages(request.params.sessionId); + } catch (error) { + return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/status`, async (request, reply) => { + try { + return await sessions.status(request.params.sessionId); + } catch (error) { + return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/commands`, async (request, reply) => { + try { + return await sessions.commands(request.params.sessionId); + } catch (error) { + return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: { text: string } }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => { + try { + await sessions.prompt(request.params.sessionId, request.body.text); + return { accepted: true }; + } catch (error) { + return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: { text: string } }>(`${prefix}/sessions/:sessionId/commands/run`, async (request, reply) => { + try { + return await sessions.runCommand(request.params.sessionId, request.body.text); + } catch (error) { + return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: { requestId: string; value: string } }>(`${prefix}/sessions/:sessionId/commands/respond`, async (request, reply) => { + try { + return await sessions.respondToCommand(request.params.sessionId, request.body.requestId, request.body.value); + } catch (error) { + return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + } + }); + + app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/abort`, async (request) => { + await sessions.abort(request.params.sessionId); + return { aborted: true }; + }); + + app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/stop`, async (request) => { + sessions.stop(request.params.sessionId); + return { stopped: true }; + }); + + app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/events`, { websocket: true }, (socket, request) => { + eventHub.add(request.params.sessionId, socket); + }); + + app.get(`${prefix}/sessions/events`, { websocket: true }, (socket) => { + eventHub.addGlobal(socket); + }); +}