diff --git a/src/client/src/api/federatedRouteContract.test.ts b/src/client/src/api/federatedRouteContract.test.ts index 748bcf0..65ab19a 100644 --- a/src/client/src/api/federatedRouteContract.test.ts +++ b/src/client/src/api/federatedRouteContract.test.ts @@ -26,6 +26,16 @@ afterEach(() => { }); describe("federated route contract", () => { + it("allowlists notification HTTP routes without adding a notification WebSocket", () => { + expect(FEDERATED_HTTP_ROUTES.filter((route) => route.path.includes("notifications"))).toEqual([ + { method: "GET", path: "/sessions/notifications" }, + { method: "GET", path: "/sessions/:sessionId/notifications" }, + { method: "POST", path: "/sessions/:sessionId/notifications/dismiss" }, + { method: "POST", path: "/sessions/:sessionId/notifications/dismiss-all" }, + ]); + expect(FEDERATED_WEBSOCKET_ROUTES.some((path) => path.includes("notifications"))).toBe(false); + }); + it("covers machine-scoped client HTTP calls with remote proxy routes", async () => { const fetchMock = vi.fn(() => Promise.resolve(jsonResponse({}))); vi.stubGlobal("fetch", fetchMock); diff --git a/src/client/src/controllers/sessionController.ts b/src/client/src/controllers/sessionController.ts index 7756d82..5c13f28 100644 --- a/src/client/src/controllers/sessionController.ts +++ b/src/client/src/controllers/sessionController.ts @@ -104,7 +104,7 @@ export class SessionController { if (event.type === "status.update") this.queueStatusUpdate(event.status); else if (event.type === "activity.update") this.queueActivityUpdate(event.activity); else if (event.type === "session.created") this.applyCreatedSession(event.session); - else this.applySessionName(event.sessionId, event.name); + else if (event.type === "session.name") this.applySessionName(event.sessionId, event.name); } dispose() { diff --git a/src/server/app.remoteProxy.test.ts b/src/server/app.remoteProxy.test.ts index 4fd8083..935ca11 100644 --- a/src/server/app.remoteProxy.test.ts +++ b/src/server/app.remoteProxy.test.ts @@ -186,6 +186,33 @@ describe("buildApp remote machine proxy routes", () => { expect(request).toHaveBeenCalledWith("POST", "/api/sessions/s1/reload", { cwd: "/repo" }); }); + it("proxies only the four allowlisted remote notification HTTP routes", async () => { + const addResponse = await appTestContext.app.inject({ method: "POST", url: "/api/machines", payload: { name: "Remote", baseUrl: "https://remote.example.test/" } }); + const remote = addResponse.json<{ id: string }>(); + const request = vi.fn((method, path, body) => Promise.resolve({ + statusCode: 200, + headers: { "content-type": "application/json" }, + body: Readable.from([JSON.stringify({ method, path, body })]), + })); + appTestContext.remoteClient = fakeRemoteClient({ request }); + + const catalog = await appTestContext.app.inject({ method: "GET", url: `/api/machines/${remote.id}/sessions/notifications` }); + const inbox = await appTestContext.app.inject({ method: "GET", url: `/api/machines/${remote.id}/sessions/${encodeURIComponent("s 1")}/notifications?cwd=${encodeURIComponent("/repo one")}` }); + const dismissBody = { cwd: "/repo one", daemonInstanceId: "daemon-test", notificationId: "notice-1" }; + const dismiss = await appTestContext.app.inject({ method: "POST", url: `/api/machines/${remote.id}/sessions/${encodeURIComponent("s 1")}/notifications/dismiss`, payload: dismissBody }); + const dismissAllBody = { cwd: "/repo one", daemonInstanceId: "daemon-test", throughOrder: 7, throughOverflowWatermark: 2 }; + const dismissAll = await appTestContext.app.inject({ method: "POST", url: `/api/machines/${remote.id}/sessions/${encodeURIComponent("s 1")}/notifications/dismiss-all`, payload: dismissAllBody }); + const wrongMethod = await appTestContext.app.inject({ method: "DELETE", url: `/api/machines/${remote.id}/sessions/s1/notifications` }); + + expect([catalog.statusCode, inbox.statusCode, dismiss.statusCode, dismissAll.statusCode]).toEqual([200, 200, 200, 200]); + expect(wrongMethod.statusCode).toBe(404); + expect(request).toHaveBeenNthCalledWith(1, "GET", "/api/sessions/notifications", undefined); + expect(request).toHaveBeenNthCalledWith(2, "GET", "/api/sessions/s%201/notifications?cwd=%2Frepo%20one", undefined); + expect(request).toHaveBeenNthCalledWith(3, "POST", "/api/sessions/s%201/notifications/dismiss", dismissBody); + expect(request).toHaveBeenNthCalledWith(4, "POST", "/api/sessions/s%201/notifications/dismiss-all", dismissAllBody); + expect(request).toHaveBeenCalledTimes(4); + }); + it("proxies remote session queue clearing through the allowlisted route", async () => { const addResponse = await appTestContext.app.inject({ method: "POST", url: "/api/machines", payload: { name: "Remote", baseUrl: "https://remote.example.test/" } }); const remote = addResponse.json<{ id: string }>(); diff --git a/src/server/realtime/sessionEventHub.test.ts b/src/server/realtime/sessionEventHub.test.ts index c397dec..26e3336 100644 --- a/src/server/realtime/sessionEventHub.test.ts +++ b/src/server/realtime/sessionEventHub.test.ts @@ -23,6 +23,36 @@ describe("SessionEventHub", () => { expect(otherSocket.send).not.toHaveBeenCalled(); }); + it("keeps notification inbox events session-scoped and sequence-stamped", () => { + const hub = new SessionEventHub(); + const sessionSocket = new FakeSocket(); + const otherSocket = new FakeSocket(); + hub.add("s1", sessionSocket); + hub.add("s2", otherSocket); + const notification = { id: "daemon-test:1", message: "notice", truncated: false, severity: "warning" as const, receivedAt: "2026-01-01T00:00:00.000Z", order: 1 }; + const summary = { sessionId: "s1", cwd: "/workspace", inboxRevision: 1, retainedCount: 1, discardedCount: 0, highestSeverity: "warning" as const }; + + hub.publish("s1", { + type: "notifications.inbox", + daemonInstanceId: "daemon-test", + catalogRevision: 1, + summary, + dismissThrough: { order: 1, overflowWatermark: 0 }, + delta: { kind: "added", notification }, + }); + + expect(sessionSocket.send).toHaveBeenCalledWith(JSON.stringify({ + type: "notifications.inbox", + daemonInstanceId: "daemon-test", + catalogRevision: 1, + summary, + dismissThrough: { order: 1, overflowWatermark: 0 }, + delta: { kind: "added", notification }, + seq: 1, + })); + expect(otherSocket.send).not.toHaveBeenCalled(); + }); + it("omits thinking signatures from final-message payloads without mutating source events", () => { const hub = new SessionEventHub(); const socket = new FakeSocket(); @@ -103,6 +133,30 @@ describe("SessionEventHub", () => { expect(sessionSocket.send).not.toHaveBeenCalled(); }); + it("publishes notification summaries only to global sockets", () => { + const hub = new SessionEventHub(); + const globalSocket = new FakeSocket(); + const sessionSocket = new FakeSocket(); + hub.addGlobal(globalSocket); + hub.add("s1", sessionSocket); + const summary = { sessionId: "s1", cwd: "/workspace", inboxRevision: 1, retainedCount: 1, discardedCount: 0, highestSeverity: "warning" as const }; + + hub.publishNotificationSummary({ + type: "notifications.summary", + daemonInstanceId: "daemon-test", + catalogRevision: 1, + summary, + }); + + expect(globalSocket.send).toHaveBeenCalledWith(JSON.stringify({ + type: "notifications.summary", + daemonInstanceId: "daemon-test", + catalogRevision: 1, + summary, + })); + expect(sessionSocket.send).not.toHaveBeenCalled(); + }); + it("contains termination failures while publishing unstamped global events", () => { const hub = new SessionEventHub(); const failed = new FakeSocket(); diff --git a/src/server/realtime/sessionEventHub.ts b/src/server/realtime/sessionEventHub.ts index 2cbb39f..e8f2144 100644 --- a/src/server/realtime/sessionEventHub.ts +++ b/src/server/realtime/sessionEventHub.ts @@ -1,4 +1,4 @@ -import type { GlobalSessionEvent, RealtimeEvent, SessionUiEvent } from "../../shared/apiTypes.js"; +import type { GlobalSessionEvent, RealtimeEvent, SessionNotificationSummaryEvent, SessionUiEvent } from "../../shared/apiTypes.js"; import { projectBrowserSessionEvent } from "../browserMessageProjection.js"; export interface RealtimeSocket { @@ -52,6 +52,11 @@ export class SessionEventHub { this.publishRealtime(event); } + publishNotificationSummary(event: SessionNotificationSummaryEvent): void { + const payload = JSON.stringify(event); + this.sendToSockets(this.globalSockets, payload); + } + publishRealtime(event: RealtimeEvent): void { const payload = JSON.stringify(event); this.sendToSockets(this.globalSockets, payload); diff --git a/src/server/sessiond.ts b/src/server/sessiond.ts index e35fde1..82278c2 100644 --- a/src/server/sessiond.ts +++ b/src/server/sessiond.ts @@ -11,6 +11,7 @@ import { registerAuthRoutes } from "./sessions/authRoutes.js"; import { PiSessionService } from "./sessions/piSessionService.js"; import { createPiSessionManagerGateway } from "./sessions/piSessionManagerGateway.js"; import { registerSessionRoutes } from "./sessions/sessionRoutes.js"; +import { SessionNotificationStore } from "./sessions/sessionNotificationStore.js"; import { ProjectScopedSpawnTargetResolver } from "./sessions/spawnTargetResolver.js"; import { ProjectService } from "./projects/projectService.js"; import { ProjectStore } from "./storage/projectStore.js"; @@ -38,6 +39,7 @@ await runSessionDaemonStartup({ logger: app.log, async createRuntime() { const eventHub = new SessionEventHub(); + const notificationStore = new SessionNotificationStore(); const workspaceActivity = new WorkspaceActivityService(eventHub); const auth = await AuthService.create({ agentDir: activeAgentProfile.dir, logger: app.log }); const spawnTargets = config.spawnSessions @@ -50,6 +52,7 @@ await runSessionDaemonStartup({ logger: app.log, ...(spawnTargets === undefined ? {} : { spawnTargets }), subsessionsEnabled: spawnTargets !== undefined && config.subsessions, + notificationStore, sessionManager: createPiSessionManagerGateway({ agentDir: activeAgentProfile.dir, env: daemonEnvironment, diff --git a/src/server/sessiond/sessionProxyRoutes.test.ts b/src/server/sessiond/sessionProxyRoutes.test.ts index 20dc14b..5fde3cf 100644 --- a/src/server/sessiond/sessionProxyRoutes.test.ts +++ b/src/server/sessiond/sessionProxyRoutes.test.ts @@ -39,6 +39,29 @@ describe("machine-scoped session proxy routes", () => { expect(daemon.requests).toEqual([{ method: "POST", path: "/sessions/session-1/queue/clear", body: { cwd: "/repo" } }]); }); + it("forwards notification snapshots and dismissal bodies unchanged", async () => { + const catalog = await app.inject({ method: "GET", url: "/api/machines/local/sessions/notifications" }); + const inbox = await app.inject({ method: "GET", url: `/api/machines/local/sessions/session-1/notifications?cwd=${encodeURIComponent("/repo")}` }); + const dismiss = await app.inject({ + method: "POST", + url: "/api/machines/local/sessions/session-1/notifications/dismiss", + payload: { cwd: "/repo", daemonInstanceId: "daemon-test", notificationId: "notice-1" }, + }); + const dismissAll = await app.inject({ + method: "POST", + url: "/api/machines/local/sessions/session-1/notifications/dismiss-all", + payload: { cwd: "/repo", daemonInstanceId: "daemon-test", throughOrder: 7, throughOverflowWatermark: 2 }, + }); + + expect([catalog.statusCode, inbox.statusCode, dismiss.statusCode, dismissAll.statusCode]).toEqual([200, 200, 200, 200]); + expect(daemon.requests).toEqual([ + { method: "GET", path: "/sessions/notifications", body: undefined }, + { method: "GET", path: "/sessions/session-1/notifications?cwd=%2Frepo", body: undefined }, + { method: "POST", path: "/sessions/session-1/notifications/dismiss", body: { cwd: "/repo", daemonInstanceId: "daemon-test", notificationId: "notice-1" } }, + { method: "POST", path: "/sessions/session-1/notifications/dismiss-all", body: { cwd: "/repo", daemonInstanceId: "daemon-test", throughOrder: 7, throughOverflowWatermark: 2 } }, + ]); + }); + it("strips the machine prefix before forwarding auth requests", async () => { const response = await app.inject({ method: "POST", url: "/api/machines/local/auth/api-key", payload: { providerId: "p", key: "k" } }); diff --git a/src/server/sessions/piSessionService.archiveCleanup.test.ts b/src/server/sessions/piSessionService.archiveCleanup.test.ts index 22433e8..6a2b8c2 100644 --- a/src/server/sessions/piSessionService.archiveCleanup.test.ts +++ b/src/server/sessions/piSessionService.archiveCleanup.test.ts @@ -1,10 +1,47 @@ import { describe, expect, it, vi } from "vitest"; import { PiSessionService } from "./piSessionService.js"; +import { SessionNotificationStore } from "./sessionNotificationStore.js"; import { CapturingSessionEventHub, fakeRuntime, fakeSessionManager, runtimeCreator, sessionGateway, sessionRecord, sessionRef, testModelRuntime } from "./piSessionService.testSupport.js"; const TEST_AGENT_DIR = "/tmp/pi-web-test-agent"; describe("PiSessionService archive and cleanup", () => { + it("clears an active notification inbox when archiving", async () => { + const store = new SessionNotificationStore({ daemonInstanceId: "daemon-archive-test" }); + const hub = new CapturingSessionEventHub(); + const fake = fakeRuntime("archive-notification-session", { + sessionManager: fakeSessionManager("/workspace", { getSessionId: () => "archive-notification-session" }), + }); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(fake.runtime), + archiveStore: { + list: () => Promise.resolve([]), + get: () => Promise.resolve(undefined), + archive: (input) => Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }), + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + }, + sessionManager: sessionGateway([sessionRecord("archive-notification-session")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("archive-notification-session")); + const generation = store.currentGeneration("archive-notification-session", "/workspace"); + if (generation === undefined) throw new Error("expected active notification generation"); + store.addNotification(generation, "archive me", "warning"); + + await service.archive(sessionRef("archive-notification-session")); + + expect(() => store.inboxSnapshot("archive-notification-session", "/workspace")).toThrow("Session not found"); + expect(hub.notificationSummaryEvents.at(-1)).toMatchObject({ + summary: { sessionId: "archive-notification-session", retainedCount: 0 }, + }); + await service.dispose(); + }); + it("archives a session subtree within the root workspace", async () => { const archivedInputs: string[] = []; const root = sessionRecord("root"); @@ -13,9 +50,13 @@ describe("PiSessionService archive and cleanup", () => { const grandchild = { ...sessionRecord("grandchild"), path: "/sessions/grandchild.jsonl", parentSessionPath: archivedChild.path }; const otherWorkspaceChild = { ...sessionRecord("other-child", "/other"), path: "/sessions/other-child.jsonl", parentSessionPath: root.path }; const fake = fakeRuntime("root", { sessionFile: root.path }); + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-tree-test" }); + const archivedRegistration = notificationStore.registerSession("archived-child", "/workspace"); + notificationStore.addNotification(archivedRegistration.generation, "residual archived child", "warning"); const service = new PiSessionService(new CapturingSessionEventHub(), { agentDir: TEST_AGENT_DIR, modelRuntime: testModelRuntime, + notificationStore, createAgentRuntime: runtimeCreator(fake.runtime), archiveStore: { list: () => Promise.resolve([{ sessionId: "archived-child", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", originalPath: archivedChild.path, archivePath: "/archive/archived-child.jsonl", created: "2026-01-01T00:00:00.000Z", modified: "2026-01-01T00:01:00.000Z", messageCount: 1, firstMessage: "archived", parentSessionPath: root.path }]), @@ -42,15 +83,92 @@ describe("PiSessionService archive and cleanup", () => { skippedAlreadyArchivedCount: 1, }); expect(archivedInputs).toEqual(["root", "direct-child", "grandchild"]); + expect(() => notificationStore.inboxSnapshot("archived-child", "/workspace")).toThrow("Session not found"); await service.dispose(); }); + it("defensively removes residual notification state while listing archived sessions", async () => { + const store = new SessionNotificationStore({ daemonInstanceId: "daemon-reconcile-test" }); + const registration = store.registerSession("archived", "/workspace"); + store.addNotification(registration.generation, "residual", "error"); + const hub = new CapturingSessionEventHub(); + const archivedRecord = { + sessionId: "archived", + cwd: "/workspace", + archivedAt: "2026-01-02T00:00:00.000Z", + originalPath: "/sessions/archived.jsonl", + archivePath: "/archive/archived.jsonl", + }; + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + archiveStore: { + list: () => Promise.resolve([archivedRecord]), + get: () => Promise.resolve(undefined), + archive: () => Promise.reject(new Error("archive should not be called")), + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(true), + }, + sessionManager: sessionGateway([]), + heartbeatIntervalMs: 60_000, + }); + + await service.list("/workspace"); + + expect(store.catalogSnapshot().sessions).toEqual([]); + expect(() => store.inboxSnapshot("archived", "/workspace")).toThrow("Session not found"); + expect(hub.notificationSummaryEvents.at(-1)).toMatchObject({ summary: { sessionId: "archived", retainedCount: 0 } }); + await service.dispose(); + }); + + it("does not register notifications while opening an archived session read-only", async () => { + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-archived-open-test" }); + const archivedRuntime = fakeRuntime("archived", { + bindExtensions: (bindings) => { + bindings.uiContext?.notify("archived startup", "error"); + return Promise.resolve(); + }, + }); + const archivedRecord = { sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/archived.jsonl" }; + const hub = new CapturingSessionEventHub(); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore, + createAgentRuntime: runtimeCreator(archivedRuntime.runtime), + archiveStore: { + list: () => Promise.resolve([archivedRecord]), + get: () => Promise.resolve(archivedRecord), + archive: () => Promise.reject(new Error("archive should not be called")), + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(true), + }, + sessionManager: sessionGateway([]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("archived")); + + expect(notificationStore.catalogSnapshot().sessions).toEqual([]); + expect(() => notificationStore.inboxSnapshot("archived", "/workspace")).toThrow("Session not found"); + expect(hub.sessionEvents).toContainEqual({ + sessionId: "archived", + event: { type: "command.output", level: "error", message: "archived startup" }, + }); + await service.dispose(); + }); + it("permanently deletes archived sessions through the archive store", async () => { const deletedSessionIds: string[] = []; + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-delete-test" }); + const registration = notificationStore.registerSession("archived", "/workspace"); + notificationStore.addNotification(registration.generation, "delete me", "info"); const service = new PiSessionService(new CapturingSessionEventHub(), { agentDir: TEST_AGENT_DIR, modelRuntime: testModelRuntime, + notificationStore, archiveStore: { list: () => Promise.resolve([]), get: (sessionId) => Promise.resolve(sessionId === "archived" || "archived".startsWith(sessionId) @@ -72,6 +190,34 @@ describe("PiSessionService archive and cleanup", () => { await expect(service.deleteArchived("active")).rejects.toThrow("Archived session not found"); expect(deletedSessionIds).toEqual(["archived"]); + expect(notificationStore.catalogSnapshot().sessions).toEqual([]); + await service.dispose(); + }); + + it("clears residual notifications before restoring an archived session", async () => { + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-restore-test" }); + const registration = notificationStore.registerSession("archived", "/workspace"); + notificationStore.addNotification(registration.generation, "restore me", "info"); + const restore = vi.fn(() => Promise.resolve()); + const service = new PiSessionService(new CapturingSessionEventHub(), { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore, + archiveStore: { + list: () => Promise.resolve([]), + get: () => Promise.resolve({ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/archived.jsonl" }), + archive: () => Promise.reject(new Error("archive should not be called")), + restore, + isArchived: () => Promise.resolve(true), + }, + sessionManager: sessionGateway([]), + heartbeatIntervalMs: 60_000, + }); + + await service.restore(sessionRef("archived")); + + expect(restore).toHaveBeenCalledWith("archived"); + expect(notificationStore.catalogSnapshot().sessions).toEqual([]); await service.dispose(); }); @@ -83,9 +229,15 @@ describe("PiSessionService archive and cleanup", () => { const listCalls: string[] = []; const open = vi.fn(() => { throw new Error("bulk archive should not open inactive runtimes"); }); const archiveMany = vi.fn((inputs: readonly { sessionId: string; cwd: string }[]) => Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" })))); + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-bulk-archive-test" }); + for (const [sessionId, cwd] of [["a", "/one"], ["b", "/one"], ["c", "/two"]] as const) { + const registration = notificationStore.registerSession(sessionId, cwd); + notificationStore.addNotification(registration.generation, `notice ${sessionId}`, "info"); + } const service = new PiSessionService(new CapturingSessionEventHub(), { agentDir: TEST_AGENT_DIR, modelRuntime: testModelRuntime, + notificationStore, archiveStore: { list: () => Promise.resolve([]), get: () => Promise.resolve(undefined), @@ -112,6 +264,7 @@ describe("PiSessionService archive and cleanup", () => { expect(open).not.toHaveBeenCalled(); expect(archiveMany).toHaveBeenCalledTimes(1); expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["a", "b", "c"]); + expect(notificationStore.catalogSnapshot().sessions).toEqual([]); await service.dispose(); }); @@ -243,9 +396,15 @@ describe("PiSessionService archive and cleanup", () => { let listAllCalls = 0; const archived = { sessionId: "archived-old", cwd: "/old-project", archivedAt: "2026-04-01T00:00:00.000Z", archivePath: "/archive/archived-old.jsonl" }; const otherArchived = { sessionId: "archived-other", cwd: "/other-project", archivedAt: "2026-04-01T00:00:00.000Z", archivePath: "/archive/archived-other.jsonl" }; + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-cleanup-test" }); + for (const sessionId of ["execute-only", "archived-old"]) { + const registration = notificationStore.registerSession(sessionId, "/old-project"); + notificationStore.addNotification(registration.generation, `notice ${sessionId}`, "warning"); + } const service = new PiSessionService(new CapturingSessionEventHub(), { agentDir: TEST_AGENT_DIR, modelRuntime: testModelRuntime, + notificationStore, now: () => new Date("2026-06-25T00:00:00.000Z"), archiveStore: { list: () => Promise.resolve([archived, otherArchived]), @@ -283,12 +442,14 @@ describe("PiSessionService archive and cleanup", () => { expect(preview.projects).toEqual([{ cwd: "/old-project", archiveCount: 1, deleteCount: 1 }]); expect(archivedInputs).toEqual([]); expect(deletedSessionIds).toEqual([]); + expect(notificationStore.catalogSnapshot().sessions).toHaveLength(2); const result = await service.cleanup({ thresholds: { archiveIdleDays: 30, deleteArchivedDays: 30 }, projectCwds: ["/old-project"] }); expect(result.archivedSessionIds).toEqual(["execute-only"]); expect(result.deletedSessionIds).toEqual(["archived-old"]); expect(archivedInputs).toEqual(["execute-only"]); expect(deletedSessionIds).toEqual(["archived-old"]); + expect(notificationStore.catalogSnapshot().sessions).toEqual([]); await service.dispose(); }); diff --git a/src/server/sessions/piSessionService.lifecycle.test.ts b/src/server/sessions/piSessionService.lifecycle.test.ts index a987bc8..cafcde0 100644 --- a/src/server/sessions/piSessionService.lifecycle.test.ts +++ b/src/server/sessions/piSessionService.lifecycle.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { describe, expect, it, vi } from "vitest"; import { PiSessionService, type PiAgentSession, type PiSessionRuntime } from "./piSessionService.js"; +import { SessionNotificationStore } from "./sessionNotificationStore.js"; import { CapturingSessionEventHub, emptyArchiveStore, fakeRuntime, fakeSessionManager, runtimeCreator, sessionGateway, sessionRecord, sessionRef, testModelRuntime, type RuntimeCreator } from "./piSessionService.testSupport.js"; const TEST_AGENT_DIR = "/tmp/pi-web-test-agent"; @@ -17,6 +18,32 @@ function deferred() { return { promise, resolve, reject }; } +function notificationStore() { + let tick = 0; + return new SessionNotificationStore({ + daemonInstanceId: "daemon-lifecycle-test", + now: () => new Date(Date.UTC(2026, 0, 1, 0, 0, tick++)), + }); +} + +function boundNotify(fake: { calls: { bindExtensions: unknown[] } }, index = -1) { + const bindings = fake.calls.bindExtensions.at(index); + if (typeof bindings !== "object" || bindings === null || !("uiContext" in bindings) || !hasNotify(bindings.uiContext)) { + throw new Error("Expected bound extension UI context"); + } + const uiContext = bindings.uiContext; + return (message: string, type?: "info" | "warning" | "error") => { uiContext.notify(message, type); }; +} + +function hasNotify(value: unknown): value is { notify(message: string, type?: "info" | "warning" | "error"): void } { + return typeof value === "object" && value !== null && "notify" in value && typeof value.notify === "function"; +} + +function currentNotify(fake: { session: Pick }) { + const uiContext = fake.session.extensionRunner.getUIContext(); + return (message: string, type?: "info" | "warning" | "error") => { uiContext.notify(message, type); }; +} + describe("PiSessionService lifecycle, listing, and reload", () => { it("starts sessions through an injected runtime creator", async () => { const hub = new CapturingSessionEventHub(); @@ -351,10 +378,235 @@ describe("PiSessionService lifecycle, listing, and reload", () => { await expect(service.runCommand(sessionRef("extension-command-session"), "/ctx-stats")).resolves.toEqual({ type: "done" }); expect(extensionMode).toBe("rpc"); - expect(hub.sessionEvents).toContainEqual({ + const legacyEvent = hub.sessionEvents.find(({ event }) => event.type === "command.output" && event.message === "context-mode stats"); + expect(legacyEvent).toMatchObject({ sessionId: "extension-command-session", event: { type: "command.output", level: "info", message: "context-mode stats" }, }); + expect(legacyEvent?.event.type === "command.output" ? typeof legacyEvent.event.notificationId : undefined).toBe("string"); + const inboxEvent = hub.sessionEvents.find(({ event }) => event.type === "notifications.inbox"); + expect(inboxEvent).toMatchObject({ + sessionId: "extension-command-session", + event: { + type: "notifications.inbox", + delta: { kind: "added", notification: { message: "context-mode stats", severity: "info" } }, + }, + }); + expect(hub.notificationSummaryEvents.at(-1)).toMatchObject({ + type: "notifications.summary", + summary: { sessionId: "extension-command-session", retainedCount: 1, highestSeverity: "info" }, + }); + + await service.dispose(); + }); + + it("stores every extension notification without touching Pi session history", async () => { + const hub = new CapturingSessionEventHub(); + const store = notificationStore(); + const branch = [{ type: "message", message: { role: "user", content: "existing" } }]; + const fake = fakeRuntime("notification-session", { + sessionManager: fakeSessionManager("/workspace", { + getSessionId: () => "notification-session", + getBranch: () => branch, + }), + }); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(fake.runtime), + sessionManager: sessionGateway([]), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + const notify = boundNotify(fake); + notify("duplicate", "warning"); + notify("duplicate", "error"); + + const snapshot = service.notificationInbox(sessionRef("notification-session")); + expect(snapshot.notifications).toMatchObject([ + { id: "daemon-lifecycle-test:2", message: "duplicate", severity: "error" }, + { id: "daemon-lifecycle-test:1", message: "duplicate", severity: "warning" }, + ]); + expect(fake.session.sessionManager.getBranch()).toBe(branch); + expect(fake.session.messages).toEqual([]); + expect(hub.sessionEvents.filter(({ event }) => event.type === "command.output")).toHaveLength(2); + expect(hub.sessionEvents.filter(({ event }) => event.type === "notifications.inbox")).toHaveLength(2); + + await service.dispose(); + }); + + it("commits Pi /reload only after replacement session_start notifications are bound", async () => { + const hub = new CapturingSessionEventHub(); + const store = notificationStore(); + const fake = fakeRuntime("runtime-reload-notifications"); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(fake.runtime), + sessionManager: sessionGateway([sessionRecord("runtime-reload-notifications")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("runtime-reload-notifications")); + const oldNotify = boundNotify(fake); + oldNotify("old notification", "warning"); + fake.session.reload = async (options) => { + oldNotify("shutdown notification", "info"); + await options?.beforeSessionStart?.(); + currentNotify(fake)("replacement startup", "error"); + }; + + await expect(service.runCommand(sessionRef("runtime-reload-notifications"), "/reload")).resolves.toMatchObject({ type: "done" }); + + expect(service.notificationInbox(sessionRef("runtime-reload-notifications"))).toMatchObject({ + summary: { retainedCount: 1, discardedCount: 0, highestSeverity: "error" }, + notifications: [{ message: "replacement startup", severity: "error" }], + }); + expect(fake.calls.bindExtensions).toHaveLength(1); + const revision = service.notificationInbox(sessionRef("runtime-reload-notifications")).summary.inboxRevision; + oldNotify("stale old runner", "error"); + expect(service.notificationInbox(sessionRef("runtime-reload-notifications")).summary.inboxRevision).toBe(revision); + + await service.dispose(); + }); + + it("preserves prior and candidate notifications when Pi /reload fails after rotation", async () => { + const hub = new CapturingSessionEventHub(); + const store = notificationStore(); + const fake = fakeRuntime("failed-runtime-reload"); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(fake.runtime), + sessionManager: sessionGateway([sessionRecord("failed-runtime-reload")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("failed-runtime-reload")); + boundNotify(fake)("prior", "info"); + fake.session.reload = async (options) => { + await options?.beforeSessionStart?.(); + currentNotify(fake)("candidate before failure", "warning"); + throw new Error("reload failed after rotation"); + }; + + await expect(service.runCommand(sessionRef("failed-runtime-reload"), "/reload")).resolves.toEqual({ + type: "unsupported", + message: "Reload failed: reload failed after rotation", + }); + expect(service.notificationInbox(sessionRef("failed-runtime-reload")).notifications.map((notification) => notification.message)).toEqual([ + "candidate before failure", + "prior", + ]); + currentNotify(fake)("after failed reload", "error"); + expect(service.notificationInbox(sessionRef("failed-runtime-reload")).notifications[0]).toMatchObject({ + message: "after failed reload", + severity: "error", + }); + + await service.dispose(); + }); + + it("leaves the prior inbox unchanged when Pi /reload fails before rotation", async () => { + const store = notificationStore(); + const fake = fakeRuntime("failed-before-rotation"); + const service = new PiSessionService(new CapturingSessionEventHub(), { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(fake.runtime), + sessionManager: sessionGateway([sessionRecord("failed-before-rotation")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("failed-before-rotation")); + boundNotify(fake)("prior", "warning"); + const before = service.notificationInbox(sessionRef("failed-before-rotation")); + fake.session.reload = () => Promise.reject(new Error("reload failed before rotation")); + + await expect(service.runCommand(sessionRef("failed-before-rotation"), "/reload")).resolves.toEqual({ + type: "unsupported", + message: "Reload failed: reload failed before rotation", + }); + expect(service.notificationInbox(sessionRef("failed-before-rotation"))).toEqual(before); + + await service.dispose(); + }); + + it("commits changed-id SDK rebind notifications only after binding succeeds", async () => { + const store = notificationStore(); + const first = fakeRuntime("session-1"); + const replacement = fakeRuntime("session-2", { + bindExtensions: (bindings) => { + bindings.uiContext?.notify("replacement startup", "error"); + return Promise.resolve(); + }, + }); + let rebindSession: ((session: PiAgentSession) => Promise) | undefined; + first.runtime.setRebindSession = (callback) => { rebindSession = callback; }; + const service = new PiSessionService(new CapturingSessionEventHub(), { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(first.runtime), + sessionManager: sessionGateway([]), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + const staleNotify = boundNotify(first); + staleNotify("old", "warning"); + Object.defineProperty(first.runtime, "session", { configurable: true, value: replacement.session }); + await rebindSession?.(replacement.session); + + expect(() => service.notificationInbox(sessionRef("session-1"))).toThrow("Session not found"); + expect(service.notificationInbox(sessionRef("session-2"))).toMatchObject({ + notifications: [{ message: "replacement startup", severity: "error" }], + }); + const revision = service.notificationInbox(sessionRef("session-2")).summary.inboxRevision; + staleNotify("stale", "error"); + expect(service.notificationInbox(sessionRef("session-2")).summary.inboxRevision).toBe(revision); + + await service.dispose(); + }); + + it("preserves changed-id SDK rebind notifications on the applied replacement when binding fails", async () => { + const store = notificationStore(); + const first = fakeRuntime("session-1"); + const replacement = fakeRuntime("session-2"); + replacement.session.bindExtensions = (bindings) => { + replacement.session.extensionRunner.setUIContext(bindings.uiContext, "rpc"); + bindings.uiContext?.notify("candidate before bind failure", "warning"); + return Promise.reject(new Error("replacement bind failed")); + }; + let rebindSession: ((session: PiAgentSession) => Promise) | undefined; + first.runtime.setRebindSession = (callback) => { rebindSession = callback; }; + const service = new PiSessionService(new CapturingSessionEventHub(), { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(first.runtime), + sessionManager: sessionGateway([]), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + boundNotify(first)("prior", "info"); + Object.defineProperty(first.runtime, "session", { configurable: true, value: replacement.session }); + await expect(rebindSession?.(replacement.session)).rejects.toThrow("replacement bind failed"); + + expect(() => service.notificationInbox(sessionRef("session-1"))).toThrow("Session not found"); + expect(service.notificationInbox(sessionRef("session-2")).notifications.map((notification) => notification.message)).toEqual([ + "candidate before bind failure", + "prior", + ]); + await expect(service.status(sessionRef("session-2"))).resolves.toMatchObject({ sessionId: "session-2" }); + currentNotify(replacement)("after failed rebind", "error"); + expect(service.notificationInbox(sessionRef("session-2")).notifications[0]).toMatchObject({ message: "after failed rebind", severity: "error" }); await service.dispose(); }); @@ -489,6 +741,34 @@ describe("PiSessionService lifecycle, listing, and reload", () => { }); + it("keeps notifications on abort but clears and unregisters them on stop", async () => { + const hub = new CapturingSessionEventHub(); + const store = notificationStore(); + const fake = fakeRuntime("stop-notification-session"); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(fake.runtime), + sessionManager: sessionGateway([sessionRecord("stop-notification-session")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("stop-notification-session")); + boundNotify(fake)("keep through abort", "warning"); + await service.abort(sessionRef("stop-notification-session")); + expect(service.notificationInbox(sessionRef("stop-notification-session")).summary.retainedCount).toBe(1); + await expect(service.stop(sessionRef("stop-notification-session", "/other"))).rejects.toThrow("Session cwd mismatch"); + expect(service.activeCount()).toBe(1); + + await service.stop(sessionRef("stop-notification-session")); + expect(() => service.notificationInbox(sessionRef("stop-notification-session"))).toThrow("Session not found"); + expect(service.notificationCatalog().sessions).toEqual([]); + expect(hub.notificationSummaryEvents.at(-1)).toMatchObject({ summary: { sessionId: "stop-notification-session", retainedCount: 0 } }); + + await service.dispose(); + }); + it("runs /reload by refreshing the active runtime resources in place", async () => { const hub = new CapturingSessionEventHub(); const fake = fakeRuntime("runtime-reload-session"); @@ -549,6 +829,123 @@ describe("PiSessionService lifecycle, listing, and reload", () => { await service.dispose(); }); + it("reload-from-disk keeps replacement startup notifications and clears the old inbox on success", async () => { + const store = notificationStore(); + const first = fakeRuntime("reload-notification-session"); + const second = fakeRuntime("reload-notification-session", { + bindExtensions: (bindings) => { + bindings.uiContext?.notify("replacement startup", "error"); + return Promise.resolve(); + }, + }); + const runtimes = [first.runtime, second.runtime]; + let createCalls = 0; + const hub = new CapturingSessionEventHub(); + const service = new PiSessionService(hub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: () => { + const runtime = runtimes[createCalls++]; + return runtime === undefined ? Promise.reject(new Error("unexpected runtime creation")) : Promise.resolve(runtime); + }, + sessionManager: sessionGateway([sessionRecord("reload-notification-session")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("reload-notification-session")); + const oldNotify = boundNotify(first); + oldNotify("old", "warning"); + const disposeFirst = first.runtime.dispose.bind(first.runtime); + first.runtime.dispose = async () => { + oldNotify("old shutdown", "info"); + await disposeFirst(); + }; + await service.reload(sessionRef("reload-notification-session")); + + expect(service.notificationInbox(sessionRef("reload-notification-session"))).toMatchObject({ + summary: { retainedCount: 1, highestSeverity: "error" }, + notifications: [{ message: "replacement startup", severity: "error" }], + }); + expect(hub.sessionEvents.some(({ event }) => event.type === "notifications.inbox" && event.delta.kind === "added" && event.delta.notification.message === "old shutdown")).toBe(true); + await service.dispose(); + }); + + it("reload-from-disk preserves prior and candidate notifications when replacement binding fails", async () => { + const store = notificationStore(); + const first = fakeRuntime("failed-disk-reload"); + const failed = fakeRuntime("failed-disk-reload", { + bindExtensions: (bindings) => { + bindings.uiContext?.notify("candidate before open failure", "warning"); + return Promise.reject(new Error("replacement open failed")); + }, + }); + const runtimes = [first.runtime, failed.runtime]; + let createCalls = 0; + const service = new PiSessionService(new CapturingSessionEventHub(), { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: () => { + const runtime = runtimes[createCalls++]; + return runtime === undefined ? Promise.reject(new Error("unexpected runtime creation")) : Promise.resolve(runtime); + }, + sessionManager: sessionGateway([sessionRecord("failed-disk-reload")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("failed-disk-reload")); + const oldNotify = boundNotify(first); + oldNotify("prior", "info"); + const disposeFirst = first.runtime.dispose.bind(first.runtime); + first.runtime.dispose = async () => { + oldNotify("old shutdown", "info"); + await disposeFirst(); + }; + await expect(service.reload(sessionRef("failed-disk-reload"))).rejects.toThrow("replacement open failed"); + + expect(service.notificationInbox(sessionRef("failed-disk-reload")).notifications.map((notification) => notification.message)).toEqual([ + "candidate before open failure", + "old shutdown", + "prior", + ]); + expect(service.activeCount()).toBe(0); + await service.stop(sessionRef("failed-disk-reload")); + expect(() => service.notificationInbox(sessionRef("failed-disk-reload"))).toThrow("Session not found"); + await service.dispose(); + }); + + it("reload-from-disk preserves the prior inbox when deferred close fails", async () => { + const store = notificationStore(); + const first = fakeRuntime("failed-close-reload"); + const service = new PiSessionService(new CapturingSessionEventHub(), { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore: store, + createAgentRuntime: runtimeCreator(first.runtime), + sessionManager: sessionGateway([sessionRecord("failed-close-reload")]), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("failed-close-reload")); + const oldNotify = boundNotify(first); + oldNotify("prior", "warning"); + first.runtime.dispose = () => { + oldNotify("shutdown before close failure", "info"); + return Promise.reject(new Error("close failed")); + }; + + await expect(service.reload(sessionRef("failed-close-reload"))).rejects.toThrow("close failed"); + + expect(service.notificationInbox(sessionRef("failed-close-reload")).notifications.map((notification) => notification.message)).toEqual([ + "shutdown before close failure", + "prior", + ]); + expect(store.currentGeneration("failed-close-reload", "/workspace")).toBeDefined(); + await service.stop(sessionRef("failed-close-reload")); + await service.dispose(); + }); + it("refuses to reload a session that has active work in progress", async () => { const fake = fakeRuntime("busy-session", { isStreaming: true }); const service = new PiSessionService(new CapturingSessionEventHub(), { diff --git a/src/server/sessions/piSessionService.promptQueue.test.ts b/src/server/sessions/piSessionService.promptQueue.test.ts index 20d8314..84f22fa 100644 --- a/src/server/sessions/piSessionService.promptQueue.test.ts +++ b/src/server/sessions/piSessionService.promptQueue.test.ts @@ -456,7 +456,7 @@ describe("PiSessionService prompt, queue, and auth warnings", () => { }); await service.status(sessionRef("stop-session")); - service.stop(sessionRef("stop-session")); + await service.stop(sessionRef("stop-session")); expect(fake.calls.clearQueue).toBe(1); await service.dispose(); diff --git a/src/server/sessions/piSessionService.testSupport.ts b/src/server/sessions/piSessionService.testSupport.ts index cee3352..c25556b 100644 --- a/src/server/sessions/piSessionService.testSupport.ts +++ b/src/server/sessions/piSessionService.testSupport.ts @@ -1,12 +1,13 @@ import { ModelRuntime, type ExtensionUIContext } from "@earendil-works/pi-coding-agent"; import { InMemoryCredentialStore, type Credential, type CredentialStore } from "@earendil-works/pi-ai"; -import type { GlobalSessionEvent, SessionUiEvent } from "../../shared/apiTypes.js"; +import type { GlobalSessionEvent, SessionNotificationSummaryEvent, SessionUiEvent } from "../../shared/apiTypes.js"; import { SessionEventHub } from "../realtime/sessionEventHub.js"; import type { PiAgentSession, PiSessionManager, PiSessionRuntime, PiSessionServiceDependencies } from "./piSessionService.js"; export class CapturingSessionEventHub extends SessionEventHub { readonly sessionEvents: { sessionId: string; event: SessionUiEvent }[] = []; readonly globalEvents: GlobalSessionEvent[] = []; + readonly notificationSummaryEvents: SessionNotificationSummaryEvent[] = []; private readonly seqBySessionOverride = new Map(); override publish(sessionId: string, event: SessionUiEvent): void { @@ -17,6 +18,10 @@ export class CapturingSessionEventHub extends SessionEventHub { this.globalEvents.push(event); } + override publishNotificationSummary(event: SessionNotificationSummaryEvent): void { + this.notificationSummaryEvents.push(event); + } + /** Test seam: set the per-session watermark returned by {@link currentSeq}. */ setSeq(sessionId: string, value: number): void { this.seqBySessionOverride.set(sessionId, value); @@ -30,6 +35,8 @@ export class CapturingSessionEventHub extends SessionEventHub { export type SessionGateway = NonNullable; export type RuntimeCreator = NonNullable; +type TestExtensionBindings = Parameters[0]; + export interface TestSession extends PiAgentSession { sessionName: string | undefined; model: PiAgentSession["model"]; @@ -131,6 +138,7 @@ export function fakeRuntime(sessionId = "session-1", patch: Partial const customMessageCalls: { message: { customType: string; content: string; display: boolean; details?: unknown }; options: unknown }[] = []; const bindExtensionCalls: unknown[] = []; const listeners: ((event: unknown) => void)[] = []; + let extensionUiContext = testExtensionUiContext; const calls = { abort: 0, bindExtensions: bindExtensionCalls, clearQueue: 0, dispose: 0, prompt: promptCalls, reload: 0, sendCustomMessage: customMessageCalls }; const session: TestSession = { sessionId, @@ -150,7 +158,8 @@ export function fakeRuntime(sessionId = "session-1", patch: Partial scopedModels: [], extensionRunner: { getRegisteredCommands: () => [], - getUIContext: () => testExtensionUiContext, + getUIContext: () => extensionUiContext, + setUIContext: (uiContext) => { extensionUiContext = uiContext ?? testExtensionUiContext; }, }, promptTemplates: [], resourceLoader: { getSkills: () => ({ skills: [] }) }, @@ -161,8 +170,9 @@ export function fakeRuntime(sessionId = "session-1", patch: Partial if (index !== -1) listeners.splice(index, 1); }; }, - bindExtensions: (bindings: unknown) => { + bindExtensions: (bindings: TestExtensionBindings) => { calls.bindExtensions.push(bindings); + if (bindings.uiContext !== undefined) extensionUiContext = bindings.uiContext; 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 }), diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index 0a2a56d..4ef1c62 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -32,7 +32,19 @@ import { deterministicSessionName, fallbackSessionName, generateShortSessionName import { computeEditPreview, type EditPreviewResult } from "./editPreview.js"; import { attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachmentService.js"; import { parsePromptAttachments } from "../../shared/promptAttachments.js"; -import type { SavedPromptAttachment, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkFailure, SessionBulkMutationRef, SessionWarning } from "../../shared/apiTypes.js"; +import type { + SavedPromptAttachment, + SessionBulkArchiveResponse, + SessionBulkDeleteArchivedResponse, + SessionBulkFailure, + SessionBulkMutationRef, + SessionNotificationCatalogSnapshot, + SessionNotificationClearReason, + SessionNotificationDismissAllRequest, + SessionNotificationDismissRequest, + SessionNotificationInboxSnapshot, + SessionWarning, +} from "../../shared/apiTypes.js"; import type { SessionRouteLookup, SessionRouteRef, SessionRouteService } from "./sessionService.js"; import { type AuthChange } from "./authService.js"; @@ -43,6 +55,11 @@ import { createSubsessionToolDefinitions, type SpawnSubsessionInvocation, type S import { buildTranscriptView } from "./subsessionTranscript.js"; import { planSessionCleanup, summarizeSessionCleanupExecution, type NormalizedSessionCleanupRequest, type SessionCleanupPlan } from "./sessionCleanup.js"; import type { SpawnTargetDecision, SpawnTargetResolver } from "./spawnTargetResolver.js"; +import { + SessionNotificationStore, + type SessionNotificationGeneration, + type SessionNotificationMutation, +} from "./sessionNotificationStore.js"; /** * Minimal structured-logging seam, shaped like Fastify's logger so sessiond can @@ -245,6 +262,7 @@ export interface PiAgentSession { extensionRunner: { getRegisteredCommands(): readonly { invocationName: string; description?: string }[]; getUIContext(): ExtensionUIContext; + setUIContext(uiContext?: ExtensionUIContext, mode?: "rpc"): void; }; promptTemplates: readonly { name: string; description?: string }[]; resourceLoader: { getSkills(): { skills: readonly { name: string; description?: string }[] } }; @@ -253,7 +271,7 @@ export interface PiAgentSession { 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 }; - reload(): Promise; + reload(options?: { beforeSessionStart?: () => void | Promise }): Promise; getContextUsage(): ClientSessionStatus["contextUsage"] | undefined; prompt(text: string, options?: { streamingBehavior?: "steer" | "followUp"; images?: ImageContent[] }): Promise; sendCustomMessage(message: { customType: string; content: string; display: boolean; details?: unknown }, options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }): Promise; @@ -301,6 +319,18 @@ interface PendingSessionOpen { promise: Promise>; } +interface CreateSessionRuntimeOptions extends Pick { + notificationGeneration?: SessionNotificationGeneration; + notifications?: "enabled" | "disabled"; +} + +type NotificationClosePolicy = + | { kind: "clear"; reason: SessionNotificationClearReason } + | { kind: "defer" }; + +const CLEAR_RUNTIME_NOTIFICATIONS: NotificationClosePolicy = { kind: "clear", reason: "runtime-close" }; +const DEFER_RUNTIME_NOTIFICATIONS: NotificationClosePolicy = { kind: "defer" }; + function resourceDiagnosticToWarning(diagnostic: ResourceDiagnostic, source: string): SessionWarning { return { severity: diagnostic.type === "error" ? "error" : "warning", @@ -565,6 +595,8 @@ export interface PiSessionServiceDependencies { logger?: PiSessionLogger; /** Clock seam for cleanup planning tests. */ now?: () => Date; + /** Daemon-lifetime notification state, injected by sessiond in production. */ + notificationStore?: SessionNotificationStore; } export class PiSessionService implements SessionRouteService { @@ -600,6 +632,8 @@ export class PiSessionService implements SessionRouteService { private readonly spawnTargets: SpawnTargetResolver | undefined; private readonly logger: PiSessionLogger; private readonly now: () => Date; + private readonly notificationStore: SessionNotificationStore; + private readonly notificationGenerationBySession = new WeakMap(); constructor(private readonly events: SessionEventHub, deps: PiSessionServiceDependencies) { this.archiveStore = deps.archiveStore ?? new SessionArchiveStore(); @@ -609,6 +643,7 @@ export class PiSessionService implements SessionRouteService { this.spawnTargets = deps.spawnTargets; this.logger = deps.logger ?? noopLogger; this.now = deps.now ?? (() => new Date()); + this.notificationStore = deps.notificationStore ?? new SessionNotificationStore(); // Subsessions are a beta capability gated behind their own flag, and they // also require the spawn capability (they share its project-scope resolver). const subsessionsActive = this.spawnTargets !== undefined && deps.subsessionsEnabled === true; @@ -649,6 +684,43 @@ export class PiSessionService implements SessionRouteService { return this.active.size; } + notificationCatalog(): SessionNotificationCatalogSnapshot { + return this.notificationStore.catalogSnapshot(); + } + + notificationInbox(ref: PiSessionRef): SessionNotificationInboxSnapshot { + return this.notificationStore.inboxSnapshot(ref.id, ref.cwd); + } + + dismissNotification( + ref: PiSessionRef, + request: Omit, + ): SessionNotificationInboxSnapshot { + const result = this.notificationStore.dismissNotification( + ref.id, + ref.cwd, + request.daemonInstanceId, + request.notificationId, + ); + this.publishNotificationMutations(result.mutations); + return result.snapshot; + } + + dismissAllNotifications( + ref: PiSessionRef, + request: Omit, + ): SessionNotificationInboxSnapshot { + const result = this.notificationStore.dismissAll( + ref.id, + ref.cwd, + request.daemonInstanceId, + request.throughOrder, + request.throughOverflowWatermark, + ); + this.publishNotificationMutations(result.mutations); + return result.snapshot; + } + async cleanupPreview(request: NormalizedSessionCleanupRequest): Promise { return previewResponseFromPlan(await this.cleanupPlan(request)); } @@ -668,7 +740,7 @@ export class PiSessionService implements SessionRouteService { skippedBusySessionIds.add(input.sessionId); continue; } - await this.closeActive(input.sessionId); + await this.closeActive(input.sessionId, { kind: "clear", reason: "archive" }); readyArchiveInputs.push(input); } await this.archiveStoreArchiveMany(readyArchiveInputs); @@ -679,7 +751,7 @@ export class PiSessionService implements SessionRouteService { skippedBusySessionIds.add(record.sessionId); continue; } - await this.closeActive(record.sessionId); + await this.closeActive(record.sessionId, { kind: "clear", reason: "delete" }); readyDeleteRecords.push(record); } await this.ensureArchivedRecordsMoved(readyDeleteRecords); @@ -711,8 +783,10 @@ export class PiSessionService implements SessionRouteService { this.subsessionLinks.clear(); this.subsessionHydratedParents.clear(); this.subsessionNotifyArmed.clear(); + this.notificationStore.clearAll("service-dispose"); await Promise.all(activeSessions.map(async (active) => { active.unsubscribe(); + active.runtime.setRebindSession(undefined); this.workspaceActivity?.removeSession(active.runtime.session.sessionId, active.runtime.session.sessionManager.getCwd()); try { await active.runtime.session.abort(); @@ -731,6 +805,9 @@ export class PiSessionService implements SessionRouteService { .map((record) => this.ensureArchivedSessionMoved(record, sessionsById.get(record.sessionId))), ); const archivedById = new Map(archivedForCwd.map((record) => [record.sessionId, record])); + for (const record of archivedForCwd) { + this.publishNotificationMutations(this.notificationStore.clearSession(record.sessionId, "archive-reconcile")); + } const unarchivedSessions = sessions.filter((session) => !archivedById.has(session.id)).map(clientSessionFromListEntry); this.workspaceActivity?.reconcileSessionActivity(cwd, this.reconcilableSessionIds(cwd, unarchivedSessions.map((session) => session.id), archivedById)); const archivedSessions = archivedForCwd @@ -1347,11 +1424,29 @@ export class PiSessionService implements SessionRouteService { private async reloadSessionRuntime(session: PiAgentSession): Promise { if (this.hasActiveWork(session)) throw new Error("Stop current session activity before reloading"); this.publishActivity(session, "reloading resources", "active"); + const priorGeneration = this.notificationGenerationBySession.get(session); + let candidateGeneration: SessionNotificationGeneration | undefined; try { - await session.reload(); + await session.reload(priorGeneration === undefined ? undefined : { + beforeSessionStart: () => { + candidateGeneration = this.notificationStore.beginReplacement(priorGeneration, { + sessionId: session.sessionId, + cwd: session.sessionManager.getCwd(), + }); + this.notificationGenerationBySession.set(session, candidateGeneration); + this.replaceSessionNotificationContext(session, candidateGeneration); + }, + }); + if (candidateGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.commitReplacement(candidateGeneration)); + } this.publishActivity(session, "resources reloaded", "idle"); this.publishStatus(session); } catch (error: unknown) { + if (candidateGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.abortReplacement(candidateGeneration, "candidate")); + this.notificationGenerationBySession.set(session, candidateGeneration); + } const message = error instanceof Error ? error.message : String(error); this.publishActivity(session, "reload failed", "error", message); this.events.publish(session.sessionId, { type: "session.error", message }); @@ -1364,7 +1459,7 @@ export class PiSessionService implements SessionRouteService { const session = await this.getOrOpen(ref); if (this.hasActiveWork(session)) throw new Error("Stop current session activity before archiving"); const archiveInput = await this.archiveInputForSession(session); - await this.closeActive(session.sessionId); + await this.closeActive(session.sessionId, { kind: "clear", reason: "archive" }); await this.archiveStore.archive(archiveInput); } @@ -1381,6 +1476,7 @@ export class PiSessionService implements SessionRouteService { for (const ref of uniqueRefs) { const archived = findArchivedRecordForBulkRef(archivedRecords, ref); if (archived !== undefined) { + this.publishNotificationMutations(this.notificationStore.clearSession(archived.sessionId, "archive")); alreadyArchivedSessionIds.push(archived.sessionId); continue; } @@ -1409,7 +1505,7 @@ export class PiSessionService implements SessionRouteService { const readyInputs: ArchiveSessionInput[] = []; for (const item of planItems) { try { - await this.closeActive(item.input.sessionId); + await this.closeActive(item.input.sessionId, { kind: "clear", reason: "archive" }); readyInputs.push(item.input); } catch (error: unknown) { failures.push({ sessionId: item.input.sessionId, error: errorMessage(error) }); @@ -1440,8 +1536,11 @@ export class PiSessionService implements SessionRouteService { const busy = plan.targets.map((target) => target.activeSession).find((target) => target !== undefined && this.hasActiveWork(target)); if (busy !== undefined) throw new Error(`Stop current session activity before archiving ${sessionDisplayName(busy)}`); + for (const target of plan.targets) { + if (target.archived) this.publishNotificationMutations(this.notificationStore.clearSession(target.id, "archive")); + } const archiveInputs = plan.unarchivedTargets.map((target) => archiveInputFromCandidate(target)); - for (const input of archiveInputs) await this.closeActive(input.sessionId); + for (const input of archiveInputs) await this.closeActive(input.sessionId, { kind: "clear", reason: "archive" }); await this.archiveStoreArchiveMany(archiveInputs); return { @@ -1455,7 +1554,7 @@ export class PiSessionService implements SessionRouteService { async restore(ref: PiSessionLookup): Promise { const archived = await this.getArchived(ref); if (archived === undefined) throw new Error("Session not found"); - await this.closeActive(archived.sessionId); + await this.closeActive(archived.sessionId, { kind: "clear", reason: "restore" }); await this.archiveStore.restore(archived.sessionId); } @@ -1464,7 +1563,7 @@ export class PiSessionService implements SessionRouteService { 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); + await this.closeActive(record.sessionId, { kind: "clear", reason: "delete" }); if (record.archivePath === undefined) await this.ensureArchivedRecordMoved(record); await this.archiveStore.deleteArchived(record.sessionId); } @@ -1495,7 +1594,7 @@ export class PiSessionService implements SessionRouteService { const readyRecords: ArchivedSessionRecord[] = []; for (const item of planItems) { try { - await this.closeActive(item.record.sessionId); + await this.closeActive(item.record.sessionId, { kind: "clear", reason: "delete" }); readyRecords.push(item.record); } catch (error: unknown) { failures.push({ sessionId: item.record.sessionId, error: errorMessage(error) }); @@ -1528,9 +1627,30 @@ export class PiSessionService implements SessionRouteService { await this.assertWritable(ref); const session = await this.getOrOpen(ref); if (this.hasActiveWork(session)) throw new Error("Stop current session activity before reloading"); - await this.closeActive(session.sessionId); - const reopened = await this.getActive(ref); - this.publishStatus(reopened.runtime.session); + + const priorGeneration = this.notificationGenerationBySession.get(session); + const sessionId = session.sessionId; + const cwd = session.sessionManager.getCwd(); + let candidateGeneration: SessionNotificationGeneration | undefined; + try { + await this.closeActive( + sessionId, + priorGeneration === undefined ? CLEAR_RUNTIME_NOTIFICATIONS : DEFER_RUNTIME_NOTIFICATIONS, + ); + candidateGeneration = priorGeneration === undefined + ? undefined + : this.notificationStore.beginReplacement(priorGeneration, { sessionId, cwd }); + const reopened = await this.getActive(ref, candidateGeneration === undefined ? {} : { notificationGeneration: candidateGeneration }); + if (candidateGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.commitReplacement(candidateGeneration)); + } + this.publishStatus(reopened.runtime.session); + } catch (error: unknown) { + if (candidateGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.abortReplacement(candidateGeneration)); + } + throw error; + } } async detachParent(ref: PiSessionLookup): Promise { @@ -1569,12 +1689,17 @@ export class PiSessionService implements SessionRouteService { this.publishStatus(active.runtime.session); } - stop(ref: PiSessionLookup): void { + async stop(ref: PiSessionLookup): Promise { const active = this.activeForLookup(ref); - if (active === undefined) return; - void this.closeActive(active.runtime.session.sessionId).catch(() => { - // Best-effort shutdown; callers that need errors await closeActive directly. - }); + if (active !== undefined) { + await this.closeActive(active.runtime.session.sessionId); + return; + } + if (isPiSessionRef(ref)) { + this.publishNotificationMutations(this.notificationStore.clearSessionIdentity(ref.id, ref.cwd, "runtime-close")); + return; + } + await this.closeActive(ref); } private async bulkSessionLookupContext(refs: readonly SessionBulkMutationRef[]): Promise { @@ -1760,10 +1885,17 @@ export class PiSessionService implements SessionRouteService { return [...names]; } - private async closeActive(sessionId: string): Promise { + private async closeActive(sessionId: string, notificationPolicy: NotificationClosePolicy = CLEAR_RUNTIME_NOTIFICATIONS): Promise { const pendingOpens = this.pendingSessionOpenPromises(sessionId); if (pendingOpens.length > 0) await Promise.allSettled(pendingOpens); const active = this.active.get(sessionId); + if (notificationPolicy.kind === "clear") { + const generation = active === undefined ? undefined : this.notificationGenerationBySession.get(active.runtime.session); + const mutations = generation === undefined + ? this.notificationStore.clearSession(sessionId, notificationPolicy.reason) + : this.notificationStore.clearGeneration(generation, notificationPolicy.reason); + this.publishNotificationMutations(mutations); + } if (!active) return; this.active.delete(sessionId); this.activities.delete(sessionId); @@ -1776,6 +1908,7 @@ export class PiSessionService implements SessionRouteService { if (this.subsessionLinkForActiveChild(active.runtime.session) !== undefined) this.subsessionNotifyArmed.delete(sessionId); clearSessionQueue(active.runtime.session); active.unsubscribe(); + active.runtime.setRebindSession(undefined); try { await active.runtime.session.abort(); } finally { @@ -1791,7 +1924,7 @@ export class PiSessionService implements SessionRouteService { return (await this.getActive(ref)).runtime.session; } - private async getActive(ref: PiSessionLookup): Promise> { + private async getActive(ref: PiSessionLookup, options: Pick = {}): Promise> { const active = this.activeForLookup(ref); if (active !== undefined) return active; @@ -1802,6 +1935,7 @@ export class PiSessionService implements SessionRouteService { archived.sessionId, archived.cwd, () => this.sessionManager.open(archivePath), + { notifications: "disabled" }, ); } @@ -1809,13 +1943,14 @@ export class PiSessionService implements SessionRouteService { ? (await this.sessionManager.list(ref.cwd)).find((s) => s.id === ref.id || s.id.startsWith(ref.id)) : (await this.sessionManager.listAll?.() ?? []).find((s) => s.id === ref || s.id.startsWith(ref)); if (!match) throw new Error("Session not found"); - return this.openExistingSession(match.id, match.cwd, () => this.sessionManager.open(match.path)); + return this.openExistingSession(match.id, match.cwd, () => this.sessionManager.open(match.path), options); } private openExistingSession( sessionId: string, cwd: string, openSessionManager: () => PiSessionManager, + options: Pick = {}, ): Promise> { const active = this.activeForLookup({ id: sessionId, cwd }); if (active !== undefined) return Promise.resolve(active); @@ -1826,7 +1961,7 @@ export class PiSessionService implements SessionRouteService { const pending: PendingSessionOpen = { sessionId, - promise: this.create(openSessionManager(), cwd), + promise: this.create(openSessionManager(), cwd, options), }; pending.promise = pending.promise.finally(() => { if (this.pendingSessionOpens.get(key) === pending) this.pendingSessionOpens.delete(key); @@ -1861,7 +1996,7 @@ export class PiSessionService implements SessionRouteService { private async create( sessionManager: PiSessionManager, cwd: string, - options: Pick = {}, + options: CreateSessionRuntimeOptions = {}, ): Promise> { const delegationToolsEnabled = options.creationProvenance !== "tracked-subsession" && await sessionAllowsDelegationTools(sessionManager, this.sessionManager); @@ -1873,19 +2008,78 @@ export class PiSessionService implements SessionRouteService { ...(options.initialModel === undefined ? {} : { initialModel: options.initialModel }), }); const active: ActiveSession = { runtime, unsubscribe: noop }; + let notificationGeneration = options.notificationGeneration; + let notificationOwnership: "disabled" | "external" | "registered" | "replacement" = options.notifications === "disabled" + ? "disabled" + : notificationGeneration === undefined + ? "registered" + : "external"; + + if (notificationOwnership === "registered") { + const existingCandidate = this.notificationStore.beginReplacementForSession( + runtime.session.sessionId, + runtime.session.sessionManager.getCwd(), + ); + if (existingCandidate !== undefined) { + notificationGeneration = existingCandidate; + notificationOwnership = "replacement"; + } else { + const registration = this.notificationStore.registerSession( + runtime.session.sessionId, + runtime.session.sessionManager.getCwd(), + ); + notificationGeneration = registration.generation; + this.publishNotificationMutations(registration.mutations); + } + } + if (notificationGeneration !== undefined) this.notificationGenerationBySession.set(runtime.session, notificationGeneration); + try { - await this.bindSessionExtensions(runtime.session); + await this.bindSessionExtensions(runtime.session, notificationGeneration); this.bindRuntime(active); runtime.setRebindSession(async (session) => { - await this.bindSessionExtensions(session); - this.bindRuntime(active); - await this.recoverSubsessionTrackingForOpenedSession(session); + const priorGeneration = notificationGeneration; + let candidateGeneration: SessionNotificationGeneration | undefined; + try { + if (priorGeneration !== undefined) { + candidateGeneration = this.notificationStore.beginReplacement(priorGeneration, { + sessionId: session.sessionId, + cwd: session.sessionManager.getCwd(), + }); + this.notificationGenerationBySession.set(session, candidateGeneration); + } + this.bindRuntime(active, session); + await this.bindSessionExtensions(session, candidateGeneration); + await this.recoverSubsessionTrackingForOpenedSession(session); + if (candidateGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.commitReplacement(candidateGeneration)); + notificationGeneration = candidateGeneration; + } + } catch (error: unknown) { + if (candidateGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.abortReplacement(candidateGeneration, "candidate")); + notificationGeneration = candidateGeneration; + this.notificationGenerationBySession.set(session, candidateGeneration); + } + throw error; + } }); this.active.set(runtime.session.sessionId, active); await this.recoverSubsessionTrackingForOpenedSession(runtime.session); + if (notificationOwnership === "replacement" && notificationGeneration !== undefined) { + this.publishNotificationMutations(this.notificationStore.commitReplacement(notificationGeneration)); + notificationOwnership = "external"; + } this.publishStatus(runtime.session); return active; } catch (error: unknown) { + if (notificationGeneration !== undefined) { + if (notificationOwnership === "registered") { + this.publishNotificationMutations(this.notificationStore.clearSession(runtime.session.sessionId, "initialization-failed")); + } else if (notificationOwnership === "replacement") { + this.publishNotificationMutations(this.notificationStore.abortReplacement(notificationGeneration)); + } + } active.unsubscribe(); let removedActive = false; for (const [sessionId, candidate] of this.active.entries()) { @@ -1908,25 +2102,11 @@ export class PiSessionService implements SessionRouteService { } } - private async bindSessionExtensions(session: PiAgentSession): Promise { - const baseUiContext = session.extensionRunner.getUIContext(); - const notify: ExtensionUIContext["notify"] = (message, type) => { - this.events.publish(session.sessionId, { - type: "command.output", - level: type === "error" ? "error" : "info", - message, - }); - }; - // PI WEB is a remote UI host, but currently only extension notifications - // cross this boundary. Delegate every other UI method to Pi's headless - // defaults so unsupported dialogs cancel safely instead of hanging. - const uiContext = new Proxy(baseUiContext, { - get(target, property, receiver): unknown { - if (property === "notify") return notify; - const value: unknown = Reflect.get(target, property, receiver); - return value; - }, - }); + private async bindSessionExtensions( + session: PiAgentSession, + generation: SessionNotificationGeneration | undefined, + ): Promise { + const uiContext = this.sessionUiContext(session, generation); await session.bindExtensions({ uiContext, mode: "rpc", @@ -1938,9 +2118,55 @@ export class PiSessionService implements SessionRouteService { }); } - private bindRuntime(active: ActiveSession): void { + private replaceSessionNotificationContext(session: PiAgentSession, generation: SessionNotificationGeneration): void { + session.extensionRunner.setUIContext(this.sessionUiContext(session, generation), "rpc"); + } + + private sessionUiContext( + session: PiAgentSession, + generation: SessionNotificationGeneration | undefined, + ): ExtensionUIContext { + const baseUiContext = session.extensionRunner.getUIContext(); + const notify: ExtensionUIContext["notify"] = (message, type) => { + if (generation === undefined) { + this.events.publish(session.sessionId, { + type: "command.output", + level: type === "error" ? "error" : "info", + message, + }); + return; + } + const added = this.notificationStore.addNotification(generation, message, type); + this.publishNotificationMutations(added.mutations); + if (added.notification === undefined) return; + this.events.publish(session.sessionId, { + type: "command.output", + level: type === "error" ? "error" : "info", + message, + notificationId: added.notification.id, + }); + }; + // PI WEB is a remote UI host, but currently only extension notifications + // cross this boundary. Delegate every other UI method to Pi's headless + // defaults so unsupported dialogs cancel safely instead of hanging. + return new Proxy(baseUiContext, { + get(target, property, receiver): unknown { + if (property === "notify") return notify; + const value: unknown = Reflect.get(target, property, receiver); + return value; + }, + }); + } + + private publishNotificationMutations(mutations: readonly SessionNotificationMutation[]): void { + for (const mutation of mutations) { + this.events.publish(mutation.sessionId, mutation.inboxEvent); + this.events.publishNotificationSummary(mutation.summaryEvent); + } + } + + private bindRuntime(active: ActiveSession, session: PiAgentSession = active.runtime.session): void { active.unsubscribe(); - const { session } = active.runtime; for (const [sessionId, candidate] of this.active.entries()) { if (candidate === active) { this.active.delete(sessionId); diff --git a/src/server/sessions/sessionNotificationStore.test.ts b/src/server/sessions/sessionNotificationStore.test.ts new file mode 100644 index 0000000..8358299 --- /dev/null +++ b/src/server/sessions/sessionNotificationStore.test.ts @@ -0,0 +1,339 @@ +import { describe, expect, it } from "vitest"; +import { + SESSION_NOTIFICATION_LIMIT, + SESSION_NOTIFICATION_MESSAGE_BYTES, + SessionNotificationStore, + truncateSessionNotificationMessage, +} from "./sessionNotificationStore.js"; + +const identity = { sessionId: "session-1", cwd: "/workspace" }; + +function testStore() { + let tick = 0; + return new SessionNotificationStore({ + daemonInstanceId: "daemon-test", + now: () => new Date(Date.UTC(2026, 0, 1, 0, 0, tick++)), + }); +} + +function register(store: SessionNotificationStore) { + return store.registerSession(identity.sessionId, identity.cwd).generation; +} + +describe("SessionNotificationStore", () => { + it("keeps duplicate calls distinct, newest-first, and recomputes severity", () => { + const store = testStore(); + const generation = register(store); + + const first = store.addNotification(generation, "same", undefined).notification; + const second = store.addNotification(generation, "same", "warning").notification; + const third = store.addNotification(generation, "same", "error").notification; + const unknown = store.addNotification(generation, "unknown severity", "fatal").notification; + + expect([first?.id, second?.id, third?.id, unknown?.id]).toEqual([ + "daemon-test:1", + "daemon-test:2", + "daemon-test:3", + "daemon-test:4", + ]); + expect(store.inboxSnapshot(identity.sessionId, identity.cwd)).toMatchObject({ + summary: { retainedCount: 4, discardedCount: 0, highestSeverity: "error", inboxRevision: 4 }, + notifications: [ + { id: "daemon-test:4", severity: "info" }, + { id: "daemon-test:3", severity: "error" }, + { id: "daemon-test:2", severity: "warning" }, + { id: "daemon-test:1", severity: "info" }, + ], + dismissThrough: { order: 4, overflowWatermark: 0 }, + }); + + store.dismissNotification(identity.sessionId, identity.cwd, store.daemonInstanceId, third?.id ?? ""); + expect(store.inboxSnapshot(identity.sessionId, identity.cwd).summary.highestSeverity).toBe("warning"); + }); + + it("truncates UTF-8 only between code points and marks exact overflow", () => { + const exact = "a".repeat(SESSION_NOTIFICATION_MESSAGE_BYTES); + expect(truncateSessionNotificationMessage(exact)).toEqual({ message: exact, truncated: false }); + + const astral = `${"a".repeat(SESSION_NOTIFICATION_MESSAGE_BYTES - 1)}😀tail`; + const astralResult = truncateSessionNotificationMessage(astral); + expect(astralResult).toEqual({ message: "a".repeat(SESSION_NOTIFICATION_MESSAGE_BYTES - 1), truncated: true }); + expect(new TextEncoder().encode(astralResult.message).byteLength).toBe(SESSION_NOTIFICATION_MESSAGE_BYTES - 1); + + const bidi = `${"a".repeat(SESSION_NOTIFICATION_MESSAGE_BYTES - 3)}\u202etail`; + const bidiResult = truncateSessionNotificationMessage(bidi); + expect(bidiResult.message.endsWith("\u202e")).toBe(true); + expect(new TextEncoder().encode(bidiResult.message).byteLength).toBe(SESSION_NOTIFICATION_MESSAGE_BYTES); + expect(bidiResult.truncated).toBe(true); + }); + + it("retains exactly the newest 100 and reports exact overflow", () => { + const store = testStore(); + const generation = register(store); + + for (let index = 1; index <= 105; index += 1) store.addNotification(generation, `message ${String(index)}`, "info"); + + const snapshot = store.inboxSnapshot(identity.sessionId, identity.cwd); + expect(snapshot.notifications).toHaveLength(SESSION_NOTIFICATION_LIMIT); + expect(snapshot.notifications[0]).toMatchObject({ id: "daemon-test:105", message: "message 105" }); + expect(snapshot.notifications.at(-1)).toMatchObject({ id: "daemon-test:6", message: "message 6" }); + expect(snapshot.summary).toMatchObject({ retainedCount: 100, discardedCount: 5 }); + expect(snapshot.dismissThrough).toEqual({ order: 105, overflowWatermark: 5 }); + }); + + it("makes individual dismissal idempotent and clears overflow with the final retained entry", () => { + const store = testStore(); + const generation = register(store); + for (let index = 0; index <= SESSION_NOTIFICATION_LIMIT; index += 1) store.addNotification(generation, String(index), "info"); + + const firstSnapshot = store.inboxSnapshot(identity.sessionId, identity.cwd); + const firstId = firstSnapshot.notifications[0]?.id ?? ""; + const firstDismiss = store.dismissNotification(identity.sessionId, identity.cwd, store.daemonInstanceId, firstId); + const afterFirstRevision = firstDismiss.snapshot.summary.inboxRevision; + const duplicateDismiss = store.dismissNotification(identity.sessionId, identity.cwd, store.daemonInstanceId, firstId); + expect(duplicateDismiss.mutations).toEqual([]); + expect(duplicateDismiss.snapshot.summary.inboxRevision).toBe(afterFirstRevision); + expect(duplicateDismiss.snapshot.summary.discardedCount).toBe(1); + + for (const notification of duplicateDismiss.snapshot.notifications) { + store.dismissNotification(identity.sessionId, identity.cwd, store.daemonInstanceId, notification.id); + } + expect(store.inboxSnapshot(identity.sessionId, identity.cwd).summary).toMatchObject({ retainedCount: 0, discardedCount: 0 }); + expect(store.catalogSnapshot().sessions).toEqual([]); + }); + + it("dismisses only through captured order and overflow cutoffs", () => { + const store = testStore(); + const generation = register(store); + for (let index = 0; index <= SESSION_NOTIFICATION_LIMIT; index += 1) store.addNotification(generation, String(index), "info"); + const clicked = store.inboxSnapshot(identity.sessionId, identity.cwd); + + const later = store.addNotification(generation, "later", "error").notification; + const result = store.dismissAll( + identity.sessionId, + identity.cwd, + store.daemonInstanceId, + clicked.dismissThrough.order, + clicked.dismissThrough.overflowWatermark, + ); + + expect(result.snapshot.notifications).toEqual([later]); + expect(result.snapshot.summary).toMatchObject({ retainedCount: 1, discardedCount: 1, highestSeverity: "error" }); + const revision = result.snapshot.summary.inboxRevision; + const replay = store.dismissAll( + identity.sessionId, + identity.cwd, + store.daemonInstanceId, + clicked.dismissThrough.order, + clicked.dismissThrough.overflowWatermark, + ); + expect(replay.mutations).toEqual([]); + expect(replay.snapshot.summary.inboxRevision).toBe(revision); + }); + + it("requests resync when dismissal reveals an entry hidden by the replacement projection cap", () => { + const store = testStore(); + const oldGeneration = register(store); + for (let index = 1; index <= 100; index += 1) store.addNotification(oldGeneration, `old ${String(index)}`, "info"); + const candidate = store.beginReplacement(oldGeneration, identity); + const added = store.addNotification(candidate, "candidate", "warning"); + const candidateId = added.notification?.id ?? ""; + expect(added.mutations[0]?.inboxEvent.delta).toMatchObject({ kind: "added", evictedNotificationId: "daemon-test:1" }); + + const result = store.dismissNotification(identity.sessionId, identity.cwd, store.daemonInstanceId, candidateId); + + expect(result.mutations[0]?.inboxEvent.delta).toEqual({ kind: "resync" }); + expect(result.snapshot.notifications).toHaveLength(100); + expect(result.snapshot.notifications.at(-1)).toMatchObject({ id: "daemon-test:1", message: "old 1" }); + expect(result.snapshot.summary.discardedCount).toBe(0); + }); + + it("keeps dismiss-all deltas bounded to the public 100-entry projection", () => { + const store = testStore(); + const oldGeneration = register(store); + for (let index = 0; index < 100; index += 1) store.addNotification(oldGeneration, `old ${String(index)}`, "info"); + const candidate = store.beginReplacement(oldGeneration, identity); + for (let index = 0; index < 100; index += 1) store.addNotification(candidate, `candidate ${String(index)}`, "warning"); + const clicked = store.inboxSnapshot(identity.sessionId, identity.cwd); + + const result = store.dismissAll( + identity.sessionId, + identity.cwd, + store.daemonInstanceId, + clicked.dismissThrough.order, + clicked.dismissThrough.overflowWatermark, + ); + + const delta = result.mutations[0]?.inboxEvent.delta; + expect(delta?.kind).toBe("dismissed"); + expect(delta?.kind === "dismissed" ? delta.notificationIds : []).toHaveLength(100); + expect(result.snapshot.summary.retainedCount).toBe(0); + store.abortReplacement(candidate); + }); + + it("treats stale daemon and unknown notification identifiers as no-ops", () => { + const store = testStore(); + const generation = register(store); + store.addNotification(generation, "keep", "warning"); + const before = store.inboxSnapshot(identity.sessionId, identity.cwd); + + const stale = store.dismissAll(identity.sessionId, identity.cwd, "old-daemon", Number.MAX_SAFE_INTEGER, Number.MAX_SAFE_INTEGER); + const unknown = store.dismissNotification(identity.sessionId, identity.cwd, store.daemonInstanceId, "missing"); + + expect(stale.mutations).toEqual([]); + expect(unknown.mutations).toEqual([]); + expect(unknown.snapshot).toEqual(before); + }); + + it("advances catalog and inbox revisions only for visible mutations and emits zero cleanup", () => { + const store = testStore(); + const generation = register(store); + expect(store.catalogSnapshot()).toMatchObject({ daemonInstanceId: "daemon-test", catalogRevision: 0, sessions: [] }); + + const added = store.addNotification(generation, "notice", "info").mutations[0]; + expect(added).toMatchObject({ + sessionId: "session-1", + inboxEvent: { type: "notifications.inbox", catalogRevision: 1, summary: { inboxRevision: 1, retainedCount: 1 }, delta: { kind: "added" } }, + summaryEvent: { type: "notifications.summary", catalogRevision: 1, summary: { inboxRevision: 1, retainedCount: 1 } }, + }); + + const cleared = store.clearSession(identity.sessionId, "archive"); + expect(cleared).toHaveLength(1); + expect(cleared[0]).toMatchObject({ + inboxEvent: { catalogRevision: 2, summary: { inboxRevision: 2, retainedCount: 0, discardedCount: 0 }, delta: { kind: "cleared", reason: "archive" } }, + summaryEvent: { catalogRevision: 2, summary: { retainedCount: 0, discardedCount: 0 } }, + }); + expect(store.catalogSnapshot()).toMatchObject({ catalogRevision: 2, sessions: [] }); + expect(store.addNotification(generation, "stale", "error")).toEqual({ mutations: [] }); + }); + + it("keeps inbox revisions and overflow watermarks monotonic across same-daemon reopen", () => { + const store = testStore(); + const firstGeneration = register(store); + for (let index = 0; index <= SESSION_NOTIFICATION_LIMIT; index += 1) store.addNotification(firstGeneration, `first ${String(index)}`, "info"); + const oldSnapshot = store.inboxSnapshot(identity.sessionId, identity.cwd); + store.clearSession(identity.sessionId, "runtime-close"); + + const reopenedGeneration = register(store); + const reopenedEmpty = store.inboxSnapshot(identity.sessionId, identity.cwd); + expect(reopenedEmpty.summary.inboxRevision).toBeGreaterThan(oldSnapshot.summary.inboxRevision); + expect(reopenedEmpty.dismissThrough.overflowWatermark).toBe(oldSnapshot.dismissThrough.overflowWatermark); + for (let index = 0; index <= SESSION_NOTIFICATION_LIMIT; index += 1) store.addNotification(reopenedGeneration, `second ${String(index)}`, "warning"); + const beforeReplay = store.inboxSnapshot(identity.sessionId, identity.cwd); + + const replay = store.dismissAll( + identity.sessionId, + identity.cwd, + store.daemonInstanceId, + oldSnapshot.dismissThrough.order, + oldSnapshot.dismissThrough.overflowWatermark, + ); + + expect(replay.mutations).toEqual([]); + expect(replay.snapshot).toEqual(beforeReplay); + expect(replay.snapshot.summary.discardedCount).toBe(1); + expect(replay.snapshot.dismissThrough.overflowWatermark).toBeGreaterThan(oldSnapshot.dismissThrough.overflowWatermark); + }); + + it("commits replacement notifications while dropping the old generation and overflow", () => { + const store = testStore(); + const oldGeneration = register(store); + for (let index = 0; index <= SESSION_NOTIFICATION_LIMIT; index += 1) store.addNotification(oldGeneration, `old ${String(index)}`, "warning"); + + const candidate = store.beginReplacement(oldGeneration, identity); + const replacementOneResult = store.addNotification(candidate, "replacement one", "info"); + const replacementOne = replacementOneResult.notification; + const replacementTwo = store.addNotification(candidate, "replacement two", "error").notification; + expect(replacementOneResult.mutations[0]?.inboxEvent.delta).toMatchObject({ + kind: "added", + evictedNotificationId: "daemon-test:2", + }); + expect(store.addNotification(oldGeneration, "stale shutdown callback", "error")).toEqual({ mutations: [] }); + + const mutations = store.commitReplacement(candidate); + const snapshot = store.inboxSnapshot(identity.sessionId, identity.cwd); + expect(mutations.at(-1)?.inboxEvent.delta).toEqual({ kind: "resync" }); + expect(snapshot.notifications.map((notification) => notification.id)).toEqual([replacementTwo?.id, replacementOne?.id]); + expect(snapshot.summary).toMatchObject({ retainedCount: 2, discardedCount: 0, highestSeverity: "error" }); + expect(store.addNotification(oldGeneration, "stale", "info")).toEqual({ mutations: [] }); + }); + + it("aborts replacement without cleanup and keeps the 100-entry bound", () => { + const store = testStore(); + const oldGeneration = register(store); + for (let index = 0; index < 100; index += 1) store.addNotification(oldGeneration, `old ${String(index)}`, "info"); + const candidate = store.beginReplacement(oldGeneration, identity); + const replacementIds: string[] = []; + for (let index = 0; index < 100; index += 1) { + const notification = store.addNotification(candidate, `candidate ${String(index)}`, "warning").notification; + if (notification !== undefined) replacementIds.push(notification.id); + } + + const mutations = store.abortReplacement(candidate); + const snapshot = store.inboxSnapshot(identity.sessionId, identity.cwd); + expect(mutations.at(-1)?.inboxEvent.delta).toEqual({ kind: "resync" }); + expect(snapshot.notifications).toHaveLength(100); + expect(snapshot.notifications.every((notification) => replacementIds.includes(notification.id))).toBe(true); + expect(snapshot.summary.discardedCount).toBe(100); + expect(store.addNotification(candidate, "stale candidate", "error")).toEqual({ mutations: [] }); + + const afterAbort = store.addNotification(oldGeneration, "old runtime recovered", "error"); + expect(afterAbort.notification).toBeDefined(); + expect(store.inboxSnapshot(identity.sessionId, identity.cwd)).toMatchObject({ + summary: { retainedCount: 100, discardedCount: 101, highestSeverity: "error" }, + }); + }); + + it("can keep a rotated candidate binding active after aborting cleanup", () => { + const store = testStore(); + const oldGeneration = register(store); + store.addNotification(oldGeneration, "old", "info"); + const candidate = store.beginReplacement(oldGeneration, identity); + store.addNotification(candidate, "candidate", "warning"); + + store.abortReplacement(candidate, "candidate"); + + expect(store.addNotification(oldGeneration, "stale old runner", "error")).toEqual({ mutations: [] }); + expect(store.addNotification(candidate, "current runner", "error").notification).toMatchObject({ message: "current runner", severity: "error" }); + expect(store.inboxSnapshot(identity.sessionId, identity.cwd).notifications.map((notification) => notification.message)).toEqual([ + "current runner", + "candidate", + "old", + ]); + }); + + it("moves preserved calls to the active changed-id candidate after aborting cleanup", () => { + const store = testStore(); + const oldGeneration = register(store); + store.addNotification(oldGeneration, "old", "info"); + const candidate = store.beginReplacement(oldGeneration, { sessionId: "session-2", cwd: identity.cwd }); + store.addNotification(candidate, "candidate", "warning"); + + const mutations = store.abortReplacement(candidate, "candidate"); + + expect(mutations.map((mutation) => [mutation.sessionId, mutation.inboxEvent.delta.kind])).toEqual([ + ["session-1", "cleared"], + ["session-2", "resync"], + ]); + expect(() => store.inboxSnapshot("session-1", identity.cwd)).toThrow("Session not found"); + expect(store.inboxSnapshot("session-2", identity.cwd).notifications.map((notification) => notification.message)).toEqual(["candidate", "old"]); + expect(store.addNotification(candidate, "after failure", "error").notification).toMatchObject({ message: "after failure" }); + }); + + it("retags a failed changed-id replacement back to the prior inbox", () => { + const store = testStore(); + const oldGeneration = register(store); + store.addNotification(oldGeneration, "old", "info"); + const candidate = store.beginReplacement(oldGeneration, { sessionId: "session-2", cwd: identity.cwd }); + const replacement = store.addNotification(candidate, "replacement", "error").notification; + + const mutations = store.abortReplacement(candidate); + + expect(mutations.map((mutation) => [mutation.sessionId, mutation.inboxEvent.delta.kind])).toEqual([ + ["session-2", "cleared"], + ["session-1", "resync"], + ]); + expect(() => store.inboxSnapshot("session-2", identity.cwd)).toThrow("Session not found"); + expect(store.inboxSnapshot(identity.sessionId, identity.cwd).notifications.map((notification) => notification.id)).toContain(replacement?.id); + }); +}); diff --git a/src/server/sessions/sessionNotificationStore.ts b/src/server/sessions/sessionNotificationStore.ts new file mode 100644 index 0000000..4f48a9e --- /dev/null +++ b/src/server/sessions/sessionNotificationStore.ts @@ -0,0 +1,597 @@ +import { randomUUID } from "node:crypto"; +import { + SESSION_NOTIFICATION_LIMIT, + SESSION_NOTIFICATION_MESSAGE_BYTES, + type SessionNotification, + type SessionNotificationCatalogSnapshot, + type SessionNotificationClearReason, + type SessionNotificationInboxDelta, + type SessionNotificationInboxEvent, + type SessionNotificationInboxSnapshot, + type SessionNotificationSeverity, + type SessionNotificationSummary, + type SessionNotificationSummaryEvent, +} from "../../shared/apiTypes.js"; + +export { SESSION_NOTIFICATION_LIMIT, SESSION_NOTIFICATION_MESSAGE_BYTES } from "../../shared/apiTypes.js"; + +export type SessionNotificationGeneration = symbol; + +export interface SessionNotificationMutation { + sessionId: string; + inboxEvent: SessionNotificationInboxEvent; + summaryEvent: SessionNotificationSummaryEvent; +} + +export interface SessionNotificationRegistration { + generation: SessionNotificationGeneration; + mutations: SessionNotificationMutation[]; +} + +export interface SessionNotificationAddResult { + notification?: SessionNotification; + mutations: SessionNotificationMutation[]; +} + +export interface SessionNotificationSnapshotResult { + snapshot: SessionNotificationInboxSnapshot; + mutations: SessionNotificationMutation[]; +} + +interface NotificationBucket { + generation: SessionNotificationGeneration; + entries: SessionNotification[]; + discardedCount: number; +} + +interface NotificationProjection { + sessionId: string; + cwd: string; + inboxRevision: number; + overflowWatermark: number; + buckets: NotificationBucket[]; +} + +interface NotificationReplacement { + generation: SessionNotificationGeneration; + projection: NotificationProjection; + bucket: NotificationBucket; +} + +interface NotificationRuntimeState { + activeGeneration: SessionNotificationGeneration; + activeProjection: NotificationProjection; + activeBucket: NotificationBucket; + candidate?: NotificationReplacement; +} + +interface GenerationBinding { + state: NotificationRuntimeState; + role: "active" | "candidate"; +} + +export interface SessionNotificationStoreOptions { + daemonInstanceId?: string; + now?: () => Date; +} + +/** + * Daemon-owned, bounded, in-memory notification state. + * + * The store deliberately knows nothing about Fastify, Pi session persistence, + * sockets, or browser state. Runtime generations are opaque capabilities: once + * a generation is replaced or cleared, stale extension callbacks become no-ops. + */ +export class SessionNotificationStore { + readonly daemonInstanceId: string; + private readonly now: () => Date; + private readonly statesBySessionId = new Map(); + private readonly bindings = new Map(); + private readonly lastInboxRevisionBySessionId = new Map(); + private readonly lastOverflowWatermarkBySessionId = new Map(); + private catalogRevision = 0; + private nextOrder = 0; + + constructor(options: SessionNotificationStoreOptions = {}) { + this.daemonInstanceId = options.daemonInstanceId ?? randomUUID(); + this.now = options.now ?? (() => new Date()); + } + + registerSession(sessionId: string, cwd: string): SessionNotificationRegistration { + requireIdentity(sessionId, cwd); + const mutations = this.clearSession(sessionId, "replacement"); + const generation = Symbol(`notifications:${sessionId}`); + const bucket = emptyBucket(generation); + const projection: NotificationProjection = { + sessionId, + cwd, + inboxRevision: this.lastInboxRevisionBySessionId.get(sessionId) ?? 0, + overflowWatermark: this.lastOverflowWatermarkBySessionId.get(sessionId) ?? 0, + buckets: [bucket], + }; + const state: NotificationRuntimeState = { + activeGeneration: generation, + activeProjection: projection, + activeBucket: bucket, + }; + this.statesBySessionId.set(sessionId, state); + this.bindings.set(generation, { state, role: "active" }); + return { generation, mutations }; + } + + currentGeneration(sessionId: string, cwd: string): SessionNotificationGeneration | undefined { + const state = this.statesBySessionId.get(sessionId); + if (state?.activeProjection.sessionId !== sessionId || state.activeProjection.cwd !== cwd || state.candidate !== undefined) return undefined; + return state.activeGeneration; + } + + beginReplacement( + activeGeneration: SessionNotificationGeneration, + target: { sessionId: string; cwd: string }, + ): SessionNotificationGeneration { + requireIdentity(target.sessionId, target.cwd); + const binding = this.bindings.get(activeGeneration); + if (binding?.role !== "active") throw new Error("Notification runtime generation is no longer active"); + const state = binding.state; + if (state.candidate !== undefined) throw new Error("Notification runtime replacement is already in progress"); + + const sameIdentity = state.activeProjection.sessionId === target.sessionId && state.activeProjection.cwd === target.cwd; + let projection: NotificationProjection; + if (sameIdentity) { + projection = state.activeProjection; + } else { + const existing = this.statesBySessionId.get(target.sessionId); + if (existing !== undefined && existing !== state) throw new Error("Notification target session is already registered"); + projection = { + sessionId: target.sessionId, + cwd: target.cwd, + inboxRevision: this.lastInboxRevisionBySessionId.get(target.sessionId) ?? 0, + overflowWatermark: this.lastOverflowWatermarkBySessionId.get(target.sessionId) ?? 0, + buckets: [], + }; + this.statesBySessionId.set(target.sessionId, state); + } + + const generation = Symbol(`notifications:${target.sessionId}:candidate`); + const bucket = emptyBucket(generation); + projection.buckets.push(bucket); + state.candidate = { generation, projection, bucket }; + this.bindings.set(generation, { state, role: "candidate" }); + return generation; + } + + beginReplacementForSession(sessionId: string, cwd: string): SessionNotificationGeneration | undefined { + const generation = this.currentGeneration(sessionId, cwd); + return generation === undefined ? undefined : this.beginReplacement(generation, { sessionId, cwd }); + } + + commitReplacement(candidateGeneration: SessionNotificationGeneration): SessionNotificationMutation[] { + const binding = this.requireCandidate(candidateGeneration); + const state = binding.state; + const candidate = state.candidate; + if (candidate === undefined) return []; + + const oldProjection = state.activeProjection; + const sameProjection = oldProjection === candidate.projection; + const before = sameProjection ? projectionFingerprint(oldProjection) : undefined; + const mutations: SessionNotificationMutation[] = []; + + this.bindings.delete(state.activeGeneration); + if (sameProjection) { + oldProjection.buckets = [candidate.bucket]; + } else { + mutations.push(...this.clearProjection(oldProjection, "replacement")); + this.statesBySessionId.delete(oldProjection.sessionId); + } + + state.activeGeneration = candidate.generation; + state.activeProjection = candidate.projection; + state.activeBucket = candidate.bucket; + delete state.candidate; + this.bindings.set(candidate.generation, { state, role: "active" }); + this.statesBySessionId.set(candidate.projection.sessionId, state); + + if (sameProjection && before !== projectionFingerprint(candidate.projection)) { + mutations.push(this.mutation(candidate.projection, { kind: "resync" })); + } + return mutations; + } + + abortReplacement( + candidateGeneration: SessionNotificationGeneration, + survivingGeneration: "prior" | "candidate" = "prior", + ): SessionNotificationMutation[] { + const binding = this.requireCandidate(candidateGeneration); + const state = binding.state; + const candidate = state.candidate; + if (candidate === undefined) return []; + + const oldProjection = state.activeProjection; + const oldGeneration = state.activeGeneration; + const oldBucket = state.activeBucket; + const candidateProjection = candidate.projection; + const sameProjection = oldProjection === candidateProjection; + const targetProjection = sameProjection || survivingGeneration === "prior" ? oldProjection : candidateProjection; + const before = projectionFingerprint(targetProjection); + const oldEntries = [...oldBucket.entries]; + const candidateEntries = [...candidate.bucket.entries]; + const oldDiscardedCount = oldBucket.discardedCount; + const candidateDiscardedCount = candidate.bucket.discardedCount; + const mutations: SessionNotificationMutation[] = []; + + if (!sameProjection) { + const sourceProjection = survivingGeneration === "candidate" ? oldProjection : candidateProjection; + mutations.push(...this.clearProjection(sourceProjection, "replacement")); + this.statesBySessionId.delete(sourceProjection.sessionId); + const transferredDiscardedCount = survivingGeneration === "candidate" ? oldDiscardedCount : candidateDiscardedCount; + targetProjection.overflowWatermark = addSafe( + targetProjection.overflowWatermark, + transferredDiscardedCount, + "Notification overflow watermark exhausted", + ); + } + + const merged = [...oldEntries, ...candidateEntries].sort((left, right) => left.order - right.order); + const overflow = Math.max(0, merged.length - SESSION_NOTIFICATION_LIMIT); + const targetGeneration = survivingGeneration === "candidate" ? candidate.generation : oldGeneration; + const targetBucket = survivingGeneration === "candidate" ? candidate.bucket : oldBucket; + targetBucket.entries = overflow === 0 ? merged : merged.slice(overflow); + targetBucket.discardedCount = addSafe(oldDiscardedCount, candidateDiscardedCount, "Notification discarded count exhausted"); + addDiscardedCount(overflow, targetProjection, targetBucket); + targetProjection.buckets = [targetBucket]; + + this.bindings.delete(survivingGeneration === "candidate" ? oldGeneration : candidate.generation); + state.activeGeneration = targetGeneration; + state.activeProjection = targetProjection; + state.activeBucket = targetBucket; + delete state.candidate; + this.statesBySessionId.set(targetProjection.sessionId, state); + this.bindings.set(targetGeneration, { state, role: "active" }); + if (before !== projectionFingerprint(targetProjection)) { + mutations.push(this.mutation(targetProjection, { kind: "resync" })); + } + return mutations; + } + + addNotification( + generation: SessionNotificationGeneration, + message: string, + severity: unknown, + ): SessionNotificationAddResult { + const binding = this.bindings.get(generation); + if (binding === undefined) return { mutations: [] }; + const { state } = binding; + // Once the replacement runner is bound, old callbacks are stale. Suppressing + // them also keeps generation overflow ordered as a bounded suffix. + if (binding.role === "active" && state.candidate !== undefined) return { mutations: [] }; + const projection = binding.role === "candidate" ? state.candidate?.projection : state.activeProjection; + const bucket = binding.role === "candidate" ? state.candidate?.bucket : state.activeBucket; + if (projection === undefined || bucket === undefined) return { mutations: [] }; + + const previouslyRetained = retainedEntries(projection); + const order = incrementSafe(this.nextOrder, "Notification order exhausted"); + this.nextOrder = order; + const truncatedMessage = truncateSessionNotificationMessage(message); + const notification: SessionNotification = Object.freeze({ + id: `${this.daemonInstanceId}:${String(order)}`, + message: truncatedMessage.message, + truncated: truncatedMessage.truncated, + severity: normalizeSeverity(severity), + receivedAt: this.now().toISOString(), + order, + }); + bucket.entries.push(notification); + const bucketEviction = bucket.entries.length > SESSION_NOTIFICATION_LIMIT ? bucket.entries.shift() : undefined; + if (bucketEviction !== undefined) addDiscardedCount(1, projection, bucket); + const retainedIds = new Set(retainedEntries(projection).map((entry) => entry.id)); + const projectionEviction = previouslyRetained.find((entry) => !retainedIds.has(entry.id)); + const delta: SessionNotificationInboxDelta = { + kind: "added", + notification, + ...(projectionEviction === undefined ? {} : { evictedNotificationId: projectionEviction.id }), + }; + return { notification, mutations: [this.mutation(projection, delta)] }; + } + + catalogSnapshot(): SessionNotificationCatalogSnapshot { + const sessions = uniqueProjections(this.statesBySessionId.values()) + .map((projection) => this.summary(projection)) + .filter((summary) => summary.retainedCount > 0 || summary.discardedCount > 0); + return { + daemonInstanceId: this.daemonInstanceId, + catalogRevision: this.catalogRevision, + sessions, + }; + } + + inboxSnapshot(sessionId: string, cwd: string): SessionNotificationInboxSnapshot { + return this.snapshot(this.requireProjection(sessionId, cwd)); + } + + dismissNotification( + sessionId: string, + cwd: string, + daemonInstanceId: string, + notificationId: string, + ): SessionNotificationSnapshotResult { + const projection = this.requireProjection(sessionId, cwd); + if (daemonInstanceId !== this.daemonInstanceId) return { snapshot: this.snapshot(projection), mutations: [] }; + + const previouslyRetained = retainedEntries(projection); + const previouslyRetainedIds = new Set(previouslyRetained.map((entry) => entry.id)); + const before = projectionFingerprint(projection); + let dismissed = false; + for (const bucket of projection.buckets) { + const index = bucket.entries.findIndex((entry) => entry.id === notificationId); + if (index === -1) continue; + bucket.entries.splice(index, 1); + dismissed = true; + break; + } + if (!dismissed) return { snapshot: this.snapshot(projection), mutations: [] }; + + if (projection.buckets.every((bucket) => bucket.entries.length === 0)) clearDiscardedCount(projection); + const newlyRevealed = retainedEntries(projection).some((entry) => !previouslyRetainedIds.has(entry.id)); + const after = projectionFingerprint(projection); + if (!previouslyRetainedIds.has(notificationId) && before === after) { + return { snapshot: this.snapshot(projection), mutations: [] }; + } + const delta: SessionNotificationInboxDelta = newlyRevealed + ? { kind: "resync" } + : { kind: "dismissed", notificationIds: [notificationId] }; + const mutation = this.mutation(projection, delta); + return { snapshot: this.snapshot(projection), mutations: [mutation] }; + } + + dismissAll( + sessionId: string, + cwd: string, + daemonInstanceId: string, + throughOrder: number, + throughOverflowWatermark: number, + ): SessionNotificationSnapshotResult { + const projection = this.requireProjection(sessionId, cwd); + if (daemonInstanceId !== this.daemonInstanceId) return { snapshot: this.snapshot(projection), mutations: [] }; + + const visibleIds = new Set(retainedEntries(projection).map((entry) => entry.id)); + const dismissedIds: string[] = []; + for (const bucket of projection.buckets) { + bucket.entries = bucket.entries.filter((entry) => { + if (entry.order > throughOrder) return true; + if (visibleIds.has(entry.id)) dismissedIds.push(entry.id); + return false; + }); + } + const acknowledgedOverflow = acknowledgeDiscardedThrough(projection, throughOverflowWatermark); + if (dismissedIds.length === 0 && acknowledgedOverflow === 0) return { snapshot: this.snapshot(projection), mutations: [] }; + + const mutation = this.mutation(projection, { kind: "dismissed", notificationIds: dismissedIds }); + return { snapshot: this.snapshot(projection), mutations: [mutation] }; + } + + clearGeneration(generation: SessionNotificationGeneration, reason: SessionNotificationClearReason): SessionNotificationMutation[] { + const binding = this.bindings.get(generation); + return binding === undefined ? [] : this.clearSession(binding.state.activeProjection.sessionId, reason); + } + + clearSessionIdentity(sessionId: string, cwd: string, reason: SessionNotificationClearReason): SessionNotificationMutation[] { + const state = this.statesBySessionId.get(sessionId); + if (state === undefined) return []; + const projection = state.activeProjection.sessionId === sessionId ? state.activeProjection : state.candidate?.projection; + if (projection?.sessionId !== sessionId) return []; + if (projection.cwd !== cwd) throw new Error("Session cwd mismatch"); + return this.clearSession(sessionId, reason); + } + + clearSession(sessionId: string, reason: SessionNotificationClearReason): SessionNotificationMutation[] { + const state = this.statesBySessionId.get(sessionId); + if (state === undefined) return []; + const projections = state.candidate === undefined + ? [state.activeProjection] + : uniqueProjectionList([state.activeProjection, state.candidate.projection]); + const mutations = projections.flatMap((projection) => this.clearProjection(projection, reason)); + this.statesBySessionId.delete(state.activeProjection.sessionId); + if (state.candidate !== undefined) { + this.statesBySessionId.delete(state.candidate.projection.sessionId); + this.bindings.delete(state.candidate.generation); + } + this.bindings.delete(state.activeGeneration); + return mutations; + } + + clearAll(reason: SessionNotificationClearReason = "service-dispose"): SessionNotificationMutation[] { + const states = new Set(this.statesBySessionId.values()); + const mutations: SessionNotificationMutation[] = []; + for (const state of states) mutations.push(...this.clearSession(state.activeProjection.sessionId, reason)); + this.statesBySessionId.clear(); + this.bindings.clear(); + this.lastInboxRevisionBySessionId.clear(); + this.lastOverflowWatermarkBySessionId.clear(); + return mutations; + } + + private requireCandidate(candidateGeneration: SessionNotificationGeneration): GenerationBinding { + const binding = this.bindings.get(candidateGeneration); + if (binding?.role !== "candidate") throw new Error("Notification replacement generation is no longer active"); + return binding; + } + + private requireProjection(sessionId: string, cwd: string): NotificationProjection { + const state = this.statesBySessionId.get(sessionId); + if (state === undefined) throw new Error("Session not found"); + const projection = state.activeProjection.sessionId === sessionId + ? state.activeProjection + : state.candidate?.projection.sessionId === sessionId + ? state.candidate.projection + : undefined; + if (projection === undefined) throw new Error("Session not found"); + if (projection.cwd !== cwd) throw new Error("Session cwd mismatch"); + return projection; + } + + private clearProjection(projection: NotificationProjection, reason: SessionNotificationClearReason): SessionNotificationMutation[] { + const wasVisible = retainedEntries(projection).length > 0 || discardedCount(projection) > 0; + for (const bucket of projection.buckets) { + bucket.entries = []; + bucket.discardedCount = 0; + } + return wasVisible ? [this.mutation(projection, { kind: "cleared", reason })] : []; + } + + private mutation(projection: NotificationProjection, delta: SessionNotificationInboxDelta): SessionNotificationMutation { + projection.inboxRevision = incrementSafe(projection.inboxRevision, "Notification inbox revision exhausted"); + this.lastInboxRevisionBySessionId.set(projection.sessionId, projection.inboxRevision); + this.lastOverflowWatermarkBySessionId.set(projection.sessionId, projection.overflowWatermark); + this.catalogRevision = incrementSafe(this.catalogRevision, "Notification catalog revision exhausted"); + const summary = this.summary(projection); + const common = { + daemonInstanceId: this.daemonInstanceId, + catalogRevision: this.catalogRevision, + summary, + }; + return { + sessionId: projection.sessionId, + inboxEvent: { type: "notifications.inbox", ...common, dismissThrough: dismissThrough(projection), delta }, + summaryEvent: { type: "notifications.summary", ...common }, + }; + } + + private summary(projection: NotificationProjection): SessionNotificationSummary { + const notifications = retainedEntries(projection); + const highestSeverity = highestSeverityOf(notifications); + return { + sessionId: projection.sessionId, + cwd: projection.cwd, + inboxRevision: projection.inboxRevision, + retainedCount: notifications.length, + discardedCount: discardedCount(projection), + ...(highestSeverity === undefined ? {} : { highestSeverity }), + }; + } + + private snapshot(projection: NotificationProjection): SessionNotificationInboxSnapshot { + const notifications = retainedEntries(projection).reverse(); + return { + daemonInstanceId: this.daemonInstanceId, + catalogRevision: this.catalogRevision, + summary: this.summary(projection), + notifications, + dismissThrough: dismissThrough(projection), + }; + } +} + +export function truncateSessionNotificationMessage( + message: string, + maxBytes = SESSION_NOTIFICATION_MESSAGE_BYTES, +): { message: string; truncated: boolean } { + if (!Number.isSafeInteger(maxBytes) || maxBytes < 0) throw new Error("maxBytes must be a non-negative safe integer"); + const encoder = new TextEncoder(); + if (encoder.encode(message).byteLength <= maxBytes) return { message, truncated: false }; + let bytes = 0; + let truncated = ""; + for (const codePoint of message) { + const codePointBytes = encoder.encode(codePoint).byteLength; + if (bytes + codePointBytes > maxBytes) break; + truncated += codePoint; + bytes += codePointBytes; + } + return { message: truncated, truncated: true }; +} + +function emptyBucket(generation: SessionNotificationGeneration): NotificationBucket { + return { generation, entries: [], discardedCount: 0 }; +} + +function normalizeSeverity(value: unknown): SessionNotificationSeverity { + return value === "warning" || value === "error" ? value : "info"; +} + +function retainedEntries(projection: NotificationProjection): SessionNotification[] { + return projection.buckets + .flatMap((bucket) => bucket.entries) + .sort((left, right) => left.order - right.order) + .slice(-SESSION_NOTIFICATION_LIMIT); +} + +function discardedCount(projection: NotificationProjection): number { + return projection.buckets.reduce((total, bucket) => total + bucket.discardedCount, 0); +} + +function highestSeverityOf(notifications: readonly SessionNotification[]): SessionNotificationSeverity | undefined { + let highest: SessionNotificationSeverity | undefined; + for (const notification of notifications) { + if (notification.severity === "error") return "error"; + if (notification.severity === "warning") highest = "warning"; + else highest ??= "info"; + } + return highest; +} + +function addDiscardedCount(count: number, projection: NotificationProjection, bucket: NotificationBucket): void { + if (count === 0) return; + projection.overflowWatermark = addSafe(projection.overflowWatermark, count, "Notification overflow watermark exhausted"); + bucket.discardedCount = addSafe(bucket.discardedCount, count, "Notification discarded count exhausted"); +} + +function clearDiscardedCount(projection: NotificationProjection): void { + for (const bucket of projection.buckets) bucket.discardedCount = 0; +} + +function acknowledgeDiscardedThrough(projection: NotificationProjection, throughWatermark: number): number { + const count = discardedCount(projection); + if (count === 0) return 0; + const firstWatermark = projection.overflowWatermark - count + 1; + const acknowledged = Math.max(0, Math.min(count, throughWatermark - firstWatermark + 1)); + let remaining = acknowledged; + for (const bucket of projection.buckets) { + if (remaining === 0) break; + const removed = Math.min(bucket.discardedCount, remaining); + bucket.discardedCount -= removed; + remaining -= removed; + } + return acknowledged; +} + +function dismissThrough(projection: NotificationProjection): { order: number; overflowWatermark: number } { + const entries = retainedEntries(projection); + return { + order: entries.at(-1)?.order ?? 0, + overflowWatermark: projection.overflowWatermark, + }; +} + +function projectionFingerprint(projection: NotificationProjection): string { + return JSON.stringify({ + ids: retainedEntries(projection).map((entry) => entry.id), + discardedCount: discardedCount(projection), + }); +} + +function uniqueProjections(states: Iterable): NotificationProjection[] { + const projections: NotificationProjection[] = []; + for (const state of new Set(states)) { + projections.push(state.activeProjection); + if (state.candidate !== undefined && state.candidate.projection !== state.activeProjection) projections.push(state.candidate.projection); + } + return projections; +} + +function uniqueProjectionList(projections: NotificationProjection[]): NotificationProjection[] { + return [...new Set(projections)]; +} + +function requireIdentity(sessionId: string, cwd: string): void { + if (sessionId === "") throw new Error("sessionId must not be empty"); + if (cwd === "") throw new Error("cwd must not be empty"); +} + +function incrementSafe(value: number, message: string): number { + return addSafe(value, 1, message); +} + +function addSafe(value: number, increment: number, message: string): number { + const next = value + increment; + if (!Number.isSafeInteger(next)) throw new Error(message); + return next; +} diff --git a/src/server/sessions/sessionRoutes.test.ts b/src/server/sessions/sessionRoutes.test.ts index 2507177..4296049 100644 --- a/src/server/sessions/sessionRoutes.test.ts +++ b/src/server/sessions/sessionRoutes.test.ts @@ -2,10 +2,24 @@ import { resolve } from "node:path"; import Fastify, { type FastifyInstance } from "fastify"; import fastifyWebsocket from "@fastify/websocket"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import type { MessagePage, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkMutationRef, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse, SessionStatus, SessionStreamSnapshot } from "../../shared/apiTypes.js"; +import type { + MessagePage, + SessionBulkArchiveResponse, + SessionBulkDeleteArchivedResponse, + SessionBulkMutationRef, + SessionCleanupExecuteResponse, + SessionCleanupPreviewResponse, + SessionNotificationDismissAllRequest, + SessionNotificationDismissRequest, + SessionNotificationInboxSnapshot, + SessionRef, + SessionStatus, + SessionStreamSnapshot, +} from "../../shared/apiTypes.js"; import { SessionEventHub } from "../realtime/sessionEventHub.js"; import { PiSessionService, type PiSessionManagerGateway } from "./piSessionService.js"; import { testModelRuntime } from "./piSessionService.testSupport.js"; +import { SessionNotificationStore } from "./sessionNotificationStore.js"; import type { SessionRouteLookup, SessionRouteService } from "./sessionService.js"; import { registerSessionRoutes } from "./sessionRoutes.js"; import type { NormalizedSessionCleanupRequest } from "./sessionCleanup.js"; @@ -31,6 +45,130 @@ afterEach(async () => { }); describe("session routes", () => { + it("returns notification catalog and selected-inbox snapshots with required cwd context", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const requestCwd = resolve("/repo"); + const catalog = await routeApp.inject({ method: "GET", url: "/sessions/notifications" }); + const inbox = await routeApp.inject({ method: "GET", url: `/sessions/session-1/notifications?cwd=${encodeURIComponent(requestCwd)}` }); + + expect(catalog.statusCode).toBe(200); + expect(catalog.json()).toEqual({ daemonInstanceId: "daemon-test", catalogRevision: 0, sessions: [] }); + expect(inbox.statusCode).toBe(200); + expect(inbox.json()).toMatchObject({ daemonInstanceId: "daemon-test", summary: { sessionId: "session-1", cwd: requestCwd } }); + expect(routeService.notificationInboxCalls).toEqual([{ id: "session-1", cwd: requestCwd }]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + + it("validates and forwards idempotent notification dismissal cutoffs", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const requestCwd = resolve("/repo"); + const dismiss = await routeApp.inject({ + method: "POST", + url: "/sessions/session-1/notifications/dismiss", + payload: { cwd: requestCwd, daemonInstanceId: "daemon-test", notificationId: "notice-1" }, + }); + const dismissAll = await routeApp.inject({ + method: "POST", + url: "/sessions/session-1/notifications/dismiss-all", + payload: { cwd: requestCwd, daemonInstanceId: "daemon-test", throughOrder: 12, throughOverflowWatermark: 3 }, + }); + + expect(dismiss.statusCode).toBe(200); + expect(dismissAll.statusCode).toBe(200); + expect(routeService.dismissNotificationCalls).toEqual([{ + ref: { id: "session-1", cwd: requestCwd }, + request: { daemonInstanceId: "daemon-test", notificationId: "notice-1" }, + }]); + expect(routeService.dismissAllNotificationCalls).toEqual([{ + ref: { id: "session-1", cwd: requestCwd }, + request: { daemonInstanceId: "daemon-test", throughOrder: 12, throughOverflowWatermark: 3 }, + }]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + + it("keeps stale notification mutations harmless and rejects mismatched ownership", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const requestCwd = resolve("/repo"); + const notificationStore = new SessionNotificationStore({ daemonInstanceId: "daemon-current" }); + const registration = notificationStore.registerSession("session-1", requestCwd); + notificationStore.addNotification(registration.generation, "keep", "warning"); + const routeService = new PiSessionService(eventHub, { + agentDir: TEST_AGENT_DIR, + modelRuntime: testModelRuntime, + notificationStore, + sessionManager: new RejectingSessionManager(), + heartbeatIntervalMs: 60_000, + }); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const stale = await routeApp.inject({ + method: "POST", + url: "/sessions/session-1/notifications/dismiss-all", + payload: { cwd: requestCwd, daemonInstanceId: "daemon-old", throughOrder: Number.MAX_SAFE_INTEGER, throughOverflowWatermark: Number.MAX_SAFE_INTEGER }, + }); + const mismatch = await routeApp.inject({ method: "GET", url: `/sessions/session-1/notifications?cwd=${encodeURIComponent(resolve("/other"))}` }); + const missing = await routeApp.inject({ method: "GET", url: `/sessions/missing/notifications?cwd=${encodeURIComponent(requestCwd)}` }); + + expect(stale.statusCode).toBe(200); + expect(stale.json()).toMatchObject({ summary: { retainedCount: 1, inboxRevision: 1 } }); + expect(mismatch.statusCode).toBe(400); + expect(mismatch.json()).toEqual({ error: "Session cwd mismatch" }); + expect(missing.statusCode).toBe(404); + expect(missing.json()).toEqual({ error: "Session not found" }); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + + it("rejects malformed notification requests before calling the service", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const missingCwd = await routeApp.inject({ method: "GET", url: "/sessions/session-1/notifications" }); + const unsafeCutoff = await routeApp.inject({ + method: "POST", + url: "/sessions/session-1/notifications/dismiss-all", + payload: { cwd: "/repo", daemonInstanceId: "daemon-test", throughOrder: Number.MAX_SAFE_INTEGER + 1, throughOverflowWatermark: 0 }, + }); + + expect(missingCwd.statusCode).toBe(400); + expect(missingCwd.json()).toEqual({ error: "cwd field must be a string" }); + expect(unsafeCutoff.statusCode).toBe(400); + expect(unsafeCutoff.json()).toEqual({ error: "throughOrder field must be a non-negative safe integer" }); + expect(routeService.notificationInboxCalls).toEqual([]); + expect(routeService.dismissAllNotificationCalls).toEqual([]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + 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: { body: "Build the thing" } }); @@ -385,6 +523,9 @@ class CapturingRouteSessionService implements SessionRouteService { readonly reloadCalls: SessionRouteLookup[] = []; readonly clearQueueCalls: SessionRouteLookup[] = []; readonly dismissWarningCalls: { lookup: SessionRouteLookup; dismissId: string }[] = []; + readonly notificationInboxCalls: SessionRef[] = []; + readonly dismissNotificationCalls: { ref: SessionRef; request: Omit }[] = []; + readonly dismissAllNotificationCalls: { ref: SessionRef; request: Omit }[] = []; dismissWarningError: Error | undefined; messagesResponse: unknown[] | MessagePage = []; streamSnapshotResponse: SessionStreamSnapshot = { seq: 0, partial: null }; @@ -426,6 +567,25 @@ class CapturingRouteSessionService implements SessionRouteService { return Promise.resolve(); } + notificationCatalog() { + return { daemonInstanceId: "daemon-test", catalogRevision: 0, sessions: [] }; + } + + notificationInbox(ref: SessionRef): SessionNotificationInboxSnapshot { + this.notificationInboxCalls.push(ref); + return notificationSnapshot(ref); + } + + dismissNotification(ref: SessionRef, request: Omit): SessionNotificationInboxSnapshot { + this.dismissNotificationCalls.push({ ref, request }); + return notificationSnapshot(ref); + } + + dismissAllNotifications(ref: SessionRef, request: Omit): SessionNotificationInboxSnapshot { + this.dismissAllNotificationCalls.push({ ref, request }); + return notificationSnapshot(ref); + } + list(): never { throw unusedRouteMethod("list"); } start(): never { throw unusedRouteMethod("start"); } @@ -542,6 +702,16 @@ class RejectingSessionManager implements PiSessionManagerGateway { } } +function notificationSnapshot(ref: SessionRef): SessionNotificationInboxSnapshot { + return { + daemonInstanceId: "daemon-test", + catalogRevision: 0, + summary: { sessionId: ref.id, cwd: ref.cwd, inboxRevision: 0, retainedCount: 0, discardedCount: 0 }, + notifications: [], + dismissThrough: { order: 0, overflowWatermark: 0 }, + }; +} + function sessionIdFromLookup(lookup: SessionRouteLookup): string { return typeof lookup === "string" ? lookup : lookup.id; } diff --git a/src/server/sessions/sessionRoutes.ts b/src/server/sessions/sessionRoutes.ts index 074ef18..4905e26 100644 --- a/src/server/sessions/sessionRoutes.ts +++ b/src/server/sessions/sessionRoutes.ts @@ -30,6 +30,11 @@ interface AttachmentsRequestBody { folder?: unknown; } +const MAX_NOTIFICATION_SESSION_ID_LENGTH = 512; +const MAX_NOTIFICATION_CWD_LENGTH = 32 * 1024; +const MAX_NOTIFICATION_DAEMON_ID_LENGTH = 512; +const MAX_NOTIFICATION_ID_LENGTH = 1024; + export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRouteService, 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" }); @@ -49,6 +54,14 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRou } }); + app.get(`${prefix}/sessions/notifications`, async (_request, reply) => { + try { + return await sessions.notificationCatalog(); + } catch (error) { + return reply.code(400).send({ error: errorMessage(error) }); + } + }); + app.post<{ Body: SessionCleanupRequest | undefined }>(`${prefix}/sessions/cleanup/preview`, async (request, reply) => { try { return await sessions.cleanupPreview(normalizeSessionCleanupRequest(optionalRecord(request.body))); @@ -81,6 +94,41 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRou } }); + app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/notifications`, async (request, reply) => { + try { + return await sessions.notificationInbox(notificationRefFromQuery(request.params.sessionId, request.query)); + } catch (error) { + return reply.code(notificationErrorStatus(error)).send({ error: errorMessage(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: Record | undefined }>(`${prefix}/sessions/:sessionId/notifications/dismiss`, async (request, reply) => { + try { + const body = requireRecord(request.body); + const ref = notificationRefFromBody(request.params.sessionId, body); + return await sessions.dismissNotification(ref, { + daemonInstanceId: requireNonEmptyBoundedString(body["daemonInstanceId"], "daemonInstanceId", MAX_NOTIFICATION_DAEMON_ID_LENGTH), + notificationId: requireNonEmptyBoundedString(body["notificationId"], "notificationId", MAX_NOTIFICATION_ID_LENGTH), + }); + } catch (error) { + return reply.code(notificationErrorStatus(error)).send({ error: errorMessage(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: Record | undefined }>(`${prefix}/sessions/:sessionId/notifications/dismiss-all`, async (request, reply) => { + try { + const body = requireRecord(request.body); + const ref = notificationRefFromBody(request.params.sessionId, body); + return await sessions.dismissAllNotifications(ref, { + daemonInstanceId: requireNonEmptyBoundedString(body["daemonInstanceId"], "daemonInstanceId", MAX_NOTIFICATION_DAEMON_ID_LENGTH), + throughOrder: requireNonNegativeSafeInteger(body["throughOrder"], "throughOrder"), + throughOverflowWatermark: requireNonNegativeSafeInteger(body["throughOverflowWatermark"], "throughOverflowWatermark"), + }); + } catch (error) { + return reply.code(notificationErrorStatus(error)).send({ error: errorMessage(error) }); + } + }); + app.get<{ Params: { sessionId: string }; Querystring: MessageQuery }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => { try { const page = { ...optionalField("before", optionalNumber(request.query.before)), ...optionalField("limit", optionalNumber(request.query.limit)) }; @@ -341,6 +389,23 @@ function parseBulkMutationRef(value: unknown): SessionBulkMutationRef { return { id, cwd: normalizeRequestCwd(cwd) }; } +function notificationRefFromQuery(id: string, query: SessionQuery): { id: string; cwd: string } { + const cwd = requireNonEmptyBoundedString(query.cwd, "cwd", MAX_NOTIFICATION_CWD_LENGTH); + return notificationRef(id, cwd); +} + +function notificationRefFromBody(id: string, body: Record): { id: string; cwd: string } { + const cwd = requireNonEmptyBoundedString(body["cwd"], "cwd", MAX_NOTIFICATION_CWD_LENGTH); + return notificationRef(id, cwd); +} + +function notificationRef(id: string, cwd: string): { id: string; cwd: string } { + return { + id: requireNonEmptyBoundedString(id, "sessionId", MAX_NOTIFICATION_SESSION_ID_LENGTH), + cwd: normalizeRequestCwd(cwd), + }; +} + function sessionLookupFromQuery(id: string, query: SessionQuery): SessionLookup { return sessionLookupFromCwd(id, query.cwd); } @@ -374,6 +439,20 @@ function requireString(record: Record, field: string): string { return value; } +function requireNonEmptyBoundedString(value: unknown, field: string, maxLength: number): string { + if (typeof value !== "string") throw new Error(`${field} field must be a string`); + if (value === "") throw new Error(`${field} field must not be empty`); + if (value.length > maxLength) throw new Error(`${field} field is too long`); + return value; +} + +function requireNonNegativeSafeInteger(value: unknown, field: string): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) { + throw new Error(`${field} field must be a non-negative safe integer`); + } + return value; +} + function requireThinkingLevel(value: unknown): string { if (typeof value !== "string" || value === "") throw new Error("level field is invalid"); return value; @@ -397,6 +476,10 @@ function mutationErrorStatus(error: unknown): 400 | 404 { return isSessionNotFoundError(error) ? 404 : 400; } +function notificationErrorStatus(error: unknown): 400 | 404 { + return isSessionNotFoundError(error) ? 404 : 400; +} + function isSessionNotFoundError(error: unknown): boolean { const message = errorMessage(error); return message === "Session not found" || message === "Archived session not found"; diff --git a/src/server/sessions/sessionService.ts b/src/server/sessions/sessionService.ts index 9958262..0466634 100644 --- a/src/server/sessions/sessionService.ts +++ b/src/server/sessions/sessionService.ts @@ -3,6 +3,10 @@ import type { SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkMutationRef, + SessionNotificationCatalogSnapshot, + SessionNotificationDismissAllRequest, + SessionNotificationDismissRequest, + SessionNotificationInboxSnapshot, } from "../../shared/apiTypes.js"; import type { ClientArchiveSessionsResponse, @@ -36,6 +40,10 @@ export interface SessionRouteService { messages(ref: SessionRouteLookup, page?: { before?: number; limit?: number }): Promise; status(ref: SessionRouteLookup): Promise; streamSnapshot(ref: SessionRouteLookup): Promise; + notificationCatalog(): SessionNotificationCatalogSnapshot | Promise; + notificationInbox(ref: SessionRouteRef): SessionNotificationInboxSnapshot | Promise; + dismissNotification(ref: SessionRouteRef, request: Omit): SessionNotificationInboxSnapshot | Promise; + dismissAllNotifications(ref: SessionRouteRef, request: Omit): SessionNotificationInboxSnapshot | Promise; clearQueue(ref: SessionRouteLookup): Promise; dismissWarning(ref: SessionRouteLookup, dismissId: string): Promise; availableModels(ref: SessionRouteLookup): Promise; diff --git a/src/shared/apiTypes.ts b/src/shared/apiTypes.ts index d687d43..2e74b8e 100644 --- a/src/shared/apiTypes.ts +++ b/src/shared/apiTypes.ts @@ -8,6 +8,7 @@ export const PI_WEB_CAPABILITIES = { sessionsReload: "sessions.reload", sessionsClearQueue: "sessions.clearQueue", sessionsPersistedState: "sessions.persistedState", + sessionsNotifications: "sessions.notifications", promptAttachments: "prompt.attachments", workspaceFileSuggestions: "workspace.fileSuggestions", piPackagesManage: "piPackages.manage", @@ -203,6 +204,93 @@ export interface SessionRef { cwd: string; } +export const SESSION_NOTIFICATION_LIMIT = 100; +export const SESSION_NOTIFICATION_MESSAGE_BYTES = 8 * 1024; + +export type SessionNotificationSeverity = "info" | "warning" | "error"; + +export interface SessionNotification { + id: string; + message: string; + truncated: boolean; + severity: SessionNotificationSeverity; + receivedAt: string; + order: number; +} + +export interface SessionNotificationSummary { + sessionId: string; + cwd: string; + inboxRevision: number; + retainedCount: number; + discardedCount: number; + highestSeverity?: SessionNotificationSeverity; +} + +export interface SessionNotificationDismissThrough { + order: number; + overflowWatermark: number; +} + +export interface SessionNotificationInboxSnapshot { + daemonInstanceId: string; + catalogRevision: number; + summary: SessionNotificationSummary; + notifications: SessionNotification[]; + dismissThrough: SessionNotificationDismissThrough; +} + +export interface SessionNotificationCatalogSnapshot { + daemonInstanceId: string; + catalogRevision: number; + sessions: SessionNotificationSummary[]; +} + +export interface SessionNotificationDismissRequest { + cwd: string; + daemonInstanceId: string; + notificationId: string; +} + +export interface SessionNotificationDismissAllRequest { + cwd: string; + daemonInstanceId: string; + throughOrder: number; + throughOverflowWatermark: number; +} + +export type SessionNotificationClearReason = + | "runtime-close" + | "archive" + | "delete" + | "restore" + | "archive-reconcile" + | "replacement" + | "initialization-failed" + | "service-dispose"; + +export type SessionNotificationInboxDelta = + | { kind: "added"; notification: SessionNotification; evictedNotificationId?: string } + | { kind: "dismissed"; notificationIds: string[] } + | { kind: "cleared"; reason: SessionNotificationClearReason } + | { kind: "resync" }; + +export interface SessionNotificationInboxEvent { + type: "notifications.inbox"; + daemonInstanceId: string; + catalogRevision: number; + summary: SessionNotificationSummary; + dismissThrough: SessionNotificationDismissThrough; + delta: SessionNotificationInboxDelta; +} + +export interface SessionNotificationSummaryEvent { + type: "notifications.summary"; + daemonInstanceId: string; + catalogRevision: number; + summary: SessionNotificationSummary; +} + export interface SessionInfo extends SessionRef { path: string; /** True when the server has verified a backing session file exists; false when known transient. */ @@ -777,11 +865,14 @@ type SessionUiEventBody = | { type: "message.end"; message?: unknown } | { type: "status.update"; status: SessionStatus } | { type: "activity.update"; activity: SessionActivity } - | { type: "command.output"; level: "info" | "success" | "error"; message: string } + | { type: "command.output"; level: "info" | "success" | "error"; message: string; notificationId?: string } + | SessionNotificationInboxEvent | { type: "session.error"; message: string } | { type: "session.name"; sessionId: string; name?: string } | { type: "session.created"; session: SessionInfo } | { type: "pi.event"; eventType: string }; -export type GlobalSessionEvent = Extract; +export type GlobalSessionEvent = + | Extract + | SessionNotificationSummaryEvent; export type RealtimeEvent = GlobalSessionEvent | TerminalUiEvent | WorkspaceActivityUiEvent; diff --git a/src/shared/capabilities.test.ts b/src/shared/capabilities.test.ts index 72d1d53..0c06fe9 100644 --- a/src/shared/capabilities.test.ts +++ b/src/shared/capabilities.test.ts @@ -50,6 +50,26 @@ describe("PI WEB capabilities", () => { })).toContain(clearQueue); }); + it("requires both web and session daemon support for notification inboxes", () => { + const notifications = PI_WEB_CAPABILITIES.sessionsNotifications; + expect(WEB_RUNTIME_CAPABILITIES).toContain(notifications); + expect(SESSIOND_RUNTIME_CAPABILITIES).toContain(notifications); + expect(parseKnownPiWebCapabilities([notifications, "future.capability"])).toEqual([notifications]); + + expect(effectivePiWebCapabilities({ + web: { available: true, capabilities: [notifications] }, + sessiond: { available: true, capabilities: [] }, + })).not.toContain(notifications); + expect(effectivePiWebCapabilities({ + web: { available: true, capabilities: [] }, + sessiond: { available: true, capabilities: [notifications] }, + })).not.toContain(notifications); + expect(effectivePiWebCapabilities({ + web: { available: true, capabilities: [notifications] }, + sessiond: { available: true, capabilities: [notifications] }, + })).toContain(notifications); + }); + it("keeps only known string capabilities when parsing runtime data", () => { expect(parseKnownPiWebCapabilities([PI_WEB_CAPABILITIES.piPackagesManage, PI_WEB_CAPABILITIES.selectedMachineSettings, "future.capability"])).toEqual([PI_WEB_CAPABILITIES.piPackagesManage, PI_WEB_CAPABILITIES.selectedMachineSettings]); expect(parseKnownPiWebCapabilities([PI_WEB_CAPABILITIES.piPackagesManage, 1])).toBeUndefined(); diff --git a/src/shared/capabilities.ts b/src/shared/capabilities.ts index 4c83425..d1e8a0c 100644 --- a/src/shared/capabilities.ts +++ b/src/shared/capabilities.ts @@ -13,6 +13,7 @@ export const WEB_RUNTIME_CAPABILITIES = [ PI_WEB_CAPABILITIES.sessionsReload, PI_WEB_CAPABILITIES.sessionsClearQueue, PI_WEB_CAPABILITIES.sessionsPersistedState, + PI_WEB_CAPABILITIES.sessionsNotifications, PI_WEB_CAPABILITIES.promptAttachments, PI_WEB_CAPABILITIES.workspaceFileSuggestions, PI_WEB_CAPABILITIES.piPackagesManage, @@ -27,6 +28,7 @@ export const SESSIOND_RUNTIME_CAPABILITIES = [ PI_WEB_CAPABILITIES.sessionsReload, PI_WEB_CAPABILITIES.sessionsClearQueue, PI_WEB_CAPABILITIES.sessionsPersistedState, + PI_WEB_CAPABILITIES.sessionsNotifications, PI_WEB_CAPABILITIES.promptAttachments, ] as const satisfies readonly PiWebCapability[]; @@ -37,6 +39,7 @@ const EFFECTIVE_CAPABILITY_REQUIREMENTS = { [PI_WEB_CAPABILITIES.sessionsReload]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.sessionsClearQueue]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.sessionsPersistedState]: ["web", "sessiond"], + [PI_WEB_CAPABILITIES.sessionsNotifications]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.promptAttachments]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.workspaceFileSuggestions]: ["web"], [PI_WEB_CAPABILITIES.piPackagesManage]: ["web"], diff --git a/src/shared/federatedRoutes.ts b/src/shared/federatedRoutes.ts index 6614c79..1018fab 100644 --- a/src/shared/federatedRoutes.ts +++ b/src/shared/federatedRoutes.ts @@ -45,11 +45,15 @@ export const FEDERATED_HTTP_ROUTES = [ { method: "GET", path: "/activity" }, { method: "GET", path: "/sessions" }, { method: "POST", path: "/sessions" }, + { method: "GET", path: "/sessions/notifications" }, { method: "POST", path: "/sessions/cleanup/preview" }, { method: "POST", path: "/sessions/cleanup" }, { method: "POST", path: "/sessions/bulk/archive" }, { method: "POST", path: "/sessions/bulk/delete-archived" }, { method: "GET", path: "/sessions/:sessionId/messages" }, + { method: "GET", path: "/sessions/:sessionId/notifications" }, + { method: "POST", path: "/sessions/:sessionId/notifications/dismiss" }, + { method: "POST", path: "/sessions/:sessionId/notifications/dismiss-all" }, { method: "GET", path: "/sessions/:sessionId/status" }, { method: "GET", path: "/sessions/:sessionId/stream-snapshot" }, { method: "GET", path: "/sessions/:sessionId/models" },