Split session daemon from web server

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 <http://localhost:5173>.
 
-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.

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",

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",

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" });

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 }));
+}

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;
+}

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<string, string>; 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<string, string>; 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();
+    });
+  }
+}

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<void> {
+  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);
+}

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<void> {
+  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);
+  });
+}