Archived
feat: add bulk session mutations
This commit is contained in:
Federico Jaramillo Martinez
@@ -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.
|
||||
@@ -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";
|
||||
|
||||
@@ -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 });
|
||||
|
||||
|
||||
@@ -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<string, unknown> = {
|
||||
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),
|
||||
|
||||
@@ -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)),
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<string, unknown> {
|
||||
@@ -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 {
|
||||
|
||||
@@ -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" };
|
||||
|
||||
@@ -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<void> {
|
||||
@@ -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<BulkSessionMutationResult> {
|
||||
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<BulkSessionMutationResult> {
|
||||
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<void> {
|
||||
@@ -581,8 +616,7 @@ export class SessionController {
|
||||
}
|
||||
}
|
||||
|
||||
private applyBulkSessionError(action: string, results: readonly PromiseSettledResult<string>[]): 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<T>(results: readonly PromiseSettledResult<T>[]): T[] {
|
||||
return results.filter(isFulfilled).map((result) => result.value);
|
||||
}
|
||||
|
||||
function rejectedReasons(results: readonly PromiseSettledResult<unknown>[]): 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<unknown>[]): string[] {
|
||||
return results.flatMap((result, index) => {
|
||||
if (!isRejected(result)) return [];
|
||||
const sessionId = sessions[index]?.id ?? "unknown";
|
||||
return [`${sessionId}: ${errorMessage(result.reason)}`];
|
||||
});
|
||||
}
|
||||
|
||||
async function allSettledWithConcurrency<T, R>(items: readonly T[], concurrency: number, worker: (item: T) => Promise<R>): Promise<PromiseSettledResult<R>[]> {
|
||||
const indexedItems = items.map((item, index) => ({ item, index }));
|
||||
const results: PromiseSettledResult<R>[] = [];
|
||||
let nextIndex = 0;
|
||||
|
||||
async function runWorker(): Promise<void> {
|
||||
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<T>(result: PromiseSettledResult<T>): result is PromiseFulfilledResult<T> {
|
||||
|
||||
@@ -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[] = [];
|
||||
|
||||
@@ -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<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived"> & { deleteArchived?: (sessionId: string) => Promise<void> };
|
||||
type SessionArchiveRepository = Pick<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived"> & {
|
||||
archiveMany?: (sessions: readonly ArchiveSessionInput[]) => Promise<ArchivedSessionRecord[]>;
|
||||
deleteArchived?: (sessionId: string) => Promise<void>;
|
||||
deleteArchivedMany?: (sessionIds: readonly string[]) => Promise<string[]>;
|
||||
};
|
||||
|
||||
export type PiSessionRef = ClientSessionRef;
|
||||
|
||||
@@ -143,6 +147,19 @@ interface WorkspaceArchiveCandidate extends SessionArchiveTreeCandidate {
|
||||
activeSession?: PiAgentSession;
|
||||
}
|
||||
|
||||
interface BulkSessionLookupContext {
|
||||
sessionsByCwd: Map<string, PiSessionListEntry[]>;
|
||||
allSessions?: readonly PiSessionListEntry[];
|
||||
}
|
||||
|
||||
interface BulkArchivePlanItem {
|
||||
input: ArchiveSessionInput;
|
||||
}
|
||||
|
||||
interface BulkDeletePlanItem {
|
||||
record: ArchivedSessionRecord;
|
||||
}
|
||||
|
||||
type AgentModel = NonNullable<SpawnSessionInvocation["model"]>;
|
||||
type ModelRegistryInstance = ReturnType<typeof ModelRegistry.create>;
|
||||
|
||||
@@ -421,10 +438,12 @@ export class PiSessionService {
|
||||
|
||||
async cleanup(request: NormalizedSessionCleanupRequest): Promise<ClientSessionCleanupExecuteResponse> {
|
||||
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<SessionBulkArchiveResponse> {
|
||||
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<ClientArchiveSessionsResponse> {
|
||||
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<SessionBulkDeleteArchivedResponse> {
|
||||
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<void> {
|
||||
await this.assertWritable(ref);
|
||||
const session = await this.getOrOpen(ref);
|
||||
@@ -1179,6 +1319,71 @@ export class PiSessionService {
|
||||
});
|
||||
}
|
||||
|
||||
private async bulkSessionLookupContext(refs: readonly SessionBulkMutationRef[]): Promise<BulkSessionLookupContext> {
|
||||
const cwdSet = new Set<string>();
|
||||
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<Map<string, PiSessionListEntry[]>> {
|
||||
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<ArchivedSessionRecord[]> {
|
||||
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<string[]> {
|
||||
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<SessionBulkFailure[]> {
|
||||
const legacyRecords = records.filter((record) => record.archivePath === undefined);
|
||||
if (legacyRecords.length === 0) return [];
|
||||
|
||||
let sessionsByCwd: Map<string, PiSessionListEntry[]>;
|
||||
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<ArchivedSessionRecord> {
|
||||
const session = (await this.sessionManager.list(record.cwd)).find((candidate) => candidate.id === record.sessionId);
|
||||
if (session === undefined) return record;
|
||||
return this.archiveStore.archive(archiveInputFromListEntry(session));
|
||||
const [moved] = await this.archiveStoreArchiveMany([archiveInputFromListEntry(session)]);
|
||||
return moved ?? record;
|
||||
}
|
||||
|
||||
private async ensureArchivedRecordsMoved(records: readonly ArchivedSessionRecord[]): Promise<void> {
|
||||
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<ArchiveSessionInput> {
|
||||
@@ -1648,6 +1866,43 @@ function previewResponseFromPlan(plan: SessionCleanupPlan): ClientSessionCleanup
|
||||
};
|
||||
}
|
||||
|
||||
function uniqueBulkSessionRefs(refs: readonly SessionBulkMutationRef[]): SessionBulkMutationRef[] {
|
||||
const seen = new Set<string>();
|
||||
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");
|
||||
|
||||
@@ -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<boolean> {
|
||||
|
||||
@@ -53,24 +53,39 @@ export class SessionArchiveStore {
|
||||
}
|
||||
|
||||
async archive(session: ArchiveSessionInput): Promise<ArchivedSessionRecord> {
|
||||
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<ArchivedSessionRecord[]> {
|
||||
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<void> {
|
||||
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<string[]> {
|
||||
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<string, unknown> {
|
||||
return typeof value === "object" && value !== null;
|
||||
}
|
||||
|
||||
@@ -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<SessionBulkArchiveResponse> {
|
||||
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<SessionBulkDeleteArchivedResponse> {
|
||||
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<void> {
|
||||
this.reloadCalls.push(lookup);
|
||||
if (this.reloadError !== undefined) return Promise.reject(this.reloadError);
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -6,11 +6,12 @@ export type { PiWebCapability };
|
||||
export const KNOWN_PI_WEB_CAPABILITIES = Object.values(PI_WEB_CAPABILITIES);
|
||||
const knownPiWebCapabilities: ReadonlySet<string> = 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"],
|
||||
|
||||
@@ -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" },
|
||||
|
||||
Reference in New Issue
Block a user