diff --git a/.changeset/bulk-session-mutations.md b/.changeset/bulk-session-mutations.md new file mode 100644 index 0000000..ac5beb8 --- /dev/null +++ b/.changeset/bulk-session-mutations.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Improve bulk session archive and delete reliability by adding true bulk mutation support for large session selections. diff --git a/src/client/src/api.ts b/src/client/src/api.ts index 93d758e..b929ea7 100644 --- a/src/client/src/api.ts +++ b/src/client/src/api.ts @@ -2,4 +2,4 @@ export { activityApi, api, configApi, filesApi, gitApi, machinesApi, piWebApi, p export { globalSessionEvents, realtimeEvents, sessionEvents, terminalSocket } from "./api/sockets"; export { DEFAULT_WORKSPACE_UPLOADS_FOLDER, effectiveWorkspaceUploadFolder, uploadWorkspaceFile, uploadWorkspaceFiles, workspaceEffectiveUploadFolder, workspaceUploadPath, WorkspaceUploadBatchError, WorkspaceUploadCancelledError } from "./api/workspaceUploads"; export type { UploadWorkspaceFileOptions, UploadWorkspaceFilesOptions, WorkspaceFileUploadProgress, WorkspaceUploadBatchFileProgress, WorkspaceUploadBatchProgress, WorkspaceUploadFileFailure, WorkspaceUploadFileInput, WorkspaceUploadFolderConfig, WorkspaceUploadTask, WorkspaceUploadXhr, WorkspaceUploadXhrFactory } from "./api/workspaceUploads"; -export type { ArchiveSessionsResponse, AuthProviderOption, AuthProviderStatus, AuthProvidersResponse, AuthStatusSource, AuthType, CommandOption, CommandResult, DeleteWorkspaceFileResponse, FileContentMediaType, FileContentResponse, FileSuggestion, FileTreeEntry, FileTreeResponse, GitDiffResponse, GitFileState, GitStatusFile, GitStatusResponse, Machine, MachineHealth, MachineKind, MachineRuntime, MachineStatus, MessagePage, ModelSelectionResponse, MoveWorkspaceFileOptions, MoveWorkspaceFileResponse, OAuthFlowState, PiWebCapability, PiWebComponentStatus, PiWebConfigEnvOverrides, PiWebConfigResponse, PiWebConfigValues, PiWebInstallationInfo, PiWebPluginConfig, PiWebPluginConfigMap, PiWebPluginInfo, PiWebPluginsResponse, PiWebPluginScope, PiWebPluginSettings, PiWebReleaseStatus, PiWebRuntimeComponent, PiWebRuntimeResponse, PiWebShortcutConfig, PiWebStatusMessage, PiWebStatusResponse, PiWebUploadsConfig, Project, PromptAttachment, QueuedSessionMessage, RealtimeEvent, RunTerminalCommandInput, SavedPromptAttachment, SessionActivity, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse, SessionCleanupProjectSummary, SessionCleanupRequest, SessionCleanupThresholds, SessionCleanupTotals, SessionInfo, SessionModel, SessionRef, SessionStatus, SlashCommand, SessionUiEvent, TerminalCommandRun, TerminalCommandRunFilter, TerminalCommandRunHandle, TerminalCommandRunStatus, TerminalInfo, TerminalUiEvent, ThinkingLevel, ThinkingLevelsResponse, WriteWorkspaceFileOptions, WriteWorkspaceFileResponse, Workspace, WorkspaceActivity, WorkspaceActivityResponse, WorkspaceActivityUiEvent } from "../../shared/apiTypes"; +export type { ArchiveSessionsResponse, AuthProviderOption, AuthProviderStatus, AuthProvidersResponse, AuthStatusSource, AuthType, CommandOption, CommandResult, DeleteWorkspaceFileResponse, FileContentMediaType, FileContentResponse, FileSuggestion, FileTreeEntry, FileTreeResponse, GitDiffResponse, GitFileState, GitStatusFile, GitStatusResponse, Machine, MachineHealth, MachineKind, MachineRuntime, MachineStatus, MessagePage, ModelSelectionResponse, MoveWorkspaceFileOptions, MoveWorkspaceFileResponse, OAuthFlowState, PiWebCapability, PiWebComponentStatus, PiWebConfigEnvOverrides, PiWebConfigResponse, PiWebConfigValues, PiWebInstallationInfo, PiWebPluginConfig, PiWebPluginConfigMap, PiWebPluginInfo, PiWebPluginsResponse, PiWebPluginScope, PiWebPluginSettings, PiWebReleaseStatus, PiWebRuntimeComponent, PiWebRuntimeResponse, PiWebShortcutConfig, PiWebStatusMessage, PiWebStatusResponse, PiWebUploadsConfig, Project, PromptAttachment, QueuedSessionMessage, RealtimeEvent, RunTerminalCommandInput, SavedPromptAttachment, SessionActivity, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkFailure, SessionBulkMutationRef, SessionBulkMutationRequest, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse, SessionCleanupProjectSummary, SessionCleanupRequest, SessionCleanupThresholds, SessionCleanupTotals, SessionInfo, SessionModel, SessionRef, SessionStatus, SlashCommand, SessionUiEvent, TerminalCommandRun, TerminalCommandRunFilter, TerminalCommandRunHandle, TerminalCommandRunStatus, TerminalInfo, TerminalUiEvent, ThinkingLevel, ThinkingLevelsResponse, WriteWorkspaceFileOptions, WriteWorkspaceFileResponse, Workspace, WorkspaceActivity, WorkspaceActivityResponse, WorkspaceActivityUiEvent } from "../../shared/apiTypes"; diff --git a/src/client/src/api/clients.test.ts b/src/client/src/api/clients.test.ts index f824327..943977b 100644 --- a/src/client/src/api/clients.test.ts +++ b/src/client/src/api/clients.test.ts @@ -78,6 +78,23 @@ describe("session API compatibility", () => { expect(JSON.parse(requestBody(fetchCall(fetchMock, 1)[1]))).toEqual({ archiveIdleDays: 7, projectCwds: ["/repo"] }); }); + it("posts bulk session mutation requests through the selected machine", async () => { + const archived = { archived: true, archivedSessionIds: ["s 1"], failures: [{ sessionId: "s 2", error: "busy" }], generatedAt: "now" }; + const deleted = { deleted: true, deletedSessionIds: ["s 1"], failures: [], generatedAt: "later" }; + const fetchMock = stubSequenceFetch([jsonResponse(archived), jsonResponse(deleted)]); + + await expect(sessionsApi.archiveMany([{ id: "s 1", cwd: "/repo" }, "s 2"], "remote a")).resolves.toEqual(archived); + await expect(sessionsApi.deleteArchivedMany([{ id: "s 1", cwd: "/repo" }], "remote a")).resolves.toEqual(deleted); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchCall(fetchMock, 0)[0]).toBe("/api/machines/remote%20a/sessions/bulk/archive"); + expect(fetchCall(fetchMock, 0)[1]?.method).toBe("POST"); + expect(JSON.parse(requestBody(fetchCall(fetchMock, 0)[1]))).toEqual({ sessions: [{ id: "s 1", cwd: "/repo" }, { id: "s 2" }] }); + expect(fetchCall(fetchMock, 1)[0]).toBe("/api/machines/remote%20a/sessions/bulk/delete-archived"); + expect(fetchCall(fetchMock, 1)[1]?.method).toBe("POST"); + expect(JSON.parse(requestBody(fetchCall(fetchMock, 1)[1]))).toEqual({ sessions: [{ id: "s 1", cwd: "/repo" }] }); + }); + it("keeps legacy session-id calls free of cwd context", async () => { const fetchMock = stubJsonFetch({ accepted: true }); diff --git a/src/client/src/api/clients.ts b/src/client/src/api/clients.ts index a0d015b..e3184f3 100644 --- a/src/client/src/api/clients.ts +++ b/src/client/src/api/clients.ts @@ -1,4 +1,4 @@ -import type { DeleteWorkspaceFileResponse, FileSuggestion, MoveWorkspaceFileOptions, PiWebConfigValues, PromptAttachment, RunTerminalCommandInput, SessionCleanupRequest, SessionRef, TerminalCommandRun, TerminalCommandRunFilter, WriteWorkspaceFileOptions } from "../../../shared/apiTypes"; +import type { DeleteWorkspaceFileResponse, FileSuggestion, MoveWorkspaceFileOptions, PiWebConfigValues, PromptAttachment, RunTerminalCommandInput, SessionBulkMutationRef, SessionCleanupRequest, SessionRef, TerminalCommandRun, TerminalCommandRunFilter, WriteWorkspaceFileOptions } from "../../../shared/apiTypes"; import { request } from "./http"; import { arrayOf, @@ -32,6 +32,8 @@ import { parseReloaded, parseRestored, parseSavedAttachments, + parseSessionBulkArchiveResponse, + parseSessionBulkDeleteArchivedResponse, parseSessionCleanupExecuteResponse, parseSessionCleanupPreviewResponse, parseSessionInfo, @@ -85,6 +87,16 @@ function sessionBody(session: SessionLookup, fields: Record = { return JSON.stringify(cwd === undefined || cwd === "" ? fields : { cwd, ...fields }); } +function sessionBulkMutationBody(sessions: readonly SessionLookup[]): string { + return JSON.stringify({ sessions: sessions.map(sessionBulkMutationRef) }); +} + +function sessionBulkMutationRef(session: SessionLookup): SessionBulkMutationRef { + const id = sessionId(session); + const cwd = sessionCwd(session); + return cwd === undefined || cwd === "" ? { id } : { id, cwd }; +} + export const piWebApi = { piWebStatus: (machineId = "local") => request(machineId === "local" ? "/api/pi-web/status" : `${machinePrefix(machineId)}/pi-web/status`, parsePiWebStatusResponse), piWebRuntime: () => request("/api/pi-web/runtime", parsePiWebRuntimeResponse), @@ -156,6 +168,8 @@ export const sessionsApi = { startSession: (cwd: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions`, parseSessionInfo, { method: "POST", body: JSON.stringify({ cwd }) }), cleanupPreview: (input: SessionCleanupRequest, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/cleanup/preview`, parseSessionCleanupPreviewResponse, { method: "POST", body: JSON.stringify(input) }), cleanup: (input: SessionCleanupRequest, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/cleanup`, parseSessionCleanupExecuteResponse, { method: "POST", body: JSON.stringify(input) }), + archiveMany: (sessions: readonly SessionLookup[], machineId = "local") => request(`${machinePrefix(machineId)}/sessions/bulk/archive`, parseSessionBulkArchiveResponse, { method: "POST", body: sessionBulkMutationBody(sessions) }), + deleteArchivedMany: (sessions: readonly SessionLookup[], machineId = "local") => request(`${machinePrefix(machineId)}/sessions/bulk/delete-archived`, parseSessionBulkDeleteArchivedResponse, { method: "POST", body: sessionBulkMutationBody(sessions) }), messages: (session: SessionLookup, options?: { limit?: number; before?: number }, machineId = "local") => request(messageUrl(session, options, machineId), parseMessagePage), status: (session: SessionLookup, machineId = "local") => request(sessionQueryUrl(session, "status", machineId), parseSessionStatus), models: (session: SessionLookup, machineId = "local") => request(sessionQueryUrl(session, "models", machineId), parseModelSelectionResponse), diff --git a/src/client/src/api/federatedRouteContract.test.ts b/src/client/src/api/federatedRouteContract.test.ts index 2d83412..1cde8ff 100644 --- a/src/client/src/api/federatedRouteContract.test.ts +++ b/src/client/src/api/federatedRouteContract.test.ts @@ -48,6 +48,8 @@ describe("federated route contract", () => { ignoreParseFailure(sessionsApi.startSession("/repo", machineId)), ignoreParseFailure(sessionsApi.cleanupPreview({ archiveIdleDays: 14 }, machineId)), ignoreParseFailure(sessionsApi.cleanup({ archiveIdleDays: 14, deleteArchivedDays: 30, projectCwds: ["/repo"] }, machineId)), + ignoreParseFailure(sessionsApi.archiveMany([session], machineId)), + ignoreParseFailure(sessionsApi.deleteArchivedMany([session], machineId)), ignoreParseFailure(sessionsApi.messages(session, { limit: 20, before: 10 }, machineId)), ignoreParseFailure(sessionsApi.status(session, machineId)), ignoreParseFailure(sessionsApi.models(session, machineId)), diff --git a/src/client/src/api/parsers.test.ts b/src/client/src/api/parsers.test.ts index 3d157e1..b9e1902 100644 --- a/src/client/src/api/parsers.test.ts +++ b/src/client/src/api/parsers.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import { PI_WEB_CAPABILITIES } from "../../../shared/capabilities"; -import { parseCommandResult, parseFileContentResponse, parseFileSuggestion, parseGitStatusResponse, parseMessagePage, parsePiWebConfigResponse, parsePiWebPluginsResponse, parsePiWebRuntimeResponse, parseSessionCleanupExecuteResponse, parseSessionCleanupPreviewResponse, parseSessionStatus, parseSlashCommand, parseTerminalCommandRun, parseTerminalInfo, parseWorkspace, parseWorkspaceActivityResponse } from "./parsers"; +import { parseCommandResult, parseFileContentResponse, parseFileSuggestion, parseGitStatusResponse, parseMessagePage, parsePiWebConfigResponse, parsePiWebPluginsResponse, parsePiWebRuntimeResponse, parseSessionBulkArchiveResponse, parseSessionBulkDeleteArchivedResponse, parseSessionCleanupExecuteResponse, parseSessionCleanupPreviewResponse, parseSessionStatus, parseSlashCommand, parseTerminalCommandRun, parseTerminalInfo, parseWorkspace, parseWorkspaceActivityResponse } from "./parsers"; describe("API parsers", () => { it("parses PI WEB config responses", () => { @@ -69,6 +69,27 @@ describe("API parsers", () => { expect(() => parseSessionCleanupExecuteResponse({ generatedAt: "now", thresholds: {}, projects: [], totals: { archiveCount: 0, deleteCount: 0 }, archivedSessionIds: ["s1"], deletedSessionIds: [1] })).toThrow("Expected string array field: deletedSessionIds"); }); + it("parses bulk session mutation responses", () => { + const failure = { sessionId: "busy", error: "Session is busy" }; + expect(parseSessionBulkArchiveResponse({ archived: true, archivedSessionIds: ["s1"], failures: [failure], generatedAt: "now" })).toEqual({ + archived: true, + archivedSessionIds: ["s1"], + failures: [failure], + generatedAt: "now", + }); + expect(parseSessionBulkDeleteArchivedResponse({ deleted: true, deletedSessionIds: ["s2"], failures: [], generatedAt: "later" })).toEqual({ + deleted: true, + deletedSessionIds: ["s2"], + failures: [], + generatedAt: "later", + }); + }); + + it("rejects malformed bulk session mutation responses", () => { + expect(() => parseSessionBulkArchiveResponse({ archived: true, archivedSessionIds: ["s1"], failures: [{ sessionId: "s2" }], generatedAt: "now" })).toThrow("Expected string field: error"); + expect(() => parseSessionBulkDeleteArchivedResponse({ deleted: true, deletedSessionIds: [1], failures: [], generatedAt: "now" })).toThrow("Expected string array field: deletedSessionIds"); + }); + it("validates session status including optional model and nullable context usage", () => { expect(parseSessionStatus({ sessionId: "s1", diff --git a/src/client/src/api/parsers.ts b/src/client/src/api/parsers.ts index 580db98..dd56e2f 100644 --- a/src/client/src/api/parsers.ts +++ b/src/client/src/api/parsers.ts @@ -1,4 +1,4 @@ -import type { ArchiveSessionsResponse, AuthProviderOption, AuthProviderStatus, AuthProvidersResponse, AuthStatusSource, AuthType, CommandOption, CommandResult, DeleteWorkspaceFileResponse, FileContentResponse, FileSuggestion, FileTreeEntry, FileTreeResponse, GitDiffResponse, GitFileState, GitStatusFile, GitStatusResponse, Machine, MachineHealth, MachineKind, MachineRuntime, MachineStatus, MessagePage, ModelSelectionResponse, MoveWorkspaceFileResponse, OAuthFlowState, PiWebCapability, PiWebComponentStatus, PiWebConfigEnvOverrides, PiWebConfigResponse, PiWebConfigValues, PiWebInstallationInfo, PiWebPluginConfigMap, PiWebPluginInfo, PiWebPluginsResponse, PiWebPluginScope, PiWebReleaseStatus, PiWebRuntimeComponent, PiWebRuntimeResponse, PiWebServiceComponent, PiWebShortcutConfig, PiWebStatusMessage, PiWebStatusResponse, PiWebStatusSeverity, Project, QueuedSessionMessage, SavedPromptAttachment, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse, SessionCleanupProjectSummary, SessionCleanupThresholds, SessionCleanupTotals, SessionInfo, SessionModel, SessionStatus, SlashCommand, TerminalCommandRun, TerminalCommandRunStatus, TerminalInfo, ThinkingLevelsResponse, WriteWorkspaceFileResponse, Workspace, WorkspaceActivity, WorkspaceActivityResponse } from "../../../shared/apiTypes"; +import type { ArchiveSessionsResponse, AuthProviderOption, AuthProviderStatus, AuthProvidersResponse, AuthStatusSource, AuthType, CommandOption, CommandResult, DeleteWorkspaceFileResponse, FileContentResponse, FileSuggestion, FileTreeEntry, FileTreeResponse, GitDiffResponse, GitFileState, GitStatusFile, GitStatusResponse, Machine, MachineHealth, MachineKind, MachineRuntime, MachineStatus, MessagePage, ModelSelectionResponse, MoveWorkspaceFileResponse, OAuthFlowState, PiWebCapability, PiWebComponentStatus, PiWebConfigEnvOverrides, PiWebConfigResponse, PiWebConfigValues, PiWebInstallationInfo, PiWebPluginConfigMap, PiWebPluginInfo, PiWebPluginsResponse, PiWebPluginScope, PiWebReleaseStatus, PiWebRuntimeComponent, PiWebRuntimeResponse, PiWebServiceComponent, PiWebShortcutConfig, PiWebStatusMessage, PiWebStatusResponse, PiWebStatusSeverity, Project, QueuedSessionMessage, SavedPromptAttachment, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkFailure, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse, SessionCleanupProjectSummary, SessionCleanupThresholds, SessionCleanupTotals, SessionInfo, SessionModel, SessionStatus, SlashCommand, TerminalCommandRun, TerminalCommandRunStatus, TerminalInfo, ThinkingLevelsResponse, WriteWorkspaceFileResponse, Workspace, WorkspaceActivity, WorkspaceActivityResponse } from "../../../shared/apiTypes"; import { isPiWebCapability } from "../../../shared/capabilities"; function isRecord(value: unknown): value is Record { @@ -212,6 +212,33 @@ export function parseSessionCleanupExecuteResponse(value: unknown): SessionClean }; } +export function parseSessionBulkArchiveResponse(value: unknown): SessionBulkArchiveResponse { + const record = requireRecord(value); + if (record["archived"] !== true) throw new Error("Expected bulk archived response"); + return { + archived: true, + archivedSessionIds: arrayOfString(record["archivedSessionIds"], "archivedSessionIds"), + failures: arrayOf(parseSessionBulkFailure)(record["failures"]), + generatedAt: requireString(record, "generatedAt"), + }; +} + +export function parseSessionBulkDeleteArchivedResponse(value: unknown): SessionBulkDeleteArchivedResponse { + const record = requireRecord(value); + if (record["deleted"] !== true) throw new Error("Expected bulk deleted response"); + return { + deleted: true, + deletedSessionIds: arrayOfString(record["deletedSessionIds"], "deletedSessionIds"), + failures: arrayOf(parseSessionBulkFailure)(record["failures"]), + generatedAt: requireString(record, "generatedAt"), + }; +} + +function parseSessionBulkFailure(value: unknown): SessionBulkFailure { + const record = requireRecord(value); + return { sessionId: requireString(record, "sessionId"), error: requireString(record, "error") }; +} + function parseSessionCleanupThresholds(value: unknown): SessionCleanupThresholds { const record = requireRecord(value); return { diff --git a/src/client/src/controllers/sessionController.test.ts b/src/client/src/controllers/sessionController.test.ts index a58f6c5..f189a2b 100644 --- a/src/client/src/controllers/sessionController.test.ts +++ b/src/client/src/controllers/sessionController.test.ts @@ -649,6 +649,90 @@ describe("SessionController", () => { expect(state.selectedSession?.id).toBe(nextSession.id); }); + it("uses true bulk archive when the selected runtime supports it and applies partial failures", async () => { + const failedSession = { ...oldSession, id: "failed-session", path: "/tmp/failed-session.jsonl" }; + const archiveCalls: { ids: string[]; machineId: string }[] = []; + let state: AppState = { + ...initialAppState(), + selectedWorkspace: workspace, + sessions: [oldSession, failedSession], + machineRuntimes: { local: { machineId: "local", ok: true, checkedAt: "now", capabilities: [PI_WEB_CAPABILITIES.sessionsBulkMutations] } }, + }; + const api: typeof defaultApi = { + ...defaultApi, + archiveMany: (sessions, machineId) => { + archiveCalls.push({ ids: sessions.map(sessionLookupId), machineId: machineId ?? "local" }); + return Promise.resolve({ archived: true, archivedSessionIds: [oldSession.id], failures: [{ sessionId: failedSession.id, error: "busy" }], generatedAt: "now" }); + }, + archive: () => { throw new Error("single archive should not be used"); }, + messages: () => Promise.resolve(emptyPage), + status: (session) => Promise.resolve(status(sessionLookupId(session))), + }; + const controller = new SessionController( + () => state, + (patch) => { state = { ...state, ...patch }; }, + () => undefined, + new InMemorySessionSelectionMemory(), + { api, socket: new FakeSocket() }, + ); + + await controller.selectSession(oldSession, { updateUrl: false }); + await controller.archiveSessions([oldSession, failedSession]); + + expect(archiveCalls).toEqual([{ ids: [oldSession.id, failedSession.id], machineId: "local" }]); + expect(state.sessions.find((session) => session.id === oldSession.id)).toMatchObject({ archived: true }); + expect(state.sessions.find((session) => session.id === failedSession.id)?.archived).toBeUndefined(); + expect(state.selectedSession?.id).toBe(failedSession.id); + expect(state.error).toBe("Archive failed for 1 session: failed-session: busy"); + }); + + it("throttles per-session archive fallback when bulk mutations are unsupported", async () => { + const sessions = Array.from({ length: 6 }, (_value, index) => ({ ...oldSession, id: `session-${String(index)}`, path: `/tmp/session-${String(index)}.jsonl` })); + const resolvers: (() => void)[] = []; + const startedIds: string[] = []; + let activeCount = 0; + let maxActiveCount = 0; + let state: AppState = { ...initialAppState(), selectedWorkspace: workspace, sessions }; + const api: typeof defaultApi = { + ...defaultApi, + archive: (session) => new Promise((resolve) => { + activeCount += 1; + maxActiveCount = Math.max(maxActiveCount, activeCount); + startedIds.push(sessionLookupId(session)); + resolvers.push(() => { + activeCount -= 1; + resolve({ archived: true }); + }); + }), + messages: () => Promise.resolve(emptyPage), + status: (session) => Promise.resolve(status(sessionLookupId(session))), + }; + const controller = new SessionController( + () => state, + (patch) => { state = { ...state, ...patch }; }, + () => undefined, + new InMemorySessionSelectionMemory(), + { api, socket: new FakeSocket() }, + ); + + const archive = controller.archiveSessions(sessions); + await Promise.resolve(); + + expect(startedIds).toHaveLength(4); + resolvers.shift()?.(); + await Promise.resolve(); + await Promise.resolve(); + expect(startedIds).toHaveLength(5); + for (const resolve of resolvers.splice(0)) resolve(); + await Promise.resolve(); + await Promise.resolve(); + for (const resolve of resolvers.splice(0)) resolve(); + await archive; + + expect(maxActiveCount).toBe(4); + expect(state.sessions.every((session) => session.archived === true)).toBe(true); + }); + it("deletes selected archived sessions in bulk and selects the next current session", async () => { const archivedSession = { ...oldSession, archived: true, archivedAt: "later" }; const nextSession = { ...oldSession, id: "next-session", path: "/tmp/next-session.jsonl" }; @@ -684,6 +768,42 @@ describe("SessionController", () => { expect(state.selectedSession?.id).toBe(nextSession.id); }); + it("uses true bulk delete when supported and keeps partial failures visible", async () => { + const deletedSession = { ...oldSession, archived: true, archivedAt: "later" }; + const failedSession = { ...oldSession, id: "failed-archived", path: "/tmp/failed-archived.jsonl", archived: true, archivedAt: "later" }; + const deleteCalls: { ids: string[]; machineId: string }[] = []; + let state: AppState = { + ...initialAppState(), + selectedWorkspace: workspace, + selectedSession: deletedSession, + sessions: [deletedSession, failedSession], + machineRuntimes: { local: { machineId: "local", ok: true, checkedAt: "now", capabilities: [PI_WEB_CAPABILITIES.sessionsDeleteArchived, PI_WEB_CAPABILITIES.sessionsBulkMutations] } }, + }; + const api: typeof defaultApi = { + ...defaultApi, + deleteArchivedMany: (sessions, machineId) => { + deleteCalls.push({ ids: sessions.map(sessionLookupId), machineId: machineId ?? "local" }); + return Promise.resolve({ deleted: true, deletedSessionIds: [deletedSession.id], failures: [{ sessionId: failedSession.id, error: "busy" }], generatedAt: "now" }); + }, + deleteArchived: () => { throw new Error("single delete should not be used"); }, + messages: () => Promise.resolve(emptyPage), + }; + const controller = new SessionController( + () => state, + (patch) => { state = { ...state, ...patch }; }, + () => undefined, + new InMemorySessionSelectionMemory(), + { api, socket: new FakeSocket() }, + ); + + await controller.deleteArchivedSessions([deletedSession, failedSession]); + + expect(deleteCalls).toEqual([{ ids: [deletedSession.id, failedSession.id], machineId: "local" }]); + expect(state.sessions.map((session) => session.id)).toEqual([failedSession.id]); + expect(state.selectedSession?.id).toBe(failedSession.id); + expect(state.error).toBe("Delete failed for 1 session: failed-archived: busy"); + }); + it("applies cleanup execution results and refreshes the current workspace sessions", async () => { const archivedAt = "2026-06-25T12:00:00.000Z"; const deletedArchived = { ...oldSession, id: "deleted-archived", path: "/tmp/deleted-archived.jsonl", archived: true, archivedAt: "2026-05-01T00:00:00.000Z" }; diff --git a/src/client/src/controllers/sessionController.ts b/src/client/src/controllers/sessionController.ts index a8b5976..93533b6 100644 --- a/src/client/src/controllers/sessionController.ts +++ b/src/client/src/controllers/sessionController.ts @@ -1,4 +1,4 @@ -import { api as defaultApi, type CommandResult, type PromptAttachment, type SessionActivity, type SessionCleanupExecuteResponse, type SessionInfo, type SessionRef, type SessionStatus } from "../api"; +import { api as defaultApi, type CommandResult, type PromptAttachment, type SessionActivity, type SessionBulkFailure, type SessionCleanupExecuteResponse, type SessionInfo, type SessionRef, type SessionStatus } from "../api"; import type { AppState } from "../appState"; import { forgetCachedNewSession, isCachedNewSessionInfo, markCachedNewSessionInfo, mergeCachedNewSessions, rememberCachedNewSession, stripCachedNewSessionMarker } from "../cachedNewSessions"; import { textMessage } from "../chatMessages"; @@ -14,6 +14,7 @@ import { InMemorySessionSelectionMemory, markSessionArchived, markSessionsArchiv import { selectedMachineId, type GetState, type SetState, type UpdateUrl } from "./types"; const MESSAGE_PAGE_SIZE = 100; +const BULK_FALLBACK_CONCURRENCY = 4; export interface SessionEventSocket { connect(session: SessionRef, onEvent: (event: SessionUiEvent) => void, onReconnect?: () => void, machineId?: string): void; @@ -27,6 +28,12 @@ export interface SessionControllerDependencies { transcripts?: ChatTranscriptStore; } +interface BulkSessionMutationResult { + succeededIds: string[]; + failures: string[]; + generatedAt?: string; +} + export class SessionController { private readonly socket: SessionEventSocket; private readonly api: typeof defaultApi; @@ -313,22 +320,22 @@ export class SessionController { const candidates = uniqueSessionsById(sessions).filter((session) => session.archived !== true && !isCachedNewSessionInfo(session)); if (candidates.length === 0) return; - const machineId = selectedMachineId(this.getState()); - const results = await Promise.allSettled(candidates.map(async (session) => { - await this.api.archive(session, machineId); - return session.id; - })); - const archivedIds = fulfilledValues(results); - if (archivedIds.length > 0) { - const state = this.getState(); - const nextSessions = markSessionsArchived(state.sessions, archivedIds, new Date().toISOString()); - const selectionChange = selectionAfterArchivingSessions(nextSessions, state.selectedSession?.id, archivedIds); - this.setState({ sessions: nextSessions }); + try { + const machineId = selectedMachineId(this.getState()); + const { succeededIds: archivedIds, failures, generatedAt } = await this.archiveSessionBatch(candidates, machineId); + if (archivedIds.length > 0) { + const state = this.getState(); + const nextSessions = markSessionsArchived(state.sessions, archivedIds, generatedAt ?? new Date().toISOString()); + const selectionChange = selectionAfterArchivingSessions(nextSessions, state.selectedSession?.id, archivedIds); + this.setState({ sessions: nextSessions }); - if (selectionChange.type === "select") await this.selectSession(selectionChange.session); - else if (selectionChange.type === "clear") this.deselectSession({ forgetRememberedSelection: true }); + if (selectionChange.type === "select") await this.selectSession(selectionChange.session); + else if (selectionChange.type === "clear") this.deselectSession({ forgetRememberedSelection: true }); + } + this.applyBulkSessionFailures("Archive", failures); + } catch (error) { + this.setState({ error: `Archive failed: ${errorMessage(error)}` }); } - this.applyBulkSessionError("Archive", results); } async deleteArchivedSessions(sessions: readonly SessionInfo[]): Promise { @@ -341,23 +348,51 @@ export class SessionController { this.setState({ error: "Deleting archived sessions requires an updated Pi-Web runtime on this machine." }); return; } - const results = await Promise.allSettled(candidates.map(async (session) => { + try { + const { succeededIds: deletedIds, failures } = await this.deleteArchivedSessionBatch(candidates, machineId); + if (deletedIds.length > 0) { + const deletedIdSet = new Set(deletedIds); + const state = this.getState(); + const nextSessions = state.sessions.filter((session) => !deletedIdSet.has(session.id)); + this.setState({ sessions: nextSessions }); + if (state.selectedSession !== undefined && deletedIdSet.has(state.selectedSession.id)) { + const next = nextSessions.find((session) => session.archived !== true) ?? nextSessions[0]; + if (next !== undefined) await this.selectSession(next); + else this.deselectSession({ forgetRememberedSelection: true }); + } + } + this.applyBulkSessionFailures("Delete", failures); + } catch (error) { + this.setState({ error: `Delete failed: ${errorMessage(error)}` }); + } + } + + private async archiveSessionBatch(sessions: readonly SessionInfo[], machineId: string): Promise { + const runtime = this.getState().machineRuntimes[machineId]; + if (runtime?.ok === true && supportsPiWebCapability(runtime, PI_WEB_CAPABILITIES.sessionsBulkMutations)) { + const response = await this.api.archiveMany(sessions, machineId); + return { succeededIds: response.archivedSessionIds, failures: bulkFailureMessages(response.failures), generatedAt: response.generatedAt }; + } + + const results = await allSettledWithConcurrency(sessions, BULK_FALLBACK_CONCURRENCY, async (session) => { + await this.api.archive(session, machineId); + return session.id; + }); + return { succeededIds: fulfilledValues(results), failures: settledSessionFailureMessages(sessions, results) }; + } + + private async deleteArchivedSessionBatch(sessions: readonly SessionInfo[], machineId: string): Promise { + const runtime = this.getState().machineRuntimes[machineId]; + if (runtime?.ok === true && supportsPiWebCapability(runtime, PI_WEB_CAPABILITIES.sessionsBulkMutations)) { + const response = await this.api.deleteArchivedMany(sessions, machineId); + return { succeededIds: response.deletedSessionIds, failures: bulkFailureMessages(response.failures) }; + } + + const results = await allSettledWithConcurrency(sessions, BULK_FALLBACK_CONCURRENCY, async (session) => { await this.api.deleteArchived(session, machineId); return session.id; - })); - const deletedIds = fulfilledValues(results); - if (deletedIds.length > 0) { - const deletedIdSet = new Set(deletedIds); - const state = this.getState(); - const nextSessions = state.sessions.filter((session) => !deletedIdSet.has(session.id)); - this.setState({ sessions: nextSessions }); - if (state.selectedSession !== undefined && deletedIdSet.has(state.selectedSession.id)) { - const next = nextSessions.find((session) => session.archived !== true) ?? nextSessions[0]; - if (next !== undefined) await this.selectSession(next); - else this.deselectSession({ forgetRememberedSelection: true }); - } - } - this.applyBulkSessionError("Delete", results); + }); + return { succeededIds: fulfilledValues(results), failures: settledSessionFailureMessages(sessions, results) }; } async applySessionCleanupResult(result: SessionCleanupExecuteResponse, machineId = selectedMachineId(this.getState())): Promise { @@ -581,8 +616,7 @@ export class SessionController { } } - private applyBulkSessionError(action: string, results: readonly PromiseSettledResult[]): void { - const failures = rejectedReasons(results); + private applyBulkSessionFailures(action: string, failures: readonly string[]): void { if (failures.length === 0) return; this.setState({ error: `${action} failed for ${String(failures.length)} session${failures.length === 1 ? "" : "s"}: ${failures.join("; ")}` }); } @@ -845,8 +879,39 @@ function fulfilledValues(results: readonly PromiseSettledResult[]): T[] { return results.filter(isFulfilled).map((result) => result.value); } -function rejectedReasons(results: readonly PromiseSettledResult[]): string[] { - return results.filter(isRejected).map((result) => errorMessage(result.reason)); +function bulkFailureMessages(failures: readonly SessionBulkFailure[]): string[] { + return failures.map((failure) => `${failure.sessionId}: ${failure.error}`); +} + +function settledSessionFailureMessages(sessions: readonly SessionInfo[], results: readonly PromiseSettledResult[]): string[] { + return results.flatMap((result, index) => { + if (!isRejected(result)) return []; + const sessionId = sessions[index]?.id ?? "unknown"; + return [`${sessionId}: ${errorMessage(result.reason)}`]; + }); +} + +async function allSettledWithConcurrency(items: readonly T[], concurrency: number, worker: (item: T) => Promise): Promise[]> { + const indexedItems = items.map((item, index) => ({ item, index })); + const results: PromiseSettledResult[] = []; + let nextIndex = 0; + + async function runWorker(): Promise { + while (nextIndex < indexedItems.length) { + const entry = indexedItems[nextIndex]; + if (entry === undefined) return; + nextIndex += 1; + try { + results[entry.index] = { status: "fulfilled", value: await worker(entry.item) }; + } catch (reason) { + results[entry.index] = { status: "rejected", reason }; + } + } + } + + const workerCount = Math.min(Math.max(1, concurrency), indexedItems.length); + await Promise.all(Array.from({ length: workerCount }, () => runWorker())); + return results; } function isFulfilled(result: PromiseSettledResult): result is PromiseFulfilledResult { diff --git a/src/server/sessions/piSessionService.test.ts b/src/server/sessions/piSessionService.test.ts index fe63392..b99cbab 100644 --- a/src/server/sessions/piSessionService.test.ts +++ b/src/server/sessions/piSessionService.test.ts @@ -452,6 +452,160 @@ describe("PiSessionService", () => { await service.dispose(); }); + it("bulk archives inactive sessions by cwd without opening runtimes", async () => { + const recordsByCwd = new Map([ + ["/one", [sessionRecord("a", "/one"), sessionRecord("b", "/one")]], + ["/two", [sessionRecord("c", "/two")]], + ]); + 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 service = new PiSessionService(new CapturingSessionEventHub(), { + 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" }), + archiveMany, + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + }, + sessionManager: { + create: () => fakeSessionManager(), + list: (cwd) => { + listCalls.push(cwd); + return Promise.resolve(recordsByCwd.get(cwd) ?? []); + }, + open, + }, + heartbeatIntervalMs: 60_000, + }); + + const result = await service.archiveMany([{ id: "a", cwd: "/one" }, { id: "b", cwd: "/one" }, { id: "c", cwd: "/two" }]); + + expect(result).toMatchObject({ archived: true, archivedSessionIds: ["a", "b", "c"], failures: [] }); + expect(listCalls).toEqual(["/one", "/two"]); + expect(open).not.toHaveBeenCalled(); + expect(archiveMany).toHaveBeenCalledTimes(1); + expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["a", "b", "c"]); + await service.dispose(); + }); + + it("bulk archive reports per-session failures without aborting other archives", async () => { + const busy = fakeRuntime("busy", { isStreaming: true }); + let createCalls = 0; + 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 service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + createCalls += 1; + return Promise.resolve(busy.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" }), + archiveMany, + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + }, + sessionManager: { + create: () => fakeSessionManager(), + list: () => Promise.resolve([sessionRecord("busy"), sessionRecord("ok")]), + open: () => fakeSessionManager(), + }, + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("busy")); + const result = await service.archiveMany([{ id: "busy", cwd: "/workspace" }, { id: "ok", cwd: "/workspace" }, { id: "missing", cwd: "/workspace" }]); + + expect(createCalls).toBe(1); + expect(busy.calls.abort).toBe(0); + expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["ok"]); + expect(result.archivedSessionIds).toEqual(["ok"]); + expect(result.failures).toEqual([ + { sessionId: "busy", error: "Stop current session activity before archiving" }, + { sessionId: "missing", error: "Session not found" }, + ]); + await service.dispose(); + }); + + it("bulk deletes only archived sessions and skips busy active archived runtimes", async () => { + const busyRecord = { sessionId: "busy-archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/busy.jsonl" }; + const idleRecord = { sessionId: "idle-archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/idle.jsonl" }; + const busy = fakeRuntime("busy-archived", { isStreaming: true }); + const deleteArchivedMany = vi.fn((sessionIds: readonly string[]) => Promise.resolve([...sessionIds])); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(busy.runtime), + archiveStore: { + list: () => Promise.resolve([busyRecord, idleRecord]), + get: (sessionId) => Promise.resolve(sessionId === "busy-archived" ? busyRecord : undefined), + archive: () => { throw new Error("archive should not be called for records that already have archive files"); }, + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + deleteArchived: () => Promise.resolve(), + deleteArchivedMany, + }, + sessionManager: { + create: () => fakeSessionManager(), + list: () => Promise.resolve([sessionRecord("unarchived")]), + open: () => fakeSessionManager(), + }, + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("busy-archived")); + const result = await service.deleteArchivedMany([{ id: "busy-archived", cwd: "/workspace" }, { id: "idle-archived", cwd: "/workspace" }, { id: "unarchived", cwd: "/workspace" }]); + + expect(busy.calls.abort).toBe(0); + expect(deleteArchivedMany).toHaveBeenCalledWith(["idle-archived"]); + expect(result.deletedSessionIds).toEqual(["idle-archived"]); + expect(result.failures).toEqual([ + { sessionId: "busy-archived", error: "Stop current session activity before deleting archived session" }, + { sessionId: "unarchived", error: "Archived session not found" }, + ]); + await service.dispose(); + }); + + it("bulk delete moves legacy archived records with one workspace scan before deleting", async () => { + 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", archivePath: `/archive/${input.sessionId}.jsonl` })))); + const deleteArchivedMany = vi.fn((sessionIds: readonly string[]) => Promise.resolve([...sessionIds])); + const listCalls: string[] = []; + const service = new PiSessionService(new CapturingSessionEventHub(), { + archiveStore: { + list: () => Promise.resolve([ + { sessionId: "legacy-a", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z" }, + { sessionId: "legacy-b", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z" }, + { sessionId: "moved", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/moved.jsonl" }, + ]), + get: () => Promise.resolve(undefined), + archive: (input) => Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }), + archiveMany, + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + deleteArchived: () => Promise.resolve(), + deleteArchivedMany, + }, + sessionManager: { + create: () => fakeSessionManager(), + list: (cwd) => { + listCalls.push(cwd); + return Promise.resolve([sessionRecord("legacy-a"), sessionRecord("legacy-b"), sessionRecord("unarchived")]); + }, + open: () => fakeSessionManager(), + }, + heartbeatIntervalMs: 60_000, + }); + + const result = await service.deleteArchivedMany([{ id: "legacy-a", cwd: "/workspace" }, { id: "legacy-b", cwd: "/workspace" }, { id: "moved", cwd: "/workspace" }]); + + expect(listCalls).toEqual(["/workspace"]); + expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["legacy-a", "legacy-b"]); + expect(deleteArchivedMany).toHaveBeenCalledWith(["legacy-a", "legacy-b", "moved"]); + expect(result.deletedSessionIds).toEqual(["legacy-a", "legacy-b", "moved"]); + expect(result.failures).toEqual([]); + await service.dispose(); + }); + it("previews session cleanup without mutating and executes a recomputed plan", async () => { const archivedInputs: string[] = []; const deletedSessionIds: string[] = []; @@ -463,15 +617,17 @@ describe("PiSessionService", () => { archiveStore: { list: () => Promise.resolve([archived, otherArchived]), get: () => Promise.resolve(undefined), - archive: (input) => { - archivedInputs.push(input.sessionId); - return Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-06-25T00:00:00.000Z" }); + archive: () => Promise.reject(new Error("cleanup should use archiveMany")), + archiveMany: (inputs) => { + archivedInputs.push(...inputs.map((input) => input.sessionId)); + return Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-06-25T00:00:00.000Z" }))); }, restore: () => Promise.resolve(), isArchived: () => Promise.resolve(false), - deleteArchived: (sessionId) => { - deletedSessionIds.push(sessionId); - return Promise.resolve(); + deleteArchived: () => Promise.reject(new Error("cleanup should use deleteArchivedMany")), + deleteArchivedMany: (sessionIds) => { + deletedSessionIds.push(...sessionIds); + return Promise.resolve([...sessionIds]); }, }, sessionManager: { @@ -504,6 +660,48 @@ describe("PiSessionService", () => { await service.dispose(); }); + it("moves legacy cleanup delete records with one workspace scan before batch deleting", async () => { + const listCalls: string[] = []; + const archiveMany = vi.fn((inputs: readonly { sessionId: string; cwd: string }[]) => Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-06-25T00:00:00.000Z", archivePath: `/archive/${input.sessionId}.jsonl` })))); + const deleteArchivedMany = vi.fn((sessionIds: readonly string[]) => Promise.resolve([...sessionIds])); + const service = new PiSessionService(new CapturingSessionEventHub(), { + now: () => new Date("2026-06-25T00:00:00.000Z"), + archiveStore: { + list: () => Promise.resolve([ + { sessionId: "legacy-a", cwd: "/old-project", archivedAt: "2026-04-01T00:00:00.000Z" }, + { sessionId: "legacy-b", cwd: "/old-project", archivedAt: "2026-04-01T00:00:00.000Z" }, + ]), + get: () => Promise.resolve(undefined), + archive: () => Promise.reject(new Error("cleanup should use archiveMany")), + archiveMany, + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + deleteArchived: () => Promise.reject(new Error("cleanup should use deleteArchivedMany")), + deleteArchivedMany, + }, + sessionManager: { + create: () => fakeSessionManager(), + list: (cwd) => { + listCalls.push(cwd); + return Promise.resolve([sessionRecord("legacy-a", cwd), sessionRecord("legacy-b", cwd)]); + }, + listAll: () => Promise.resolve([]), + open: () => fakeSessionManager(), + }, + heartbeatIntervalMs: 60_000, + }); + + const result = await service.cleanup({ thresholds: { deleteArchivedDays: 30 }, projectCwds: ["/old-project"] }); + + expect(listCalls).toEqual(["/old-project"]); + expect(archiveMany).toHaveBeenCalledTimes(1); + expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["legacy-a", "legacy-b"]); + expect(deleteArchivedMany).toHaveBeenCalledWith(["legacy-a", "legacy-b"]); + expect(result.deletedSessionIds).toEqual(["legacy-a", "legacy-b"]); + + await service.dispose(); + }); + it("skips busy active sessions during cleanup execution", async () => { const fake = fakeRuntime("busy-open", { isStreaming: true, sessionManager: fakeSessionManager("/old-project"), sessionFile: "/sessions/busy-open.jsonl" }); const archivedInputs: string[] = []; diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index df36016..391e6ae 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -27,7 +27,7 @@ import { computeEditPreview, type EditPreviewResult } from "./editPreview.js"; import { createPiSessionManagerGateway } from "./piSessionManagerGateway.js"; import { attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachmentService.js"; import { parsePromptAttachments } from "../../shared/promptAttachments.js"; -import type { SavedPromptAttachment } from "../../shared/apiTypes.js"; +import type { SavedPromptAttachment, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkFailure, SessionBulkMutationRef } from "../../shared/apiTypes.js"; import { cwdPathsEqual } from "../workingDirectory.js"; import type { WorkspaceActivityService } from "../activity/workspaceActivityService.js"; @@ -118,7 +118,11 @@ function parsePromptStreamingBehavior(value: unknown): QueuedPromptKind | undefi throw new Error('Prompt streamingBehavior must be "steer" or "followUp"'); } -type SessionArchiveRepository = Pick & { deleteArchived?: (sessionId: string) => Promise }; +type SessionArchiveRepository = Pick & { + archiveMany?: (sessions: readonly ArchiveSessionInput[]) => Promise; + deleteArchived?: (sessionId: string) => Promise; + deleteArchivedMany?: (sessionIds: readonly string[]) => Promise; +}; export type PiSessionRef = ClientSessionRef; @@ -143,6 +147,19 @@ interface WorkspaceArchiveCandidate extends SessionArchiveTreeCandidate { activeSession?: PiAgentSession; } +interface BulkSessionLookupContext { + sessionsByCwd: Map; + allSessions?: readonly PiSessionListEntry[]; +} + +interface BulkArchivePlanItem { + input: ArchiveSessionInput; +} + +interface BulkDeletePlanItem { + record: ArchivedSessionRecord; +} + type AgentModel = NonNullable; type ModelRegistryInstance = ReturnType; @@ -421,10 +438,12 @@ export class PiSessionService { async cleanup(request: NormalizedSessionCleanupRequest): Promise { const plan = await this.cleanupPlan(request); - if (plan.deleteRecords.length > 0 && this.archiveStore.deleteArchived === undefined) throw new Error("Archive store does not support deletion"); + if (plan.deleteRecords.length > 0 && this.archiveStore.deleteArchived === undefined && this.archiveStore.deleteArchivedMany === undefined) throw new Error("Archive store does not support deletion"); const archiveInputs: ArchiveSessionInput[] = []; + const readyArchiveInputs: ArchiveSessionInput[] = []; const deleteRecords: ArchivedSessionRecord[] = []; + const readyDeleteRecords: ArchivedSessionRecord[] = []; const skippedBusySessionIds = new Set(plan.skippedBusySessionIds); for (const input of plan.archiveInputs) { @@ -433,9 +452,10 @@ export class PiSessionService { continue; } await this.closeActive(input.sessionId); - await this.archiveStore.archive(input); - archiveInputs.push(input); + readyArchiveInputs.push(input); } + await this.archiveStoreArchiveMany(readyArchiveInputs); + archiveInputs.push(...readyArchiveInputs); for (const record of plan.deleteRecords) { if (this.activeSessionHasWork(record.sessionId)) { @@ -443,10 +463,11 @@ export class PiSessionService { continue; } await this.closeActive(record.sessionId); - if (record.archivePath === undefined) await this.ensureArchivedRecordMoved(record); - await this.archiveStore.deleteArchived?.(record.sessionId); - deleteRecords.push(record); + readyDeleteRecords.push(record); } + await this.ensureArchivedRecordsMoved(readyDeleteRecords); + const deletedSessionIds = new Set(await this.archiveStoreDeleteArchivedMany(readyDeleteRecords.map((record) => record.sessionId))); + deleteRecords.push(...readyDeleteRecords.filter((record) => deletedSessionIds.has(record.sessionId))); return summarizeSessionCleanupExecution({ archiveInputs, @@ -1105,6 +1126,70 @@ export class PiSessionService { await this.archiveStore.archive(archiveInput); } + async archiveMany(refs: readonly SessionBulkMutationRef[]): Promise { + const uniqueRefs = uniqueBulkSessionRefs(refs); + const [archivedRecords, sessionContext] = await Promise.all([ + this.archiveStore.list(), + this.bulkSessionLookupContext(uniqueRefs), + ]); + const failures: SessionBulkFailure[] = []; + const alreadyArchivedSessionIds: string[] = []; + const planItems: BulkArchivePlanItem[] = []; + + for (const ref of uniqueRefs) { + const archived = findArchivedRecordForBulkRef(archivedRecords, ref); + if (archived !== undefined) { + alreadyArchivedSessionIds.push(archived.sessionId); + continue; + } + + const active = this.activeForLookup(bulkRefToLookup(ref)); + const listed = findListedSessionForBulkRef(sessionContext, ref); + const resolvedSessionId = active?.runtime.session.sessionId ?? listed?.id ?? ref.id; + if (active !== undefined && this.hasActiveWork(active.runtime.session)) { + failures.push({ sessionId: resolvedSessionId, error: "Stop current session activity before archiving" }); + continue; + } + + try { + if (listed !== undefined) { + planItems.push({ input: archiveInputFromListEntry(listed) }); + } else if (active !== undefined) { + planItems.push({ input: archiveInputFromActiveSession(active.runtime.session) }); + } else { + failures.push({ sessionId: ref.id, error: "Session not found" }); + } + } catch (error: unknown) { + failures.push({ sessionId: resolvedSessionId, error: errorMessage(error) }); + } + } + + const readyInputs: ArchiveSessionInput[] = []; + for (const item of planItems) { + try { + await this.closeActive(item.input.sessionId); + readyInputs.push(item.input); + } catch (error: unknown) { + failures.push({ sessionId: item.input.sessionId, error: errorMessage(error) }); + } + } + + const archivedSessionIds = [...alreadyArchivedSessionIds]; + try { + const archived = await this.archiveStoreArchiveMany(readyInputs); + archivedSessionIds.push(...archived.map((record) => record.sessionId)); + } catch (error: unknown) { + for (const input of readyInputs) failures.push({ sessionId: input.sessionId, error: errorMessage(error) }); + } + + return { + archived: true, + archivedSessionIds: uniqueStrings(archivedSessionIds), + failures, + generatedAt: new Date().toISOString(), + }; + } + async archiveTree(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); const catalog = await this.workspaceArchiveCandidates(session.sessionManager.getCwd()); @@ -1115,7 +1200,7 @@ export class PiSessionService { 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.archiveStore.archive(input); + await this.archiveStoreArchiveMany(archiveInputs); return { archived: true, @@ -1142,6 +1227,61 @@ export class PiSessionService { await this.archiveStore.deleteArchived(record.sessionId); } + async deleteArchivedMany(refs: readonly SessionBulkMutationRef[]): Promise { + if (this.archiveStore.deleteArchived === undefined && this.archiveStore.deleteArchivedMany === undefined) throw new Error("Archive store does not support deletion"); + + const uniqueRefs = uniqueBulkSessionRefs(refs); + const archivedRecords = await this.archiveStore.list(); + const failures: SessionBulkFailure[] = []; + const planItems: BulkDeletePlanItem[] = []; + + for (const ref of uniqueRefs) { + const record = findArchivedRecordForBulkRef(archivedRecords, ref); + if (record === undefined) { + failures.push({ sessionId: ref.id, error: "Archived session not found" }); + continue; + } + + const active = this.activeForLookup({ id: record.sessionId, cwd: record.cwd }); + if (active !== undefined && this.hasActiveWork(active.runtime.session)) { + failures.push({ sessionId: record.sessionId, error: "Stop current session activity before deleting archived session" }); + continue; + } + planItems.push({ record }); + } + + const readyRecords: ArchivedSessionRecord[] = []; + for (const item of planItems) { + try { + await this.closeActive(item.record.sessionId); + readyRecords.push(item.record); + } catch (error: unknown) { + failures.push({ sessionId: item.record.sessionId, error: errorMessage(error) }); + } + } + + const moveFailures = await this.moveLegacyArchivedRecordsForDelete(readyRecords); + failures.push(...moveFailures); + const moveFailureIds = new Set(moveFailures.map((failure) => failure.sessionId)); + const deleteIds = readyRecords + .map((record) => record.sessionId) + .filter((sessionId) => !moveFailureIds.has(sessionId)); + + let deletedSessionIds: string[] = []; + try { + deletedSessionIds = await this.archiveStoreDeleteArchivedMany(deleteIds); + } catch (error: unknown) { + for (const sessionId of deleteIds) failures.push({ sessionId, error: errorMessage(error) }); + } + + return { + deleted: true, + deletedSessionIds, + failures, + generatedAt: new Date().toISOString(), + }; + } + async reload(ref: PiSessionLookup): Promise { await this.assertWritable(ref); const session = await this.getOrOpen(ref); @@ -1179,6 +1319,71 @@ export class PiSessionService { }); } + private async bulkSessionLookupContext(refs: readonly SessionBulkMutationRef[]): Promise { + const cwdSet = new Set(); + let needsAllSessions = false; + for (const ref of refs) { + if (ref.cwd === undefined) needsAllSessions = true; + else cwdSet.add(ref.cwd); + } + + const [sessionsByCwd, allSessions] = await Promise.all([ + this.listSessionsByCwd([...cwdSet]), + needsAllSessions ? this.sessionManager.listAll?.() ?? Promise.resolve([]) : Promise.resolve(undefined), + ]); + return allSessions === undefined ? { sessionsByCwd } : { sessionsByCwd, allSessions }; + } + + private async listSessionsByCwd(cwds: readonly string[]): Promise> { + const uniqueCwds = uniqueStrings(cwds); + const entries = await Promise.all(uniqueCwds.map(async (cwd) => [cwd, await this.sessionManager.list(cwd)] as const)); + return new Map(entries); + } + + private async archiveStoreArchiveMany(inputs: readonly ArchiveSessionInput[]): Promise { + if (inputs.length === 0) return []; + if (this.archiveStore.archiveMany !== undefined) return this.archiveStore.archiveMany(inputs); + const records: ArchivedSessionRecord[] = []; + for (const input of inputs) records.push(await this.archiveStore.archive(input)); + return records; + } + + private async archiveStoreDeleteArchivedMany(sessionIds: readonly string[]): Promise { + if (sessionIds.length === 0) return []; + if (this.archiveStore.deleteArchivedMany !== undefined) return this.archiveStore.deleteArchivedMany(sessionIds); + if (this.archiveStore.deleteArchived === undefined) throw new Error("Archive store does not support deletion"); + for (const sessionId of sessionIds) await this.archiveStore.deleteArchived(sessionId); + return [...sessionIds]; + } + + private async moveLegacyArchivedRecordsForDelete(records: readonly ArchivedSessionRecord[]): Promise { + const legacyRecords = records.filter((record) => record.archivePath === undefined); + if (legacyRecords.length === 0) return []; + + let sessionsByCwd: Map; + try { + sessionsByCwd = await this.listSessionsByCwd(legacyRecords.map((record) => record.cwd)); + } catch (error: unknown) { + return legacyRecords.map((record) => ({ sessionId: record.sessionId, error: errorMessage(error) })); + } + + const moveInputs = legacyRecords + .map((record) => findSessionByIdOrPrefix(sessionsByCwd.get(record.cwd) ?? [], record.sessionId)) + .filter(isDefined) + .map(archiveInputFromListEntry); + if (moveInputs.length === 0) return []; + + try { + await this.archiveStoreArchiveMany(moveInputs); + return []; + } catch (error: unknown) { + const failedIds = new Set(moveInputs.map((input) => input.sessionId)); + return legacyRecords + .filter((record) => failedIds.has(record.sessionId)) + .map((record) => ({ sessionId: record.sessionId, error: errorMessage(error) })); + } + } + private async cleanupPlan(request: NormalizedSessionCleanupRequest) { const [sessions, archivedRecords] = await Promise.all([this.sessionManager.listAll?.() ?? [], this.archiveStore.list()]); return planSessionCleanup({ @@ -1224,7 +1429,20 @@ export class PiSessionService { private async ensureArchivedRecordMoved(record: ArchivedSessionRecord): Promise { const session = (await this.sessionManager.list(record.cwd)).find((candidate) => candidate.id === record.sessionId); if (session === undefined) return record; - return this.archiveStore.archive(archiveInputFromListEntry(session)); + const [moved] = await this.archiveStoreArchiveMany([archiveInputFromListEntry(session)]); + return moved ?? record; + } + + private async ensureArchivedRecordsMoved(records: readonly ArchivedSessionRecord[]): Promise { + const legacyRecords = records.filter((record) => record.archivePath === undefined); + if (legacyRecords.length === 0) return; + + const sessionsByCwd = await this.listSessionsByCwd(legacyRecords.map((record) => record.cwd)); + const moveInputs = legacyRecords + .map((record) => sessionsByCwd.get(record.cwd)?.find((candidate) => candidate.id === record.sessionId)) + .filter(isDefined) + .map(archiveInputFromListEntry); + await this.archiveStoreArchiveMany(moveInputs); } private async archiveInputForSession(session: PiAgentSession): Promise { @@ -1648,6 +1866,43 @@ function previewResponseFromPlan(plan: SessionCleanupPlan): ClientSessionCleanup }; } +function uniqueBulkSessionRefs(refs: readonly SessionBulkMutationRef[]): SessionBulkMutationRef[] { + const seen = new Set(); + const unique: SessionBulkMutationRef[] = []; + for (const ref of refs) { + const key = `${ref.cwd ?? ""}\0${ref.id}`; + if (seen.has(key)) continue; + seen.add(key); + unique.push(ref); + } + return unique; +} + +function bulkRefToLookup(ref: SessionBulkMutationRef): PiSessionLookup { + return ref.cwd === undefined ? ref.id : { id: ref.id, cwd: ref.cwd }; +} + +function findArchivedRecordForBulkRef(records: readonly ArchivedSessionRecord[], ref: SessionBulkMutationRef): ArchivedSessionRecord | undefined { + return records.find((record) => (ref.cwd === undefined || record.cwd === ref.cwd) && (record.sessionId === ref.id || record.sessionId.startsWith(ref.id))); +} + +function findListedSessionForBulkRef(context: BulkSessionLookupContext, ref: SessionBulkMutationRef): PiSessionListEntry | undefined { + if (ref.cwd !== undefined) return findSessionByIdOrPrefix(context.sessionsByCwd.get(ref.cwd) ?? [], ref.id); + return context.allSessions === undefined ? undefined : findSessionByIdOrPrefix(context.allSessions, ref.id); +} + +function findSessionByIdOrPrefix(sessions: readonly PiSessionListEntry[], sessionId: string): PiSessionListEntry | undefined { + return sessions.find((session) => session.id === sessionId) ?? sessions.find((session) => session.id.startsWith(sessionId)); +} + +function uniqueStrings(values: readonly string[]): string[] { + return [...new Set(values)]; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + function modelToClientModel(model: PiAgentSession["model"]): ClientSessionModel { if (model === undefined) return {}; const name = getString(model, "name"); diff --git a/src/server/sessions/sessionArchiveStore.test.ts b/src/server/sessions/sessionArchiveStore.test.ts index ac441fa..d08186f 100644 --- a/src/server/sessions/sessionArchiveStore.test.ts +++ b/src/server/sessions/sessionArchiveStore.test.ts @@ -71,6 +71,53 @@ describe("SessionArchiveStore", () => { expect(await exists(record.archivePath)).toBe(false); await expect(store.list()).resolves.toEqual([]); }); + + it("archives and permanently deletes sessions in batches", async () => { + const root = await mkdtemp(join(tmpdir(), "pi-web-archive-batch-")); + tempRoots.push(root); + const activeDir = join(root, "active"); + await mkdir(activeDir, { recursive: true }); + const sourceA = join(activeDir, "2026-01-01_a.jsonl"); + const sourceB = join(activeDir, "2026-01-01_b.jsonl"); + await writeFile(sourceA, "a\n", "utf8"); + await writeFile(sourceB, "b\n", "utf8"); + + const store = new SessionArchiveStore(join(root, "archived-sessions.json"), join(root, "archived-files")); + const records = await store.archiveMany([ + { + sessionId: "a", + cwd: "/workspace", + path: sourceA, + created: "2026-01-01T00:00:00.000Z", + modified: "2026-01-01T00:01:00.000Z", + messageCount: 1, + firstMessage: "a", + }, + { + sessionId: "b", + cwd: "/workspace", + path: sourceB, + created: "2026-01-01T00:00:00.000Z", + modified: "2026-01-01T00:02:00.000Z", + messageCount: 2, + firstMessage: "b", + }, + ]); + + expect(records.map((record) => record.sessionId)).toEqual(["a", "b"]); + expect(await exists(sourceA)).toBe(false); + expect(await exists(sourceB)).toBe(false); + await expect(store.list()).resolves.toMatchObject([{ sessionId: "a" }, { sessionId: "b" }]); + + const archivePaths = records.map((record) => record.archivePath); + if (archivePaths.some((path) => path === undefined)) throw new Error("Expected archive paths"); + await expect(store.deleteArchivedMany(["a", "b", "missing"])).resolves.toEqual(["a", "b"]); + for (const archivePath of archivePaths) { + if (archivePath === undefined) throw new Error("Expected archive path"); + expect(await exists(archivePath)).toBe(false); + } + await expect(store.list()).resolves.toEqual([]); + }); }); async function exists(path: string): Promise { diff --git a/src/server/sessions/sessionArchiveStore.ts b/src/server/sessions/sessionArchiveStore.ts index cb55715..cefbb0f 100644 --- a/src/server/sessions/sessionArchiveStore.ts +++ b/src/server/sessions/sessionArchiveStore.ts @@ -53,24 +53,39 @@ export class SessionArchiveStore { } async archive(session: ArchiveSessionInput): Promise { + const [record] = await this.archiveMany([session]); + if (record === undefined) throw new Error("Archive operation did not produce a record"); + return record; + } + + async archiveMany(sessions: readonly ArchiveSessionInput[]): Promise { + if (sessions.length === 0) return []; return this.exclusive(async () => { const data = await this.read(); - const existingIndex = data.sessions.findIndex((record) => record.sessionId === session.sessionId); - const existing = existingIndex === -1 ? undefined : data.sessions[existingIndex]; - const archivePath = existing?.archivePath ?? this.archivePathFor(session); - const record = archiveRecordFromInput(session, { - archivedAt: existing?.archivedAt ?? new Date().toISOString(), - originalPath: existing?.originalPath ?? session.path, - archivePath, - }); + const records: ArchivedSessionRecord[] = []; + const filesToRemove: { source: string; archivePath: string }[] = []; - await copySessionFileToArchive(session.path, archivePath); + for (const session of sessions) { + const existingIndex = data.sessions.findIndex((record) => record.sessionId === session.sessionId); + const existing = existingIndex === -1 ? undefined : data.sessions[existingIndex]; + const archivePath = existing?.archivePath ?? this.archivePathFor(session); + const record = archiveRecordFromInput(session, { + archivedAt: existing?.archivedAt ?? new Date().toISOString(), + originalPath: existing?.originalPath ?? session.path, + archivePath, + }); + + await copySessionFileToArchive(session.path, archivePath); + + if (existingIndex === -1) data.sessions.push(record); + else data.sessions[existingIndex] = record; + records.push(record); + filesToRemove.push({ source: session.path, archivePath }); + } - if (existingIndex === -1) data.sessions.push(record); - else data.sessions[existingIndex] = record; await this.write(data); - await removeActiveSessionFile(session.path, archivePath); - return record; + for (const file of filesToRemove) await removeActiveSessionFile(file.source, file.archivePath); + return records; }); } @@ -90,14 +105,25 @@ export class SessionArchiveStore { } async deleteArchived(sessionId: string): Promise { - await this.exclusive(async () => { - const data = await this.read(); - const record = data.sessions.find((session) => session.sessionId === sessionId); - if (record === undefined) return; + await this.deleteArchivedMany([sessionId]); + } - if (record.archivePath !== undefined && await pathExists(record.archivePath)) await unlink(record.archivePath); - const sessions = data.sessions.filter((session) => session.sessionId !== sessionId); + async deleteArchivedMany(sessionIds: readonly string[]): Promise { + const targetIds = uniqueStrings(sessionIds); + if (targetIds.length === 0) return []; + return this.exclusive(async () => { + const data = await this.read(); + const targetIdSet = new Set(targetIds); + const records = data.sessions.filter((session) => targetIdSet.has(session.sessionId)); + if (records.length === 0) return []; + + for (const record of records) { + if (record.archivePath !== undefined && await pathExists(record.archivePath)) await unlink(record.archivePath); + } + const sessions = data.sessions.filter((session) => !targetIdSet.has(session.sessionId)); await this.write({ sessions }); + const deletedIds = new Set(records.map((record) => record.sessionId)); + return targetIds.filter((sessionId) => deletedIds.has(sessionId)); }); } @@ -254,6 +280,10 @@ function safeFileName(value: string): string { return value.replace(/[^a-zA-Z0-9._-]/g, "_") || "session"; } +function uniqueStrings(values: readonly string[]): string[] { + return [...new Set(values)]; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } diff --git a/src/server/sessions/sessionRoutes.test.ts b/src/server/sessions/sessionRoutes.test.ts index 2f2e730..abef639 100644 --- a/src/server/sessions/sessionRoutes.test.ts +++ b/src/server/sessions/sessionRoutes.test.ts @@ -2,7 +2,7 @@ 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 { SessionCleanupExecuteResponse, SessionCleanupPreviewResponse } from "../../shared/apiTypes.js"; +import type { SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkMutationRef, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse } from "../../shared/apiTypes.js"; import { SessionEventHub } from "../realtime/sessionEventHub.js"; import { PiSessionService, type PiSessionManagerGateway, type PiSessionRef } from "./piSessionService.js"; import { registerSessionRoutes } from "./sessionRoutes.js"; @@ -178,6 +178,49 @@ describe("session routes", () => { await routeApp.close(); } }); + + it("routes bulk archive and delete requests with normalized session refs", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(eventHub); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const requestCwd = resolve("/repo"); + const archiveResponse = await routeApp.inject({ method: "POST", url: "/sessions/bulk/archive", payload: { sessions: [{ id: "s1", cwd: requestCwd }, { id: "s2" }] } }); + const deleteResponse = await routeApp.inject({ method: "POST", url: "/sessions/bulk/delete-archived", payload: { sessions: [{ id: "s1", cwd: requestCwd }] } }); + + expect(archiveResponse.statusCode).toBe(200); + expect(archiveResponse.json()).toMatchObject({ archived: true, archivedSessionIds: ["s1", "s2"], failures: [] }); + expect(deleteResponse.statusCode).toBe(200); + expect(deleteResponse.json()).toMatchObject({ deleted: true, deletedSessionIds: ["s1"], failures: [] }); + expect(routeService.bulkArchiveCalls).toEqual([[{ id: "s1", cwd: requestCwd }, { id: "s2" }]]); + expect(routeService.bulkDeleteCalls).toEqual([[{ id: "s1", cwd: requestCwd }]]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + + it("rejects malformed bulk mutation bodies before calling the service", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(eventHub); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const response = await routeApp.inject({ method: "POST", url: "/sessions/bulk/archive", payload: { sessions: [{ cwd: "/repo" }] } }); + + expect(response.statusCode).toBe(400); + expect(response.json()).toEqual({ error: "id field must be a string" }); + expect(routeService.bulkArchiveCalls).toEqual([]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); }); class CapturingRouteSessionService extends PiSessionService { @@ -185,6 +228,8 @@ class CapturingRouteSessionService extends PiSessionService { readonly reloadCalls: (string | PiSessionRef)[] = []; readonly cleanupPreviewCalls: NormalizedSessionCleanupRequest[] = []; readonly cleanupCalls: NormalizedSessionCleanupRequest[] = []; + readonly bulkArchiveCalls: SessionBulkMutationRef[][] = []; + readonly bulkDeleteCalls: SessionBulkMutationRef[][] = []; reloadError: Error | undefined; constructor(eventHub: SessionEventHub) { @@ -201,6 +246,16 @@ class CapturingRouteSessionService extends PiSessionService { return Promise.resolve({ generatedAt: "2026-06-25T00:00:00.000Z", thresholds: request.thresholds, projects: [], totals: { archiveCount: 0, deleteCount: 0 }, archivedSessionIds: [], deletedSessionIds: [] }); } + override archiveMany(refs: readonly SessionBulkMutationRef[]): Promise { + this.bulkArchiveCalls.push([...refs]); + return Promise.resolve({ archived: true, archivedSessionIds: refs.map((ref) => ref.id), failures: [], generatedAt: "2026-06-25T00:00:00.000Z" }); + } + + override deleteArchivedMany(refs: readonly SessionBulkMutationRef[]): Promise { + this.bulkDeleteCalls.push([...refs]); + return Promise.resolve({ deleted: true, deletedSessionIds: refs.map((ref) => ref.id), failures: [], generatedAt: "2026-06-25T00:00:00.000Z" }); + } + override reload(lookup: string | PiSessionRef): Promise { this.reloadCalls.push(lookup); if (this.reloadError !== undefined) return Promise.reject(this.reloadError); diff --git a/src/server/sessions/sessionRoutes.ts b/src/server/sessions/sessionRoutes.ts index 9e0c8a2..3aeef8a 100644 --- a/src/server/sessions/sessionRoutes.ts +++ b/src/server/sessions/sessionRoutes.ts @@ -1,5 +1,5 @@ import type { FastifyInstance } from "fastify"; -import type { SessionCleanupRequest } from "../../shared/apiTypes.js"; +import type { SessionBulkMutationRequest, SessionBulkMutationRef, SessionCleanupRequest } from "../../shared/apiTypes.js"; import { normalizeRequestCwd } from "../workingDirectory.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; import type { PiSessionRef, PiSessionService } from "./piSessionService.js"; @@ -64,6 +64,22 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS } }); + app.post<{ Body: SessionBulkMutationRequest | undefined }>(`${prefix}/sessions/bulk/archive`, async (request, reply) => { + try { + return await sessions.archiveMany(bulkMutationRefsFromBody(request.body)); + } catch (error) { + return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) }); + } + }); + + app.post<{ Body: SessionBulkMutationRequest | undefined }>(`${prefix}/sessions/bulk/delete-archived`, async (request, reply) => { + try { + return await sessions.deleteArchivedMany(bulkMutationRefsFromBody(request.body)); + } catch (error) { + return reply.code(mutationErrorStatus(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)) }; @@ -281,6 +297,23 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS }); } +function bulkMutationRefsFromBody(body: SessionBulkMutationRequest | undefined): SessionBulkMutationRef[] { + const record = requireRecord(body); + const sessions = record["sessions"]; + if (!Array.isArray(sessions)) throw new Error("sessions field must be an array"); + return sessions.map(parseBulkMutationRef); +} + +function parseBulkMutationRef(value: unknown): SessionBulkMutationRef { + const record = requireRecord(value); + const id = requireString(record, "id").trim(); + if (id === "") throw new Error("id field must not be empty"); + const cwd = record["cwd"]; + if (cwd === undefined || cwd === "") return { id }; + if (typeof cwd !== "string") throw new Error("cwd field must be a string"); + return { id, cwd: normalizeRequestCwd(cwd) }; +} + function sessionLookupFromQuery(id: string, query: SessionQuery): SessionLookup { return sessionLookupFromCwd(id, query.cwd); } diff --git a/src/shared/apiTypes.ts b/src/shared/apiTypes.ts index cf028dc..093504b 100644 --- a/src/shared/apiTypes.ts +++ b/src/shared/apiTypes.ts @@ -3,6 +3,7 @@ export type MachineStatus = "unknown" | "online" | "offline" | "error"; export const PI_WEB_CAPABILITIES = { sessionsDeleteArchived: "sessions.deleteArchived", + sessionsBulkMutations: "sessions.bulkMutations", sessionsCleanup: "sessions.cleanup", sessionsReload: "sessions.reload", promptAttachments: "prompt.attachments", @@ -163,6 +164,34 @@ export interface ArchiveSessionsResponse { skippedAlreadyArchivedCount?: number; } +export interface SessionBulkMutationRef { + id: string; + cwd?: string; +} + +export interface SessionBulkMutationRequest { + sessions: SessionBulkMutationRef[]; +} + +export interface SessionBulkFailure { + sessionId: string; + error: string; +} + +export interface SessionBulkArchiveResponse { + archived: true; + archivedSessionIds: string[]; + failures: SessionBulkFailure[]; + generatedAt: string; +} + +export interface SessionBulkDeleteArchivedResponse { + deleted: true; + deletedSessionIds: string[]; + failures: SessionBulkFailure[]; + generatedAt: string; +} + export interface SessionCleanupRequest { /** Archive non-archived sessions whose modified time is older than this many days. Omit/null to disable. */ archiveIdleDays?: number | null; diff --git a/src/shared/capabilities.ts b/src/shared/capabilities.ts index 87df6f4..11b83d6 100644 --- a/src/shared/capabilities.ts +++ b/src/shared/capabilities.ts @@ -6,11 +6,12 @@ export type { PiWebCapability }; export const KNOWN_PI_WEB_CAPABILITIES = Object.values(PI_WEB_CAPABILITIES); const knownPiWebCapabilities: ReadonlySet = new Set(KNOWN_PI_WEB_CAPABILITIES); -export const WEB_RUNTIME_CAPABILITIES = [PI_WEB_CAPABILITIES.sessionsDeleteArchived, PI_WEB_CAPABILITIES.sessionsCleanup, PI_WEB_CAPABILITIES.sessionsReload, PI_WEB_CAPABILITIES.promptAttachments, PI_WEB_CAPABILITIES.workspaceFileSuggestions] as const satisfies readonly PiWebCapability[]; -export const SESSIOND_RUNTIME_CAPABILITIES = [PI_WEB_CAPABILITIES.sessionsDeleteArchived, PI_WEB_CAPABILITIES.sessionsCleanup, PI_WEB_CAPABILITIES.sessionsReload, PI_WEB_CAPABILITIES.promptAttachments] as const satisfies readonly PiWebCapability[]; +export const WEB_RUNTIME_CAPABILITIES = [PI_WEB_CAPABILITIES.sessionsDeleteArchived, PI_WEB_CAPABILITIES.sessionsBulkMutations, PI_WEB_CAPABILITIES.sessionsCleanup, PI_WEB_CAPABILITIES.sessionsReload, PI_WEB_CAPABILITIES.promptAttachments, PI_WEB_CAPABILITIES.workspaceFileSuggestions] as const satisfies readonly PiWebCapability[]; +export const SESSIOND_RUNTIME_CAPABILITIES = [PI_WEB_CAPABILITIES.sessionsDeleteArchived, PI_WEB_CAPABILITIES.sessionsBulkMutations, PI_WEB_CAPABILITIES.sessionsCleanup, PI_WEB_CAPABILITIES.sessionsReload, PI_WEB_CAPABILITIES.promptAttachments] as const satisfies readonly PiWebCapability[]; const EFFECTIVE_CAPABILITY_REQUIREMENTS = { [PI_WEB_CAPABILITIES.sessionsDeleteArchived]: ["web", "sessiond"], + [PI_WEB_CAPABILITIES.sessionsBulkMutations]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.sessionsCleanup]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.sessionsReload]: ["web", "sessiond"], [PI_WEB_CAPABILITIES.promptAttachments]: ["web", "sessiond"], diff --git a/src/shared/federatedRoutes.ts b/src/shared/federatedRoutes.ts index a78fc16..8efdc92 100644 --- a/src/shared/federatedRoutes.ts +++ b/src/shared/federatedRoutes.ts @@ -37,6 +37,8 @@ export const FEDERATED_HTTP_ROUTES = [ { method: "POST", path: "/sessions" }, { 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/status" }, { method: "GET", path: "/sessions/:sessionId/models" },