Merge remote-tracking branch 'origin/main' into investigate/issue-12-session-dir

# Conflicts:
#	src/client/src/api.ts
#	src/client/src/api/clients.ts
#	src/client/src/api/federatedRouteContract.test.ts
#	src/server/sessions/piSessionService.ts
#	src/server/sessions/sessionRoutes.ts

src/server/sessions/piSessionService.test.ts

@@ -49,8 +49,9 @@ function sessionRef(id: string, cwd = "/workspace") {
 
 function fakeRuntime(sessionId = "session-1", patch: Partial<TestSession> = {}) {
   const promptCalls: { text: string; options: unknown }[] = [];
+  const bindExtensionCalls: unknown[] = [];
   const listeners: ((event: unknown) => void)[] = [];
-  const calls = { abort: 0, clearQueue: 0, dispose: 0, prompt: promptCalls };
+  const calls = { abort: 0, bindExtensions: bindExtensionCalls, clearQueue: 0, dispose: 0, prompt: promptCalls };
   const session: TestSession = {
     sessionId,
     sessionFile: `/tmp/${sessionId}.jsonl`,
@@ -75,6 +76,10 @@ function fakeRuntime(sessionId = "session-1", patch: Partial<TestSession> = {})
         if (index !== -1) listeners.splice(index, 1);
       };
     },
+    bindExtensions: (bindings: unknown) => {
+      calls.bindExtensions.push(bindings);
+      return Promise.resolve();
+    },
     getSessionStats: () => ({ sessionId, totalMessages: 0, userMessages: 0, assistantMessages: 0, toolCalls: 0, tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, cost: 0 }),
     getContextUsage: () => undefined,
     prompt: (text: string, options: unknown) => {
@@ -149,6 +154,7 @@ describe("PiSessionService", () => {
     const session = await service.start("/workspace");
 
     expect(createCalls).toBe(1);
+    expect(fake.calls.bindExtensions).toHaveLength(1);
     expect(session).toMatchObject({ id: "session-1", cwd: "/workspace", messageCount: 0 });
     expect(service.activeCount()).toBe(1);
     expect(hub.globalEvents.some((event) => event.type === "status.update" && event.status.sessionId === "session-1")).toBe(true);
@@ -158,6 +164,59 @@ describe("PiSessionService", () => {
     expect(fake.calls.dispose).toBe(1);
   });
 
+  it("binds extensions again when the SDK runtime replaces the active session", async () => {
+    const hub = new CapturingSessionEventHub();
+    const fake = fakeRuntime("session-1");
+    const replacement = fakeRuntime("session-2");
+    let rebindSession: ((session: PiAgentSession) => Promise<void>) | undefined;
+    fake.runtime.setRebindSession = (callback) => { rebindSession = callback; };
+    const service = new PiSessionService(hub, {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.start("/workspace");
+    Object.defineProperty(fake.runtime, "session", { configurable: true, value: replacement.session });
+    await rebindSession?.(replacement.session);
+
+    expect(fake.calls.bindExtensions).toHaveLength(1);
+    expect(replacement.calls.bindExtensions).toHaveLength(1);
+    expect(service.activeCount()).toBe(1);
+    expect(await service.status("session-2")).toMatchObject({ sessionId: "session-2" });
+
+    await service.dispose();
+  });
+
+  it("publishes extension errors reported while binding session extensions", async () => {
+    const hub = new CapturingSessionEventHub();
+    const fake = fakeRuntime("extension-session", {
+      bindExtensions: (bindings) => {
+        bindings.onError?.({ extensionPath: "pi-mcp-adapter", event: "session_start", error: "MCP failed" });
+        return Promise.resolve();
+      },
+    });
+    const service = new PiSessionService(hub, {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.start("/workspace");
+
+    expect(hub.sessionEvents).toContainEqual({
+      sessionId: "extension-session",
+      event: { type: "session.error", message: "pi-mcp-adapter: MCP failed" },
+    });
+    const extensionErrorActivity = hub.globalEvents.find((event) => event.type === "activity.update" && event.activity.sessionId === "extension-session");
+    expect(extensionErrorActivity).toMatchObject({
+      type: "activity.update",
+      activity: { sessionId: "extension-session", phase: "error", label: "extension error", detail: "pi-mcp-adapter: MCP failed" },
+    });
+
+    await service.dispose();
+  });
+
   it("clears stale active activity once a previously active session becomes idle", async () => {
     vi.useFakeTimers();
     let service: PiSessionService | undefined;
@@ -318,6 +377,33 @@ describe("PiSessionService", () => {
     await service.dispose();
   });
 
+  it("permanently deletes archived sessions through the archive store", async () => {
+    const deletedSessionIds: string[] = [];
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      archiveStore: {
+        list: () => Promise.resolve([]),
+        get: (sessionId) => Promise.resolve(sessionId === "archived" || "archived".startsWith(sessionId)
+          ? { sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/archived.jsonl" }
+          : undefined),
+        archive: () => { throw new Error("archive should not be called for records that already have archive files"); },
+        restore: () => Promise.resolve(),
+        isArchived: () => Promise.resolve(false),
+        deleteArchived: (sessionId) => {
+          deletedSessionIds.push(sessionId);
+          return Promise.resolve();
+        },
+      },
+      sessionManager: sessionGateway([sessionRecord("active")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await expect(service.deleteArchived("arch")).resolves.toBeUndefined();
+    await expect(service.deleteArchived("active")).rejects.toThrow("Archived session not found");
+
+    expect(deletedSessionIds).toEqual(["archived"]);
+    await service.dispose();
+  });
+
   it("reconciles workspace activity when listing only archived sessions", async () => {
     const reconciliations: { cwd: string; sessionIds: string[] }[] = [];
     const service = new PiSessionService(new CapturingSessionEventHub(), {
@@ -365,6 +451,20 @@ describe("PiSessionService", () => {
     await service.dispose();
   });
 
+  it("rejects malformed prompt text before opening the runtime", async () => {
+    const fake = fakeRuntime("prompt-session");
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("prompt-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await expect(service.prompt("prompt-session", undefined)).rejects.toThrow("Prompt text is required");
+
+    expect(fake.calls.prompt).toEqual([]);
+    await service.dispose();
+  });
+
   it("includes queued message details in session status", async () => {
     const fake = fakeRuntime("status-session", {
       messages: [{ role: "user", content: "hello" }, { role: "assistant", content: "hi" }],

src/server/sessions/piSessionService.ts

@@ -54,7 +54,18 @@ interface QueuedPrompt {
   text: string;
 }
 
-type SessionArchiveRepository = Pick<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived">;
+function requirePromptText(value: unknown): string {
+  if (typeof value !== "string") throw new Error("Prompt text is required");
+  return value;
+}
+
+function parsePromptStreamingBehavior(value: unknown): QueuedPromptKind | undefined {
+  if (value === undefined) return undefined;
+  if (value === "steer" || value === "followUp") return value;
+  throw new Error('Prompt streamingBehavior must be "steer" or "followUp"');
+}
+
+type SessionArchiveRepository = Pick<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived"> & { deleteArchived?: (sessionId: string) => Promise<void> };
 
 export type PiSessionRef = ClientSessionRef;
 
@@ -95,6 +106,17 @@ export interface PiSessionManagerGateway {
   open(path: string): PiSessionManager;
 }
 
+interface PiExtensionError {
+  extensionPath: string;
+  event: string;
+  error: string;
+  stack?: string;
+}
+
+interface PiExtensionBindings {
+  onError?: (error: PiExtensionError) => void;
+}
+
 export interface PiAgentSession {
   modelRegistry: ModelRegistryInstance;
   sessionManager: PiSessionManager;
@@ -113,6 +135,7 @@ export interface PiAgentSession {
   promptTemplates: readonly { name: string; description?: string }[];
   resourceLoader: { getSkills(): { skills: readonly { name: string; description?: string }[] } };
   subscribe(listener: (event: unknown) => void): () => void;
+  bindExtensions(bindings: PiExtensionBindings): Promise<void>;
   compact(instructions?: string): Promise<{ summary: string; tokensBefore: number }>;
   getUserMessagesForForking(): readonly { entryId: string; text: string }[];
   getSessionStats(): { sessionId: string; totalMessages: number; userMessages: number; assistantMessages: number; toolCalls: number; tokens: ClientSessionStatus["tokens"]; cost: number };
@@ -377,22 +400,24 @@ export class PiSessionService {
     return commands.sort((a, b) => a.name.localeCompare(b.name));
   }
 
-  async prompt(ref: PiSessionLookup, text: string, streamingBehavior?: "steer" | "followUp"): Promise<void> {
+  async prompt(ref: PiSessionLookup, text: unknown, streamingBehavior?: unknown): Promise<void> {
+    const promptText = requirePromptText(text);
+    const requestedBehavior = parsePromptStreamingBehavior(streamingBehavior);
     await this.assertWritable(ref);
     const session = await this.getOrOpen(ref);
-    this.maybeGenerateSessionName(session, text);
+    this.maybeGenerateSessionName(session, promptText);
     const isQueued = session.isStreaming || session.isCompacting;
-    const behavior = isQueued ? streamingBehavior ?? "followUp" : undefined;
-    if (isQueued && this.hasQueuedMessageText(session, text)) {
+    const behavior = isQueued ? requestedBehavior ?? "followUp" : undefined;
+    if (isQueued && this.hasQueuedMessageText(session, promptText)) {
       this.publishActivity(session, "duplicate queued message ignored", "active");
       this.publishStatus(session);
       return;
     }
     if (session.isCompacting) {
-      this.enqueuePromptDuringCompaction(session, text, behavior ?? "followUp");
+      this.enqueuePromptDuringCompaction(session, promptText, behavior ?? "followUp");
       return;
     }
-    void this.submitPrompt(session, text, behavior);
+    void this.submitPrompt(session, promptText, behavior);
   }
 
   private submitPrompt(session: PiAgentSession, text: string, behavior: QueuedPromptKind | undefined): Promise<void> {
@@ -497,6 +522,16 @@ export class PiSessionService {
     await this.archiveStore.restore(archived.sessionId);
   }
 
+  async deleteArchived(ref: PiSessionLookup): Promise<void> {
+    const record = await this.getArchived(ref);
+    if (record === undefined) throw new Error("Archived session not found");
+    if (this.archiveStore.deleteArchived === undefined) throw new Error("Archive store does not support deletion");
+
+    await this.closeActive(record.sessionId);
+    if (record.archivePath === undefined) await this.ensureArchivedRecordMoved(record);
+    await this.archiveStore.deleteArchived(record.sessionId);
+  }
+
   async detachParent(ref: PiSessionLookup): Promise<void> {
     const session = await this.getOrOpen(ref);
     const sessionFile = session.sessionFile;
@@ -541,6 +576,12 @@ export class PiSessionService {
     }
   }
 
+  private async ensureArchivedRecordMoved(record: ArchivedSessionRecord): Promise<ArchivedSessionRecord> {
+    const session = (await this.sessionManager.list(record.cwd)).find((candidate) => candidate.id === record.sessionId);
+    if (session === undefined) return record;
+    return this.archiveStore.archive(archiveInputFromListEntry(session));
+  }
+
   private async archiveInputForSession(session: PiAgentSession): Promise<ArchiveSessionInput> {
     const cwd = session.sessionManager.getCwd();
     const sessionFile = session.sessionFile;
@@ -655,17 +696,28 @@ export class PiSessionService {
 
   private async create(sessionManager: PiSessionManager, cwd: string): Promise<ActiveSession<PiSessionRuntime>> {
     const runtime = await this.createAgentRuntime(this.createRuntime, { cwd, agentDir: this.agentDir, sessionManager });
+    await this.bindSessionExtensions(runtime.session);
     const active: ActiveSession<PiSessionRuntime> = { runtime, unsubscribe: noop };
     this.bindRuntime(active);
-    runtime.setRebindSession(() => {
+    runtime.setRebindSession(async (session) => {
+      await this.bindSessionExtensions(session);
       this.bindRuntime(active);
-      return Promise.resolve();
     });
     this.active.set(runtime.session.sessionId, active);
     this.publishStatus(runtime.session);
     return active;
   }
 
+  private async bindSessionExtensions(session: PiAgentSession): Promise<void> {
+    await session.bindExtensions({
+      onError: (error) => {
+        const message = `${error.extensionPath}: ${error.error}`;
+        this.publishActivity(session, "extension error", "error", message);
+        this.events.publish(session.sessionId, { type: "session.error", message });
+      },
+    });
+  }
+
   private bindRuntime(active: ActiveSession<PiSessionRuntime>): void {
     active.unsubscribe();
     const { session } = active.runtime;

src/server/sessions/sessionArchiveStore.test.ts

@@ -44,6 +44,33 @@ describe("SessionArchiveStore", () => {
     expect(await exists(record.archivePath)).toBe(false);
     await expect(store.list()).resolves.toEqual([]);
   });
+
+  it("permanently deletes archived session files and records", async () => {
+    const root = await mkdtemp(join(tmpdir(), "pi-web-archive-delete-"));
+    tempRoots.push(root);
+    const activeDir = join(root, "active");
+    await mkdir(activeDir, { recursive: true });
+    const sourcePath = join(activeDir, "2026-01-01_s1.jsonl");
+    await writeFile(sourcePath, "session contents\n", "utf8");
+
+    const store = new SessionArchiveStore(join(root, "archived-sessions.json"), join(root, "archived-files"));
+    const record = await store.archive({
+      sessionId: "s1",
+      cwd: "/workspace",
+      path: sourcePath,
+      created: "2026-01-01T00:00:00.000Z",
+      modified: "2026-01-01T00:01:00.000Z",
+      messageCount: 2,
+      firstMessage: "hello",
+    });
+
+    if (record.archivePath === undefined) throw new Error("Expected archive path");
+    await store.deleteArchived("s1");
+
+    expect(await exists(sourcePath)).toBe(false);
+    expect(await exists(record.archivePath)).toBe(false);
+    await expect(store.list()).resolves.toEqual([]);
+  });
 });
 
 async function exists(path: string): Promise<boolean> {

src/server/sessions/sessionArchiveStore.ts

@@ -88,6 +88,18 @@ export class SessionArchiveStore {
     });
   }
 
+  async deleteArchived(sessionId: string): Promise<void> {
+    await this.exclusive(async () => {
+      const data = await this.read();
+      const record = data.sessions.find((session) => session.sessionId === sessionId);
+      if (record === undefined) return;
+
+      if (record.archivePath !== undefined && await pathExists(record.archivePath)) await unlink(record.archivePath);
+      const sessions = data.sessions.filter((session) => session.sessionId !== sessionId);
+      await this.write({ sessions });
+    });
+  }
+
   async isArchived(sessionId: string): Promise<boolean> {
     return (await this.get(sessionId)) !== undefined;
   }

src/server/sessions/sessionNameGenerator.test.ts

@@ -15,4 +15,8 @@ describe("sessionNameGenerator", () => {
     expect(fallbackSessionName('<skill name="x" location="/x">\nDo x\n</skill>\n\nCheck the UI now'))
       .toBe("Check the UI now");
   });
+
+  it("skips fallback names when the first request is missing", () => {
+    expect(fallbackSessionName(undefined)).toBeUndefined();
+  });
 });

src/server/sessions/sessionNameGenerator.ts

@@ -43,7 +43,9 @@ export async function generateShortSessionName<TApi extends Api>(modelRegistry:
   return cleanSessionName(finalMessage === undefined ? streamedText : textFromAssistant(finalMessage));
 }
 
-export function fallbackSessionName(firstMessage: string): string | undefined {
+export function fallbackSessionName(firstMessage: unknown): string | undefined {
+  if (typeof firstMessage !== "string") return undefined;
+
   return cleanSessionName(firstMessage
     .replace(/<skill name="[^"]+" location="[^"]+">[\s\S]*?<\/skill>/g, "")
     .replace(/```[\s\S]*?```/g, " ")

src/server/sessions/sessionRoutes.test.ts

@@ -0,0 +1,58 @@
+import Fastify, { type FastifyInstance } from "fastify";
+import fastifyWebsocket from "@fastify/websocket";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { SessionEventHub } from "../realtime/sessionEventHub.js";
+import { PiSessionService, type PiSessionManagerGateway } from "./piSessionService.js";
+import { registerSessionRoutes } from "./sessionRoutes.js";
+
+let app: FastifyInstance;
+let service: PiSessionService;
+let sessionManager: RejectingSessionManager;
+
+beforeEach(async () => {
+  app = Fastify({ logger: false });
+  await app.register(fastifyWebsocket);
+  sessionManager = new RejectingSessionManager();
+  const eventHub = new SessionEventHub();
+  service = new PiSessionService(eventHub, { sessionManager, heartbeatIntervalMs: 60_000 });
+  registerSessionRoutes(app, service, eventHub);
+});
+
+afterEach(async () => {
+  await service.dispose();
+  await app.close();
+});
+
+describe("session routes", () => {
+  it("rejects prompt payloads that omit text without opening a session", async () => {
+    const response = await app.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { cwd: "/repo", body: "Build the thing" } });
+
+    expect(response.statusCode).toBe(400);
+    expect(response.json()).toEqual({ error: "Prompt text is required" });
+    expect(sessionManager.calls).toEqual({ create: 0, list: 0, listAll: 0, open: 0 });
+  });
+});
+
+class RejectingSessionManager implements PiSessionManagerGateway {
+  readonly calls = { create: 0, list: 0, listAll: 0, open: 0 };
+
+  list() {
+    this.calls.list += 1;
+    return Promise.resolve([]);
+  }
+
+  create(): never {
+    this.calls.create += 1;
+    throw new Error("Session manager should not create sessions for invalid prompt payloads");
+  }
+
+  listAll() {
+    this.calls.listAll += 1;
+    return Promise.resolve([]);
+  }
+
+  open(): never {
+    this.calls.open += 1;
+    throw new Error("Session manager should not open sessions for invalid prompt payloads");
+  }
+}

src/server/sessions/sessionRoutes.ts

@@ -13,6 +13,12 @@ interface MessageQuery extends SessionQuery {
 
 class SessionRouteValidationError extends Error {}
 
+interface PromptRequestBody {
+  cwd?: unknown;
+  text?: unknown;
+  streamingBehavior?: unknown;
+}
+
 export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionService, eventHub: SessionEventHub, prefix = ""): void {
   app.get<{ Querystring: SessionQuery }>(`${prefix}/sessions`, async (request, reply) => {
     if (request.query.cwd === undefined || request.query.cwd === "") return reply.code(400).send({ error: "cwd query parameter is required" });
@@ -106,12 +112,10 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS
     }
   });
 
-  app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown; streamingBehavior?: "steer" | "followUp" } }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => {
+  app.post<{ Params: { sessionId: string }; Body: PromptRequestBody | undefined }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => {
     try {
       const body = requireRecord(request.body);
-      const streamingBehavior = body["streamingBehavior"];
-      if (streamingBehavior !== undefined && streamingBehavior !== "steer" && streamingBehavior !== "followUp") throw new Error("streamingBehavior must be steer or followUp");
-      await sessions.prompt(sessionRefFromBody(request.params.sessionId, body), requireString(body, "text"), streamingBehavior);
+      await sessions.prompt(sessionRefFromBody(request.params.sessionId, body), body["text"], body["streamingBehavior"]);
       return { accepted: true };
     } catch (error) {
       return reply.code(400).send({ error: errorMessage(error) });
@@ -190,6 +194,15 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS
     }
   });
 
+  app.delete<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId`, async (request, reply) => {
+    try {
+      await sessions.deleteArchived(sessionRefFromQuery(request.params.sessionId, request.query));
+      return { deleted: true };
+    } catch (error) {
+      return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) });
+    }
+  });
+
   app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/detach-parent`, async (request, reply) => {
     try {
       await sessions.detachParent(sessionRefFromBody(request.params.sessionId, requireRecord(request.body)));