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.
This commit is contained in:
@@ -45,7 +45,15 @@ mock.module("@/lib/runtime-switch", () => ({
|
|||||||
return () => undefined
|
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 => ({
|
const buildSession = (title: string, time: Session["time"]): Session => ({
|
||||||
id: "ses_1",
|
id: "ses_1",
|
||||||
@@ -146,4 +154,35 @@ describe("applySessionEventToGlobalSessions", () => {
|
|||||||
expect(mutationCalls).toBe(1)
|
expect(mutationCalls).toBe(1)
|
||||||
expect(upsertedSessions).toHaveLength(1_000)
|
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)
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -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<string, Map<string, { archivedAt: number; expiresAt: number }>>()
|
||||||
|
|
||||||
|
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<string>): 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
|
||||||
|
}
|
||||||
@@ -92,6 +92,16 @@ describe("persisted directory sessions", () => {
|
|||||||
expect(readManagedChatSessions()).toEqual([])
|
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 () => {
|
test("keeps the 50 most recently updated sessions across restart reads", async () => {
|
||||||
const sessions = Array.from({ length: 60 }, (_, updated) => session(59 - updated, updated))
|
const sessions = Array.from({ length: 60 }, (_, updated) => session(59 - updated, updated))
|
||||||
|
|
||||||
|
|||||||
@@ -177,7 +177,7 @@ function scheduleSessionCacheWrite(directory: string, sessions: Session[]): void
|
|||||||
if (pending.runtimeKey !== runtimeKey) pendingSessionWrites.delete(pendingKey)
|
if (pending.runtimeKey !== runtimeKey) pendingSessionWrites.delete(pendingKey)
|
||||||
}
|
}
|
||||||
pendingSessionWrites.set(key, { runtimeKey, key, legacyKey: legacyCacheKey(directory, "sessions"), sessions })
|
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)
|
pendingSessionWriteTimer = setTimeout(flushPendingSessionWrites, SESSION_PERSIST_DEBOUNCE_MS)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -21,7 +21,16 @@ let sessionDeleteError: unknown | null = null
|
|||||||
let beforeSessionUpdateResolve: ((sessionId: string) => void) | null = null
|
let beforeSessionUpdateResolve: ((sessionId: string) => void) | null = null
|
||||||
let beforeSessionDeleteResolve: ((sessionId: string) => void) | null = null
|
let beforeSessionDeleteResolve: ((sessionId: string) => void) | null = null
|
||||||
const globalUpsertedSessions: unknown[] = []
|
const globalUpsertedSessions: unknown[] = []
|
||||||
|
const globalUpsertedSessionBatches: Session[][] = []
|
||||||
const globalRemovedSessionIds: string[] = []
|
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 deletedCleanupIdentities: Array<{ runtimeKey: string; directory: string; sessionId: string }> = []
|
||||||
const movedSessionDirectories: Array<{ sessionID: string; directory: string }> = []
|
const movedSessionDirectories: Array<{ sessionID: string; directory: string }> = []
|
||||||
|
|
||||||
@@ -243,11 +252,15 @@ mock.module("@/stores/useGlobalSessionsStore", () => ({
|
|||||||
},
|
},
|
||||||
useGlobalSessionsStore: {
|
useGlobalSessionsStore: {
|
||||||
getState: () => ({
|
getState: () => ({
|
||||||
activeSessions: [],
|
activeSessions: globalActiveSessions,
|
||||||
archivedSessions: [],
|
archivedSessions: [],
|
||||||
upsertSession: (session: unknown) => {
|
upsertSession: (session: unknown) => {
|
||||||
globalUpsertedSessions.push(session)
|
globalUpsertedSessions.push(session)
|
||||||
},
|
},
|
||||||
|
upsertSessions: (sessions: Session[]) => {
|
||||||
|
globalUpsertedSessionBatches.push(sessions)
|
||||||
|
globalUpsertedSessions.push(...sessions)
|
||||||
|
},
|
||||||
removeSessions: (ids: Iterable<string>) => {
|
removeSessions: (ids: Iterable<string>) => {
|
||||||
globalRemovedSessionIds.push(...ids)
|
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", () => ({
|
mock.module("./session-deletion-cleanup", () => ({
|
||||||
cleanupPersistedSessionState: (identity: { runtimeKey: string; directory: string; sessionId: string }) => {
|
cleanupPersistedSessionState: (identity: { runtimeKey: string; directory: string; sessionId: string }) => {
|
||||||
deletedCleanupIdentities.push(identity)
|
deletedCleanupIdentities.push(identity)
|
||||||
@@ -396,6 +421,10 @@ describe("confirmed session removal", () => {
|
|||||||
sessionUpdateResult = {}
|
sessionUpdateResult = {}
|
||||||
beforeSessionUpdateResolve = null
|
beforeSessionUpdateResolve = null
|
||||||
beforeSessionDeleteResolve = 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 () => {
|
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<string, unknown>): 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)", () => {
|
describe("session restore (unarchive)", () => {
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
replyCalls.length = 0
|
replyCalls.length = 0
|
||||||
|
|||||||
@@ -33,6 +33,8 @@ import { getBtwOriginalSessionID, getBtwSessionID, isBtwSession, withoutBtwSessi
|
|||||||
import { withLinkedIssue, type LinkedIssue } from "@/lib/linkedIssues"
|
import { withLinkedIssue, type LinkedIssue } from "@/lib/linkedIssues"
|
||||||
import { getImperativeSessionMessageLoader } from "./session-message-loader"
|
import { getImperativeSessionMessageLoader } from "./session-message-loader"
|
||||||
import { cleanupPersistedSessionState } from "./session-deletion-cleanup"
|
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 { getRuntimeKey } from "@/lib/runtime-switch"
|
||||||
import { markAmbiguousTransportFailure } from "@/lib/relay/transport-error"
|
import { markAmbiguousTransportFailure } from "@/lib/relay/transport-error"
|
||||||
import { getErrorStatus, isAmbiguousSendFailure } from "./send-failure-classification"
|
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 _optimisticRemove: ((input: OptimisticRemoveInput) => void) | null = null
|
||||||
let _optimisticConfirm: ((input: OptimisticConfirmInput) => 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<DirectoryStoreApi["getState"]>,
|
state: ReturnType<DirectoryStoreApi["getState"]>,
|
||||||
sessionId: string,
|
sessionIds: Iterable<string>,
|
||||||
deleted: boolean,
|
deleted: boolean,
|
||||||
) {
|
) {
|
||||||
const revision = (state.sessionRevision ?? 0) + 1
|
const revision = (state.sessionRevision ?? 0) + 1
|
||||||
const sessionEventRevision = { ...(state.sessionEventRevision ?? {}) }
|
const sessionEventRevision = { ...(state.sessionEventRevision ?? {}) }
|
||||||
const sessionDeletedRevision = { ...(state.sessionDeletedRevision ?? {}) }
|
const sessionDeletedRevision = { ...(state.sessionDeletedRevision ?? {}) }
|
||||||
if (deleted) {
|
for (const sessionId of sessionIds) {
|
||||||
sessionDeletedRevision[sessionId] = revision
|
if (deleted) {
|
||||||
delete sessionEventRevision[sessionId]
|
sessionDeletedRevision[sessionId] = revision
|
||||||
} else {
|
delete sessionEventRevision[sessionId]
|
||||||
sessionEventRevision[sessionId] = revision
|
} else {
|
||||||
delete sessionDeletedRevision[sessionId]
|
sessionEventRevision[sessionId] = revision
|
||||||
|
delete sessionDeletedRevision[sessionId]
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return {
|
return {
|
||||||
sessionListSource: "live" as const,
|
sessionListSource: "live" as const,
|
||||||
@@ -98,6 +109,14 @@ function sessionMutationPatch(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function sessionMutationPatch(
|
||||||
|
state: ReturnType<DirectoryStoreApi["getState"]>,
|
||||||
|
sessionId: string,
|
||||||
|
deleted: boolean,
|
||||||
|
) {
|
||||||
|
return sessionsMutationPatch(state, [sessionId], deleted)
|
||||||
|
}
|
||||||
|
|
||||||
function invalidateSessionLoads(sessionId: string, directories: Iterable<string | null | undefined>): void {
|
function invalidateSessionLoads(sessionId: string, directories: Iterable<string | null | undefined>): void {
|
||||||
const loader = getImperativeSessionMessageLoader()
|
const loader = getImperativeSessionMessageLoader()
|
||||||
if (!loader) return
|
if (!loader) return
|
||||||
@@ -1039,6 +1058,50 @@ function removeSessionFromLiveStores(sessionId: string, preferredDirectory?: str
|
|||||||
return snapshots
|
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<string>, preferredDirectory?: string): SessionListSnapshot[] {
|
||||||
|
const ids = new Set(sessionIds)
|
||||||
|
if (!_childStores || ids.size === 0) return []
|
||||||
|
|
||||||
|
const snapshots: SessionListSnapshot[] = []
|
||||||
|
const visited = new Set<string>()
|
||||||
|
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 {
|
function cleanupSessionWorktreeMetadata(sessionId: string): void {
|
||||||
useSessionUIStore.getState().setWorktreeMetadata(sessionId, null)
|
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
|
* One failed session never blocks or erases the others: it is reported in
|
||||||
* `failedIds` while the remaining IDs are still attempted. When
|
* `failedIds` while the remaining IDs are still attempted. When
|
||||||
@@ -1269,10 +1339,55 @@ export async function archiveSessions(
|
|||||||
const archivedIds: string[] = []
|
const archivedIds: string[] = []
|
||||||
const failedIds: string[] = []
|
const failedIds: string[] = []
|
||||||
const expectedRuntimeKey = options?.expectedRuntimeKey ?? getRuntimeKey()
|
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)) {
|
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
|
break
|
||||||
}
|
}
|
||||||
if (await archiveSession(id, expectedRuntimeKey)) archivedIds.push(id)
|
if (await archiveSession(id, expectedRuntimeKey)) archivedIds.push(id)
|
||||||
@@ -1282,6 +1397,82 @@ export async function archiveSessions(
|
|||||||
return { archivedIds, failedIds }
|
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<string, Session>()
|
||||||
|
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<string, string[]>()
|
||||||
|
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.
|
* Sentinel written to `time.archived` when restoring a session.
|
||||||
*
|
*
|
||||||
|
|||||||
@@ -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<SessionArchiveBatchResult> {
|
||||||
|
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,
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -38,7 +38,11 @@ import { setSyncRefs, getAllSyncSessions } from "./sync-refs"
|
|||||||
import { useSessionUIStore } from "./session-ui-store"
|
import { useSessionUIStore } from "./session-ui-store"
|
||||||
import { stripSessionDiffSnapshots } from "./sanitize"
|
import { stripSessionDiffSnapshots } from "./sanitize"
|
||||||
import { upsertSessionRecord } from "./session-records"
|
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 { syncDebug } from "./debug"
|
||||||
import { getReconnectCandidateSessionIds, mergeBootstrapSessions } from "./reconnect-recovery"
|
import { getReconnectCandidateSessionIds, mergeBootstrapSessions } from "./reconnect-recovery"
|
||||||
import { messagesBefore } from "./message-ordering"
|
import { messagesBefore } from "./message-ordering"
|
||||||
@@ -1586,6 +1590,8 @@ export function handleEvent(
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (shouldConsumeBulkArchiveEcho(payload, expectedRuntimeKey)) return
|
||||||
|
|
||||||
const directory = resolveDirectoryFromRoutingIndex(routingIndex, rawDirectory, payload, childStores, batch)
|
const directory = resolveDirectoryFromRoutingIndex(routingIndex, rawDirectory, payload, childStores, batch)
|
||||||
|
|
||||||
if (payload.type === "session.deleted" && expectedRuntimeKey === getRuntimeKey()) {
|
if (payload.type === "session.deleted" && expectedRuntimeKey === getRuntimeKey()) {
|
||||||
|
|||||||
@@ -384,6 +384,14 @@ const handleLocalApiRequest = async (input: RequestInfo | URL, url: URL, init: R
|
|||||||
return unsupportedWebRouteResponse('Remote tunnel settings');
|
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)) {
|
if (/^\/api\/projects\/[^/]+\/scheduled-tasks(?:\/[^/]+)?$/.test(normalizedPathname)) {
|
||||||
return unsupportedWebRouteResponse('Scheduled tasks');
|
return unsupportedWebRouteResponse('Scheduled tasks');
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -253,6 +253,40 @@ const latestCompletedAssistantMessageID = async ({ client, sessionID, directory
|
|||||||
return asNonEmptyString(latest?.id);
|
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 resolveRequestedDirectory = async ({ payload, readSettingsFromDiskMigrated, sanitizeProjects, validateDirectoryPath }) => {
|
||||||
const projectID = asNonEmptyString(payload?.projectId) || asNonEmptyString(payload?.projectID);
|
const projectID = asNonEmptyString(payload?.projectId) || asNonEmptyString(payload?.projectID);
|
||||||
if (projectID) {
|
if (projectID) {
|
||||||
@@ -579,6 +613,64 @@ export const createOpenChamberSessionService = (dependencies) => {
|
|||||||
return { model, agent, variant, promptDispatched: true, dispatchedAsCommand: Boolean(resolvedCommand) };
|
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 create = async (payload = {}) => {
|
||||||
const title = asNonEmptyString(payload.title);
|
const title = asNonEmptyString(payload.title);
|
||||||
const prompt = asNonEmptyString(payload.prompt);
|
const prompt = asNonEmptyString(payload.prompt);
|
||||||
@@ -813,6 +905,7 @@ export const createOpenChamberSessionService = (dependencies) => {
|
|||||||
|
|
||||||
return {
|
return {
|
||||||
create,
|
create,
|
||||||
|
archive,
|
||||||
send: (sessionID, payload) => runExisting('send', sessionID, payload),
|
send: (sessionID, payload) => runExisting('send', sessionID, payload),
|
||||||
fork: (sessionID, payload) => runExisting('fork', 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(
|
app.post(
|
||||||
'/api/openchamber/sessions/:sessionId/send',
|
'/api/openchamber/sessions/:sessionId/send',
|
||||||
express.json({ limit: '1mb' }),
|
express.json({ limit: '1mb' }),
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ const getWorktreeBootstrapStatusMock = vi.fn(async () => ({
|
|||||||
const sessionCreateMock = vi.fn(async () => ({ data: { id: 'ses_123' } }));
|
const sessionCreateMock = vi.fn(async () => ({ data: { id: 'ses_123' } }));
|
||||||
const sessionForkMock = vi.fn(async () => ({ data: { id: 'ses_fork', title: 'Forked session' } }));
|
const sessionForkMock = vi.fn(async () => ({ data: { id: 'ses_fork', title: 'Forked session' } }));
|
||||||
const sessionMessagesMock = vi.fn(async () => ({ data: [] }));
|
const sessionMessagesMock = vi.fn(async () => ({ data: [] }));
|
||||||
|
const sessionUpdateMock = vi.fn(async ({ sessionID }) => ({ data: { id: sessionID, time: { archived: 1 } } }));
|
||||||
|
|
||||||
let existingSessionMessages = [];
|
let existingSessionMessages = [];
|
||||||
let dispatchedUserMessageSeq = 0;
|
let dispatchedUserMessageSeq = 0;
|
||||||
@@ -78,6 +79,7 @@ vi.mock('@opencode-ai/sdk/v2', () => ({
|
|||||||
fork: sessionForkMock,
|
fork: sessionForkMock,
|
||||||
messages: sessionMessagesMock,
|
messages: sessionMessagesMock,
|
||||||
command: sessionCommandMock,
|
command: sessionCommandMock,
|
||||||
|
update: sessionUpdateMock,
|
||||||
},
|
},
|
||||||
command: {
|
command: {
|
||||||
list: commandListMock,
|
list: commandListMock,
|
||||||
@@ -132,6 +134,92 @@ describe('openchamber session routes', () => {
|
|||||||
sessionCommandMock.mockResolvedValue({ data: {} });
|
sessionCommandMock.mockResolvedValue({ data: {} });
|
||||||
commandListMock.mockReset();
|
commandListMock.mockReset();
|
||||||
commandListMock.mockResolvedValue({ data: [] });
|
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 () => {
|
it('creates a session for a directory', async () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user