diff --git a/packages/app/src/app.tsx b/packages/app/src/app.tsx index 9062b715cc..181f7c1594 100644 --- a/packages/app/src/app.tsx +++ b/packages/app/src/app.tsx @@ -31,8 +31,8 @@ import { CommandProvider } from "@/context/command" import { CommentsProvider } from "@/context/comments" import { FileProvider } from "@/context/file" import { ServerSDKProvider, useServerSDK } from "@/context/server-sdk" -import { ServerSyncProvider } from "@/context/server-sync" -import { GlobalProvider, useGlobal } from "@/context/global" +import { ServerSyncProvider, useServerSync } from "@/context/server-sync" +import { GlobalProvider } from "@/context/global" import { HighlightsProvider } from "@/context/highlights" import { LanguageProvider, type Locale, useLanguage } from "@/context/language" import { LayoutProvider } from "@/context/layout" @@ -51,7 +51,7 @@ import LegacyLayout from "@/pages/layout" import NewLayout from "@/pages/layout-new" import { ErrorPage } from "./pages/error" import { useCheckServerHealth } from "./utils/server-health" -import { legacySessionHref, requireServerKey, rootSession, sessionHref } from "./utils/session-route" +import { legacySessionHref, requireServerKey, sessionHref } from "./utils/session-route" import Session from "@/pages/session" import { NewHome, LegacyHome } from "@/pages/home" @@ -109,37 +109,26 @@ function ResolvedTargetSessionRoute() { const params = useParams<{ serverKey: string; id: string }>() const settings = useSettings() const tabs = useTabs() - const global = useGlobal() - const serverSDK = useServerSDK() + const sync = useServerSync() const serverKey = createMemo(() => requireServerKey(params.serverKey)) - const placement = createMemo(() => global.sessionPlacement.get(serverKey(), params.id)) + const cached = createMemo(() => sync().session.lineage.peek(params.id)) const [resolved] = createResource( () => { - if (placement()) return - return { id: params.id, sdk: serverSDK() } - }, - async ({ id, sdk }) => { - const session = (await sdk.client.session.get({ sessionID: id })).data! - const root = await rootSession(session, (sessionID) => - sdk.client.session.get({ sessionID }).then((result) => result.data!), - ) - return global.sessionPlacement.set({ - server: serverKey(), - leafID: session.id, - rootID: root.id, - directory: session.directory, - }) + if (cached()) return + return { id: params.id, sync: sync() } }, + ({ id, sync }) => sync.session.lineage.resolve(id), ) - const directory = createMemo(() => placement()?.directory ?? resolved()?.directory) + const current = createMemo(() => cached() ?? resolved()) + const directory = createMemo(() => current()?.session.directory) const targetDirectory = () => directory()! createEffect(() => { - const current = placement() ?? resolved() - if (!current) return + const session = current() + if (!session) return tabs.addSessionTab({ server: serverKey(), - sessionId: current.rootID, + sessionId: session.root.id, }) }) diff --git a/packages/app/src/components/prompt-input/submit.test.ts b/packages/app/src/components/prompt-input/submit.test.ts index 94a2488044..43bdafefa4 100644 --- a/packages/app/src/components/prompt-input/submit.test.ts +++ b/packages/app/src/components/prompt-input/submit.test.ts @@ -185,6 +185,10 @@ beforeAll(async () => { mock.module("@/context/server-sync", () => ({ useServerSync: () => () => ({ + session: { + remember: () => undefined, + set: () => undefined, + }, child: (directory: string) => { syncedDirectories.push(directory) storedSessions[directory] ??= [] diff --git a/packages/app/src/components/prompt-input/submit.ts b/packages/app/src/components/prompt-input/submit.ts index 621723ae11..814c7c8409 100644 --- a/packages/app/src/components/prompt-input/submit.ts +++ b/packages/app/src/components/prompt-input/submit.ts @@ -56,16 +56,14 @@ const draftImages = (prompt: Prompt) => prompt.filter((part): part is ImageAttac export async function sendFollowupDraft(input: FollowupSendInput) { const text = draftText(input.draft.prompt) const images = draftImages(input.draft.prompt) - const [, setStore] = input.serverSync.child(input.draft.sessionDirectory) - const setBusy = () => { if (!input.optimisticBusy) return - setStore("session_status", input.draft.sessionID, { type: "busy" }) + input.serverSync.session.set("session_status", input.draft.sessionID, { type: "busy" }) } const setIdle = () => { if (!input.optimisticBusy) return - setStore("session_status", input.draft.sessionID, { type: "idle" }) + input.serverSync.session.set("session_status", input.draft.sessionID, { type: "idle" }) } const wait = async () => { @@ -234,9 +232,7 @@ export function createPromptSubmit(input: PromptSubmitInput) { const sessionID = params.id if (!sessionID) return Promise.resolve() - serverSync().todo.set(sessionID, []) - const [, setStore] = serverSync().child(sdk().directory) - setStore("todo", sessionID, []) + serverSync().session.set("todo", sessionID, []) input.onAbort?.() @@ -282,6 +278,7 @@ export function createPromptSubmit(input: PromptSubmitInput) { } const seed = (dir: string, info: Session) => { + serverSync().session.remember(info) const [, setStore] = serverSync().child(dir) setStore("session", (list: Session[]) => { const result = Binary.search(list, info.id, (item) => item.id) diff --git a/packages/app/src/components/titlebar.tsx b/packages/app/src/components/titlebar.tsx index ff8bb55743..0aed18fe97 100644 --- a/packages/app/src/components/titlebar.tsx +++ b/packages/app/src/components/titlebar.tsx @@ -512,38 +512,15 @@ export function Titlebar(props: { update?: TitlebarUpdate }) { return conn ? global.ensureServerCtx(conn) : undefined }) const sdk = createMemo(() => serverCtx()?.sdk ?? null) - const cachedSession = createMemo(() => { - const placement = global.sessionPlacement.get(tab.server, tab.sessionId) - const ctx = serverCtx() - if (!placement || !ctx) return - return ctx.sync - .child(placement.directory, { bootstrap: false })[0] - .session.find((session) => session.id === tab.sessionId) - }) + const cachedSession = createMemo(() => serverCtx()?.sync.session.peek(tab.sessionId)) const [loadedSession] = createResource( () => { - if (cachedSession()) return null const id = tab.sessionId const ctx = serverCtx() return ctx ? { id, ctx } : null }, - ({ id, ctx }) => - ctx.sdk.client.session - .get({ sessionID: id }) - .then((x) => { - const session = x.data - if (!session) return - if (!session.parentID) - global.sessionPlacement.set({ - server: tab.server, - leafID: session.id, - rootID: session.id, - directory: session.directory, - }) - return session - }) - .catch(() => undefined), + ({ id, ctx }) => ctx.sync.session.resolve(id).catch(() => undefined), ) const session = createMemo(() => cachedSession() ?? loadedSession()) let prefetched = false diff --git a/packages/app/src/context/directory-sync.ts b/packages/app/src/context/directory-sync.ts index 1702cfe53d..bb510c62f1 100644 --- a/packages/app/src/context/directory-sync.ts +++ b/packages/app/src/context/directory-sync.ts @@ -1,175 +1,23 @@ -import { batch, createMemo } from "solid-js" -import { createStore, produce, reconcile } from "solid-js/store" import { Binary } from "@opencode-ai/core/util/binary" -import { retry } from "@opencode-ai/core/util/retry" -import { - clearSessionPrefetch, - getSessionPrefetch, - getSessionPrefetchPromise, - setSessionPrefetch, -} from "./global-sync/session-prefetch" import type { Message, Part } from "@opencode-ai/sdk/v2/client" -import { SESSION_CACHE_LIMIT, dropSessionCaches, pickSessionCacheEvictions } from "./global-sync/session-cache" -import { diffs as list, message as clean } from "@/utils/diffs" -import { type createServerSdkContext } from "./server-sdk" -import { type createServerSyncContextInner } from "./server-sync" - -const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) - -function sortParts(parts: Part[]) { - return parts.filter((part) => !!part?.id).sort((a, b) => cmp(a.id, b.id)) -} - -function runInflight(map: Map>, key: string, task: () => Promise) { - const pending = map.get(key) - if (pending) return pending - const promise = task().finally(() => { - map.delete(key) - }) - map.set(key, promise) - return promise -} - -const keyFor = (directory: string, id: string) => `${directory}\n${id}` +import { createMemo } from "solid-js" +import { produce, reconcile, type SetStoreFunction } from "solid-js/store" +import type { createServerSdkContext } from "./server-sdk" +import type { createServerSyncContextInner } from "./server-sync" +import type { State } from "./global-sync/types" const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) - -const isNotFound = (error: unknown) => - error instanceof Error && - typeof error.cause === "object" && - error.cause !== null && - (error.cause as { status?: unknown }).status === 404 - -function merge(a: readonly T[], b: readonly T[]) { - const map = new Map(a.map((item) => [item.id, item] as const)) - for (const item of b) map.set(item.id, item) - return [...map.values()].sort((x, y) => cmp(x.id, y.id)) -} - -type OptimisticStore = { - message: Record - part: Record -} - -type OptimisticAddInput = { - sessionID: string - message: Message - parts: Part[] -} - -type OptimisticRemoveInput = { - sessionID: string - messageID: string -} - -type OptimisticItem = { - message: Message - parts: Part[] -} - -type MessagePage = { - session: Message[] - part: { id: string; part: Part[] }[] - cursor?: string - complete: boolean -} - -const hasParts = (parts: Part[] | undefined, want: Part[]) => { - if (!parts) return want.length === 0 - return want.every((part) => Binary.search(parts, part.id, (item) => item.id).found) -} - -const mergeParts = (parts: Part[] | undefined, want: Part[]) => { - if (!parts) return sortParts(want) - const next = [...parts] - let changed = false - for (const part of want) { - const result = Binary.search(next, part.id, (item) => item.id) - if (result.found) continue - next.splice(result.index, 0, part) - changed = true - } - if (!changed) return parts - return next -} - -export function mergeOptimisticPage(page: MessagePage, items: OptimisticItem[]) { - if (items.length === 0) return { ...page, confirmed: [] as string[] } - - const session = [...page.session] - const part = new Map(page.part.map((item) => [item.id, sortParts(item.part)])) - const confirmed: string[] = [] - - for (const item of items) { - const result = Binary.search(session, item.message.id, (message) => message.id) - const found = result.found - if (!found) session.splice(result.index, 0, item.message) - - const current = part.get(item.message.id) - if (found && hasParts(current, item.parts)) { - confirmed.push(item.message.id) - continue - } - - part.set(item.message.id, mergeParts(current, item.parts)) - } - - return { - cursor: page.cursor, - complete: page.complete, - session, - part: [...part.entries()].sort((a, b) => cmp(a[0], b[0])).map(([id, part]) => ({ id, part })), - confirmed, - } -} - -export function applyOptimisticAdd(draft: OptimisticStore, input: OptimisticAddInput) { - const messages = draft.message[input.sessionID] - if (messages) { - const result = Binary.search(messages, input.message.id, (m) => m.id) - messages.splice(result.index, 0, input.message) - } else { - draft.message[input.sessionID] = [input.message] - } - draft.part[input.message.id] = sortParts(input.parts) -} - -export function applyOptimisticRemove(draft: OptimisticStore, input: OptimisticRemoveInput) { - const messages = draft.message[input.sessionID] - if (messages) { - const result = Binary.search(messages, input.messageID, (m) => m.id) - if (result.found) messages.splice(result.index, 1) - } - delete draft.part[input.messageID] -} - -function setOptimisticAdd(setStore: (...args: unknown[]) => void, input: OptimisticAddInput) { - setStore("message", input.sessionID, (messages: Message[] | undefined) => { - if (!messages) return [input.message] - const result = Binary.search(messages, input.message.id, (m) => m.id) - const next = [...messages] - next.splice(result.index, 0, input.message) - return next - }) - setStore("part", input.message.id, sortParts(input.parts)) -} - -function setOptimisticRemove(setStore: (...args: unknown[]) => void, input: OptimisticRemoveInput) { - setStore("message", input.sessionID, (messages: Message[] | undefined) => { - if (!messages) return messages - const result = Binary.search(messages, input.messageID, (m) => m.id) - if (!result.found) return messages - const next = [...messages] - next.splice(result.index, 1) - return next - }) - setStore("part", (part: Record) => { - if (!(input.messageID in part)) return part - const next = { ...part } - delete next[input.messageID] - return next - }) -} +const sessionFields = new Set([ + "session_status", + "session_working", + "session_diff", + "todo", + "permission", + "question", + "message", + "part", + "part_text_accum_delta", +]) export const createDirSyncContext = ( directory: string, @@ -177,210 +25,39 @@ export const createDirSyncContext = ( serverSDK: ReturnType, ) => { const client = serverSDK.createClient({ directory, throwOnError: true }) - - type Child = ReturnType<(typeof serverSync)["child"]> - type Setter = Child[1] - const current = createMemo(() => serverSync.child(directory, { mcp: true })) - const target = (directory?: string) => { - if (!directory || directory === directory) return current() - return serverSync.child(directory) - } const absolute = (path: string) => (current()[0].path.directory + "/" + path).replace("//", "/") - const initialMessagePageSize = 2 - const historyMessagePageSize = 200 - const inflight = new Map>() - const inflightDiff = new Map>() - const inflightTodo = new Map>() - const optimistic = new Map>() - const maxDirs = 30 - const seen = new Map>() - const [meta, setMeta] = createStore({ - limit: {} as Record, - cursor: {} as Record, - complete: {} as Record, - loading: {} as Record, + const data = new Proxy({} as State, { + get(_, property: keyof State) { + if (property === "session_working") return serverSync.session.data.session_working.bind(serverSync.session.data) + if (sessionFields.has(property)) return serverSync.session.data[property as keyof typeof serverSync.session.data] + return current()[0][property] + }, }) + const set = ((...input: unknown[]) => { + if (typeof input[0] === "string" && sessionFields.has(input[0])) { + return (serverSync.session.set as (...args: unknown[]) => unknown)(...input) + } + const result = (current()[1] as (...args: unknown[]) => unknown)(...input) + if (input[0] === "session") current()[0].session.forEach(serverSync.session.remember) + return result + }) as SetStoreFunction - const getSession = (sessionID: string) => { - const store = current()[0] - const match = Binary.search(store.session, sessionID, (s) => s.id) - if (match.found) return store.session[match.index] - return undefined - } - - const setOptimistic = (directory: string, sessionID: string, item: OptimisticItem) => { - const key = keyFor(directory, sessionID) - const list = optimistic.get(key) - if (list) { - list.set(item.message.id, { message: item.message, parts: sortParts(item.parts) }) + const index = (sessionID: string) => { + const session = serverSync.session.get(sessionID) + if (!session || session.directory !== directory) return + const [store, setStore] = current() + const result = Binary.search(store.session, session.id, (item) => item.id) + if (result.found) { + setStore("session", result.index, reconcile(session)) return } - optimistic.set(key, new Map([[item.message.id, { message: item.message, parts: sortParts(item.parts) }]])) - } - - const clearOptimistic = (directory: string, sessionID: string, messageID?: string) => { - const key = keyFor(directory, sessionID) - if (!messageID) { - optimistic.delete(key) - return - } - - const list = optimistic.get(key) - if (!list) return - list.delete(messageID) - if (list.size === 0) optimistic.delete(key) - } - - const getOptimistic = (directory: string, sessionID: string) => [ - ...(optimistic.get(keyFor(directory, sessionID))?.values() ?? []), - ] - - const seenFor = (directory: string) => { - const existing = seen.get(directory) - if (existing) { - seen.delete(directory) - seen.set(directory, existing) - return existing - } - const created = new Set() - seen.set(directory, created) - while (seen.size > maxDirs) { - const first = seen.keys().next().value - if (!first) break - const stale = [...(seen.get(first) ?? [])] - seen.delete(first) - const [, setStore] = serverSync.child(first, { bootstrap: false }) - evict(first, setStore, stale) - } - return created - } - - const clearMeta = (directory: string, sessionIDs: string[]) => { - if (sessionIDs.length === 0) return - for (const sessionID of sessionIDs) { - clearOptimistic(directory, sessionID) - } - setMeta( - produce((draft) => { - for (const sessionID of sessionIDs) { - const key = keyFor(directory, sessionID) - delete draft.limit[key] - delete draft.cursor[key] - delete draft.complete[key] - delete draft.loading[key] - } - }), - ) - } - - const evict = (directory: string, setStore: Setter, sessionIDs: string[]) => { - if (sessionIDs.length === 0) return - clearSessionPrefetch(serverSDK.scope, directory, sessionIDs) - for (const sessionID of sessionIDs) { - serverSync.todo.set(sessionID, undefined) - } - setStore( - produce((draft) => { - dropSessionCaches(draft, sessionIDs) - }), - ) - clearMeta(directory, sessionIDs) - } - - const touch = (directory: string, setStore: Setter, sessionID: string) => { - const stale = pickSessionCacheEvictions({ - seen: seenFor(directory), - keep: sessionID, - limit: SESSION_CACHE_LIMIT, - }) - evict(directory, setStore, stale) - } - - const fetchMessages = async (input: { client: typeof client; sessionID: string; limit: number; before?: string }) => { - const messages = await retry(() => - input.client.session.messages({ sessionID: input.sessionID, limit: input.limit, before: input.before }), - ) - const items = (messages.data ?? []).filter((x) => !!x?.info?.id) - const session = items.map((x) => clean(x.info)).sort((a, b) => cmp(a.id, b.id)) - const part = items.map((message) => ({ id: message.info.id, part: sortParts(message.parts) })) - const cursor = messages.response.headers.get("x-next-cursor") ?? undefined - return { - session, - part, - cursor, - complete: !cursor, - } - } - - const tracked = (directory: string, sessionID: string) => seen.get(directory)?.has(sessionID) ?? false - - const loadMessages = async (input: { - directory: string - client: typeof client - setStore: Setter - sessionID: string - limit: number - before?: string - mode?: "replace" | "prepend" - }) => { - const key = keyFor(input.directory, input.sessionID) - if (meta.loading[key]) return - - setMeta("loading", key, true) - await fetchMessages(input) - .then((page) => { - if (!tracked(input.directory, input.sessionID)) return - const next = mergeOptimisticPage(page, getOptimistic(input.directory, input.sessionID)) - for (const messageID of next.confirmed) { - clearOptimistic(input.directory, input.sessionID, messageID) - } - const [store] = serverSync.child(input.directory, { bootstrap: false }) - const cached = input.mode === "prepend" ? (store.message[input.sessionID] ?? []) : [] - const message = input.mode === "prepend" ? merge(cached, next.session) : next.session - batch(() => { - input.setStore("message", input.sessionID, reconcile(message, { key: "id" })) - for (const p of next.part) { - const filtered = p.part.filter((x) => !SKIP_PARTS.has(x.type)) - if (filtered.length) input.setStore("part", p.id, filtered) - } - setMeta("limit", key, message.length) - setMeta("cursor", key, next.cursor) - setMeta("complete", key, next.complete) - setSessionPrefetch({ - scope: serverSDK.scope, - directory: input.directory, - sessionID: input.sessionID, - limit: message.length, - cursor: next.cursor, - complete: next.complete, - }) - }) - }) - .catch((error) => { - if (isNotFound(error) && !tracked(input.directory, input.sessionID)) return - throw error - }) - .finally(() => { - setMeta( - produce((draft) => { - if (!tracked(input.directory, input.sessionID)) { - delete draft.loading[key] - return - } - draft.loading[key] = false - }), - ) - }) + setStore("session", produce((draft) => void draft.splice(result.index, 0, session))) } return { - get data() { - return current()[0] - }, - get set(): Setter { - return current()[1] - }, + data, + set, get status() { return current()[0].status }, @@ -389,24 +66,20 @@ export const createDirSyncContext = ( }, get project() { const store = current()[0] - const match = Binary.search(serverSync.data.project, store.project, (p) => p.id) + const match = Binary.search(serverSync.data.project, store.project, (project) => project.id) if (match.found) return serverSync.data.project[match.index] - return undefined }, session: { - get: getSession, + get(sessionID: string) { + const session = serverSync.session.get(sessionID) + if (session?.directory === directory) return session + }, optimistic: { add(input: { directory?: string; sessionID: string; message: Message; parts: Part[] }) { - const _directory = input.directory ?? directory - const [, setStore] = target(input.directory) - setOptimistic(_directory, input.sessionID, { message: input.message, parts: input.parts }) - setOptimisticAdd(setStore as (...args: unknown[]) => void, input) + serverSync.session.optimistic.add(input) }, remove(input: { directory?: string; sessionID: string; messageID: string }) { - const _directory = input.directory ?? directory - const [, setStore] = target(input.directory) - clearOptimistic(_directory, input.sessionID, input.messageID) - setOptimisticRemove(setStore as (...args: unknown[]) => void, input) + serverSync.session.optimistic.remove(input) }, }, addOptimisticMessage(input: { @@ -417,196 +90,47 @@ export const createDirSyncContext = ( model: { providerID: string; modelID: string } variant?: string }) { - const message: Message = { - id: input.messageID, + serverSync.session.optimistic.add({ sessionID: input.sessionID, - role: "user", - time: { created: Date.now() }, - agent: input.agent, - model: { ...input.model, variant: input.variant }, - } - const [, setStore] = target() - setOptimistic(directory, input.sessionID, { message, parts: input.parts }) - setOptimisticAdd(setStore as (...args: unknown[]) => void, { - sessionID: input.sessionID, - message, + message: { + id: input.messageID, + sessionID: input.sessionID, + role: "user", + time: { created: Date.now() }, + agent: input.agent, + model: { ...input.model, variant: input.variant }, + }, parts: input.parts, }) }, - async sync(sessionID: string, opts?: { force?: boolean }) { - const [store, setStore] = serverSync.child(directory) - const key = keyFor(directory, sessionID) - - touch(directory, setStore, sessionID) - - const seeded = getSessionPrefetch(serverSDK.scope, directory, sessionID) - if (seeded && store.message[sessionID] !== undefined && meta.limit[key] === undefined) { - batch(() => { - setMeta("limit", key, seeded.limit) - setMeta("cursor", key, seeded.cursor) - setMeta("complete", key, seeded.complete) - setMeta("loading", key, false) - }) - } - - return runInflight(inflight, key, async () => { - const pending = getSessionPrefetchPromise(serverSDK.scope, directory, sessionID) - if (pending) { - await pending - const seeded = getSessionPrefetch(serverSDK.scope, directory, sessionID) - if (seeded && store.message[sessionID] !== undefined && meta.limit[key] === undefined) { - batch(() => { - setMeta("limit", key, seeded.limit) - setMeta("cursor", key, seeded.cursor) - setMeta("complete", key, seeded.complete) - setMeta("loading", key, false) - }) - } - } - - const hasSession = Binary.search(store.session, sessionID, (s) => s.id).found - const cached = store.message[sessionID] !== undefined && meta.limit[key] !== undefined - if (cached && hasSession && !opts?.force) return - - const limit = meta.limit[key] ?? initialMessagePageSize - const sessionReq = - hasSession && !opts?.force - ? Promise.resolve() - : retry(() => client.session.get({ sessionID })) - .then((session) => { - if (!tracked(directory, sessionID)) return - const data = session.data - if (!data) return - setStore( - "session", - produce((draft) => { - const match = Binary.search(draft, sessionID, (s) => s.id) - if (match.found) { - draft[match.index] = data - return - } - draft.splice(match.index, 0, data) - }), - ) - }) - .catch((error) => { - if (isNotFound(error) && !tracked(directory, sessionID)) return - throw error - }) - - const messagesReq = - cached && !opts?.force - ? Promise.resolve() - : loadMessages({ - directory, - client, - setStore, - sessionID, - limit, - }) - - await Promise.all([sessionReq, messagesReq]) - }) + async sync(sessionID: string, options?: { force?: boolean }) { + await serverSync.session.sync(sessionID, options) + index(sessionID) }, - async diff(sessionID: string, opts?: { force?: boolean }) { - const [store, setStore] = serverSync.child(directory) - touch(directory, setStore, sessionID) - if (store.session_diff[sessionID] !== undefined && !opts?.force) return - - const key = keyFor(directory, sessionID) - return runInflight(inflightDiff, key, () => - retry(() => client.session.diff({ sessionID })).then((diff) => { - if (!tracked(directory, sessionID)) return - setStore("session_diff", sessionID, reconcile(list(diff.data), { key: "file" })) - }), - ) - }, - async todo(sessionID: string, opts?: { force?: boolean }) { - const [store, setStore] = serverSync.child(directory) - touch(directory, setStore, sessionID) - const existing = store.todo[sessionID] - const cached = serverSync.data.session_todo[sessionID] - if (existing !== undefined) { - if (cached === undefined) { - serverSync.todo.set(sessionID, existing) - } - if (!opts?.force) return - } - - if (cached !== undefined) { - setStore("todo", sessionID, reconcile(cached, { key: "id" })) - } - - const key = keyFor(directory, sessionID) - return runInflight(inflightTodo, key, () => - retry(() => client.session.todo({ sessionID })).then((todo) => { - if (!tracked(directory, sessionID)) return - const list = todo.data ?? [] - setStore("todo", sessionID, reconcile(list, { key: "id" })) - serverSync.todo.set(sessionID, list) - }), - ) - }, - history: { - more(sessionID: string) { - const store = current()[0] - const key = keyFor(directory, sessionID) - if (store.message[sessionID] === undefined) return false - if (meta.limit[key] === undefined) return false - if (meta.complete[key]) return false - return !!meta.cursor[key] - }, - loading(sessionID: string) { - const key = keyFor(directory, sessionID) - return meta.loading[key] ?? false - }, - async loadMore(sessionID: string, count?: number) { - const [, setStore] = serverSync.child(directory) - touch(directory, setStore, sessionID) - const key = keyFor(directory, sessionID) - const step = count ?? historyMessagePageSize - if (meta.loading[key]) return - if (meta.complete[key]) return - const before = meta.cursor[key] - if (!before) return - - await loadMessages({ - directory, - client, - setStore, - sessionID, - limit: step, - before, - mode: "prepend", - }) - }, - }, - evict(sessionID: string, _directory = directory) { - const [, setStore] = serverSync.child(_directory) - seenFor(_directory).delete(sessionID) - evict(_directory, setStore, [sessionID]) + diff: serverSync.session.diff, + todo: serverSync.session.todo, + history: serverSync.session.history, + evict(sessionID: string) { + serverSync.session.evict(sessionID) }, fetch: async (count = 10) => { - const [store, setStore] = serverSync.child(directory) - setStore("limit", (x) => x + count) - await client.session.list().then((x) => { - const sessions = (x.data ?? []) - .filter((s) => !!s?.id) - .sort((a, b) => cmp(a.id, b.id)) - .slice(0, store.limit) - setStore("session", reconcile(sessions, { key: "id" })) - }) + const [store, setStore] = current() + setStore("limit", (value) => value + count) + const response = await client.session.list() + const sessions = (response.data ?? []) + .filter((session) => !!session?.id) + .sort((a, b) => cmp(a.id, b.id)) + .slice(0, store.limit) + sessions.forEach(serverSync.session.remember) + setStore("session", reconcile(sessions, { key: "id" })) }, more: createMemo(() => current()[0].session.length >= current()[0].limit), archive: async (sessionID: string) => { - const [, setStore] = serverSync.child(directory) - await client.session.update({ sessionID, time: { archived: Date.now() } }) - setStore( - produce((draft) => { - const match = Binary.search(draft.session, sessionID, (s) => s.id) - if (match.found) draft.session.splice(match.index, 1) - }), - ) + await serverSDK.client.session.update({ sessionID, time: { archived: Date.now() } }) + current()[1]("session", produce((draft) => { + const match = Binary.search(draft, sessionID, (session) => session.id) + if (match.found) draft.splice(match.index, 1) + })) }, }, mcp: { diff --git a/packages/app/src/context/global-sync/bootstrap.ts b/packages/app/src/context/global-sync/bootstrap.ts index b24bdda811..7055e6553b 100644 --- a/packages/app/src/context/global-sync/bootstrap.ts +++ b/packages/app/src/context/global-sync/bootstrap.ts @@ -7,14 +7,14 @@ import type { ProviderAuthResponse, QuestionRequest, Session, - Todo, } from "@opencode-ai/sdk/v2/client" import { showToast } from "@/utils/toast" import { getFilename } from "@opencode-ai/core/util/path" import { retry } from "@opencode-ai/core/util/retry" import { batch } from "solid-js" -import { reconcile, type SetStoreFunction, type Store } from "solid-js/store" +import { produce, reconcile, type SetStoreFunction, type Store } from "solid-js/store" import type { State, VcsCache } from "./types" +import type { ServerSession } from "../server-session" import { cmp, normalizeAgentList, normalizeProviderList } from "./utils" import { formatServerError } from "@/utils/server-errors" import { QueryClient, queryOptions } from "@tanstack/solid-query" @@ -26,9 +26,6 @@ type GlobalStore = { ready: boolean path: Path project: Project[] - session_todo: { - [sessionID: string]: Todo[] - } provider: NormalizedProviderListResponse provider_auth: ProviderAuthResponse config: Config @@ -215,6 +212,7 @@ export async function bootstrapDirectory(input: { provider: NormalizedProviderListResponse } queryClient: QueryClient + session?: ServerSession }) { const loading = input.store.status !== "complete" const seededProject = projectID(input.directory, input.global.project) @@ -238,7 +236,25 @@ export async function bootstrapDirectory(input: { .then((data) => input.setStore("agent", data)), () => retry(() => input.sdk.config.get().then((x) => input.setStore("config", reconcile(x.data!, { merge: false })))), - () => retry(() => input.sdk.session.status().then((x) => input.setStore("session_status", x.data!))), + () => + retry(() => + input.sdk.session.status().then(async (x) => { + if (input.session) { + const statuses = x.data ?? {} + await Promise.all(Object.keys(statuses).map((sessionID) => input.session!.resolve(sessionID).catch(() => undefined))) + input.session.set("session_status", produce((draft) => { + for (const sessionID of Object.keys(draft)) { + if (statuses[sessionID]) continue + if (input.session?.get(sessionID)?.directory === input.directory) delete draft[sessionID] + } + })) + for (const [sessionID, status] of Object.entries(statuses)) { + input.session.set("session_status", sessionID, reconcile(status)) + } + } + if (!input.session) input.setStore("session_status", x.data!) + }), + ), !seededProject && (() => retry(() => input.sdk.project.current()).then((x) => input.setStore("project", x.data!.id))), !seededPath && @@ -263,21 +279,25 @@ export async function bootstrapDirectory(input: { const grouped = groupBySession( (x.data ?? []).filter((perm): perm is PermissionRequest => !!perm?.id && !!perm.sessionID), ) - return warmSessions({ ids, store: input.store, setStore: input.setStore, sdk: input.sdk }).then(() => + const warm = input.session + ? Promise.all(ids.map((sessionID) => input.session!.resolve(sessionID))).then(() => undefined) + : warmSessions({ ids, store: input.store, setStore: input.setStore, sdk: input.sdk }) + return warm.then(() => batch(() => { - for (const sessionID of Object.keys(input.store.permission)) { + const current = input.session?.data.permission ?? input.store.permission + for (const sessionID of Object.keys(current)) { if (grouped[sessionID]) continue - input.setStore("permission", sessionID, []) + if (input.session?.get(sessionID)?.directory !== input.directory) continue + if (input.session) input.session.set("permission", sessionID, []) + if (!input.session) input.setStore("permission", sessionID, []) } for (const [sessionID, permissions] of Object.entries(grouped)) { - input.setStore( - "permission", - sessionID, - reconcile( - permissions.filter((p) => !!p?.id).sort((a, b) => cmp(a.id, b.id)), - { key: "id" }, - ), + const value = reconcile( + permissions.filter((p) => !!p?.id).sort((a, b) => cmp(a.id, b.id)), + { key: "id" }, ) + if (input.session) input.session.set("permission", sessionID, value) + if (!input.session) input.setStore("permission", sessionID, value) } }), ) @@ -288,21 +308,25 @@ export async function bootstrapDirectory(input: { input.sdk.question.list().then((x) => { const ids = (x.data ?? []).map((question) => question?.sessionID).filter((id): id is string => !!id) const grouped = groupBySession((x.data ?? []).filter((q): q is QuestionRequest => !!q?.id && !!q.sessionID)) - return warmSessions({ ids, store: input.store, setStore: input.setStore, sdk: input.sdk }).then(() => + const warm = input.session + ? Promise.all(ids.map((sessionID) => input.session!.resolve(sessionID))).then(() => undefined) + : warmSessions({ ids, store: input.store, setStore: input.setStore, sdk: input.sdk }) + return warm.then(() => batch(() => { - for (const sessionID of Object.keys(input.store.question)) { + const current = input.session?.data.question ?? input.store.question + for (const sessionID of Object.keys(current)) { if (grouped[sessionID]) continue - input.setStore("question", sessionID, []) + if (input.session?.get(sessionID)?.directory !== input.directory) continue + if (input.session) input.session.set("question", sessionID, []) + if (!input.session) input.setStore("question", sessionID, []) } for (const [sessionID, questions] of Object.entries(grouped)) { - input.setStore( - "question", - sessionID, - reconcile( - questions.filter((q) => !!q?.id).sort((a, b) => cmp(a.id, b.id)), - { key: "id" }, - ), + const value = reconcile( + questions.filter((q) => !!q?.id).sort((a, b) => cmp(a.id, b.id)), + { key: "id" }, ) + if (input.session) input.session.set("question", sessionID, value) + if (!input.session) input.setStore("question", sessionID, value) } }), ) diff --git a/packages/app/src/context/global-sync/event-reducer.ts b/packages/app/src/context/global-sync/event-reducer.ts index 9ca8f8bdad..bca3f85720 100644 --- a/packages/app/src/context/global-sync/event-reducer.ts +++ b/packages/app/src/context/global-sync/event-reducer.ts @@ -17,6 +17,21 @@ import { dropSessionCaches } from "./session-cache" import { diffs as list, message as clean } from "@/utils/diffs" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) +const SESSION_CONTENT_EVENTS = new Set([ + "session.diff", + "todo.updated", + "session.status", + "message.updated", + "message.removed", + "message.part.updated", + "message.part.removed", + "message.part.delta", + "permission.asked", + "permission.replied", + "question.asked", + "question.replied", + "question.rejected", +]) export function applyGlobalEvent(input: { event: { type: string; properties?: unknown } @@ -100,8 +115,11 @@ export function applyDirectoryEvent(input: { vcsCache?: VcsCache setSessionTodo?: (sessionID: string, todos: Todo[] | undefined) => void retainedLimit?: number + sessionContent?: boolean + permission?: State["permission"] }) { const event = input.event + if (input.sessionContent === false && SESSION_CONTENT_EVENTS.has(event.type)) return const limit = Math.max(input.store.limit, input.retainedLimit ?? 0) switch (event.type) { case "server.instance.disposed": { @@ -117,7 +135,7 @@ export function applyDirectoryEvent(input: { } const next = input.store.session.slice() next.splice(result.index, 0, info) - const trimmed = trimSessions(next, { limit, permission: input.store.permission }) + const trimmed = trimSessions(next, { limit, permission: input.permission ?? input.store.permission }) input.setStore("session", reconcile(trimmed, { key: "id" })) cleanupDroppedSessionCaches(input.store, input.setStore, trimmed, input.setSessionTodo) if (!info.parentID) input.setStore("sessionTotal", (value) => value + 1) @@ -147,7 +165,7 @@ export function applyDirectoryEvent(input: { } const next = input.store.session.slice() next.splice(result.index, 0, info) - const trimmed = trimSessions(next, { limit, permission: input.store.permission }) + const trimmed = trimSessions(next, { limit, permission: input.permission ?? input.store.permission }) input.setStore("session", reconcile(trimmed, { key: "id" })) cleanupDroppedSessionCaches(input.store, input.setStore, trimmed, input.setSessionTodo) break diff --git a/packages/app/src/context/global-sync/session-prefetch.test.ts b/packages/app/src/context/global-sync/session-prefetch.test.ts deleted file mode 100644 index cf8ee77b4b..0000000000 --- a/packages/app/src/context/global-sync/session-prefetch.test.ts +++ /dev/null @@ -1,139 +0,0 @@ -import { describe, expect, test } from "bun:test" -import { - clearSessionPrefetch, - clearSessionPrefetchDirectory, - getSessionPrefetch, - runSessionPrefetch, - setSessionPrefetch, - shouldSkipSessionPrefetch, -} from "./session-prefetch" -import { ServerScope } from "@/utils/server-scope" - -const scope = ServerScope.local - -describe("session prefetch", () => { - test("stores and clears message metadata by directory", () => { - clearSessionPrefetch(scope, "/tmp/a", ["ses_1"]) - clearSessionPrefetch(scope, "/tmp/b", ["ses_1"]) - - setSessionPrefetch({ - directory: "/tmp/a", - scope, - sessionID: "ses_1", - limit: 200, - cursor: "abc", - complete: false, - at: 123, - }) - - expect(getSessionPrefetch(scope, "/tmp/a", "ses_1")).toEqual({ - limit: 200, - cursor: "abc", - complete: false, - at: 123, - }) - expect(getSessionPrefetch(scope, "/tmp/b", "ses_1")).toBeUndefined() - - clearSessionPrefetch(scope, "/tmp/a", ["ses_1"]) - - expect(getSessionPrefetch(scope, "/tmp/a", "ses_1")).toBeUndefined() - }) - - test("dedupes inflight work", async () => { - clearSessionPrefetch(scope, "/tmp/c", ["ses_2"]) - - let calls = 0 - const run = () => - runSessionPrefetch({ - directory: "/tmp/c", - scope, - sessionID: "ses_2", - task: async () => { - calls += 1 - return { limit: 100, cursor: "next", complete: true, at: 456 } - }, - }) - - const [a, b] = await Promise.all([run(), run()]) - - expect(calls).toBe(1) - expect(a).toEqual({ limit: 100, cursor: "next", complete: true, at: 456 }) - expect(b).toEqual({ limit: 100, cursor: "next", complete: true, at: 456 }) - }) - - test("clears a whole directory", () => { - setSessionPrefetch({ - scope, - directory: "/tmp/d", - sessionID: "ses_1", - limit: 10, - cursor: "a", - complete: true, - at: 1, - }) - setSessionPrefetch({ - scope, - directory: "/tmp/d", - sessionID: "ses_2", - limit: 20, - cursor: "b", - complete: false, - at: 2, - }) - setSessionPrefetch({ - scope, - directory: "/tmp/e", - sessionID: "ses_1", - limit: 30, - cursor: "c", - complete: true, - at: 3, - }) - - clearSessionPrefetchDirectory(scope, "/tmp/d") - - expect(getSessionPrefetch(scope, "/tmp/d", "ses_1")).toBeUndefined() - expect(getSessionPrefetch(scope, "/tmp/d", "ses_2")).toBeUndefined() - expect(getSessionPrefetch(scope, "/tmp/e", "ses_1")).toEqual({ limit: 30, cursor: "c", complete: true, at: 3 }) - }) - - test("isolates identical directories and sessions by server scope", () => { - const remote = "https://debian.example" as ServerScope - setSessionPrefetch({ scope, directory: "/repo", sessionID: "ses_1", limit: 10, complete: true, at: 1 }) - setSessionPrefetch({ scope: remote, directory: "/repo", sessionID: "ses_1", limit: 20, complete: true, at: 2 }) - - expect(getSessionPrefetch(scope, "/repo", "ses_1")?.limit).toBe(10) - expect(getSessionPrefetch(remote, "/repo", "ses_1")?.limit).toBe(20) - }) - - test("refreshes stale first-page prefetched history", () => { - expect( - shouldSkipSessionPrefetch({ - message: true, - info: { limit: 200, cursor: "x", complete: false, at: 1 }, - chunk: 200, - now: 1 + 15_001, - }), - ).toBe(false) - }) - - test("keeps deeper or complete history cached", () => { - expect( - shouldSkipSessionPrefetch({ - message: true, - info: { limit: 400, cursor: "x", complete: false, at: 1 }, - chunk: 200, - now: 1 + 15_001, - }), - ).toBe(true) - - expect( - shouldSkipSessionPrefetch({ - message: true, - info: { limit: 120, complete: true, at: 1 }, - chunk: 200, - now: 1 + 15_001, - }), - ).toBe(true) - }) -}) diff --git a/packages/app/src/context/global-sync/session-prefetch.ts b/packages/app/src/context/global-sync/session-prefetch.ts deleted file mode 100644 index d2c3d86f4e..0000000000 --- a/packages/app/src/context/global-sync/session-prefetch.ts +++ /dev/null @@ -1,107 +0,0 @@ -import { ScopedKey, type ServerScope } from "@/utils/server-scope" - -const key = (scope: ServerScope, directory: string, sessionID: string) => ScopedKey.from(scope, directory, sessionID) - -export const SESSION_PREFETCH_TTL = 15_000 - -type Meta = { - limit: number - cursor?: string - complete: boolean - at: number -} - -export function shouldSkipSessionPrefetch(input: { message: boolean; info?: Meta; chunk: number; now?: number }) { - if (input.message) { - if (!input.info) return true - if (input.info.complete) return true - if (input.info.limit > input.chunk) return true - } else { - if (!input.info) return false - } - - return (input.now ?? Date.now()) - input.info.at < SESSION_PREFETCH_TTL -} - -const cache = new Map() -const inflight = new Map>() -const rev = new Map() - -const version = (id: string) => rev.get(id) ?? 0 - -export function getSessionPrefetch(scope: ServerScope, directory: string, sessionID: string) { - return cache.get(key(scope, directory, sessionID)) -} - -export function getSessionPrefetchPromise(scope: ServerScope, directory: string, sessionID: string) { - return inflight.get(key(scope, directory, sessionID)) -} - -export function clearSessionPrefetchInflight(scope: ServerScope) { - const prefix = ScopedKey.prefix(scope) - for (const id of inflight.keys()) { - if (id.startsWith(prefix)) inflight.delete(id) - } -} - -export function isSessionPrefetchCurrent(scope: ServerScope, directory: string, sessionID: string, value: number) { - return version(key(scope, directory, sessionID)) === value -} - -export function runSessionPrefetch(input: { - directory: string - scope: ServerScope - sessionID: string - task: (value: number) => Promise -}) { - const id = key(input.scope, input.directory, input.sessionID) - const pending = inflight.get(id) - if (pending) return pending - - const value = version(id) - - const promise = input.task(value).finally(() => { - if (inflight.get(id) === promise) inflight.delete(id) - }) - - inflight.set(id, promise) - return promise -} - -export function setSessionPrefetch(input: { - directory: string - scope: ServerScope - sessionID: string - limit: number - cursor?: string - complete: boolean - at?: number -}) { - cache.set(key(input.scope, input.directory, input.sessionID), { - limit: input.limit, - cursor: input.cursor, - complete: input.complete, - at: input.at ?? Date.now(), - }) -} - -export function clearSessionPrefetch(scope: ServerScope, directory: string, sessionIDs: Iterable) { - for (const sessionID of sessionIDs) { - if (!sessionID) continue - const id = key(scope, directory, sessionID) - rev.set(id, version(id) + 1) - cache.delete(id) - inflight.delete(id) - } -} - -export function clearSessionPrefetchDirectory(scope: ServerScope, directory: string) { - const prefix = ScopedKey.prefix(scope, directory) - const keys = new Set([...cache.keys(), ...inflight.keys()]) - for (const id of keys) { - if (!id.startsWith(prefix)) continue - rev.set(id, version(id) + 1) - cache.delete(id) - inflight.delete(id) - } -} diff --git a/packages/app/src/context/global.tsx b/packages/app/src/context/global.tsx index cebf811c7c..138b17c4fc 100644 --- a/packages/app/src/context/global.tsx +++ b/packages/app/src/context/global.tsx @@ -8,13 +8,11 @@ import { createServerSyncContext } from "./server-sync" import { getOwner } from "solid-js/web" import { QueryClient } from "@tanstack/solid-query" import type { ServerScope } from "@/utils/server-scope" -import { createSessionPlacementStore } from "@/utils/session-placement" export const { use: useGlobal, provider: GlobalProvider } = createSimpleContext({ name: "Global", init: () => { const server = useServer() - const sessionPlacement = createSessionPlacementStore() const serverHealth = useServerHealth( () => server.list, () => true, @@ -87,7 +85,6 @@ export const { use: useGlobal, provider: GlobalProvider } = createSimpleContext( }, }, }, - sessionPlacement, ensureServerCtx(conn: ServerConnection.Any) { return ensureServerCtx(conn) }, diff --git a/packages/app/src/context/server-session.test.ts b/packages/app/src/context/server-session.test.ts new file mode 100644 index 0000000000..555aa8e9f0 --- /dev/null +++ b/packages/app/src/context/server-session.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, test } from "bun:test" +import type { OpencodeClient, Session } from "@opencode-ai/sdk/v2/client" +import { createServerSession } from "./server-session" + +const session = (id: string, parentID?: string): Session => ({ + id, + slug: id, + projectID: "project", + directory: "/repo", + title: id, + version: "1", + parentID, + time: { created: 1, updated: 1 }, +}) + +function setup(sessions: Record) { + const get: unknown[] = [] + const messages: unknown[] = [] + const client = { + session: { + get: async (input: unknown) => { + get.push(input) + const id = (input as { sessionID: string }).sessionID + return { data: sessions[id] } + }, + messages: async (input: unknown) => { + messages.push(input) + return { data: [], response: { headers: new Headers() } } + }, + diff: async () => ({ data: [] }), + todo: async () => ({ data: [] }), + }, + } as unknown as OpencodeClient + return { get, messages, store: createServerSession(client) } +} + +describe("server session", () => { + test("resolves lineage by session ID without directory", async () => { + const ctx = setup({ child: session("child", "root"), root: session("root") }) + + const result = await ctx.store.lineage.resolve("child") + + expect(result.root.id).toBe("root") + expect(ctx.get).toEqual([{ sessionID: "child" }, { sessionID: "root" }]) + expect(ctx.store.lineage.peek("child")).toEqual(result) + }) + + test("loads session content through the server client", async () => { + const ctx = setup({ root: session("root") }) + + await ctx.store.sync("root") + + expect(ctx.get).toEqual([{ sessionID: "root" }]) + expect(ctx.messages).toEqual([{ sessionID: "root", limit: 2, before: undefined }]) + expect(ctx.store.data.message.root).toEqual([]) + }) + + test("applies events without a directory store", () => { + const ctx = setup({}) + ctx.store.apply({ type: "session.created", properties: { info: session("root") } }) + ctx.store.apply({ type: "session.status", properties: { sessionID: "root", status: { type: "busy" } } }) + + expect(ctx.store.get("root")?.directory).toBe("/repo") + expect(ctx.store.data.session_working("root")).toBe(true) + }) + + test("preserves pinned session content under server-wide cache pressure", () => { + const ctx = setup({}) + ctx.store.pin("active") + ctx.store.optimistic.add({ + sessionID: "active", + message: { id: "message", sessionID: "active", role: "assistant", time: { created: 1 }, parentID: "parent", modelID: "model", providerID: "provider", mode: "build", agent: "agent", path: { cwd: "/repo", root: "/repo" }, cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } }, + parts: [], + }) + + for (let index = 0; index < 50; index++) { + ctx.store.apply({ type: "session.status", properties: { sessionID: `session-${index}`, status: { type: "busy" } } }) + } + + expect(ctx.store.data.message.active?.map((message) => message.id)).toEqual(["message"]) + }) +}) diff --git a/packages/app/src/context/server-session.ts b/packages/app/src/context/server-session.ts new file mode 100644 index 0000000000..5566b9e7b1 --- /dev/null +++ b/packages/app/src/context/server-session.ts @@ -0,0 +1,538 @@ +import { Binary } from "@opencode-ai/core/util/binary" +import { retry } from "@opencode-ai/core/util/retry" +import type { + Message, + OpencodeClient, + Part, + PermissionRequest, + QuestionRequest, + Session, + SessionStatus, + SnapshotFileDiff, + Todo, +} from "@opencode-ai/sdk/v2/client" +import { batch } from "solid-js" +import { createStore, produce, reconcile } from "solid-js/store" +import { diffs as cleanDiffs, message as cleanMessage } from "@/utils/diffs" +import { rootSession } from "@/utils/session-route" +import { dropSessionCaches, pickSessionCacheEvictions, SESSION_CACHE_LIMIT } from "./global-sync/session-cache" + +const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) +const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) +const initialMessagePageSize = 2 +const historyMessagePageSize = 200 +const sessionInfoLimit = 2_048 + +type OptimisticItem = { + message: Message + parts: Part[] +} + +const hasParts = (parts: Part[] | undefined, want: Part[]) => { + if (!parts) return want.length === 0 + return want.every((part) => Binary.search(parts, part.id, (item) => item.id).found) +} + +function mergeOptimisticPage( + page: { session: Message[]; part: { id: string; part: Part[] }[]; cursor?: string; complete: boolean }, + items: OptimisticItem[], +) { + if (items.length === 0) return { ...page, confirmed: [] as string[] } + const session = [...page.session] + const part = new Map(page.part.map((item) => [item.id, item.part])) + const confirmed: string[] = [] + for (const item of items) { + const result = Binary.search(session, item.message.id, (message) => message.id) + if (!result.found) session.splice(result.index, 0, item.message) + const current = part.get(item.message.id) + if (result.found && hasParts(current, item.parts)) { + confirmed.push(item.message.id) + continue + } + part.set(item.message.id, merge(current ?? [], item.parts)) + } + return { + ...page, + session, + part: [...part.entries()].sort((a, b) => cmp(a[0], b[0])).map(([id, parts]) => ({ id, part: parts })), + confirmed, + } +} + +function runInflight(map: Map>, key: string, task: () => Promise) { + const pending = map.get(key) + if (pending) return pending + const promise = task().finally(() => { + if (map.get(key) === promise) map.delete(key) + }) + map.set(key, promise) + return promise +} + +function merge(a: readonly T[], b: readonly T[]) { + const items = new Map(a.map((item) => [item.id, item] as const)) + for (const item of b) items.set(item.id, item) + return [...items.values()].sort((x, y) => cmp(x.id, y.id)) +} + +export function createServerSession(client: OpencodeClient) { + const [data, setData] = createStore({ + info: {} as Record, + session_status: {} as Record, + session_diff: {} as Record, + todo: {} as Record, + permission: {} as Record, + question: {} as Record, + message: {} as Record, + part: {} as Record, + part_text_accum_delta: {} as Record, + session_working(id: string) { + return (this.session_status[id]?.type ?? "idle") !== "idle" + }, + }) + const requests = new Map>() + const inflight = new Map>() + const inflightDiff = new Map>() + const inflightTodo = new Map>() + const optimistic = new Map>() + const seen = new Set() + const infoSeen = new Set() + const pinned = new Map() + const generations = new Map() + const [meta, setMeta] = createStore({ + limit: {} as Record, + cursor: {} as Record, + complete: {} as Record, + loading: {} as Record, + at: {} as Record, + }) + + const remember = (session: Session) => { + setData("info", session.id, reconcile(session)) + infoSeen.delete(session.id) + infoSeen.add(session.id) + if (infoSeen.size > sessionInfoLimit) { + const preserve = new Set([ + ...pinned.keys(), + ...requests.keys(), + ...Object.entries(data.permission).filter(([, items]) => items.length > 0).map(([sessionID]) => sessionID), + ...Object.entries(data.question).filter(([, items]) => items.length > 0).map(([sessionID]) => sessionID), + ...Object.entries(data.session_status).filter(([, status]) => status.type !== "idle").map(([sessionID]) => sessionID), + ]) + for (const sessionID of preserve) { + let current = data.info[sessionID] + while (current) { + preserve.add(current.id) + current = current.parentID ? data.info[current.parentID] : undefined + } + } + const stale: string[] = [] + for (const sessionID of infoSeen) { + if (infoSeen.size - stale.length <= sessionInfoLimit) break + if (!preserve.has(sessionID)) stale.push(sessionID) + } + stale.forEach((sessionID) => infoSeen.delete(sessionID)) + setData("info", produce((draft) => stale.forEach((sessionID) => delete draft[sessionID]))) + } + return session + } + + const resolve = (sessionID: string, options?: { force?: boolean }) => { + const cached = data.info[sessionID] + if (cached && !options?.force) return Promise.resolve(cached) + const pending = requests.get(sessionID) + if (pending) return pending + const generation = generations.get(sessionID) ?? 0 + const request = client.session.get({ sessionID }).then((result) => { + if (!result.data) throw new Error(`Session not found: ${sessionID}`) + if ((generations.get(sessionID) ?? 0) !== generation) return result.data + return remember(result.data) + }) + requests.set(sessionID, request) + void request.then( + () => { + if (requests.get(sessionID) === request) requests.delete(sessionID) + }, + () => { + if (requests.get(sessionID) === request) requests.delete(sessionID) + }, + ) + return request + } + + const peekLineage = (sessionID: string) => { + const session = data.info[sessionID] + if (!session) return + const seen = new Set([session.id]) + let root = session + while (root.parentID) { + if (seen.has(root.parentID)) throw new Error(`Session parent cycle: ${root.parentID}`) + seen.add(root.parentID) + const parent = data.info[root.parentID] + if (!parent) return + root = parent + } + return { session, root } + } + + const clearOptimistic = (sessionID: string, messageID?: string) => { + if (!messageID) { + optimistic.delete(sessionID) + return + } + const items = optimistic.get(sessionID) + if (!items) return + items.delete(messageID) + if (items.size === 0) optimistic.delete(sessionID) + } + + const evict = (sessionIDs: string[]) => { + if (sessionIDs.length === 0) return + sessionIDs.forEach((sessionID) => { + generations.set(sessionID, (generations.get(sessionID) ?? 0) + 1) + clearOptimistic(sessionID) + requests.delete(sessionID) + inflight.delete(sessionID) + inflightDiff.delete(sessionID) + inflightTodo.delete(sessionID) + }) + setData( + produce((draft) => { + dropSessionCaches(draft, sessionIDs) + }), + ) + setMeta( + produce((draft) => { + for (const sessionID of sessionIDs) { + delete draft.limit[sessionID] + delete draft.cursor[sessionID] + delete draft.complete[sessionID] + delete draft.loading[sessionID] + delete draft.at[sessionID] + } + }), + ) + } + + const protectedSessions = () => new Set([ + ...pinned.keys(), + ...requests.keys(), + ...inflight.keys(), + ...inflightDiff.keys(), + ...inflightTodo.keys(), + ...optimistic.keys(), + ...Object.entries(data.permission).filter(([, items]) => items.length > 0).map(([sessionID]) => sessionID), + ...Object.entries(data.question).filter(([, items]) => items.length > 0).map(([sessionID]) => sessionID), + ...Object.entries(data.session_status).filter(([, status]) => status.type !== "idle").map(([sessionID]) => sessionID), + ]) + + const touch = (sessionID: string) => + evict(pickSessionCacheEvictions({ seen, keep: sessionID, limit: SESSION_CACHE_LIMIT, preserve: protectedSessions() })) + + const fetchMessages = async (sessionID: string, limit: number, before?: string) => { + const response = await retry(() => client.session.messages({ sessionID, limit, before })) + const items = (response.data ?? []).filter((item) => !!item?.info?.id) + return { + session: items.map((item) => cleanMessage(item.info)).sort((a, b) => cmp(a.id, b.id)), + part: items.map((item) => ({ id: item.info.id, part: item.parts.filter((part) => !!part?.id).sort((a, b) => cmp(a.id, b.id)) })), + cursor: response.response.headers.get("x-next-cursor") ?? undefined, + complete: !response.response.headers.get("x-next-cursor"), + } + } + + const loadMessages = async (sessionID: string, limit: number, before?: string, mode?: "replace" | "prepend") => { + if (meta.loading[sessionID]) return + const generation = generations.get(sessionID) ?? 0 + setMeta("loading", sessionID, true) + await fetchMessages(sessionID, limit, before) + .then((page) => { + if ((generations.get(sessionID) ?? 0) !== generation) return + const next = mergeOptimisticPage(page, [...(optimistic.get(sessionID)?.values() ?? [])]) + next.confirmed.forEach((messageID) => clearOptimistic(sessionID, messageID)) + const messages = mode === "prepend" ? merge(data.message[sessionID] ?? [], next.session) : next.session + batch(() => { + setData("message", sessionID, reconcile(messages, { key: "id" })) + for (const item of next.part) { + const parts = item.part.filter((part) => !SKIP_PARTS.has(part.type)) + if (parts.length) setData("part", item.id, reconcile(parts, { key: "id" })) + } + setMeta("limit", sessionID, messages.length) + setMeta("cursor", sessionID, next.cursor) + setMeta("complete", sessionID, next.complete) + setMeta("at", sessionID, Date.now()) + }) + }) + .finally(() => { + if ((generations.get(sessionID) ?? 0) === generation) setMeta("loading", sessionID, false) + }) + } + + const sync = (sessionID: string, options?: { force?: boolean; messageLimit?: number }) => { + touch(sessionID) + return runInflight(inflight, sessionID, async () => { + const cached = data.message[sessionID] !== undefined && meta.limit[sessionID] !== undefined + if (cached && data.info[sessionID] && !options?.force) return + await Promise.all([ + resolve(sessionID, options), + cached && !options?.force + ? Promise.resolve() + : loadMessages(sessionID, options?.messageLimit ?? meta.limit[sessionID] ?? initialMessagePageSize), + ]) + }) + } + + const prefetch = async (sessionID: string, limit: number) => { + touch(sessionID) + await inflight.get(sessionID) + if (Date.now() - (meta.at[sessionID] ?? 0) <= 15_000 && (meta.complete[sessionID] || (data.message[sessionID]?.length ?? 0) >= limit)) return + await runInflight(inflight, sessionID, () => loadMessages(sessionID, limit)) + } + + const eventSessionID = (event: { type: string; properties?: unknown }) => { + const properties = event.properties + if (!properties || typeof properties !== "object") return + if ("sessionID" in properties && typeof properties.sessionID === "string") return properties.sessionID + if ("info" in properties && properties.info && typeof properties.info === "object" && "sessionID" in properties.info && typeof properties.info.sessionID === "string") return properties.info.sessionID + if ("part" in properties && properties.part && typeof properties.part === "object" && "sessionID" in properties.part && typeof properties.part.sessionID === "string") return properties.part.sessionID + } + + const apply = (event: { type: string; properties?: unknown }) => { + const eventID = eventSessionID(event) + if (eventID) { + touch(eventID) + if (!data.info[eventID]) void resolve(eventID).catch(() => {}) + } + switch (event.type) { + case "session.created": + remember((event.properties as { info: Session }).info) + return + case "session.updated": { + const info = (event.properties as { info: Session }).info + remember(info) + if (info.time.archived) evict([info.id]) + return + } + case "session.deleted": { + const sessionID = (event.properties as { info: Session }).info.id + infoSeen.delete(sessionID) + setData("info", produce((draft) => void delete draft[sessionID])) + evict([sessionID]) + return + } + case "session.diff": { + const props = event.properties as { sessionID: string; diff: SnapshotFileDiff[] } + setData("session_diff", props.sessionID, reconcile(cleanDiffs(props.diff), { key: "file" })) + return + } + case "todo.updated": { + const props = event.properties as { sessionID: string; todos: Todo[] } + setData("todo", props.sessionID, reconcile(props.todos, { key: "id" })) + return + } + case "session.status": { + const props = event.properties as { sessionID: string; status: SessionStatus } + setData("session_status", props.sessionID, reconcile(props.status)) + return + } + case "message.updated": { + const info = cleanMessage((event.properties as { info: Message }).info) + const messages = data.message[info.sessionID] + if (!messages) { + setData("message", info.sessionID, [info]) + return + } + const result = Binary.search(messages, info.id, (message) => message.id) + if (result.found) setData("message", info.sessionID, result.index, reconcile(info)) + if (!result.found) setData("message", info.sessionID, (value = []) => { + const next = value.slice() + next.splice(result.index, 0, info) + return next + }) + return + } + case "message.removed": { + const props = event.properties as { sessionID: string; messageID: string } + setData(produce((draft) => { + const messages = draft.message[props.sessionID] + if (messages) { + const result = Binary.search(messages, props.messageID, (message) => message.id) + if (result.found) messages.splice(result.index, 1) + } + for (const part of draft.part[props.messageID] ?? []) delete draft.part_text_accum_delta[part.id] + delete draft.part[props.messageID] + })) + return + } + case "message.part.updated": { + const part = (event.properties as { part: Part }).part + if (SKIP_PARTS.has(part.type)) return + setData("part_text_accum_delta", produce((draft) => void delete draft[part.id])) + const parts = data.part[part.messageID] + if (!parts) { + setData("part", part.messageID, [part]) + return + } + const result = Binary.search(parts, part.id, (item) => item.id) + if (result.found) setData("part", part.messageID, result.index, reconcile(part)) + if (!result.found) setData("part", part.messageID, (value = []) => { + const next = value.slice() + next.splice(result.index, 0, part) + return next + }) + return + } + case "message.part.removed": { + const props = event.properties as { messageID: string; partID: string } + setData(produce((draft) => { + delete draft.part_text_accum_delta[props.partID] + const parts = draft.part[props.messageID] + if (!parts) return + const result = Binary.search(parts, props.partID, (part) => part.id) + if (result.found) parts.splice(result.index, 1) + if (parts.length === 0) delete draft.part[props.messageID] + })) + return + } + case "message.part.delta": { + const props = event.properties as { messageID: string; partID: string; field: string; delta: string } + const parts = data.part[props.messageID] + if (!parts) return + const result = Binary.search(parts, props.partID, (part) => part.id) + if (!result.found) return + const field = props.field as keyof (typeof parts)[number] + const current = parts[result.index]?.[field] + setData("part_text_accum_delta", props.partID, (value) => (value ?? (typeof current === "string" ? current : "")) + props.delta) + setData("part", props.messageID, produce((draft) => { + if (!draft) return + const part = draft[result.index] + const field = props.field as keyof typeof part + ;(part[field] as string) = ((part[field] as string | undefined) ?? "") + props.delta + })) + return + } + case "permission.asked": { + const permission = event.properties as PermissionRequest + const permissions = data.permission[permission.sessionID] ?? [] + const result = Binary.search(permissions, permission.id, (item) => item.id) + if (result.found) setData("permission", permission.sessionID, result.index, reconcile(permission)) + if (!result.found) setData("permission", permission.sessionID, produce((draft = []) => void draft.splice(result.index, 0, permission))) + return + } + case "permission.replied": { + const props = event.properties as { sessionID: string; requestID: string } + setData("permission", props.sessionID, produce((draft) => { + if (!draft) return + const result = Binary.search(draft, props.requestID, (item) => item.id) + if (result.found) draft.splice(result.index, 1) + })) + return + } + case "question.asked": { + const question = event.properties as QuestionRequest + const questions = data.question[question.sessionID] ?? [] + const result = Binary.search(questions, question.id, (item) => item.id) + if (result.found) setData("question", question.sessionID, result.index, reconcile(question)) + if (!result.found) setData("question", question.sessionID, produce((draft = []) => void draft.splice(result.index, 0, question))) + return + } + case "question.replied": + case "question.rejected": { + const props = event.properties as { sessionID: string; requestID: string } + setData("question", props.sessionID, produce((draft) => { + if (!draft) return + const result = Binary.search(draft, props.requestID, (item) => item.id) + if (result.found) draft.splice(result.index, 1) + })) + } + } + } + + return { + data, + set: setData, + get: (sessionID: string) => data.info[sessionID], + peek: (sessionID: string) => data.info[sessionID], + remember, + resolve, + lineage: { + peek: peekLineage, + async resolve(sessionID: string) { + const session = await resolve(sessionID) + return { session, root: await rootSession(session, resolve) } + }, + }, + sync, + prefetch, + shouldPrefetch(sessionID: string, limit: number) { + if (data.message[sessionID] === undefined) return true + if (Date.now() - (meta.at[sessionID] ?? 0) > 15_000) return true + if (meta.complete[sessionID]) return false + return (meta.limit[sessionID] ?? 0) <= limit + }, + fresh(sessionID: string, ttl: number) { + return Date.now() - (meta.at[sessionID] ?? 0) <= ttl + }, + optimistic: { + add(input: { sessionID: string; message: Message; parts: Part[] }) { + const items = optimistic.get(input.sessionID) + if (items) items.set(input.message.id, input) + if (!items) optimistic.set(input.sessionID, new Map([[input.message.id, input]])) + setData("message", input.sessionID, (messages = []) => merge(messages, [input.message])) + setData("part", input.message.id, input.parts.filter((part) => !!part?.id).sort((a, b) => cmp(a.id, b.id))) + }, + remove(input: { sessionID: string; messageID: string }) { + clearOptimistic(input.sessionID, input.messageID) + setData("message", input.sessionID, (messages) => messages?.filter((message) => message.id !== input.messageID)) + setData("part", produce((draft) => void delete draft[input.messageID])) + }, + }, + diff(sessionID: string, options?: { force?: boolean }) { + touch(sessionID) + if (data.session_diff[sessionID] !== undefined && !options?.force) return Promise.resolve() + return runInflight(inflightDiff, sessionID, () => { + const generation = generations.get(sessionID) ?? 0 + return retry(() => client.session.diff({ sessionID })).then((result) => { + if ((generations.get(sessionID) ?? 0) !== generation) return + setData("session_diff", sessionID, reconcile(cleanDiffs(result.data), { key: "file" })) + }) + }) + }, + todo(sessionID: string, options?: { force?: boolean }) { + touch(sessionID) + if (data.todo[sessionID] !== undefined && !options?.force) return Promise.resolve() + return runInflight(inflightTodo, sessionID, () => { + const generation = generations.get(sessionID) ?? 0 + return retry(() => client.session.todo({ sessionID })).then((result) => { + if ((generations.get(sessionID) ?? 0) !== generation) return + setData("todo", sessionID, reconcile(result.data ?? [], { key: "id" })) + }) + }) + }, + history: { + more: (sessionID: string) => data.message[sessionID] !== undefined && meta.limit[sessionID] !== undefined && !meta.complete[sessionID] && !!meta.cursor[sessionID], + loading: (sessionID: string) => meta.loading[sessionID] ?? false, + async loadMore(sessionID: string, count = historyMessagePageSize) { + touch(sessionID) + if (meta.loading[sessionID] || meta.complete[sessionID] || !meta.cursor[sessionID]) return + await loadMessages(sessionID, count, meta.cursor[sessionID], "prepend") + }, + }, + evict(sessionID: string) { + if (protectedSessions().has(sessionID)) return + seen.delete(sessionID) + evict([sessionID]) + }, + pin(sessionID: string) { + pinned.set(sessionID, (pinned.get(sessionID) ?? 0) + 1) + touch(sessionID) + }, + unpin(sessionID: string) { + const count = pinned.get(sessionID) + if (!count || count === 1) pinned.delete(sessionID) + if (count && count > 1) pinned.set(sessionID, count - 1) + }, + apply, + } +} + +export type ServerSession = ReturnType diff --git a/packages/app/src/context/server-sync.tsx b/packages/app/src/context/server-sync.tsx index db1a57d2b2..f2ee8869ac 100644 --- a/packages/app/src/context/server-sync.tsx +++ b/packages/app/src/context/server-sync.tsx @@ -1,4 +1,4 @@ -import type { Config, OpencodeClient, Path, Project, ProviderAuthResponse, Todo } from "@opencode-ai/sdk/v2/client" +import type { Config, OpencodeClient, Path, Project, ProviderAuthResponse } from "@opencode-ai/sdk/v2/client" import { showToast } from "@/utils/toast" import { getFilename } from "@opencode-ai/core/util/path" import { type Accessor, batch, createMemo, getOwner, onCleanup, onMount, untrack } from "solid-js" @@ -17,8 +17,7 @@ import { loadProvidersQuery, } from "./global-sync/bootstrap" import { createChildStoreManager } from "./global-sync/child-store" -import { applyDirectoryEvent, applyGlobalEvent, cleanupDroppedSessionCaches } from "./global-sync/event-reducer" -import { clearSessionPrefetchDirectory } from "./global-sync/session-prefetch" +import { applyDirectoryEvent, applyGlobalEvent } from "./global-sync/event-reducer" import { estimateRootSessionTotal, loadRootSessionsWithFallback } from "./global-sync/session-load" import { trimSessions } from "./global-sync/session-trim" import type { ProjectMeta } from "./global-sync/types" @@ -38,15 +37,13 @@ import { retry } from "@opencode-ai/core/util/retry" import type { ServerScope } from "@/utils/server-scope" import { persisted } from "@/utils/persist" import { toggleMcp } from "./global-sync/mcp" +import { createServerSession } from "./server-session" type GlobalStore = { ready: boolean error?: InitError path: Path project: Project[] - session_todo: { - [sessionID: string]: Todo[] - } provider: NormalizedProviderListResponse provider_auth: ProviderAuthResponse config: Config @@ -118,7 +115,6 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { return !bootstrap.isPending }, project: [], - session_todo: {}, provider_auth: {}, get path() { const EMPTY = { state: "", config: "", worktree: "", directory: "", home: "" } @@ -188,20 +184,6 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { return (setGlobalStore as (...args: unknown[]) => unknown)(...input) }) as typeof setGlobalStore - const setSessionTodo = (sessionID: string, todos: Todo[] | undefined) => { - if (!sessionID) return - if (!todos) { - setGlobalStore( - "session_todo", - produce((draft) => { - delete draft[sessionID] - }), - ) - return - } - setGlobalStore("session_todo", sessionID, reconcile(todos, { key: "id" })) - } - const paused = () => untrack(() => globalStore.reload) !== undefined const queue = createRefreshQueue({ @@ -211,6 +193,8 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { bootstrapInstance, }) + const session = createServerSession(serverSDK.client) + const children = createChildStoreManager({ owner, scope: serverSDK.scope, @@ -239,7 +223,6 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { sessionMeta.delete(key) sdkCache.delete(key) clearProviderRev(serverSDK.scope, key) - clearSessionPrefetchDirectory(serverSDK.scope, key) }, translate: language.t, queryOptions: queryOptionsApi, @@ -263,11 +246,10 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { if (meta && meta.limit >= retainedLimit) { const next = trimSessions(store.session, { limit: retainedLimit, - permission: store.permission, + permission: session.data.permission, }) if (next.length !== store.session.length) { setStore("session", reconcile(next, { key: "id" })) - cleanupDroppedSessionCaches(store, setStore, next, setSessionTodo) } children.unpin(key) return @@ -290,11 +272,12 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) const limit = Math.max(store.limit, options?.limit ?? 0, sessionMeta.get(key)?.limit ?? 0) const childSessions = store.session.filter((s) => !!s.parentID) - const sessions = trimSessions([...nonArchived, ...childSessions], { + const next = trimSessions([...nonArchived, ...childSessions], { limit, - permission: store.permission, + permission: session.data.permission, }) batch(() => { + next.forEach(session.remember) setStore( "sessionTotal", estimateRootSessionTotal({ @@ -303,8 +286,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { limited: x.limited, }), ) - setStore("session", reconcile(sessions, { key: "id" })) - cleanupDroppedSessionCaches(store, setStore, sessions, setSessionTodo) + setStore("session", reconcile(next, { key: "id" })) }) sessionMeta.set(key, { limit }) }) @@ -358,6 +340,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { loadSessions, translate: language.t, queryClient, + session, }) }) @@ -375,6 +358,8 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { const event = e.details const recent = bootingRoot || Date.now() - bootedAt < 1500 + session.apply(event) + if (directory === "global") { applyGlobalEvent({ event, @@ -404,8 +389,9 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { store, setStore, push: queue.push, - setSessionTodo, retainedLimit: sessionMeta.get(key)?.limit, + sessionContent: false, + permission: session.data.permission, vcsCache: children.vcsCache.get(key), loadLsp: () => { void queryClient.fetchQuery(queryOptionsApi.lsp(key)) @@ -479,9 +465,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { // bootstrap, updateConfig: updateConfigMutation.mutateAsync, project: projectApi, - todo: { - set: setSessionTodo, - }, + session, mcp: { toggle: async (directory: string, name: string) => { const key = directoryKey(directory) diff --git a/packages/app/src/pages/directory-layout.tsx b/packages/app/src/pages/directory-layout.tsx index b2704c2862..58b127a057 100644 --- a/packages/app/src/pages/directory-layout.tsx +++ b/packages/app/src/pages/directory-layout.tsx @@ -2,7 +2,7 @@ import { DataProvider } from "@opencode-ai/session-ui/context" import { showToast } from "@/utils/toast" import { base64Encode } from "@opencode-ai/core/util/encode" import { useLocation, useNavigate, useParams } from "@solidjs/router" -import { type Accessor, createEffect, createMemo, createResource, type ParentProps, Show } from "solid-js" +import { type Accessor, createEffect, createMemo, createResource, onCleanup, type ParentProps, Show } from "solid-js" import { useLanguage } from "@/context/language" import { LocalProvider } from "@/context/local" import { SDKProvider } from "@/context/sdk" @@ -11,7 +11,7 @@ import { decode64 } from "@/utils/base64" import { Schema } from "effect" import type { ServerConnection } from "@/context/server" import { sessionHref } from "@/utils/session-route" -import { useGlobal } from "@/context/global" +import { useServerSync } from "@/context/server-sync" export function DirectoryDataProvider( props: ParentProps<{ @@ -24,7 +24,7 @@ export function DirectoryDataProvider( const navigate = useNavigate() const params = useParams() const sync = useSync() - const global = useGlobal() + const serverSync = useServerSync() const directory = () => (typeof props.directory === "function" ? props.directory() : props.directory) const slug = createMemo(() => base64Encode(directory())) const href = (sessionID: string) => { @@ -50,15 +50,18 @@ export function DirectoryDataProvider( .catch(() => {}), ) + createEffect(() => { + const sessionID = params.id + if (!sessionID) return + serverSync().session.pin(sessionID) + onCleanup(() => serverSync().session.unpin(sessionID)) + }) + return ( { - const server = props.server?.() - if (server && params.id) global.sessionPlacement.inherit(server, params.id, sessionID) - navigate(href(sessionID)) - }} + onNavigateToSession={(sessionID: string) => navigate(href(sessionID))} onSessionHref={href} > {props.children} diff --git a/packages/app/src/pages/home.tsx b/packages/app/src/pages/home.tsx index 12934aebea..e46e5ad4cb 100644 --- a/packages/app/src/pages/home.tsx +++ b/packages/app/src/pages/home.tsx @@ -220,10 +220,9 @@ export function NewHome() { void directory.session .sync(record.session.id) .then(() => { - const store = ctx.sync.child(record.session.directory)[0] return Promise.all( - (store.message[record.session.id] ?? []).flatMap((message) => - (store.part[message.id] ?? []).flatMap((part) => { + (ctx.sync.session.data.message[record.session.id] ?? []).flatMap((message) => + (ctx.sync.session.data.part[message.id] ?? []).flatMap((part) => { if (part.type !== "text" || !part.text) return [] return preloadMarkdown(part.text, part.id, marked) }), @@ -343,12 +342,6 @@ export function NewHome() { if (!conn) return const directory = project?.worktree ?? session.directory const ctx = global.ensureServerCtx(conn) - global.sessionPlacement.set({ - server: ServerConnection.key(conn), - leafID: session.id, - rootID: session.id, - directory: session.directory, - }) ctx.projects.open(directory) ctx.projects.touch(directory) startTransition(() => { diff --git a/packages/app/src/pages/layout.tsx b/packages/app/src/pages/layout.tsx index ee1e7f98bd..0938163535 100644 --- a/packages/app/src/pages/layout.tsx +++ b/packages/app/src/pages/layout.tsx @@ -1,5 +1,4 @@ import { - batch, createEffect, createMemo, createResource, @@ -26,7 +25,7 @@ import { Tooltip } from "@opencode-ai/ui/tooltip" import { DropdownMenu } from "@opencode-ai/ui/dropdown-menu" import { Dialog } from "@opencode-ai/ui/dialog" import { getFilename } from "@opencode-ai/core/util/path" -import { Session, type Message } from "@opencode-ai/sdk/v2/client" +import { Session } from "@opencode-ai/sdk/v2/client" import { usePlatform } from "@/context/platform" import { useSettings } from "@/context/settings" import { createStore, produce, reconcile } from "solid-js/store" @@ -37,16 +36,7 @@ import { toaster } from "@opencode-ai/ui/toast" import { setV2Toast, showToast, ToastRegion } from "@/utils/toast" import { useServerSDK } from "@/context/server-sdk" import { clearWorkspaceTerminals } from "@/context/terminal" -import { dropSessionCaches, pickSessionCacheEvictions } from "@/context/global-sync/session-cache" -import { - clearSessionPrefetchInflight, - clearSessionPrefetch, - getSessionPrefetch, - isSessionPrefetchCurrent, - runSessionPrefetch, - setSessionPrefetch, - shouldSkipSessionPrefetch, -} from "@/context/global-sync/session-prefetch" +import { pickSessionCacheEvictions } from "@/context/global-sync/session-cache" import { useNotification } from "@/context/notification" import { usePermission } from "@/context/permission" import { Binary } from "@opencode-ai/core/util/binary" @@ -684,7 +674,6 @@ export default function LegacyLayout(props: ParentProps) { serverSDK().url prefetchToken.value += 1 - clearSessionPrefetchInflight(serverSDK().scope) prefetchQueues.clear() }) @@ -712,88 +701,10 @@ export default function LegacyLayout(props: ParentProps) { return created } - const mergeByID = (current: T[], incoming: T[]) => { - if (current.length === 0) { - return incoming.slice().sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) - } - - const map = new Map() - for (const item of current) { - map.set(item.id, item) - } - for (const item of incoming) { - map.set(item.id, item) - } - return [...map.values()].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) - } - async function prefetchMessages(directory: string, sessionID: string, token: number) { - const [store, setStore] = serverSync().child(directory, { bootstrap: false }) - - return runSessionPrefetch({ - scope: serverSDK().scope, - directory, - sessionID, - task: (rev) => - retry(() => serverSDK().client.session.messages({ directory, sessionID, limit: prefetchChunk })) - .then((messages) => { - if (prefetchToken.value !== token) return - if (!isSessionPrefetchCurrent(serverSDK().scope, directory, sessionID, rev)) return - - const items = (messages.data ?? []).filter((x) => !!x?.info?.id) - const next = items.map((x) => x.info).filter((m): m is Message => !!m?.id) - const sorted = mergeByID([], next) - const stale = markPrefetched(directory, sessionID) - const cursor = messages.response.headers.get("x-next-cursor") ?? undefined - const meta = { - limit: sorted.length, - cursor, - complete: !cursor, - at: Date.now(), - } - - if (stale.length > 0) { - clearSessionPrefetch(serverSDK().scope, directory, stale) - for (const id of stale) { - serverSync().todo.set(id, undefined) - } - } - - const current = store.message[sessionID] ?? [] - const merged = mergeByID( - current.filter((item): item is Message => !!item?.id), - sorted, - ) - - if (!isSessionPrefetchCurrent(serverSDK().scope, directory, sessionID, rev)) return - - batch(() => { - if (stale.length > 0) { - setStore( - produce((draft) => { - dropSessionCaches(draft, stale) - }), - ) - } - - setStore("message", sessionID, reconcile(merged, { key: "id" })) - setSessionPrefetch({ scope: serverSDK().scope, directory, sessionID, ...meta }) - - for (const message of items) { - const currentParts = store.part[message.info.id] ?? [] - const mergedParts = mergeByID( - currentParts.filter((item): item is (typeof currentParts)[number] & { id: string } => !!item?.id), - message.parts.filter((item): item is (typeof message.parts)[number] & { id: string } => !!item?.id), - ) - - setStore("part", message.info.id, reconcile(mergedParts, { key: "id" })) - } - }) - - return meta - }) - .catch(() => undefined), - }) + await serverSync().session.prefetch(sessionID, prefetchChunk).catch(() => {}) + if (prefetchToken.value !== token) return + for (const stale of markPrefetched(directory, sessionID)) serverSync().session.evict(stale) } const pumpPrefetch = (directory: string) => { @@ -820,15 +731,7 @@ export default function LegacyLayout(props: ParentProps) { const directory = session.directory if (!directory) return - const [store] = serverSync().child(directory, { bootstrap: false }) - const cached = untrack(() => { - const info = getSessionPrefetch(serverSDK().scope, directory, session.id) - return shouldSkipSessionPrefetch({ - message: store.message[session.id] !== undefined, - info, - chunk: prefetchChunk, - }) - }) + const cached = untrack(() => !serverSync().session.shouldPrefetch(session.id, prefetchChunk)) if (cached) return const q = queueFor(directory) diff --git a/packages/app/src/pages/layout/project-avatar-state.ts b/packages/app/src/pages/layout/project-avatar-state.ts index e9b3ec6c54..b7abd2480e 100644 --- a/packages/app/src/pages/layout/project-avatar-state.ts +++ b/packages/app/src/pages/layout/project-avatar-state.ts @@ -15,7 +15,7 @@ export function useSessionTabAvatarState( const hasPermissions = createMemo(() => { if (!active()) return false const [store] = globalSync().child(directory(), { bootstrap: false }) - return !!sessionPermissionRequest(store.session, store.permission, sessionId(), (item) => { + return !!sessionPermissionRequest(store.session, globalSync().session.data.permission, sessionId(), (item) => { return !permission.autoResponds(item, directory()) }) }) @@ -23,8 +23,7 @@ export function useSessionTabAvatarState( const loading = createMemo(() => { if (!active()) return false if (hasPermissions()) return false - const [store] = globalSync().child(directory(), { bootstrap: false }) - return store.session_working(sessionId()) + return globalSync().session.data.session_working(sessionId()) }) return { unread, loading } } diff --git a/packages/app/src/pages/layout/sidebar-items.tsx b/packages/app/src/pages/layout/sidebar-items.tsx index 44430b943e..fb89cf76e7 100644 --- a/packages/app/src/pages/layout/sidebar-items.tsx +++ b/packages/app/src/pages/layout/sidebar-items.tsx @@ -33,8 +33,10 @@ export const ProjectIcon = (props: { const hasError = createMemo(() => dirs().some((directory) => notification.project.unseenHasError(directory))) const hasPermissions = createMemo(() => dirs().some((directory) => { - const [store] = serverSync().child(directory, { bootstrap: false }) - return hasProjectPermissions(store.permission, (item) => !permission.autoResponds(item, directory)) + return hasProjectPermissions(serverSync().session.data.permission, (item) => { + if (serverSync().session.get(item.sessionID)?.directory !== directory) return false + return !permission.autoResponds(item, directory) + }) }), ) const notify = createMemo(() => props.notify && (hasPermissions() || unseenCount() > 0)) @@ -151,16 +153,16 @@ export const SessionItem = (props: SessionItemProps): JSX.Element => { const hasError = createMemo(() => notification.session.unseenHasError(props.session.id)) const [sessionStore] = serverSync().child(props.session.directory) const hasPermissions = createMemo(() => { - return !!sessionPermissionRequest(sessionStore.session, sessionStore.permission, props.session.id, (item) => { + return !!sessionPermissionRequest(sessionStore.session, serverSync().session.data.permission, props.session.id, (item) => { return !permission.autoResponds(item, props.session.directory) }) }) const isWorking = createMemo(() => { if (hasPermissions()) return false - return sessionStore.session_working(props.session.id) + return serverSync().session.data.session_working(props.session.id) }) - const tint = createMemo(() => messageAgentColor(sessionStore.message[props.session.id], sessionStore.agent)) + const tint = createMemo(() => messageAgentColor(serverSync().session.data.message[props.session.id], sessionStore.agent)) const tooltip = createMemo(() => props.showTooltip ?? (props.mobile || !props.sidebarExpanded())) const currentChild = createMemo(() => { if (!props.showChild) return diff --git a/packages/app/src/pages/layout/sidebar-project.tsx b/packages/app/src/pages/layout/sidebar-project.tsx index ecd0f7ad40..9d29a1b495 100644 --- a/packages/app/src/pages/layout/sidebar-project.tsx +++ b/packages/app/src/pages/layout/sidebar-project.tsx @@ -304,8 +304,10 @@ export const SortableProject = (props: { const projectStore = createMemo(() => serverSync().child(props.project.worktree, { bootstrap: false })[0]) const isWorking = createMemo(() => dirs().some((directory) => { - const [store] = serverSync().child(directory, { bootstrap: false }) - return Object.keys(store.session_status).some((id) => store.session_working(id)) + return Object.keys(serverSync().session.data.session_status).some((id) => { + if (serverSync().session.get(id)?.directory !== directory) return false + return serverSync().session.data.session_working(id) + }) }), ) const projectSessions = createMemo(() => sortedRootSessions(projectStore(), props.sortNow())) diff --git a/packages/app/src/pages/session.tsx b/packages/app/src/pages/session.tsx index 6c9d1732f9..577de8386d 100644 --- a/packages/app/src/pages/session.tsx +++ b/packages/app/src/pages/session.tsx @@ -517,7 +517,7 @@ export default function Page() { if (!id) return if (status === "idle" && !blocked) return const cached = untrack( - () => sync().data.todo[id] !== undefined || serverSync().data.session_todo[id] !== undefined, + () => sync().data.todo[id] !== undefined, ) todoFrame = requestAnimationFrame(() => { diff --git a/packages/app/src/pages/session/composer/session-composer-state.ts b/packages/app/src/pages/session/composer/session-composer-state.ts index edcc37ebd4..5c1cc5a1e8 100644 --- a/packages/app/src/pages/session/composer/session-composer-state.ts +++ b/packages/app/src/pages/session/composer/session-composer-state.ts @@ -50,7 +50,7 @@ export function createSessionComposerState(options?: { closeMs?: number | (() => const todos = createMemo((): Todo[] => { const id = params.id if (!id) return [] - return serverSync().data.session_todo[id] ?? [] + return serverSync().session.data.todo[id] ?? [] }) const done = createMemo( @@ -111,7 +111,6 @@ export function createSessionComposerState(options?: { closeMs?: number | (() => const clear = () => { const id = params.id if (!id) return - serverSync().todo.set(id, []) sync().set("todo", id, []) } diff --git a/packages/app/src/pages/session/timeline/message-timeline.tsx b/packages/app/src/pages/session/timeline/message-timeline.tsx index 39707fa775..9885075199 100644 --- a/packages/app/src/pages/session/timeline/message-timeline.tsx +++ b/packages/app/src/pages/session/timeline/message-timeline.tsx @@ -646,14 +646,14 @@ export function MessageTimeline(props: { } const shareMutation = useMutation(() => ({ - mutationFn: (id: string) => serverSDK().client.session.share({ sessionID: id, directory: sdk().directory }), + mutationFn: (id: string) => serverSDK().client.session.share({ sessionID: id }), onError: (err) => { console.error("Failed to share session", err) }, })) const unshareMutation = useMutation(() => ({ - mutationFn: (id: string) => serverSDK().client.session.unshare({ sessionID: id, directory: sdk().directory }), + mutationFn: (id: string) => serverSDK().client.session.unshare({ sessionID: id }), onError: (err) => { console.error("Failed to unshare session", err) }, diff --git a/packages/app/src/pages/session/timeline/model.ts b/packages/app/src/pages/session/timeline/model.ts index f34aacb5f9..91bfc0eaf7 100644 --- a/packages/app/src/pages/session/timeline/model.ts +++ b/packages/app/src/pages/session/timeline/model.ts @@ -1,37 +1,29 @@ import type { Message, UserMessage } from "@opencode-ai/sdk/v2" import { createMemo, createResource, onCleanup, untrack, type Accessor } from "solid-js" -import { getSessionPrefetch, SESSION_PREFETCH_TTL } from "@/context/global-sync/session-prefetch" -import { useSDK } from "@/context/sdk" -import { useServerSDK } from "@/context/server-sdk" +import { useServerSync } from "@/context/server-sync" import { useSync } from "@/context/sync" import { same } from "@/utils/same" const emptyUserMessages: UserMessage[] = [] +const sessionFreshness = 15_000 export function createTimelineModel(input: { sessionID: Accessor revertMessageID: Accessor }) { - const sdk = useSDK() - const serverSDK = useServerSDK() + const serverSync = useServerSync() const sync = useSync() let refreshFrame: number | undefined let refreshTimer: number | undefined const [resource] = createResource( - () => [sdk().directory, input.sessionID()] as const, - ([directory, id]) => { + () => input.sessionID(), + (id) => { clearRefresh() if (!id) return const cached = untrack(() => sync().data.message[id] !== undefined) - const stale = cached - ? (() => { - const info = getSessionPrefetch(serverSDK().scope, directory, id) - if (!info) return true - return Date.now() - info.at > SESSION_PREFETCH_TTL - })() - : false + const stale = cached && !serverSync().session.fresh(id, sessionFreshness) refreshFrame = requestAnimationFrame(() => { refreshFrame = undefined diff --git a/packages/app/src/utils/session-placement.test.ts b/packages/app/src/utils/session-placement.test.ts deleted file mode 100644 index 248889929f..0000000000 --- a/packages/app/src/utils/session-placement.test.ts +++ /dev/null @@ -1,38 +0,0 @@ -import { describe, expect, test } from "bun:test" -import { ServerConnection } from "@/context/server" -import { createSessionPlacementStore } from "./session-placement" - -describe("session placement", () => { - const local = ServerConnection.Key.make("http://localhost:4096") - const remote = ServerConnection.Key.make("https://example.com") - - test("aliases a leaf and root without crossing servers", () => { - const store = createSessionPlacementStore() - store.set({ server: local, leafID: "child", rootID: "root", directory: "/repo" }) - store.set({ server: remote, leafID: "child", rootID: "other", directory: "/remote" }) - - expect(store.get(local, "child")).toEqual({ rootID: "root", directory: "/repo" }) - expect(store.get(local, "root")).toEqual({ rootID: "root", directory: "/repo" }) - expect(store.get(remote, "child")).toEqual({ rootID: "other", directory: "/remote" }) - }) - - test("inherits known placement for in-app child navigation", () => { - const store = createSessionPlacementStore() - store.set({ server: local, leafID: "parent", rootID: "root", directory: "/repo" }) - - expect(store.inherit(local, "parent", "child")).toEqual({ rootID: "root", directory: "/repo" }) - expect(store.get(local, "child")).toEqual({ rootID: "root", directory: "/repo" }) - expect(store.inherit(local, "missing", "unknown")).toBeUndefined() - }) - - test("bounds retained placement aliases", () => { - const store = createSessionPlacementStore() - for (let index = 0; index < 300; index++) { - store.set({ server: local, leafID: `leaf-${index}`, rootID: `root-${index}`, directory: `/repo/${index}` }) - } - - expect(store.size()).toBeLessThanOrEqual(256) - expect(store.get(local, "leaf-0")).toBeUndefined() - expect(store.get(local, "leaf-299")).toEqual({ rootID: "root-299", directory: "/repo/299" }) - }) -}) diff --git a/packages/app/src/utils/session-placement.ts b/packages/app/src/utils/session-placement.ts deleted file mode 100644 index 95ecc450d7..0000000000 --- a/packages/app/src/utils/session-placement.ts +++ /dev/null @@ -1,41 +0,0 @@ -import { ServerConnection } from "@/context/server" - -export type SessionPlacement = { - rootID: string - directory: string -} - -export function createSessionPlacementStore() { - const placements = new Map() - const limit = 256 - const key = (server: ServerConnection.Key, sessionID: string) => `${server}\0${sessionID}` - const write = (id: string, placement: SessionPlacement) => { - placements.delete(id) - placements.set(id, placement) - while (placements.size > limit) placements.delete(placements.keys().next().value!) - } - - return { - get(server: ServerConnection.Key, sessionID: string) { - const id = key(server, sessionID) - const placement = placements.get(id) - if (placement) write(id, placement) - return placement - }, - set(input: SessionPlacement & { server: ServerConnection.Key; leafID: string }) { - const placement = { rootID: input.rootID, directory: input.directory } - write(key(input.server, input.leafID), placement) - write(key(input.server, input.rootID), placement) - return placement - }, - inherit(server: ServerConnection.Key, sourceID: string, leafID: string) { - const placement = placements.get(key(server, sourceID)) - if (!placement) return - write(key(server, leafID), placement) - return placement - }, - size() { - return placements.size - }, - } -} diff --git a/packages/app/src/utils/session-route.test.ts b/packages/app/src/utils/session-route.test.ts index 42ead3b5fd..b7230b8f5a 100644 --- a/packages/app/src/utils/session-route.test.ts +++ b/packages/app/src/utils/session-route.test.ts @@ -36,4 +36,13 @@ describe("session routes", () => { }), ).toBe(sessions.root) }) + + test("rejects a parent cycle", async () => { + const sessions: Record = { + child: { id: "child", parentID: "parent" }, + parent: { id: "parent", parentID: "child" }, + } + + expect(rootSession(sessions.child, async (id) => sessions[id]!)).rejects.toThrow("Session parent cycle: child") + }) }) diff --git a/packages/app/src/utils/session-route.ts b/packages/app/src/utils/session-route.ts index 1bf921954e..d4b2759e39 100644 --- a/packages/app/src/utils/session-route.ts +++ b/packages/app/src/utils/session-route.ts @@ -18,8 +18,13 @@ export function requireServerKey(segment: string | undefined) { type SessionParent = { id: string; parentID?: string } -export async function rootSession(session: SessionParent, get: (sessionID: string) => Promise) { +export async function rootSession(session: T, get: (sessionID: string) => Promise) { + const seen = new Set([session.id]) let current = session - while (current.parentID) current = await get(current.parentID) + while (current.parentID) { + if (seen.has(current.parentID)) throw new Error(`Session parent cycle: ${current.parentID}`) + seen.add(current.parentID) + current = await get(current.parentID) + } return current }