| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105 |
- 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 cmpMessage = (a: Message, b: Message) => a.time.created - b.time.created || cmp(a.id, b.id)
- const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"])
- const initialMessagePageSize = 2
- const historyMessagePageSize = 200
- const sessionInfoLimit = 2_048
- const emptyIDs: ReadonlySet<string> = new Set()
- type OptimisticItem = {
- message: Message
- parts: Part[]
- confirmedParts?: Part[]
- confirmedMessage?: boolean
- }
- type MessagePage = {
- session: Message[]
- part: { id: string; part: Part[] }[]
- cursor?: string
- complete: boolean
- }
- // Most markers describe the current HTTP attempt; deltaParts persists non-durable stream state across retries.
- type MessageLoadState = {
- touchedMessages: Set<string>
- removedMessages: Set<string>
- retainedMessages: Set<string>
- touchedParts: Map<string, Set<string>>
- deltaParts: Map<string, Set<string>>
- carriedDeltaParts: Map<string, Set<string>>
- removedParts: Map<string, Set<string>>
- optimisticParts: Map<string, Set<string>>
- orphanParents: Set<string>
- clearedMessageParts: Set<string>
- }
- function mergeOptimisticPage(page: MessagePage, items: OptimisticItem[]) {
- if (items.length === 0) return { ...page, observed: [] as { messageID: string; parts: Part[] }[] }
- const session = [...page.session]
- const part = new Map(page.part.map((item) => [item.id, item.part]))
- const observed: { messageID: string; parts: Part[] }[] = []
- 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)
- const confirmed = result.found
- ? item.parts.filter((part) => Binary.search(current ?? [], part.id, (value) => value.id).found)
- : []
- if (result.found) observed.push({ messageID: item.message.id, parts: confirmed })
- part.set(
- item.message.id,
- merge(
- result.found ? (current ?? []) : merge(item.confirmedParts ?? [], current ?? []),
- item.parts.filter((part) => !confirmed.includes(part)),
- ),
- )
- }
- return {
- ...page,
- session,
- part: [...part.entries()].sort((a, b) => cmp(a[0], b[0])).map(([id, parts]) => ({ id, part: parts })),
- observed,
- }
- }
- function runInflight(map: Map<string, Promise<void>>, key: string, task: () => Promise<void>) {
- 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<T extends { id: string }>(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))
- }
- function reconcileFetched<T extends { id: string }>(
- fetched: T[],
- current: readonly T[],
- options: {
- touched?: ReadonlySet<string>
- retained?: ReadonlySet<string>
- preserveUnfetched?: boolean | ((item: T) => boolean)
- } = {},
- ) {
- const result = new Map(fetched.map((item) => [item.id, item]))
- const live = new Map(current.map((item) => [item.id, item]))
- if (options.preserveUnfetched) {
- for (const item of current) {
- if (!result.has(item.id) && (options.preserveUnfetched === true || options.preserveUnfetched(item)))
- result.set(item.id, item)
- }
- }
- for (const id of options.retained ?? emptyIDs) {
- if (result.has(id)) continue
- const item = live.get(id)
- if (item) result.set(id, item)
- }
- // Events observed while the request is pending are the freshest client state for those identities.
- for (const id of options.touched ?? emptyIDs) {
- const item = live.get(id)
- if (item) result.set(id, item)
- if (!item) result.delete(id)
- }
- return [...result.values()].sort((a, b) => cmp(a.id, b.id))
- }
- export function createServerSession(client: OpencodeClient, options?: { retry?: typeof retry }) {
- const [data, setData] = createStore({
- info: {} as Record<string, Session | undefined>,
- session_status: {} as Record<string, SessionStatus>,
- session_diff: {} as Record<string, SnapshotFileDiff[]>,
- todo: {} as Record<string, Todo[]>,
- permission: {} as Record<string, PermissionRequest[]>,
- question: {} as Record<string, QuestionRequest[]>,
- message: {} as Record<string, Message[]>,
- part: {} as Record<string, Part[]>,
- part_text_accum_delta: {} as Record<string, string>,
- session_working(id: string) {
- return (this.session_status[id]?.type ?? "idle") !== "idle"
- },
- })
- const requests = new Map<string, Promise<Session>>()
- const inflight = new Map<string, Promise<void>>()
- const inflightDiff = new Map<string, Promise<void>>()
- const inflightTodo = new Map<string, Promise<void>>()
- const optimistic = new Map<string, Map<string, OptimisticItem>>()
- const messageLoads = new Map<string, MessageLoadState>()
- const pendingParts = new Map<string, Map<string, Set<string>>>()
- const orphanParts = new Map<string, Set<string>>()
- const removedMessages = new Map<string, Set<string>>()
- const deltaBases = new Map<string, { base: string; sessionID: string }>()
- const deleteMessageParts = (
- cache: { part: Record<string, Part[] | undefined>; part_text_accum_delta: Record<string, string | undefined> },
- messageID: string,
- ) => {
- for (const part of cache.part[messageID] ?? []) {
- delete cache.part_text_accum_delta[part.id]
- deltaBases.delete(part.id)
- }
- delete cache.part[messageID]
- }
- const seen = new Set<string>()
- const infoSeen = new Set<string>()
- const pinned = new Map<string, number>()
- const generations = new Map<string, object>()
- const generation = (sessionID: string) => {
- const current = generations.get(sessionID)
- if (current) return current
- const created = {}
- generations.set(sessionID, created)
- return created
- }
- const [meta, setMeta] = createStore({
- limit: {} as Record<string, number | undefined>,
- cursor: {} as Record<string, string | undefined>,
- complete: {} as Record<string, boolean | undefined>,
- loading: {} as Record<string, boolean | undefined>,
- at: {} as Record<string, number | undefined>,
- })
- 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(),
- ...inflight.keys(),
- ...inflightDiff.keys(),
- ...inflightTodo.keys(),
- ...messageLoads.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),
- ])
- 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))
- stale.forEach((sessionID) => generations.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 active = generation(sessionID)
- const request = client.session.get({ sessionID }).then((result) => {
- if (!result.data) throw new Error(`Session not found: ${sessionID}`)
- if (generations.get(sessionID) !== active) return result.data
- return remember(result.data)
- })
- requests.set(sessionID, request)
- const cleanup = () => {
- if (requests.get(sessionID) === request) requests.delete(sessionID)
- if (
- generations.get(sessionID) === active &&
- !data.info[sessionID] &&
- !requests.has(sessionID) &&
- !messageLoads.has(sessionID) &&
- !inflight.has(sessionID) &&
- !inflightDiff.has(sessionID) &&
- !inflightTodo.has(sessionID)
- )
- generations.delete(sessionID)
- }
- void request.then(cleanup, cleanup)
- 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 clearOptimisticPart = (sessionID: string, messageID: string, partID: string) => {
- const items = optimistic.get(sessionID)
- const item = items?.get(messageID)
- if (!items || !item) return
- const parts = item.parts.filter((part) => part.id !== partID)
- const confirmedParts = item.confirmedParts?.filter((part) => part.id !== partID)
- if (parts.length === 0) {
- clearOptimistic(sessionID, messageID)
- return
- }
- items.set(messageID, { ...item, parts, confirmedParts, confirmedMessage: true })
- }
- const confirmOptimisticPart = (sessionID: string, messageID: string, part: Part) => {
- const items = optimistic.get(sessionID)
- const item = items?.get(messageID)
- if (!items || !item) return
- const parts = item.parts.filter((value) => value.id !== part.id)
- if (parts.length === 0) {
- clearOptimistic(sessionID, messageID)
- return
- }
- items.set(messageID, {
- ...item,
- parts,
- confirmedParts: merge(item.confirmedParts ?? [], [part]),
- confirmedMessage: true,
- })
- }
- const confirmOptimistic = (sessionID: string, messageID: string, confirmedParts: Part[]) => {
- const items = optimistic.get(sessionID)
- const item = items?.get(messageID)
- if (!items || !item) return
- const confirmed = new Set(confirmedParts.map((part) => part.id))
- const parts = item.parts.filter((part) => !confirmed.has(part.id))
- if (parts.length === 0) {
- clearOptimistic(sessionID, messageID)
- return
- }
- items.set(messageID, {
- ...item,
- parts,
- confirmedParts: merge(item.confirmedParts ?? [], confirmedParts),
- confirmedMessage: true,
- })
- }
- const trackPartChange = (sessionID: string, messageID: string, partID: string) => {
- const load = messageLoads.get(sessionID)
- if (!load) return
- // A part event keeps an existing parent when the fetched page omits it without overriding fetched metadata.
- const messages = data.message[sessionID]
- if (messages && Binary.search(messages, messageID, (message) => message.id).found)
- load.retainedMessages.add(messageID)
- const parts = load.touchedParts.get(messageID)
- if (parts) {
- parts.add(partID)
- return
- }
- load.touchedParts.set(messageID, new Set([partID]))
- }
- const resetMessageLoad = (sessionID: string, load: MessageLoadState) => {
- load.touchedMessages.clear()
- load.retainedMessages.clear()
- load.touchedParts.clear()
- load.carriedDeltaParts.clear()
- load.clearedMessageParts.clear()
- for (const messageID of load.removedMessages) {
- load.touchedMessages.add(messageID)
- load.clearedMessageParts.add(messageID)
- }
- for (const [messageID, parts] of load.deltaParts) {
- load.touchedParts.set(messageID, new Set(parts))
- load.carriedDeltaParts.set(messageID, new Set(parts))
- const messages = data.message[sessionID]
- if (messages && Binary.search(messages, messageID, (message) => message.id).found)
- load.retainedMessages.add(messageID)
- }
- for (const [messageID, parts] of load.removedParts) {
- const touched = load.touchedParts.get(messageID) ?? new Set<string>()
- parts.forEach((partID) => touched.add(partID))
- load.touchedParts.set(messageID, touched)
- const messages = data.message[sessionID]
- if (messages && Binary.search(messages, messageID, (message) => message.id).found)
- load.retainedMessages.add(messageID)
- }
- for (const [messageID, parts] of load.optimisticParts) {
- load.removedMessages.delete(messageID)
- load.clearedMessageParts.add(messageID)
- load.touchedMessages.add(messageID)
- const touched = load.touchedParts.get(messageID) ?? new Set<string>()
- parts.forEach((partID) => touched.add(partID))
- load.touchedParts.set(messageID, touched)
- }
- }
- const evict = (sessionIDs: string[]) => {
- if (sessionIDs.length === 0) return
- const evicted = new Set(sessionIDs)
- for (const [partID, item] of deltaBases) {
- if (evicted.has(item.sessionID)) deltaBases.delete(partID)
- }
- sessionIDs.forEach((sessionID) => {
- generations.delete(sessionID)
- clearOptimistic(sessionID)
- requests.delete(sessionID)
- inflight.delete(sessionID)
- inflightDiff.delete(sessionID)
- inflightTodo.delete(sessionID)
- messageLoads.delete(sessionID)
- pendingParts.delete(sessionID)
- orphanParts.delete(sessionID)
- removedMessages.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(),
- ...messageLoads.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, onAttempt?: () => void) => {
- const response = await (options?.retry ?? retry)(() => {
- onAttempt?.()
- return 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 replaceMessages = (sessionID: string, messages: Message[]) => {
- const messageIDs = new Set(messages.map((message) => message.id))
- const dropped = (data.message[sessionID] ?? []).filter((message) => !messageIDs.has(message.id))
- setData("message", sessionID, reconcile(messages, { key: "id" }))
- setData(
- produce((draft) => {
- for (const message of dropped) deleteMessageParts(draft, message.id)
- }),
- )
- return messageIDs
- }
- const replaceParts = (
- sessionID: string,
- items: MessagePage["part"],
- messageIDs: Set<string>,
- load?: MessageLoadState,
- ) => {
- for (const item of items) {
- if (!messageIDs.has(item.id)) continue
- const fetched = load?.clearedMessageParts.has(item.id)
- ? []
- : item.part.filter((part) => !SKIP_PARTS.has(part.type))
- const fetchedIDs = new Set(fetched.map((part) => part.id))
- const pending = pendingParts.get(sessionID)?.get(item.id)
- const touched = new Set([...(load?.touchedParts.get(item.id) ?? []), ...(pending ?? [])])
- for (const part of fetched) {
- const accumulated = data.part_text_accum_delta[part.id]
- const base = deltaBases.get(part.id)?.base
- const preserveDelta =
- base !== undefined &&
- accumulated !== undefined &&
- "text" in part &&
- typeof part.text === "string" &&
- part.text.startsWith(base) &&
- accumulated.startsWith(part.text) &&
- accumulated !== part.text
- if (preserveDelta) touched.add(part.id)
- if (load?.carriedDeltaParts.get(item.id)?.has(part.id) && !preserveDelta) touched.delete(part.id)
- }
- for (const partID of load?.carriedDeltaParts.get(item.id) ?? []) {
- if (!fetchedIDs.has(partID)) touched.delete(partID)
- }
- const parts = reconcileFetched(fetched, data.part[item.id] ?? [], { touched })
- if (!parts.length) {
- orphanParts.get(sessionID)?.delete(item.id)
- setData(produce((draft) => deleteMessageParts(draft, item.id)))
- continue
- }
- const partIDs = new Set(parts.map((part) => part.id))
- setData(
- "part_text_accum_delta",
- produce((draft) => {
- for (const part of data.part[item.id] ?? []) {
- if (!partIDs.has(part.id) || !touched.has(part.id)) {
- delete draft[part.id]
- deltaBases.delete(part.id)
- }
- }
- }),
- )
- setData("part", item.id, reconcile(parts, { key: "id" }))
- orphanParts.get(sessionID)?.delete(item.id)
- }
- }
- const applyMessagePage = (
- sessionID: string,
- page: MessagePage,
- load: MessageLoadState | undefined,
- preserveUnfetched: boolean | ((message: Message) => boolean),
- cleanupOrphans: boolean,
- ) => {
- const merged = mergeOptimisticPage(page, [...(optimistic.get(sessionID)?.values() ?? [])])
- merged.observed.forEach((item) => {
- if (!load?.clearedMessageParts.has(item.messageID)) confirmOptimistic(sessionID, item.messageID, item.parts)
- })
- const touchedMessages = new Set([...(load?.touchedMessages ?? []), ...(removedMessages.get(sessionID) ?? [])])
- const messages = reconcileFetched(merged.session, data.message[sessionID] ?? [], {
- touched: touchedMessages,
- retained: load?.retainedMessages,
- preserveUnfetched,
- })
- batch(() => {
- const messageIDs = replaceMessages(sessionID, messages)
- replaceParts(sessionID, merged.part, messageIDs, load)
- const orphans = orphanParts.get(sessionID)
- if (cleanupOrphans && page.complete && orphans) {
- for (const messageID of orphans) {
- if (!messageIDs.has(messageID)) setData(produce((draft) => deleteMessageParts(draft, messageID)))
- }
- orphanParts.delete(sessionID)
- }
- setMeta("limit", sessionID, messages.length)
- setMeta("cursor", sessionID, merged.cursor)
- setMeta("complete", sessionID, merged.complete)
- setMeta("at", sessionID, Date.now())
- })
- }
- const loadMessages = async (sessionID: string, limit: number, before?: string, mode?: "replace" | "prepend") => {
- if (meta.loading[sessionID]) return
- const active = generation(sessionID)
- const load: MessageLoadState = {
- touchedMessages: new Set(),
- removedMessages: new Set(),
- retainedMessages: new Set(),
- touchedParts: new Map(),
- deltaParts: new Map(),
- carriedDeltaParts: new Map(),
- removedParts: new Map(),
- optimisticParts: new Map(),
- orphanParents: new Set(),
- clearedMessageParts: new Set(),
- }
- messageLoads.set(sessionID, load)
- setMeta("loading", sessionID, true)
- let applied = false
- await fetchMessages(sessionID, limit, before, () => resetMessageLoad(sessionID, load))
- .then((page) => {
- if (generations.get(sessionID) !== active) return
- const first = page.session.reduce<Message | undefined>(
- (oldest, message) => (!oldest || cmpMessage(message, oldest) < 0 ? message : oldest),
- undefined,
- )
- const preserveUnfetched =
- mode === "prepend" || (!page.complete && (!first || ((message: Message) => cmpMessage(message, first) < 0)))
- applyMessagePage(
- sessionID,
- page,
- messageLoads.get(sessionID) === load ? load : undefined,
- preserveUnfetched,
- mode !== "prepend",
- )
- applied = true
- })
- .finally(() => {
- if (!applied && generations.get(sessionID) === active && messageLoads.get(sessionID) === load) {
- for (const messageID of load.orphanParents) {
- if (!orphanParts.get(sessionID)?.has(messageID)) continue
- setData(produce((draft) => deleteMessageParts(draft, messageID)))
- orphanParts.get(sessionID)?.delete(messageID)
- }
- if (orphanParts.get(sessionID)?.size === 0) orphanParts.delete(sessionID)
- }
- if (messageLoads.get(sessionID) === load) messageLoads.delete(sessionID)
- if (generations.get(sessionID) === active) 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] &&
- event.type !== "session.created" &&
- event.type !== "session.updated" &&
- event.type !== "session.deleted"
- )
- 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 load = messageLoads.get(info.sessionID)
- load?.touchedMessages.add(info.id)
- load?.removedMessages.delete(info.id)
- const items = optimistic.get(info.sessionID)
- const item = items?.get(info.id)
- if (items && item) {
- if (item.parts.length === 0) clearOptimistic(info.sessionID, info.id)
- if (item.parts.length > 0) items.set(info.id, { ...item, confirmedMessage: true })
- }
- const orphans = orphanParts.get(info.sessionID)
- orphans?.delete(info.id)
- if (orphans?.size === 0) orphanParts.delete(info.sessionID)
- const removedMessagesForSession = removedMessages.get(info.sessionID)
- removedMessagesForSession?.delete(info.id)
- if (removedMessagesForSession?.size === 0) removedMessages.delete(info.sessionID)
- 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 }
- const load = messageLoads.get(props.sessionID)
- load?.touchedMessages.add(props.messageID)
- load?.removedMessages.add(props.messageID)
- load?.clearedMessageParts.add(props.messageID)
- load?.deltaParts.delete(props.messageID)
- load?.carriedDeltaParts.delete(props.messageID)
- load?.removedParts.delete(props.messageID)
- load?.optimisticParts.delete(props.messageID)
- pendingParts.get(props.sessionID)?.delete(props.messageID)
- if (pendingParts.get(props.sessionID)?.size === 0) pendingParts.delete(props.sessionID)
- const removedMessagesForSession = removedMessages.get(props.sessionID) ?? new Set<string>()
- removedMessagesForSession.add(props.messageID)
- removedMessages.set(props.sessionID, removedMessagesForSession)
- clearOptimistic(props.sessionID, props.messageID)
- 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)
- }
- deleteMessageParts(draft, props.messageID)
- }),
- )
- return
- }
- case "message.part.updated": {
- const part = (event.properties as { part: Part }).part
- if (SKIP_PARTS.has(part.type)) return
- const messages = data.message[part.sessionID]
- const load = messageLoads.get(part.sessionID)
- const missing = !messages || !Binary.search(messages, part.messageID, (message) => message.id).found
- // Outside a page load, accepting a part without its ordered parent event would create an unbounded orphan.
- if (
- missing &&
- (!load ||
- load.clearedMessageParts.has(part.messageID) ||
- removedMessages.get(part.sessionID)?.has(part.messageID))
- )
- return
- if (missing) {
- const orphans = orphanParts.get(part.sessionID) ?? new Set<string>()
- orphans.add(part.messageID)
- orphanParts.set(part.sessionID, orphans)
- load?.orphanParents.add(part.messageID)
- }
- const deltas = load?.deltaParts.get(part.messageID)
- deltas?.delete(part.id)
- if (deltas?.size === 0) load?.deltaParts.delete(part.messageID)
- const carried = load?.carriedDeltaParts.get(part.messageID)
- carried?.delete(part.id)
- if (carried?.size === 0) load?.carriedDeltaParts.delete(part.messageID)
- const removed = load?.removedParts.get(part.messageID)
- removed?.delete(part.id)
- if (removed?.size === 0) load?.removedParts.delete(part.messageID)
- const pending = pendingParts.get(part.sessionID)?.get(part.messageID)
- pending?.delete(part.id)
- if (pending?.size === 0) pendingParts.get(part.sessionID)?.delete(part.messageID)
- if (pendingParts.get(part.sessionID)?.size === 0) pendingParts.delete(part.sessionID)
- const optimistic = load?.optimisticParts.get(part.messageID)
- optimistic?.delete(part.id)
- if (optimistic?.size === 0) load?.optimisticParts.delete(part.messageID)
- deltaBases.delete(part.id)
- trackPartChange(part.sessionID, part.messageID, part.id)
- confirmOptimisticPart(part.sessionID, part.messageID, part)
- 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 { sessionID: string; messageID: string; partID: string }
- // Part removal is event-only on the server, so its tombstone lasts until a later update or eviction.
- const pending = pendingParts.get(props.sessionID) ?? new Map<string, Set<string>>()
- const parts = pending.get(props.messageID) ?? new Set<string>()
- parts.add(props.partID)
- pending.set(props.messageID, parts)
- pendingParts.set(props.sessionID, pending)
- const deltas = messageLoads.get(props.sessionID)?.deltaParts.get(props.messageID)
- deltas?.delete(props.partID)
- if (deltas?.size === 0) messageLoads.get(props.sessionID)?.deltaParts.delete(props.messageID)
- const load = messageLoads.get(props.sessionID)
- const carried = load?.carriedDeltaParts.get(props.messageID)
- carried?.delete(props.partID)
- if (carried?.size === 0) load?.carriedDeltaParts.delete(props.messageID)
- if (load) {
- const parts = load.removedParts.get(props.messageID) ?? new Set<string>()
- parts.add(props.partID)
- load.removedParts.set(props.messageID, parts)
- const optimistic = load.optimisticParts.get(props.messageID)
- optimistic?.delete(props.partID)
- if (optimistic?.size === 0) load.optimisticParts.delete(props.messageID)
- }
- trackPartChange(props.sessionID, props.messageID, props.partID)
- clearOptimisticPart(props.sessionID, props.messageID, props.partID)
- setData(
- produce((draft) => {
- delete draft.part_text_accum_delta[props.partID]
- deltaBases.delete(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 {
- sessionID: string
- 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
- trackPartChange(props.sessionID, props.messageID, props.partID)
- const load = messageLoads.get(props.sessionID)
- if (load) {
- const parts = load.deltaParts.get(props.messageID) ?? new Set<string>()
- parts.add(props.partID)
- load.deltaParts.set(props.messageID, parts)
- const carried = load.carriedDeltaParts.get(props.messageID)
- carried?.delete(props.partID)
- if (carried?.size === 0) load.carriedDeltaParts.delete(props.messageID)
- }
- const field = props.field as keyof (typeof parts)[number]
- const current = parts[result.index]?.[field]
- if (!deltaBases.has(props.partID) && typeof current === "string")
- deltaBases.set(props.partID, { base: current, sessionID: props.sessionID })
- 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]
- if (!permissions) {
- setData("permission", permission.sessionID, [permission])
- return
- }
- 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]
- if (!questions) {
- setData("question", question.sessionID, [question])
- return
- }
- 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 parts = input.parts
- .filter((part) => !!part?.id && !SKIP_PARTS.has(part.type))
- .sort((a, b) => cmp(a.id, b.id))
- const load = messageLoads.get(input.sessionID)
- if (load?.clearedMessageParts.has(input.message.id)) {
- const touched = load.touchedParts.get(input.message.id) ?? new Set<string>()
- parts.forEach((part) => touched.add(part.id))
- load.touchedParts.set(input.message.id, touched)
- }
- if (load) {
- load.removedMessages.delete(input.message.id)
- load.optimisticParts.set(input.message.id, new Set(parts.map((part) => part.id)))
- }
- const items = optimistic.get(input.sessionID)
- const removedMessagesForSession = removedMessages.get(input.sessionID)
- removedMessagesForSession?.delete(input.message.id)
- if (removedMessagesForSession?.size === 0) removedMessages.delete(input.sessionID)
- if (items) items.set(input.message.id, { ...input, parts, confirmedParts: [] })
- if (!items)
- optimistic.set(input.sessionID, new Map([[input.message.id, { ...input, parts, confirmedParts: [] }]]))
- setData("message", input.sessionID, (messages = []) => merge(messages, [input.message]))
- setData(
- "part_text_accum_delta",
- produce((draft) => {
- for (const part of [...(data.part[input.message.id] ?? []), ...parts]) {
- delete draft[part.id]
- deltaBases.delete(part.id)
- }
- }),
- )
- setData("part", input.message.id, parts)
- },
- remove(input: { sessionID: string; messageID: string }) {
- const item = optimistic.get(input.sessionID)?.get(input.messageID)
- if (!item) return
- messageLoads.get(input.sessionID)?.optimisticParts.delete(input.messageID)
- clearOptimistic(input.sessionID, input.messageID)
- if (item.confirmedMessage) {
- const partIDs = new Set(item.parts.map((part) => part.id))
- setData(
- produce((draft) => {
- for (const part of item.parts) {
- delete draft.part_text_accum_delta[part.id]
- deltaBases.delete(part.id)
- }
- const parts = draft.part[input.messageID]
- if (!parts) return
- draft.part[input.messageID] = parts.filter((part) => !partIDs.has(part.id))
- if (draft.part[input.messageID]?.length === 0) delete draft.part[input.messageID]
- }),
- )
- return
- }
- setData("message", input.sessionID, (messages) => messages?.filter((message) => message.id !== input.messageID))
- setData(produce((draft) => deleteMessageParts(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 active = generation(sessionID)
- return retry(() => client.session.diff({ sessionID })).then((result) => {
- if (generations.get(sessionID) !== active) 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 active = generation(sessionID)
- return retry(() => client.session.todo({ sessionID })).then((result) => {
- if (generations.get(sessionID) !== active) 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<typeof createServerSession>
|