From 0ef189f3c8455987b29d9119e71ebb715c237a2c Mon Sep 17 00:00:00 2001 From: Iuliia Ivashko Date: Fri, 4 Sep 2026 16:50:03 +0300 Subject: [PATCH] perf(sessions): archive a worktree's sessions through one server batch Removing a worktree archived its sessions one SDK call at a time and then re-rendered the whole sidebar once per streamed session.updated echo. On a worktree with 121 sessions that meant 14.8s of main-thread work, 121 requests, and 328 localStorage writes. - Add POST /api/openchamber/sessions/archive: validates a batch (max 500 ids, per-request archivedAt), archives sequentially, and reports partial failures instead of dropping the batch. VS Code serves no such route and answers 501; the shared UI then falls back to the per-session path. - Plan batches from the sessions this client actually holds, live directory stores first, so worktree-only sessions still batch. - Claim (id, archivedAt) pairs before the request and consume the matching session.updated echoes, so the server's own confirmations no longer fan out into 121 store publications. Runtime-scoped, TTL 30s, released on response or fallback; non-matching updates pass. - Make the managed-chats persistence a real trailing debounce instead of a 50ms throttle, so a burst of publications coalesces into one localStorage write. Benchmark (121 sessions, production build, real Chrome): 14785ms -> ~1030ms, long tasks 100 -> 1, global store publications 236 -> 1, persistence writes 328 -> 3. --- .../sync-context-session-events.test.ts | 41 +++- packages/ui/src/sync/bulk-archive-echo.ts | 49 ++++ packages/ui/src/sync/persist-cache.test.ts | 10 + packages/ui/src/sync/persist-cache.ts | 2 +- packages/ui/src/sync/session-actions.test.ts | 186 ++++++++++++++- packages/ui/src/sync/session-actions.ts | 213 +++++++++++++++++- packages/ui/src/sync/session-archive-batch.ts | 77 +++++++ packages/ui/src/sync/sync-context.tsx | 8 +- packages/vscode/webview/main.tsx | 8 + .../server/lib/openchamber-sessions/routes.js | 102 +++++++++ .../lib/openchamber-sessions/routes.test.js | 88 ++++++++ 11 files changed, 769 insertions(+), 15 deletions(-) create mode 100644 packages/ui/src/sync/bulk-archive-echo.ts create mode 100644 packages/ui/src/sync/session-archive-batch.ts diff --git a/packages/ui/src/sync/__tests__/sync-context-session-events.test.ts b/packages/ui/src/sync/__tests__/sync-context-session-events.test.ts index d846e868..5a42c6c5 100644 --- a/packages/ui/src/sync/__tests__/sync-context-session-events.test.ts +++ b/packages/ui/src/sync/__tests__/sync-context-session-events.test.ts @@ -45,7 +45,15 @@ mock.module("@/lib/runtime-switch", () => ({ return () => undefined }, })) -import { applySessionEventsToGlobalSessions, applySessionEventToGlobalSessions } from "../session-event-router" +import { + applySessionEventsToGlobalSessions, + applySessionEventToGlobalSessions, +} from "../session-event-router" +import { + registerBulkArchiveEchoes, + releaseBulkArchiveEchoes, + shouldConsumeBulkArchiveEcho, +} from "../bulk-archive-echo" const buildSession = (title: string, time: Session["time"]): Session => ({ id: "ses_1", @@ -146,4 +154,35 @@ describe("applySessionEventToGlobalSessions", () => { expect(mutationCalls).toBe(1) expect(upsertedSessions).toHaveLength(1_000) }) + + test("consumes only the matching bulk archive echo", () => { + registerBulkArchiveEchoes(runtimeKey, [{ id: "ses_1", archivedAt: 20 }], 100) + + expect(shouldConsumeBulkArchiveEcho(buildEvent(buildSession("Initial", { + created: 1, + updated: 20, + archived: 20, + })), runtimeKey, 101)).toBe(true) + expect(shouldConsumeBulkArchiveEcho(buildEvent(buildSession("Initial", { + created: 1, + updated: 21, + archived: 21, + })), runtimeKey, 101)).toBe(false) + expect(shouldConsumeBulkArchiveEcho(buildEvent(buildSession("Initial", { + created: 1, + updated: 20, + archived: 20, + })), "runtime-b", 101)).toBe(false) + }) + + test("does not consume an expired or released bulk archive echo", () => { + registerBulkArchiveEchoes(runtimeKey, [{ id: "ses_1", archivedAt: 20 }], 100) + const event = buildEvent(buildSession("Initial", { created: 1, updated: 20, archived: 20 })) + + expect(shouldConsumeBulkArchiveEcho(event, runtimeKey, 30_101)).toBe(false) + + registerBulkArchiveEchoes(runtimeKey, [{ id: "ses_1", archivedAt: 20 }], 100) + releaseBulkArchiveEchoes(runtimeKey, ["ses_1"]) + expect(shouldConsumeBulkArchiveEcho(event, runtimeKey, 101)).toBe(false) + }) }) diff --git a/packages/ui/src/sync/bulk-archive-echo.ts b/packages/ui/src/sync/bulk-archive-echo.ts new file mode 100644 index 00000000..0e9f469e --- /dev/null +++ b/packages/ui/src/sync/bulk-archive-echo.ts @@ -0,0 +1,49 @@ +import type { Event } from "@opencode-ai/sdk/v2/client" +import { subscribeRuntimeEndpointWillChange } from "@/lib/runtime-switch" + +const BULK_ARCHIVE_ECHO_TTL_MS = 30_000 +const pendingEchoes = new Map>() + +subscribeRuntimeEndpointWillChange(() => pendingEchoes.clear()) + +export const registerBulkArchiveEchoes = ( + runtimeKey: string, + sessions: Iterable<{ id: string; archivedAt: number }>, + now = Date.now(), +): void => { + let runtimeEchoes = pendingEchoes.get(runtimeKey) + if (!runtimeEchoes) { + runtimeEchoes = new Map() + pendingEchoes.set(runtimeKey, runtimeEchoes) + } + for (const session of sessions) { + runtimeEchoes.set(session.id, { + archivedAt: session.archivedAt, + expiresAt: now + BULK_ARCHIVE_ECHO_TTL_MS, + }) + } +} + +export const releaseBulkArchiveEchoes = (runtimeKey: string, sessionIds: Iterable): void => { + const runtimeEchoes = pendingEchoes.get(runtimeKey) + if (!runtimeEchoes) return + for (const sessionId of sessionIds) runtimeEchoes.delete(sessionId) + if (runtimeEchoes.size === 0) pendingEchoes.delete(runtimeKey) +} + +export const shouldConsumeBulkArchiveEcho = ( + event: Event, + runtimeKey: string, + now = Date.now(), +): boolean => { + if (event.type !== "session.updated") return false + const runtimeEchoes = pendingEchoes.get(runtimeKey) + const expected = runtimeEchoes?.get(event.properties.info.id) + if (!expected) return false + if (expected.expiresAt < now) { + runtimeEchoes?.delete(event.properties.info.id) + if (runtimeEchoes?.size === 0) pendingEchoes.delete(runtimeKey) + return false + } + return event.properties.info.time.archived === expected.archivedAt +} diff --git a/packages/ui/src/sync/persist-cache.test.ts b/packages/ui/src/sync/persist-cache.test.ts index 2afc3889..95427e33 100644 --- a/packages/ui/src/sync/persist-cache.test.ts +++ b/packages/ui/src/sync/persist-cache.test.ts @@ -92,6 +92,16 @@ describe("persisted directory sessions", () => { expect(readManagedChatSessions()).toEqual([]) }) + test("coalesces a continuing burst into one trailing session write", async () => { + persistSessions(directory, [session(1, 1)]) + await new Promise((resolve) => setTimeout(resolve, 30)) + persistSessions(directory, [session(1, 2)]) + await waitForPersistence() + + expect(storage.writes).toBe(1) + expect(readDirCache(directory).sessions?.[0]?.time.updated).toBe(2) + }) + test("keeps the 50 most recently updated sessions across restart reads", async () => { const sessions = Array.from({ length: 60 }, (_, updated) => session(59 - updated, updated)) diff --git a/packages/ui/src/sync/persist-cache.ts b/packages/ui/src/sync/persist-cache.ts index 51d3e580..f2a43af5 100644 --- a/packages/ui/src/sync/persist-cache.ts +++ b/packages/ui/src/sync/persist-cache.ts @@ -177,7 +177,7 @@ function scheduleSessionCacheWrite(directory: string, sessions: Session[]): void if (pending.runtimeKey !== runtimeKey) pendingSessionWrites.delete(pendingKey) } pendingSessionWrites.set(key, { runtimeKey, key, legacyKey: legacyCacheKey(directory, "sessions"), sessions }) - if (pendingSessionWriteTimer !== undefined) return + if (pendingSessionWriteTimer !== undefined) clearTimeout(pendingSessionWriteTimer) pendingSessionWriteTimer = setTimeout(flushPendingSessionWrites, SESSION_PERSIST_DEBOUNCE_MS) } diff --git a/packages/ui/src/sync/session-actions.test.ts b/packages/ui/src/sync/session-actions.test.ts index d4611074..01e703a1 100644 --- a/packages/ui/src/sync/session-actions.test.ts +++ b/packages/ui/src/sync/session-actions.test.ts @@ -21,7 +21,16 @@ let sessionDeleteError: unknown | null = null let beforeSessionUpdateResolve: ((sessionId: string) => void) | null = null let beforeSessionDeleteResolve: ((sessionId: string) => void) | null = null const globalUpsertedSessions: unknown[] = [] +const globalUpsertedSessionBatches: Session[][] = [] const globalRemovedSessionIds: string[] = [] +// Sessions this client is holding. `archiveSessions` reads them to decide which +// sessions can be archived by the server in one batch. +let globalActiveSessions: Session[] = [] +const archiveBatchRequests: Array<{ directory: string; ids: string[] }> = [] +let archiveBatchResponse: { status: number; body: unknown } = { + status: 404, + body: { error: 'not found' }, +} const deletedCleanupIdentities: Array<{ runtimeKey: string; directory: string; sessionId: string }> = [] const movedSessionDirectories: Array<{ sessionID: string; directory: string }> = [] @@ -243,11 +252,15 @@ mock.module("@/stores/useGlobalSessionsStore", () => ({ }, useGlobalSessionsStore: { getState: () => ({ - activeSessions: [], + activeSessions: globalActiveSessions, archivedSessions: [], upsertSession: (session: unknown) => { globalUpsertedSessions.push(session) }, + upsertSessions: (sessions: Session[]) => { + globalUpsertedSessionBatches.push(sessions) + globalUpsertedSessions.push(...sessions) + }, removeSessions: (ids: Iterable) => { globalRemovedSessionIds.push(...ids) }, @@ -255,6 +268,18 @@ mock.module("@/stores/useGlobalSessionsStore", () => ({ }, })) +mock.module("@/lib/runtime-fetch", () => ({ + runtimeFetch: async (path: string, init?: { body?: string }) => { + const payload = JSON.parse(String(init?.body ?? "{}")) + archiveBatchRequests.push({ directory: payload.directory, ids: payload.ids }) + void path + return new Response(JSON.stringify(archiveBatchResponse.body), { + status: archiveBatchResponse.status, + headers: { "content-type": "application/json" }, + }) + }, +})) + mock.module("./session-deletion-cleanup", () => ({ cleanupPersistedSessionState: (identity: { runtimeKey: string; directory: string; sessionId: string }) => { deletedCleanupIdentities.push(identity) @@ -396,6 +421,10 @@ describe("confirmed session removal", () => { sessionUpdateResult = {} beforeSessionUpdateResolve = null beforeSessionDeleteResolve = null + globalUpsertedSessionBatches.length = 0 + globalActiveSessions = [] + archiveBatchRequests.length = 0 + archiveBatchResponse = { status: 404, body: { error: 'not found' } } }) test("does not remove live or persisted state when delete fails", async () => { @@ -636,6 +665,161 @@ describe("confirmed session removal", () => { }) }) +describe("archiving a batch through the server", () => { + const liveSession = (id: string, metadata?: Record): Session => ({ + id, + directory: "/test/project", + time: { created: 1 }, + ...(metadata ? { metadata } : {}), + } as unknown as Session) + + const archivedSession = (id: string): Session => ({ + id, + directory: "/test/project", + time: { created: 1, archived: 2 }, + } as unknown as Session) + + beforeEach(() => { + replyCalls.length = 0 + globalUpsertedSessions.length = 0 + globalUpsertedSessionBatches.length = 0 + globalActiveSessions = [] + archiveBatchRequests.length = 0 + archiveBatchResponse = { status: 404, body: { error: "not found" } } + sessionUpdateResult = {} + beforeSessionUpdateResolve = null + }) + + test("archives held sessions in one request and reconciles the stores once", async () => { + globalActiveSessions = [liveSession("session-a"), liveSession("session-b")] + archiveBatchResponse = { + status: 200, + body: { archived: [archivedSession("session-a"), archivedSession("session-b")], failedIds: [] }, + } + const source = createStore({}, { session: [liveSession("session-a"), liveSession("session-b")] }) + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + const result = await archiveSessions(["session-a", "session-b"]) + + expect(result).toEqual({ archivedIds: ["session-a", "session-b"], failedIds: [] }) + expect(archiveBatchRequests).toEqual([{ directory: "/test/project", ids: ["session-a", "session-b"] }]) + // The point of the batch: no per-session SDK call, and one store write for + // the whole set instead of one per session. + expect(replyCalls.filter((call) => call.method === "session.update")).toEqual([]) + expect(globalUpsertedSessionBatches).toHaveLength(1) + expect(source.getState().session).toEqual([]) + expect(source.getState().sessionRevision).toBe(1) + }) + + test("batches sessions held only by the live directory store", async () => { + archiveBatchResponse = { + status: 200, + body: { archived: [archivedSession("session-a")], failedIds: [] }, + } + const source = createStore({}, { session: [liveSession("session-a")] }) + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + const result = await archiveSessions(["session-a"]) + + expect(result).toEqual({ archivedIds: ["session-a"], failedIds: [] }) + expect(archiveBatchRequests).toEqual([{ directory: "/test/project", ids: ["session-a"] }]) + expect(replyCalls.filter((call) => call.method === "session.update")).toEqual([]) + }) + + test("reports the sessions the server could not archive without losing the rest", async () => { + globalActiveSessions = [liveSession("session-a"), liveSession("session-b")] + archiveBatchResponse = { + status: 200, + body: { archived: [archivedSession("session-a")], failedIds: ["session-b"] }, + } + const source = createStore({}, { session: [liveSession("session-a"), liveSession("session-b")] }) + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + const result = await archiveSessions(["session-a", "session-b"]) + + expect(result).toEqual({ archivedIds: ["session-a"], failedIds: ["session-b"] }) + expect(source.getState().session.map((item) => item.id)).toEqual(["session-b"]) + }) + + test("falls back to archiving one by one when the runtime does not serve the route", async () => { + globalActiveSessions = [liveSession("session-a"), liveSession("session-b")] + archiveBatchResponse = { status: 501, body: { error: "not supported in VS Code" } } + sessionUpdateResult = { data: archivedSession("session-a") } + const source = createStore({}, { session: [liveSession("session-a"), liveSession("session-b")] }) + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + const result = await archiveSessions(["session-a", "session-b"]) + + expect(result).toEqual({ archivedIds: ["session-a", "session-b"], failedIds: [] }) + expect(replyCalls.filter((call) => call.method === "session.update").map((call) => call.params.sessionID)) + .toEqual(["session-a", "session-b"]) + expect(source.getState().session).toEqual([]) + }) + + test("treats a malformed batch answer as unavailable instead of as an empty success", async () => { + globalActiveSessions = [liveSession("session-a")] + archiveBatchResponse = { status: 200, body: { archived: [{ title: "no id" }], failedIds: [] } } + sessionUpdateResult = { data: archivedSession("session-a") } + const source = createStore({}, { session: [liveSession("session-a")] }) + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + const result = await archiveSessions(["session-a"]) + + expect(result).toEqual({ archivedIds: ["session-a"], failedIds: [] }) + expect(replyCalls.filter((call) => call.method === "session.update").map((call) => call.params.sessionID)) + .toEqual(["session-a"]) + }) + + test("keeps review and btw sessions on the per-session path", async () => { + const review = liveSession("session-review", { openchamber: { kind: "review", originalSessionID: "session-parent" } }) + const parentWithFork = liveSession("session-parent", { openchamber: { btwSessionID: "session-fork" } }) + globalActiveSessions = [liveSession("session-plain"), review, parentWithFork] + archiveBatchResponse = { + status: 200, + body: { archived: [archivedSession("session-plain")], failedIds: [] }, + } + sessionUpdateResult = { data: archivedSession("session-review") } + const source = createStore({}, { session: [liveSession("session-plain"), review, parentWithFork] }) + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + await archiveSessions(["session-plain", "session-review", "session-parent"]) + + // Unlinking a partner rewrites another session's metadata, so those two + // never travel in the batch. + expect(archiveBatchRequests).toEqual([{ directory: "/test/project", ids: ["session-plain"] }]) + expect(replyCalls.filter((call) => call.method === "session.update").map((call) => call.params.sessionID)) + .toEqual(["session-review", "session-parent"]) + }) + + test("does not reconcile a batch answered after a runtime switch", async () => { + globalActiveSessions = [liveSession("session-a")] + archiveBatchResponse = { + status: 200, + body: { archived: [archivedSession("session-a")], failedIds: [] }, + } + const source = createStore({}, { session: [liveSession("session-a")] }) + const { getRuntimeKey, switchRuntimeEndpoint } = await import("../lib/runtime-switch") + switchRuntimeEndpoint({ apiBaseUrl: "http://archive-bulk-a.test", runtimeKey: "archive-bulk-a" }) + const capturedRuntimeKey = getRuntimeKey() + const { archiveSessions, setActionRefs } = await import("./session-actions") + setActionRefs(mockSdk as unknown as OpencodeClient, createChildStores([["/test/project", source]]), () => "/test/project") + + const pending = archiveSessions(["session-a"], { expectedRuntimeKey: capturedRuntimeKey }) + switchRuntimeEndpoint({ apiBaseUrl: "http://archive-bulk-b.test", runtimeKey: "archive-bulk-b" }) + const result = await pending + + expect(result).toEqual({ archivedIds: [], failedIds: ["session-a"] }) + expect(source.getState().session.map((item) => item.id)).toEqual(["session-a"]) + expect(globalUpsertedSessionBatches).toEqual([]) + }) +}) + describe("session restore (unarchive)", () => { beforeEach(() => { replyCalls.length = 0 diff --git a/packages/ui/src/sync/session-actions.ts b/packages/ui/src/sync/session-actions.ts index c208e0cc..85ddbd37 100644 --- a/packages/ui/src/sync/session-actions.ts +++ b/packages/ui/src/sync/session-actions.ts @@ -33,6 +33,8 @@ import { getBtwOriginalSessionID, getBtwSessionID, isBtwSession, withoutBtwSessi import { withLinkedIssue, type LinkedIssue } from "@/lib/linkedIssues" import { getImperativeSessionMessageLoader } from "./session-message-loader" import { cleanupPersistedSessionState } from "./session-deletion-cleanup" +import { requestSessionArchiveBatch } from "./session-archive-batch" +import { registerBulkArchiveEchoes, releaseBulkArchiveEchoes } from "./bulk-archive-echo" import { getRuntimeKey } from "@/lib/runtime-switch" import { markAmbiguousTransportFailure } from "@/lib/relay/transport-error" import { getErrorStatus, isAmbiguousSendFailure } from "./send-failure-classification" @@ -75,20 +77,29 @@ let _optimisticAdd: ((input: OptimisticAddInput) => void) | null = null let _optimisticRemove: ((input: OptimisticRemoveInput) => void) | null = null let _optimisticConfirm: ((input: OptimisticConfirmInput) => void) | null = null -function sessionMutationPatch( +/** + * Revision patch for one or more sessions changing in the same store write. + * + * A batch bumps the revision once, because it is one state change: consumers + * compare revisions to decide whether their view of the list is stale, and a + * batch leaves them stale exactly once rather than once per session. + */ +function sessionsMutationPatch( state: ReturnType, - sessionId: string, + sessionIds: Iterable, deleted: boolean, ) { const revision = (state.sessionRevision ?? 0) + 1 const sessionEventRevision = { ...(state.sessionEventRevision ?? {}) } const sessionDeletedRevision = { ...(state.sessionDeletedRevision ?? {}) } - if (deleted) { - sessionDeletedRevision[sessionId] = revision - delete sessionEventRevision[sessionId] - } else { - sessionEventRevision[sessionId] = revision - delete sessionDeletedRevision[sessionId] + for (const sessionId of sessionIds) { + if (deleted) { + sessionDeletedRevision[sessionId] = revision + delete sessionEventRevision[sessionId] + } else { + sessionEventRevision[sessionId] = revision + delete sessionDeletedRevision[sessionId] + } } return { sessionListSource: "live" as const, @@ -98,6 +109,14 @@ function sessionMutationPatch( } } +function sessionMutationPatch( + state: ReturnType, + sessionId: string, + deleted: boolean, +) { + return sessionsMutationPatch(state, [sessionId], deleted) +} + function invalidateSessionLoads(sessionId: string, directories: Iterable): void { const loader = getImperativeSessionMessageLoader() if (!loader) return @@ -1039,6 +1058,50 @@ function removeSessionFromLiveStores(sessionId: string, preferredDirectory?: str return snapshots } +/** + * Remove a batch of server-confirmed sessions from every live child store. + * + * Each affected store is written once for the whole batch. Removing the + * sessions one at a time notified every subscriber — and therefore re-rendered + * the sidebar — once per session, which is what made archiving a worktree's + * sessions block the main thread for seconds. + */ +function removeSessionsFromLiveStores(sessionIds: Iterable, preferredDirectory?: string): SessionListSnapshot[] { + const ids = new Set(sessionIds) + if (!_childStores || ids.size === 0) return [] + + const snapshots: SessionListSnapshot[] = [] + const visited = new Set() + const candidates: Array<[string, DirectoryStoreApi]> = [] + + if (preferredDirectory) { + const preferredStore = _childStores.children.get(preferredDirectory) + if (preferredStore) { + candidates.push([preferredDirectory, preferredStore]) + visited.add(preferredDirectory) + } + } + + for (const entry of _childStores.children.entries()) { + if (visited.has(entry[0])) continue + candidates.push(entry) + } + + for (const [directory, store] of candidates) { + const current = store.getState() + const removed = current.session.filter((session) => ids.has(session.id)).map((session) => session.id) + if (removed.length === 0) continue + + snapshots.push({ directory }) + store.setState({ + session: current.session.filter((session) => !ids.has(session.id)), + ...sessionsMutationPatch(current, removed, true), + }) + } + + return snapshots +} + function cleanupSessionWorktreeMetadata(sessionId: string): void { useSessionUIStore.getState().setWorktreeMetadata(sessionId, null) } @@ -1252,7 +1315,14 @@ export type ArchiveSessionsOptions = { } /** - * Archive several sessions sequentially, preserving partial results. + * Archive several sessions, preserving partial results. + * + * Sessions that carry no review or btw link are archived by their directory's + * server in one request, and the whole answer is reconciled with a single store + * write. The remainder — review sessions, btw forks, sessions with an active + * btw fork, and any session this client does not hold — keep the per-session + * path, because unlinking a partner is UI-owned work that reads and rewrites + * another session's metadata. * * One failed session never blocks or erases the others: it is reported in * `failedIds` while the remaining IDs are still attempted. When @@ -1269,10 +1339,55 @@ export async function archiveSessions( const archivedIds: string[] = [] const failedIds: string[] = [] const expectedRuntimeKey = options?.expectedRuntimeKey ?? getRuntimeKey() + if (ids.length === 0) return { archivedIds, failedIds } - for (const [index, id] of ids.entries()) { + const plan = planArchiveBatches(ids) + + for (const [directory, batchIds] of plan.batchesByDirectory) { if (isStaleRuntime(expectedRuntimeKey)) { - failedIds.push(...ids.slice(index)) + failedIds.push(...batchIds) + continue + } + + const archivedAt = Date.now() + registerBulkArchiveEchoes( + expectedRuntimeKey, + batchIds.map((id) => ({ id, archivedAt })), + ) + const result = await requestSessionArchiveBatch(directory, batchIds, archivedAt) + if (isStaleRuntime(expectedRuntimeKey)) { + failedIds.push(...batchIds) + continue + } + + if (result.outcome === "archived") { + releaseBulkArchiveEchoes(expectedRuntimeKey, batchIds) + registerBulkArchiveEchoes( + expectedRuntimeKey, + result.archived.flatMap((session) => ( + session.time?.archived === undefined + ? [] + : [{ id: session.id, archivedAt: session.time.archived }] + )), + ) + commitArchivedSessions(result.archived, directory) + archivedIds.push(...result.archived.map((session) => session.id)) + failedIds.push(...result.failedIds) + continue + } + + // The runtime does not serve the batch route, or its answer could not be + // trusted. Archiving each session individually is slower but reaches the + // same state, and re-archiving a session the server already archived writes + // the same field again. + console.warn("[session-actions] archive batch unavailable, archiving one by one", result.reason) + releaseBulkArchiveEchoes(expectedRuntimeKey, batchIds) + plan.individualIds.push(...batchIds) + } + + for (const [index, id] of plan.individualIds.entries()) { + if (isStaleRuntime(expectedRuntimeKey)) { + failedIds.push(...plan.individualIds.slice(index)) break } if (await archiveSession(id, expectedRuntimeKey)) archivedIds.push(id) @@ -1282,6 +1397,82 @@ export async function archiveSessions( return { archivedIds, failedIds } } +/** + * A session whose archive also has to rewrite another session's metadata. + * + * Review sessions and btw forks point at a parent that must be unlinked, and a + * parent with an active btw fork has to delete that fork. Those are + * read-modify-write pairs on a second session, so they stay on the per-session + * path instead of the server batch. + */ +function hasLinkedSessionCleanup(session: Session): boolean { + return isReviewSession(session) || isBtwSession(session) || Boolean(getBtwSessionID(session)) +} + +/** + * Split the requested IDs into per-directory server batches and the sessions + * that must be archived individually. + * + * Link classification reads this client's session records rather than + * refetching each session: those records are kept current by the same + * `session.updated` events that publish a link created anywhere else, so a + * fetch per session would buy no authority the store does not already have. + * A session this client does not hold is classified as individual, which + * restores the per-session fetch for exactly the cases where the store has + * nothing to say. + */ +function planArchiveBatches(ids: string[]) { + const global = useGlobalSessionsStore.getState() + const knownSessions = new Map() + for (const session of [...global.activeSessions, ...global.archivedSessions]) { + knownSessions.set(session.id, session) + } + for (const store of _childStores?.children.values() ?? []) { + for (const session of store.getState().session) knownSessions.set(session.id, session) + } + + const batchesByDirectory = new Map() + const individualIds: string[] = [] + + for (const id of ids) { + const session = knownSessions.get(id) + const directory = session + ? resolveGlobalSessionDirectory(session) ?? getSessionDirectory(id) + : undefined + if (!session || !directory || hasLinkedSessionCleanup(session)) { + individualIds.push(id) + continue + } + const batch = batchesByDirectory.get(directory) + if (batch) batch.push(id) + else batchesByDirectory.set(directory, [id]) + } + + return { batchesByDirectory, individualIds } +} + +/** + * Reconcile a server-confirmed archive batch with one write per store. + * + * This mirrors what `archiveSession` does for a single session — drop it from + * the live directory stores, invalidate its cached messages, move it to the + * archived bucket, and clear it if it was open — with the per-session store + * notifications collapsed into one. + */ +function commitArchivedSessions(sessions: Session[], directory: string): void { + if (sessions.length === 0) return + + const ids = sessions.map((session) => session.id) + const snapshots = removeSessionsFromLiveStores(ids, directory) + const directories = [...snapshots.map((snapshot) => snapshot.directory), directory] + for (const id of ids) invalidateSessionLoads(id, directories) + + useGlobalSessionsStore.getState().upsertSessions(sessions) + + const ui = useSessionUIStore.getState() + if (ui.currentSessionId && ids.includes(ui.currentSessionId)) ui.setCurrentSession(null) +} + /** * Sentinel written to `time.archived` when restoring a session. * diff --git a/packages/ui/src/sync/session-archive-batch.ts b/packages/ui/src/sync/session-archive-batch.ts new file mode 100644 index 00000000..7774a43e --- /dev/null +++ b/packages/ui/src/sync/session-archive-batch.ts @@ -0,0 +1,77 @@ +/** + * Server-side archive batch. + * + * Archiving the sessions linked to a worktree one request at a time is what + * made removing a worktree with many sessions take tens of seconds: every + * session cost its own round trip and its own store reconciliation. This asks + * the OpenChamber server to archive the whole batch next to OpenCode, so the + * browser spends one request and reconciles once. + * + * The route is an OpenChamber capability, not an OpenCode one. Runtimes that do + * not serve it (the VS Code webview has no server process) answer with a stable + * unsupported status, and callers fall back to archiving session by session. + */ + +import type { Session } from '@opencode-ai/sdk/v2/client'; +import { z } from 'zod'; + +import { runtimeFetch } from '@/lib/runtime-fetch'; + +/** + * The route answers with sessions OpenCode itself returned from + * `session.update`. Only the identity this layer routes on is asserted here; + * every other field is carried through to the stores exactly as the server + * sent it, the same as for any other session response. + */ +const archiveResponseSchema = z.object({ + archived: z.array(z.looseObject({ id: z.string().min(1) })), + failedIds: z.array(z.string().min(1)), +}); + +export type SessionArchiveBatchResult = + | { outcome: 'archived'; archived: Session[]; failedIds: string[] } + | { outcome: 'unavailable'; reason: string }; + +export async function requestSessionArchiveBatch( + directory: string, + ids: string[], + archivedAt: number, +): Promise { + let response: Response; + try { + response = await runtimeFetch('/api/openchamber/sessions/archive', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ directory, ids, archivedAt }), + }); + } catch (error) { + return { outcome: 'unavailable', reason: error instanceof Error ? error.message : 'archive request failed' }; + } + + if (!response.ok) { + return { outcome: 'unavailable', reason: `archive request failed with ${response.status}` }; + } + + let body: unknown; + try { + body = await response.json(); + } catch (error) { + return { outcome: 'unavailable', reason: error instanceof Error ? error.message : 'archive response was not JSON' }; + } + + const parsed = archiveResponseSchema.safeParse(body); + if (!parsed.success) { + // A body this layer cannot read is reported as unavailable rather than as + // an empty success, so a caller never mistakes "the response made no + // sense" for "nothing needed archiving" and drops the sessions. + return { outcome: 'unavailable', reason: `malformed archive response: ${parsed.error.issues[0]?.message ?? 'unknown shape'}` }; + } + + return { + outcome: 'archived', + // SAFETY: the schema guarantees the non-empty string `id` this layer keys + // on; the remaining fields are the server's own session payload. + archived: parsed.data.archived as Session[], + failedIds: parsed.data.failedIds, + }; +} diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index ed2772ed..158f340f 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -38,7 +38,11 @@ import { setSyncRefs, getAllSyncSessions } from "./sync-refs" import { useSessionUIStore } from "./session-ui-store" import { stripSessionDiffSnapshots } from "./sanitize" import { upsertSessionRecord } from "./session-records" -import { applySessionEventToGlobalSessions, applySessionEventsToGlobalSessions } from "./session-event-router" +import { + applySessionEventToGlobalSessions, + applySessionEventsToGlobalSessions, +} from "./session-event-router" +import { shouldConsumeBulkArchiveEcho } from "./bulk-archive-echo" import { syncDebug } from "./debug" import { getReconnectCandidateSessionIds, mergeBootstrapSessions } from "./reconnect-recovery" import { messagesBefore } from "./message-ordering" @@ -1586,6 +1590,8 @@ export function handleEvent( return } + if (shouldConsumeBulkArchiveEcho(payload, expectedRuntimeKey)) return + const directory = resolveDirectoryFromRoutingIndex(routingIndex, rawDirectory, payload, childStores, batch) if (payload.type === "session.deleted" && expectedRuntimeKey === getRuntimeKey()) { diff --git a/packages/vscode/webview/main.tsx b/packages/vscode/webview/main.tsx index 03e8b1d1..4a4bc409 100644 --- a/packages/vscode/webview/main.tsx +++ b/packages/vscode/webview/main.tsx @@ -384,6 +384,14 @@ const handleLocalApiRequest = async (input: RequestInfo | URL, url: URL, init: R return unsupportedWebRouteResponse('Remote tunnel settings'); } + // Archiving a batch of sessions server-side needs an OpenChamber server + // process; the extension host has none. Answering explicitly keeps the + // shared UI on its per-session archive path instead of leaving the request + // to the generic proxy. + if (normalizedPathname === '/api/openchamber/sessions/archive') { + return unsupportedWebRouteResponse('Server-side session archiving'); + } + if (/^\/api\/projects\/[^/]+\/scheduled-tasks(?:\/[^/]+)?$/.test(normalizedPathname)) { return unsupportedWebRouteResponse('Scheduled tasks'); } diff --git a/packages/web/server/lib/openchamber-sessions/routes.js b/packages/web/server/lib/openchamber-sessions/routes.js index 6c58b333..b9b768e0 100644 --- a/packages/web/server/lib/openchamber-sessions/routes.js +++ b/packages/web/server/lib/openchamber-sessions/routes.js @@ -253,6 +253,40 @@ const latestCompletedAssistantMessageID = async ({ client, sessionID, directory return asNonEmptyString(latest?.id); }; +/** + * Upper bound on one archive batch. + * + * The batch is applied one session at a time against OpenCode, so an unbounded + * list would hold a request open for as long as the list is large. Callers with + * more sessions than this send several batches and keep their own partial + * results. + */ +const MAX_ARCHIVE_BATCH = 500; + +const parseArchiveRequest = (payload) => { + const rawIds = payload?.ids; + if (!Array.isArray(rawIds) || rawIds.length === 0) { + return { ok: false, error: 'ids must be a non-empty array of session ids' }; + } + if (rawIds.length > MAX_ARCHIVE_BATCH) { + return { ok: false, error: `ids must contain at most ${MAX_ARCHIVE_BATCH} session ids` }; + } + + const ids = []; + for (const value of rawIds) { + const id = asNonEmptyString(value); + if (!id) return { ok: false, error: 'ids must contain non-empty session ids' }; + ids.push(id); + } + + const archivedAt = payload?.archivedAt; + if (archivedAt !== undefined && (!Number.isSafeInteger(archivedAt) || archivedAt <= 0)) { + return { ok: false, error: 'archivedAt must be a positive integer timestamp' }; + } + + return { ok: true, ids, archivedAt: archivedAt ?? Date.now() }; +}; + const resolveRequestedDirectory = async ({ payload, readSettingsFromDiskMigrated, sanitizeProjects, validateDirectoryPath }) => { const projectID = asNonEmptyString(payload?.projectId) || asNonEmptyString(payload?.projectID); if (projectID) { @@ -579,6 +613,64 @@ export const createOpenChamberSessionService = (dependencies) => { return { model, agent, variant, promptDispatched: true, dispatchedAsCommand: Boolean(resolvedCommand) }; }; + /** + * Archive a batch of sessions in one request. + * + * The UI archives every session linked to a worktree before removing it. + * Doing that from the browser costs one request per session plus a store + * reconciliation between each of them, which is what made deleting a + * worktree with many sessions take tens of seconds. Here the batch stays on + * the server, next to OpenCode, and the client reconciles once. + * + * Sessions are updated one at a time on purpose: they are archived against a + * single OpenCode instance, and a fan-out of concurrent writes would trade a + * UI stall for server event-loop starvation. One failed session never stops + * the batch — it is reported in `failedIds` while the rest still archive, so + * callers keep the partial-failure behaviour they already show. + */ + const archive = async (payload = {}) => { + const parsed = parseArchiveRequest(payload); + if (!parsed.ok) { + throw new OpenChamberControlError(parsed.error, 400); + } + + const resolvedDirectory = await resolveRequestedDirectory({ + payload, + readSettingsFromDiskMigrated, + sanitizeProjects, + validateDirectoryPath, + }); + if (!resolvedDirectory.ok) { + throw new OpenChamberControlError(resolvedDirectory.error, resolvedDirectory.status || 400); + } + + if (typeof waitForOpenCodeReady === 'function') await waitForOpenCodeReady(10_000, 250); + + const directory = resolvedDirectory.directory; + const baseUrl = buildOpenCodeUrl('/', '').replace(/\/$/, ''); + const client = createOpencodeClient({ baseUrl, headers: getOpenCodeAuthHeaders() }); + + const archived = []; + const failedIds = []; + for (const sessionID of parsed.ids) { + try { + const response = await client.session.update({ + sessionID, + directory, + time: { archived: parsed.archivedAt }, + }); + const session = response?.data; + if (session?.id) archived.push(session); + else failedIds.push(sessionID); + } catch (error) { + console.warn('[OpenChamberSessions] failed to archive session', sessionID, error); + failedIds.push(sessionID); + } + } + + return { directory, archived, failedIds }; + }; + const create = async (payload = {}) => { const title = asNonEmptyString(payload.title); const prompt = asNonEmptyString(payload.prompt); @@ -813,6 +905,7 @@ export const createOpenChamberSessionService = (dependencies) => { return { create, + archive, send: (sessionID, payload) => runExisting('send', sessionID, payload), fork: (sessionID, payload) => runExisting('fork', sessionID, payload), }; @@ -843,6 +936,15 @@ export const registerOpenChamberSessionRoutes = (app, dependencies) => { } }); + app.post('/api/openchamber/sessions/archive', express.json({ limit: '1mb' }), async (req, res) => { + try { + return res.json(await service.archive(req.body && typeof req.body === 'object' ? req.body : {})); + } catch (error) { + console.error('[OpenChamberSessions] failed to archive sessions:', error); + return sendServiceError(res, error, 'Failed to archive sessions'); + } + }); + app.post( '/api/openchamber/sessions/:sessionId/send', express.json({ limit: '1mb' }), diff --git a/packages/web/server/lib/openchamber-sessions/routes.test.js b/packages/web/server/lib/openchamber-sessions/routes.test.js index c95fd6e2..6a60ce73 100644 --- a/packages/web/server/lib/openchamber-sessions/routes.test.js +++ b/packages/web/server/lib/openchamber-sessions/routes.test.js @@ -17,6 +17,7 @@ const getWorktreeBootstrapStatusMock = vi.fn(async () => ({ const sessionCreateMock = vi.fn(async () => ({ data: { id: 'ses_123' } })); const sessionForkMock = vi.fn(async () => ({ data: { id: 'ses_fork', title: 'Forked session' } })); const sessionMessagesMock = vi.fn(async () => ({ data: [] })); +const sessionUpdateMock = vi.fn(async ({ sessionID }) => ({ data: { id: sessionID, time: { archived: 1 } } })); let existingSessionMessages = []; let dispatchedUserMessageSeq = 0; @@ -78,6 +79,7 @@ vi.mock('@opencode-ai/sdk/v2', () => ({ fork: sessionForkMock, messages: sessionMessagesMock, command: sessionCommandMock, + update: sessionUpdateMock, }, command: { list: commandListMock, @@ -132,6 +134,92 @@ describe('openchamber session routes', () => { sessionCommandMock.mockResolvedValue({ data: {} }); commandListMock.mockReset(); commandListMock.mockResolvedValue({ data: [] }); + sessionUpdateMock.mockReset(); + sessionUpdateMock.mockImplementation(async ({ sessionID }) => ({ data: { id: sessionID, time: { archived: 1 } } })); + }); + + describe('archiving a batch of sessions', () => { + it('archives every id against the resolved directory and returns the archived sessions', async () => { + const { app } = createApp(); + const response = await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/repo/app', ids: ['ses_a', 'ses_b'], archivedAt: 1700 }) + .expect(200); + + expect(response.body.directory).toBe('/repo/app'); + expect(response.body.archived.map((session) => session.id)).toEqual(['ses_a', 'ses_b']); + expect(response.body.failedIds).toEqual([]); + expect(sessionUpdateMock).toHaveBeenCalledTimes(2); + expect(sessionUpdateMock).toHaveBeenCalledWith({ + sessionID: 'ses_a', + directory: '/repo/app', + time: { archived: 1700 }, + }); + }); + + it('keeps archiving after a failed session and reports it as failed', async () => { + sessionUpdateMock.mockImplementation(async ({ sessionID }) => { + if (sessionID === 'ses_b') throw new Error('session.update failed'); + return { data: { id: sessionID } }; + }); + + const { app } = createApp(); + const response = await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/repo/app', ids: ['ses_a', 'ses_b', 'ses_c'] }) + .expect(200); + + expect(response.body.archived.map((session) => session.id)).toEqual(['ses_a', 'ses_c']); + expect(response.body.failedIds).toEqual(['ses_b']); + }); + + it('reports a session the server did not confirm as failed instead of archived', async () => { + sessionUpdateMock.mockImplementation(async ({ sessionID }) => ( + sessionID === 'ses_b' ? { data: null } : { data: { id: sessionID } } + )); + + const { app } = createApp(); + const response = await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/repo/app', ids: ['ses_a', 'ses_b'] }) + .expect(200); + + expect(response.body.archived.map((session) => session.id)).toEqual(['ses_a']); + expect(response.body.failedIds).toEqual(['ses_b']); + }); + + it('rejects an empty batch, an oversized batch, and non-string ids', async () => { + const { app } = createApp(); + + await request(app).post('/api/openchamber/sessions/archive').send({ directory: '/repo/app', ids: [] }).expect(400); + await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/repo/app', ids: Array.from({ length: 501 }, (_, index) => `ses_${index}`) }) + .expect(400); + await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/repo/app', ids: ['ses_a', ''] }) + .expect(400); + await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/repo/app', ids: ['ses_a'], archivedAt: -1 }) + .expect(400); + + expect(sessionUpdateMock).not.toHaveBeenCalled(); + }); + + it('rejects a directory the runtime does not accept', async () => { + const { app } = createApp({ + validateDirectoryPath: async () => ({ ok: false, error: 'Invalid directory' }), + }); + + await request(app) + .post('/api/openchamber/sessions/archive') + .send({ directory: '/elsewhere', ids: ['ses_a'] }) + .expect(400); + + expect(sessionUpdateMock).not.toHaveBeenCalled(); + }); }); it('creates a session for a directory', async () => {