diff --git a/bun.lock b/bun.lock index a5f0bca0080..c1029b45670 100644 --- a/bun.lock +++ b/bun.lock @@ -188,12 +188,15 @@ "@types/bun": "catalog:", "@typescript/native-preview": "catalog:", "effect": "catalog:", + "solid-js": "catalog:", }, "peerDependencies": { "effect": "4.0.0-beta.101", + "solid-js": ">=1.9.0", }, "optionalPeers": [ "effect", + "solid-js", ], }, "packages/codemode": { diff --git a/packages/client/package.json b/packages/client/package.json index fe379b54577..8cbc7ae00c3 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -20,6 +20,7 @@ "./promise": "./src/promise/index.ts", "./promise/api": "./src/promise/api.ts", "./service": "./src/promise/service.ts", + "./solid": "./src/solid/index.ts", "./effect": "./src/effect/index.ts", "./effect/api": "./src/effect/api.ts", "./effect/service": "./src/effect/service.ts" @@ -36,11 +37,15 @@ "@opencode-ai/protocol": "workspace:*" }, "peerDependencies": { - "effect": "4.0.0-beta.101" + "effect": "4.0.0-beta.101", + "solid-js": ">=1.9.0" }, "peerDependenciesMeta": { "effect": { "optional": true + }, + "solid-js": { + "optional": true } }, "devDependencies": { @@ -49,6 +54,7 @@ "@tsconfig/bun": "catalog:", "@types/bun": "catalog:", "@typescript/native-preview": "catalog:", - "effect": "catalog:" + "effect": "catalog:", + "solid-js": "catalog:" } } diff --git a/packages/client/src/solid/connection.ts b/packages/client/src/solid/connection.ts new file mode 100644 index 00000000000..9ae9a0d6510 --- /dev/null +++ b/packages/client/src/solid/connection.ts @@ -0,0 +1,269 @@ +import { batch, onCleanup, onMount } from "solid-js" +import { createStore } from "solid-js/store" +import type { OpenCodeClient, OpenCodeEvent } from "../promise" + +export type ClientConnectionStatus = "connected" | "connecting" | "reconnecting" +export type ClientConnectionEvent = { + readonly type: "client.connection" + readonly created: number + readonly data: { + readonly status: "connecting" | "connected" | "disconnected" | "reconnecting" + readonly attempt: number + readonly error?: string + } +} + +export type ClientConnectionOptions = { + readonly reconnect?: (signal: AbortSignal) => Promise + readonly onEvent: (event: OpenCodeEvent) => void + readonly flushInterval?: number + readonly pageLifecycle?: boolean + readonly log?: { + readonly debug?: (message: string, data?: Readonly>) => void + readonly info?: (message: string, data?: Readonly>) => void + } +} + +const connectTimeout = 2_000 +const reconnectDelay = 1_000 +const connectionHistoryLimit = 50 + +type CurrentDelta = Extract< + OpenCodeEvent, + { type: "session.text.delta" | "session.reasoning.delta" | "session.tool.input.delta" | "session.compaction.delta" } +> + +export function coalesceClientEvents(events: OpenCodeEvent[]) { + return events.reduce((output, event) => { + const current = currentDelta(event) + const previous = output[output.length - 1] + const prior = currentDelta(previous) + if ( + !current || + !prior || + previous?.location?.directory !== event.location?.directory || + currentDeltaKey(prior) !== currentDeltaKey(current) + ) { + output.push(event) + return output + } + const fragment = currentDeltaFragment(prior) + currentDeltaFragment(current) + output[output.length - 1] = { + ...current, + data: + current.type === "session.compaction.delta" + ? { ...current.data, text: fragment } + : { ...current.data, delta: fragment }, + } as CurrentDelta + return output + }, []) +} + +function currentDelta(event: OpenCodeEvent | undefined): CurrentDelta | undefined { + if ( + event?.type === "session.text.delta" || + event?.type === "session.reasoning.delta" || + event?.type === "session.tool.input.delta" || + event?.type === "session.compaction.delta" + ) + return event +} + +function currentDeltaKey(event: CurrentDelta) { + if (event.type === "session.tool.input.delta") + return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.id}` + if (event.type === "session.compaction.delta") return `${event.type}:${event.data.sessionID}` + return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.ordinal}` +} + +function currentDeltaFragment(event: CurrentDelta) { + return event.type === "session.compaction.delta" ? event.data.text : event.data.delta +} + +export function createClientConnection(initialApi: OpenCodeClient, options: ClientConnectionOptions) { + const abort = new AbortController() + const history: ClientConnectionEvent[] = [] + const [connection, setConnection] = createStore<{ + status: ClientConnectionStatus + attempt: number + error?: string + }>({ status: "connecting", attempt: 0 }) + let api = initialApi + let pending: OpenCodeEvent[] = [] + let flushTimer: ReturnType | undefined + let stream: AbortController | undefined + let run: Promise | undefined + let started = false + let generation = 0 + + function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) { + history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } }) + if (history.length > connectionHistoryLimit) history.shift() + } + + function publish(event: OpenCodeEvent) { + pending.push(event) + if (flushTimer) return + flushTimer = setTimeout(() => { + flushTimer = undefined + const events = pending + pending = [] + batch(() => coalesceClientEvents(events).forEach(options.onEvent)) + }, options.flushInterval ?? 10) + } + + async function connect(signal: AbortSignal, attempt: number) { + let connectedAt: number | undefined + const request = new AbortController() + const cancel = () => request.abort(signal.reason) + const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout) + signal.addEventListener("abort", cancel, { once: true }) + + try { + record(attempt === 0 ? "connecting" : "reconnecting", attempt) + options.log?.info?.("event stream connecting", { attempt }) + const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]() + const first = await iterator.next() + if (signal.aborted) return { error: undefined, connectedAt } + if (first.done) + return { + error: request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected"), + connectedAt, + } + if (first.value.type !== "server.connected") + return { error: new Error("Event stream did not start with server.connected"), connectedAt } + + clearTimeout(timeout) + record("connected", attempt) + connectedAt = Date.now() + options.log?.info?.("event stream connected") + publish(first.value) + setConnection({ status: "connected", attempt: 0, error: undefined }) + + while (!signal.aborted) { + const event = await iterator.next() + if (signal.aborted) return { error: undefined, connectedAt } + if (event.done) return { error: new Error("Event stream disconnected"), connectedAt } + if ("durable" in event.value) + options.log?.debug?.("event", { + type: event.value.type, + aggregateID: event.value.durable.aggregateID, + seq: event.value.durable.seq, + }) + publish(event.value) + } + return { error: undefined, connectedAt } + } catch (error) { + return { error, connectedAt } + } finally { + request.abort() + clearTimeout(timeout) + signal.removeEventListener("abort", cancel) + } + } + + async function runStream(active: number) { + let attempt = 0 + while (!abort.signal.aborted && started && generation === active) { + setConnection({ status: attempt === 0 ? "connecting" : "reconnecting", attempt, error: undefined }) + const controller = new AbortController() + stream = controller + const cancel = () => controller.abort(abort.signal.reason) + abort.signal.addEventListener("abort", cancel) + const result = await connect(controller.signal, attempt) + abort.signal.removeEventListener("abort", cancel) + if (abort.signal.aborted || !started || generation !== active) return + if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= reconnectDelay) attempt = 0 + attempt += 1 + const message = errorMessage(result.error) + record("disconnected", attempt, message) + options.log?.info?.("event stream disconnected", { attempt, error: message }) + setConnection({ status: "reconnecting", attempt, error: message }) + + if (options.reconnect) { + const next = await options.reconnect(controller.signal).catch((error) => { + if (!controller.signal.aborted) + options.log?.info?.("server resolution failed", { attempt, error: errorMessage(error) }) + }) + if (abort.signal.aborted || controller.signal.aborted || !started || generation !== active) return + if (next) { + api = next + if (attempt === 1) continue + } + } + await wait(reconnectDelay, controller.signal) + } + } + + function start() { + if (started) return run + started = true + const active = ++generation + const previous = run + const current = (async () => { + if (previous) await previous + await runStream(active) + })().finally(() => { + if (run !== current) return + run = undefined + }) + run = current + return run + } + + function stop() { + started = false + generation += 1 + stream?.abort() + } + + onMount(() => { + if (options.pageLifecycle) { + const pagehide = () => stop() + const pageshow = (event: PageTransitionEvent) => { + if (event.persisted) void start() + } + window.addEventListener("pagehide", pagehide) + window.addEventListener("pageshow", pageshow) + onCleanup(() => { + window.removeEventListener("pagehide", pagehide) + window.removeEventListener("pageshow", pageshow) + }) + } + void start() + }) + + onCleanup(() => { + stop() + abort.abort() + if (flushTimer) clearTimeout(flushTimer) + pending = [] + }) + + return { + status: () => connection.status, + attempt: () => connection.attempt, + error: () => connection.error, + internal: { + history: () => history.slice(), + }, + } +} + +function errorMessage(error: unknown) { + if (error === undefined) return undefined + if (error instanceof Error) return error.message + return String(error) +} + +function wait(delay: number, signal: AbortSignal) { + return new Promise((resolve) => { + const timer = setTimeout(done, delay) + signal.addEventListener("abort", done, { once: true }) + function done() { + clearTimeout(timer) + signal.removeEventListener("abort", done) + resolve() + } + }) +} diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts new file mode 100644 index 00000000000..0d1c5579528 --- /dev/null +++ b/packages/client/src/solid/data.ts @@ -0,0 +1,1550 @@ +// Client data layer: apply server events and cache API reads into a Solid store. +// Prefer straightforward projection. Do not add generation counters, stale-response +// merges, live/history overlays, or other race machinery here—last write wins. +// Reconnect invalidates cached reads; active UI owners decide what to sync again. + +import type { + AgentInfo, + CommandInfo, + FormInfo, + IntegrationInfo, + LocationRef, + LocationGetOutput, + McpResource, + McpServer, + ModelInfo, + PermissionSavedInfo, + PermissionRequest, + PermissionReplyInput, + Project, + ProviderInfo, + ReferenceInfo, + SessionMessageInfo, + SessionMessageAssistant, + SessionMessageAssistantReasoning, + SessionMessageAssistantText, + SessionMessageAssistantTool, + SessionInfo, + SessionInboxInfo, + ShellInfo, + SkillInfo, + VcsInfo, + OpenCodeEvent, + OpenCodeClient, + WebSearchProvider, +} from "../promise" +import { Worktree } from "@opencode-ai/schema/worktree" +import { isPermissionNotFoundError } from "../promise" +import { createStore, produce, reconcile } from "solid-js/store" +import type { SessionInbox } from "@opencode-ai/schema/session-inbox" +import { createEffect, createSignal, onCleanup } from "solid-js" + +export type DataSessionStatus = "idle" | "running" + +export type CreateServerDataInput = { + readonly api: () => OpenCodeClient + readonly directory: string + readonly event: { + readonly on: ( + type: Type, + handler: (event: Extract) => void, + ) => () => void + readonly listen: (handler: (event: { name: OpenCodeEvent["type"]; details: OpenCodeEvent }) => void) => () => void + } + readonly connection?: { + readonly status: () => "connected" | "connecting" | "reconnecting" + } +} + +const messageIDFromEvent = (eventID: string) => eventID.replace(/^evt_/, "msg_") + +// Global MCP elicitations temporarily use "global" instead of a real session ID, so the +// server cannot recover their Location when settling them. Preserve the event Location +// until MCP elicitations carry session ownership. +export type FormWithLocation = FormInfo & { readonly location?: LocationRef } +type ShellWithLocation = ShellInfo & { readonly location: LocationRef } + +type LocationData = { + info?: LocationGetOutput + vcs?: VcsInfo + agent?: AgentInfo[] + command?: CommandInfo[] + integration?: IntegrationInfo[] + mcp?: { + server?: McpServer[] + resource?: McpResource[] + } + model?: ModelInfo[] + provider?: ProviderInfo[] + reference?: ReferenceInfo[] + websearch?: WebSearchProvider[] + // Currently running shell commands for this location, keyed by shell id. Entries are removed + // once the command exits or is deleted, so this only ever holds in-flight shells. + shell?: Record + skill?: SkillInfo[] +} + +type Store = { + session: { + info: Record + // Family index keyed by a family's root (or furthest-known-ancestor when the + // true root is not yet loaded). The value is a flat deduplicated list of every + // session ID in that family, including the key itself once its info arrives. + family: Record + active: Record + message: Record + pending: Record + input: Record + permission: Record + // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. + form: Record + } + project: { + info: Record + permission: Record + } + location: Record +} + +export function locationKey(location: LocationRef) { + return JSON.stringify([location.directory, location.workspaceID]) +} + +function locationQuery(ref?: LocationRef | null) { + return ref?.directory ? { directory: ref.directory, workspace: ref.workspaceID } : undefined +} + +function createSync() { + const state = new Map>() + return { + run(key: string, load: () => Promise) { + const active = state.get(key) + if (active === true) return Promise.resolve() + if (active) return active + const pending = load() + .then(() => { + if (state.get(key) === pending) state.set(key, true) + }) + .finally(() => { + if (state.get(key) === pending) state.delete(key) + }) + state.set(key, pending) + return pending + }, + complete(key: string) { + if (state.has(key)) return + state.set(key, true) + }, + invalidate(key?: string) { + if (key) { + state.delete(key) + return + } + state.clear() + }, + } +} + +export function createServerData(config: CreateServerDataInput) { + const api = config.api + + const [store, setStore] = createStore({ + session: { + info: {}, + family: {}, + active: {}, + message: {}, + pending: {}, + input: {}, + permission: {}, + form: {}, + }, + project: { + info: {}, + permission: {}, + }, + location: {}, + }) + + const [defaultLocation, setDefaultLocation] = createSignal({ directory: config.directory }) + const messageIndex = new Map>() + const sync = createSync() + + function setSessionActive(sessionID: string, status: DataSessionStatus) { + setStore("session", "active", sessionID, status) + } + + function addPending(item: SessionInboxInfo) { + if (store.session.pending[item.sessionID]?.some((pending) => pending.id === item.id)) return + setStore("session", "pending", item.sessionID, [...(store.session.pending[item.sessionID] ?? []), item]) + } + + function removePending(sessionID: string, inboxID?: string) { + if (!inboxID) return + if (store.session.pending[sessionID]?.some((item) => item.id === inboxID)) + setStore( + "session", + "pending", + sessionID, + (store.session.pending[sessionID] ?? []).filter((item) => item.id !== inboxID), + ) + if (store.session.input[sessionID]?.includes(inboxID)) + setStore( + "session", + "input", + sessionID, + (store.session.input[sessionID] ?? []).filter((id) => id !== inboxID), + ) + } + + function removePermission(sessionID: string, requestID: string) { + const requests = store.session.permission[sessionID] + if (!requests?.some((request) => request.id === requestID)) return + setStore( + "session", + "permission", + sessionID, + requests.filter((request) => request.id !== requestID), + ) + } + + function updatePending(sessionID: string, inboxID: string, delivery: SessionInbox.Delivery) { + const index = store.session.pending[sessionID]?.findIndex((item) => item.id === inboxID) ?? -1 + const item = store.session.pending[sessionID]?.[index] + if (index < 0 || !item || item.delivery === delivery) return + setStore("session", "pending", sessionID, index, { ...item, delivery }) + } + + const message = { + update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { + setStore( + "session", + "message", + produce((draft) => { + fn((draft[sessionID] ??= []), index(sessionID)) + }), + ) + }, + append(messages: SessionMessageInfo[], index: Map, item: SessionMessageInfo) { + if (index.has(item.id)) return + index.set(item.id, messages.length) + messages.push(item) + }, + activeAssistant(messages: SessionMessageInfo[]) { + const item = messages.findLast((item) => item.type === "assistant" && !item.time.completed) + return item?.type === "assistant" ? item : undefined + }, + assistant(messages: SessionMessageInfo[], index: Map, messageID: string) { + const position = index.get(messageID) + const item = position === undefined ? undefined : messages[position] + return item?.type === "assistant" ? item : undefined + }, + shell(messages: SessionMessageInfo[], shellID: string) { + const item = messages.findLast((item) => item.type === "shell" && item.shellID === shellID) + return item?.type === "shell" ? item : undefined + }, + compaction(messages: SessionMessageInfo[]) { + const item = messages.findLast((item) => item.type === "compaction" && item.status === "running") + return item?.type === "compaction" ? item : undefined + }, + latestTool(assistant: SessionMessageAssistant | undefined, id?: string) { + return assistant?.content.findLast( + (item): item is SessionMessageAssistantTool => item.type === "tool" && (id === undefined || item.id === id), + ) + }, + latestText(assistant: SessionMessageAssistant | undefined) { + return assistant?.content.findLast((item): item is SessionMessageAssistantText => item.type === "text") + }, + latestReasoning(assistant: SessionMessageAssistant | undefined) { + return assistant?.content.findLast( + (item): item is SessionMessageAssistantReasoning => item.type === "reasoning" && !item.time?.completed, + ) + }, + reindex(messages: SessionMessageInfo[], index: Map, start: number) { + for (let position = start; position < messages.length; position++) { + const item = messages[position] + if (item) index.set(item.id, position) + } + }, + } + + function index(sessionID: string) { + const existing = messageIndex.get(sessionID) + if (existing) return existing + const created = new Map() + messageIndex.set(sessionID, created) + return created + } + + // Walk parentID upward through loaded session info to the family root. When a + // parent's info is missing, that missing ID is the furthest-known ancestor and + // is returned so orphan subtrees group under it until the parent arrives. A + // seen set guards against parent cycles, stopping at the last non-repeating + // ancestor. + function resolveRoot(sessionID: string) { + let current = sessionID + let parentID = store.session.info[sessionID]?.parentID + const seen = new Set([sessionID]) + while (parentID) { + if (seen.has(parentID)) break + seen.add(parentID) + current = parentID + parentID = store.session.info[parentID]?.parentID + } + return current + } + + // Register one session into the family index. Idempotent: syncing an + // existing session never duplicates its ID. When a tentative family keyed by + // sessionID exists (descendants arrived while sessionID's own info was + // absent) but sessionID turns out to have a parent, fold the orphan subtree + // into the resolved root's family and drop the tentative entry. + function registerSession(sessionID: string) { + const info = store.session.info[sessionID] + if (!info) return + const rootID = resolveRoot(sessionID) + setStore( + "session", + "family", + produce((draft) => { + if (sessionID !== rootID && draft[sessionID]) { + const members = (draft[rootID] ??= []) + for (const id of draft[sessionID]) { + if (!members.includes(id)) members.push(id) + } + delete draft[sessionID] + } + const family = (draft[rootID] ??= []) + if (!family.includes(sessionID)) family.push(sessionID) + }), + ) + } + + function removeSession(sessionID: string) { + messageIndex.delete(sessionID) + sync.invalidate(`session:${sessionID}`) + sync.invalidate(`session.pending:${sessionID}`) + sync.invalidate(`session.message:${sessionID}`) + sync.invalidate(`session.permission:${sessionID}`) + sync.invalidate(`session.form:${sessionID}:`) + setStore( + "session", + produce((draft) => { + delete draft.info[sessionID] + delete draft.active[sessionID] + delete draft.message[sessionID] + delete draft.pending[sessionID] + delete draft.input[sessionID] + delete draft.permission[sessionID] + delete draft.form[sessionID] + for (const [rootID, family] of Object.entries(draft.family)) { + const next = family.filter((id) => id !== sessionID) + if (next.length === 0) delete draft.family[rootID] + else draft.family[rootID] = next + } + }), + ) + } + + function handleEvent(event: OpenCodeEvent) { + switch (event.type) { + case "server.connected": + void api() + .session.active() + .then((active) => { + setStore( + "session", + "active", + reconcile(Object.fromEntries(Object.keys(active).map((sessionID) => [sessionID, "running" as const]))), + ) + }) + .catch(() => undefined) + void api() + .location.get({ location: locationQuery(defaultLocation()) }) + .then((location) => { + const key = locationKey(location) + setStore("location", key, { ...store.location[key], info: location }) + setDefaultLocation({ directory: location.directory, workspaceID: location.workspaceID }) + }) + .catch((error) => console.error("Failed to preload location", error)) + void result.location.vcs.sync().catch((error) => console.error("Failed to preload VCS info", error)) + void result.project.sync().catch((error) => console.error("Failed to preload projects", error)) + return + case "session.created": + result.session.invalidate(event.data.sessionID) + void result.session.sync(event.data.sessionID) + // Band-aid: a newly created session starts empty, so live events can be its source of truth. + // Fetching pending inputs and projected messages separately lets promotion move an input between snapshots, + // causing both requests to miss it and overwrite event-built state. Skip those racy initial reads until + // hydration can load pending and projected messages atomically. + sync.complete(`session.pending:${event.data.sessionID}`) + sync.complete(`session.message:${event.data.sessionID}`) + return + case "session.forked": + result.session.invalidate(event.data.sessionID) + result.session.pending.invalidate(event.data.sessionID) + result.session.message.invalidate(event.data.sessionID) + void result.session.sync(event.data.sessionID) + return + case "session.deleted": + removeSession(event.data.sessionID) + return + case "session.usage.updated": + if (store.session.info[event.data.sessionID]) + setStore("session", "info", event.data.sessionID, { + cost: event.data.cost, + tokens: event.data.tokens, + }) + return + case "session.agent.selected": { + const previous = store.session.info[event.data.sessionID]?.agent + if (store.session.info[event.data.sessionID]) + setStore("session", "info", event.data.sessionID, "agent", event.data.agent) + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "agent-switched", + agent: event.data.agent, + previous, + time: { created: event.created }, + }) + }) + return + } + case "session.model.selected": + if (store.session.info[event.data.sessionID]) + setStore("session", "info", event.data.sessionID, "model", event.data.model) + if (!store.session.message[event.data.sessionID]) return + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "model-switched", + model: event.data.model, + time: { created: event.created }, + }) + }) + void api() + .session.message({ sessionID: event.data.sessionID, messageID: messageIDFromEvent(event.id) }) + .then((item) => { + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(item.id) + if (position === undefined) return message.append(draft, index, item) + draft[position] = item + }) + }) + .catch((error) => console.error("Failed to load projected model switch message", error)) + return + case "session.renamed": + // Preserve the live title when it races the session's initial read. + void result.session.sync(event.data.sessionID).then(() => { + if (store.session.info[event.data.sessionID]) + setStore("session", "info", event.data.sessionID, "title", event.data.title) + }) + return + case "session.moved": { + const current = store.session.info[event.data.sessionID] + if (current) { + const previous = { + location: { ...current.location }, + projectID: current.projectID, + subpath: current.subpath, + } + setStore("session", "info", event.data.sessionID, "location", event.data.location) + if (event.data.projectID) setStore("session", "info", event.data.sessionID, "projectID", event.data.projectID) + setStore("session", "info", event.data.sessionID, "subpath", event.data.subpath) + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "location-switched", + location: event.data.location, + projectID: event.data.projectID, + subpath: event.data.subpath, + previous, + time: { created: event.created }, + }) + }) + } + return + } + case "worktree.resolved": { + for (const [sessionID, info] of Object.entries(store.session.info)) { + const adopted = Worktree.adopt({ projectID: info.projectID, directory: info.location.directory }, event.data) + if (!adopted) continue + setStore("session", "info", sessionID, "projectID", adopted.projectID) + setStore("session", "info", sessionID, "subpath", adopted.subpath) + } + result.project.invalidate() + void result.project.sync() + return + } + case "worktree.updated": + result.project.invalidate() + void result.project.sync() + return + case "session.inbox.delivered": { + const admitted = store.session.input[event.data.sessionID]?.includes(event.data.inboxID) ?? false + removePending(event.data.sessionID, event.data.inboxID) + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(event.data.inboxID) + if (position === undefined) return + const existing = draft[position] + if (!existing || !admitted) return + existing.time.created = event.created + draft.splice(position, 1) + draft.push(existing) + message.reindex(draft, index, position) + }) + return + } + case "session.inbox.delivery.changed": + updatePending(event.data.sessionID, event.data.inboxID, event.data.delivery) + return + case "session.inbox.cancelled": { + removePending(event.data.sessionID, event.data.inboxID) + if (messageIndex.get(event.data.sessionID)?.has(event.data.inboxID)) + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(event.data.inboxID) + if (position === undefined) return + draft.splice(position, 1) + index.delete(event.data.inboxID) + message.reindex(draft, index, position) + }) + return + } + case "session.inbox.enqueued": { + const item = event.data.item + addPending({ + id: event.data.inboxID, + sessionID: event.data.sessionID, + timeCreated: event.created, + ...item, + }) + if (!store.session.input[event.data.sessionID]?.includes(event.data.inboxID)) + setStore("session", "input", event.data.sessionID, [ + ...(store.session.input[event.data.sessionID] ?? []), + event.data.inboxID, + ]) + if (item.type !== "user" && item.type !== "synthetic") return + message.update(event.data.sessionID, (draft, index) => { + message.append( + draft, + index, + item.type === "user" + ? { + id: event.data.inboxID, + type: "user", + ...item.payload, + time: { created: event.created }, + } + : { + id: event.data.inboxID, + type: "synthetic", + ...item.payload, + time: { created: event.created }, + }, + ) + }) + return + } + case "session.instructions.updated": + // Mirror the projector: the initial baseline and empty-rendering deltas carry no text + // and produce no transcript message. + const updateText = event.data.text + if (updateText === undefined) return + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "system", + text: updateText, + description: `Instructions updated: ${Object.keys(event.data.delta).join(", ")}`, + metadata: event.metadata, + time: { created: event.created }, + }) + }) + return + case "session.synthetic": + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "synthetic", + text: event.data.text, + description: event.data.description, + metadata: event.data.metadata, + time: { created: event.created }, + }) + }) + return + case "session.shell.started": + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "shell", + shellID: event.data.shell.id, + command: event.data.shell.command, + status: event.data.shell.status, + exit: event.data.shell.exit, + metadata: event.metadata, + time: { created: event.created }, + }) + }) + return + case "session.shell.ended": + message.update(event.data.sessionID, (draft) => { + const match = message.shell(draft, event.data.shell.id) + if (!match) return + match.status = event.data.shell.status + match.exit = event.data.shell.exit + match.output = event.data.output + match.time.completed = event.created + }) + return + case "session.step.started": + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(event.data.assistantMessageID) + const existing = position === undefined ? undefined : draft[position] + if (existing?.type === "assistant") { + existing.agent = event.data.agent + existing.model = event.data.model + existing.retry = undefined + existing.error = undefined + existing.finish = undefined + existing.time.completed = undefined + if (event.data.snapshot) existing.snapshot = { ...existing.snapshot, start: event.data.snapshot } + return + } + const currentAssistant = message.activeAssistant(draft) + if (currentAssistant) { + currentAssistant.retry = undefined + currentAssistant.time.completed = event.created + } + message.append(draft, index, { + id: event.data.assistantMessageID, + type: "assistant", + agent: event.data.agent, + model: event.data.model, + metadata: event.metadata, + content: [], + snapshot: event.data.snapshot ? { start: event.data.snapshot } : undefined, + time: { created: event.created }, + }) + }) + return + case "session.step.ended": { + message.update(event.data.sessionID, (draft, index) => { + const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) + if (!currentAssistant) return + currentAssistant.time.completed = event.created + currentAssistant.finish = event.data.finish + currentAssistant.cost = event.data.cost + currentAssistant.tokens = event.data.tokens + if (event.data.snapshot) + currentAssistant.snapshot = { ...currentAssistant.snapshot, end: event.data.snapshot } + }) + return + } + case "session.step.failed": + message.update(event.data.sessionID, (draft, index) => { + const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) + if (!currentAssistant) return + currentAssistant.time.completed = event.created + currentAssistant.finish = "error" + currentAssistant.error = event.data.error + currentAssistant.retry = undefined + if (event.data.cost !== undefined && event.data.tokens !== undefined) { + currentAssistant.cost = event.data.cost + currentAssistant.tokens = event.data.tokens + } + }) + return + case "session.text.started": + message.update(event.data.sessionID, (draft, index) => { + message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ + type: "text", + text: "", + }) + }) + return + case "session.text.delta": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestText(message.assistant(draft, index, event.data.assistantMessageID)) + if (match) match.text += event.data.delta + }) + return + case "session.text.ended": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestText(message.assistant(draft, index, event.data.assistantMessageID)) + if (match) match.text = event.data.text + }) + return + case "session.tool.input.started": + message.update(event.data.sessionID, (draft, index) => { + message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ + type: "tool", + id: event.data.id, + name: event.data.name, + time: { created: event.created }, + state: { status: "streaming", input: "" }, + }) + }) + return + case "session.tool.input.delta": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestTool( + message.assistant(draft, index, event.data.assistantMessageID), + event.data.id, + ) + if (match?.state.status === "streaming") match.state.input += event.data.delta + }) + return + case "session.tool.input.ended": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestTool( + message.assistant(draft, index, event.data.assistantMessageID), + event.data.id, + ) + if (match?.state.status === "streaming") match.state.input = event.data.text + }) + return + case "session.tool.called": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestTool( + message.assistant(draft, index, event.data.assistantMessageID), + event.data.id, + ) + if (!match) return + match.time.ran = event.created + match.executed = event.data.executed + match.providerState = event.data.state + match.state = { status: "running", input: event.data.input, metadata: {} } + }) + return + case "session.tool.progress": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestTool( + message.assistant(draft, index, event.data.assistantMessageID), + event.data.id, + ) + if (match?.state.status !== "running") return + match.state.metadata = event.data.metadata + }) + return + case "session.tool.success": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestTool( + message.assistant(draft, index, event.data.assistantMessageID), + event.data.id, + ) + if (match?.state.status !== "running") return + match.state = { + status: "completed", + input: match.state.input, + metadata: event.data.metadata, + content: [...event.data.content], + } + match.executed = event.data.executed || match.executed === true + match.providerResultState = event.data.resultState + match.time.completed = event.created + }) + return + case "session.tool.failed": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestTool( + message.assistant(draft, index, event.data.assistantMessageID), + event.data.id, + ) + if (!match || (match.state.status !== "streaming" && match.state.status !== "running")) return + match.state = { + status: "error", + error: event.data.error, + input: typeof match.state.input === "string" ? {} : match.state.input, + metadata: event.data.metadata, + content: event.data.content, + } + match.executed = event.data.executed || match.executed === true + match.providerResultState = event.data.resultState + match.time.completed = event.created + }) + return + case "session.reasoning.started": + message.update(event.data.sessionID, (draft, index) => { + message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ + type: "reasoning", + text: "", + state: event.data.state, + time: { created: event.created }, + }) + }) + return + case "session.reasoning.delta": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestReasoning(message.assistant(draft, index, event.data.assistantMessageID)) + if (match) match.text += event.data.delta + }) + return + case "session.reasoning.ended": + message.update(event.data.sessionID, (draft, index) => { + const match = message.latestReasoning(message.assistant(draft, index, event.data.assistantMessageID)) + if (match) { + match.text = event.data.text + match.time = { created: match.time?.created ?? event.created, completed: event.created } + if (event.data.state !== undefined) match.state = event.data.state + } + }) + return + case "session.retry.scheduled": + message.update(event.data.sessionID, (draft, index) => { + const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) + if (!currentAssistant) return + currentAssistant.retry = { + attempt: event.data.attempt, + at: event.data.at, + error: event.data.error, + } + }) + return + case "session.execution.started": + setSessionActive(event.data.sessionID, "running") + return + case "session.compaction.started": + if (event.data.inputID) removePending(event.data.sessionID, event.data.inputID) + message.update(event.data.sessionID, (draft, index) => { + message.append(draft, index, { + id: event.data.inputID ?? messageIDFromEvent(event.id), + type: "compaction", + status: "running", + reason: event.data.reason, + summary: "", + recent: event.data.recent ?? "", + time: { created: event.created }, + }) + }) + return + case "session.execution.succeeded": + case "session.execution.failed": + case "session.execution.interrupted": + setSessionActive(event.data.sessionID, "idle") + message.update(event.data.sessionID, (draft) => { + const currentAssistant = message.activeAssistant(draft) + if (currentAssistant) currentAssistant.retry = undefined + }) + return + case "session.revert.staged": + if (store.session.info[event.data.sessionID]) + setStore("session", "info", event.data.sessionID, "revert", event.data.revert) + return + case "session.revert.cleared": + if (store.session.info[event.data.sessionID]) + setStore("session", "info", event.data.sessionID, "revert", undefined) + return + case "session.revert.committed": + if (store.session.info[event.data.sessionID]) { + setStore("session", "info", event.data.sessionID, "revert", undefined) + } + setStore( + "session", + "input", + event.data.sessionID, + (store.session.input[event.data.sessionID] ?? []).filter((id) => id < event.data.to), + ) + message.update(event.data.sessionID, (draft, index) => { + const position = draft.findIndex((item) => item.id >= event.data.to) + if (position === -1) return + for (const item of draft.splice(position)) index.delete(item.id) + }) + return + case "session.compaction.delta": + message.update(event.data.sessionID, (draft) => { + const current = message.compaction(draft) + if (current?.status === "running") current.summary += event.data.text + }) + return + case "session.compaction.ended": + message.update(event.data.sessionID, (draft, index) => { + const position = draft.findLastIndex((item) => item.type === "compaction" && item.status === "running") + const current = draft[position] + if (current?.type === "compaction") { + Object.assign(current, { + status: "completed", + reason: event.data.reason, + summary: event.data.text, + recent: event.data.recent, + }) + return + } + message.append(draft, index, { + id: messageIDFromEvent(event.id), + type: "compaction", + status: "completed", + reason: event.data.reason, + summary: event.data.text, + recent: event.data.recent, + time: { created: event.created }, + }) + }) + return + case "session.compaction.failed": + if (event.data.inputID) removePending(event.data.sessionID, event.data.inputID) + message.update(event.data.sessionID, (draft, index) => { + const position = draft.findLastIndex((item) => item.type === "compaction" && item.status === "running") + const current = draft[position] + const failed: Extract = { + id: current?.id ?? event.data.inputID ?? messageIDFromEvent(event.id), + type: "compaction", + status: "failed", + reason: event.data.reason ?? "manual", + error: event.data.error ?? { + type: "compaction.failed", + message: "Compaction failed before recording an error", + }, + metadata: current?.type === "compaction" ? current.metadata : event.metadata, + time: current?.type === "compaction" ? current.time : { created: event.created }, + } + if (current?.type === "compaction") { + draft[position] = failed + return + } + message.append(draft, index, failed) + }) + return + case "permission.asked": + if (store.session.permission[event.data.sessionID]?.some((request) => request.id === event.data.id)) return + setStore("session", "permission", event.data.sessionID, [ + ...(store.session.permission[event.data.sessionID] ?? []), + event.data, + ]) + return + case "permission.replied": + removePermission(event.data.sessionID, event.data.requestID) + return + case "form.replied": + case "form.cancelled": + setStore( + "session", + "form", + event.data.sessionID, + (store.session.form[event.data.sessionID] ?? []).filter((form) => form.id !== event.data.id), + ) + return + } + + if (!event.location) return + const location = event.location + switch (event.type) { + case "catalog.updated": + result.location.model.invalidate(location) + result.location.provider.invalidate(location) + void Promise.all([result.location.model.sync(location), result.location.provider.sync(location)]) + break + case "agent.updated": + result.location.agent.invalidate(location) + void result.location.agent.sync(location) + break + case "command.updated": + result.location.command.invalidate(location) + void result.location.command.sync(location) + break + case "skill.updated": + result.location.skill.invalidate(location) + void result.location.skill.sync(location) + break + case "vcs.branch.updated": + setStore("location", locationKey(location), (data) => ({ + ...data, + vcs: { + branch: { + ...data?.vcs?.branch, + current: event.data.branch, + }, + }, + })) + break + case "form.created": + if (store.session.form[event.data.form.sessionID]?.some((form) => form.id === event.data.form.id)) break + setStore("session", "form", event.data.form.sessionID, [ + ...(store.session.form[event.data.form.sessionID] ?? []), + event.data.form.sessionID === "global" ? { ...event.data.form, location } : event.data.form, + ]) + break + case "shell.created": + setStore("location", locationKey(location), (data) => ({ + ...data, + shell: { + ...data?.shell, + [event.data.info.id]: { ...event.data.info, location }, + }, + })) + break + case "shell.exited": + case "shell.deleted": + setStore("location", locationKey(location), (data) => ({ + ...data, + shell: Object.fromEntries(Object.entries(data?.shell ?? {}).filter(([id]) => id !== event.data.id)), + })) + break + case "reference.updated": + result.location.reference.invalidate(location) + void result.location.reference.sync(location) + break + case "integration.updated": + result.location.integration.invalidate(location) + result.location.model.invalidate(location) + result.location.provider.invalidate(location) + void Promise.all([ + result.location.integration.sync(location), + result.location.model.sync(location), + result.location.provider.sync(location), + ]) + break + case "config.updated": + case "websearch.updated": + void result.location.websearch.refresh(location) + break + // Authenticating an MCP integration reconnects its server, which emits mcp.status.changed, + // so the mcp list syncs here rather than off integration.updated. + case "mcp.status.changed": + result.location.mcp.server.invalidate(location) + void result.location.mcp.server.sync(location) + break + case "mcp.resources.changed": + result.location.mcp.resource.invalidate(location) + void result.location.mcp.resource.sync(location) + break + } + } + + const result = { + on: config.event.on, + listen: config.event.listen, + session: { + list() { + return Object.values(store.session.info).toSorted((a, b) => b.time.updated - a.time.updated) + }, + get(sessionID: string) { + return store.session.info[sessionID] + }, + remember(info: SessionInfo) { + setStore("session", "info", info.id, reconcile(info)) + sync.complete(`session:${info.id}`) + registerSession(info.id) + }, + setStatus(sessionID: string, status: DataSessionStatus) { + setSessionActive(sessionID, status) + }, + lineage: { + peek(sessionID: string) { + const session = store.session.info[sessionID] + if (!session) return + const seen = new Set([session.id]) + let root = session + while (root.parentID) { + if (seen.has(root.parentID)) return { session, root } + seen.add(root.parentID) + const parent = store.session.info[root.parentID] + if (!parent) return + root = parent + } + return { session, root } + }, + async resolve(sessionID: string) { + await result.session.sync(sessionID) + const session = store.session.info[sessionID] + if (!session) throw new Error(`Session not found: ${sessionID}`) + const seen = new Set([session.id]) + let root = session + while (root.parentID) { + if (seen.has(root.parentID)) return { session, root } + seen.add(root.parentID) + await result.session.sync(root.parentID) + const parent = store.session.info[root.parentID] + if (!parent) throw new Error(`Session not found: ${root.parentID}`) + root = parent + } + return { session, root } + }, + }, + root(sessionID: string) { + return resolveRoot(sessionID) + }, + family(sessionID: string) { + return store.session.family[resolveRoot(sessionID)] ?? [] + }, + cost(sessionID: string) { + const session = store.session.info[sessionID] + if (!session) return 0 + if (session.parentID) return session.cost + return (store.session.family[sessionID] ?? [sessionID]).reduce( + (total, id) => total + (store.session.info[id]?.cost ?? 0), + 0, + ) + }, + status(sessionID: string) { + return store.session.active[sessionID] ?? "idle" + }, + input: { + list(sessionID: string) { + return store.session.input[sessionID] ?? [] + }, + has(sessionID: string, inboxID: string) { + return store.session.input[sessionID]?.includes(inboxID) ?? false + }, + }, + pending: { + list(sessionID: string) { + return store.session.pending[sessionID] ?? [] + }, + sync(sessionID: string) { + return sync.run(`session.pending:${sessionID}`, async () => { + const pending = await api().session.inbox.list({ sessionID }) + setStore("session", "pending", sessionID, reconcile(pending)) + setStore( + "session", + "input", + sessionID, + reconcile(pending.filter((item) => item.type !== "compaction").map((item) => item.id)), + ) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session.pending:${sessionID}`) + }, + }, + sync(sessionID: string, options?: { children?: boolean }) { + return sync.run(options?.children ? `session.family:${sessionID}` : `session:${sessionID}`, async () => { + const [info, children] = await Promise.all([ + api().session.get({ sessionID }), + options?.children + ? api() + .session.list({ parentID: sessionID, order: "desc" }) + .then((response) => response.data) + : [], + ]) + const sessions = [info, ...children] + setStore( + "session", + "info", + produce((draft) => { + for (const session of sessions) draft[session.id] = session + }), + ) + for (const session of sessions) { + sync.complete(`session:${session.id}`) + registerSession(session.id) + } + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session:${sessionID}`) + }, + message: { + list(sessionID: string) { + return store.session.message[sessionID] ?? [] + }, + get(sessionID: string, messageID: string) { + const messages = store.session.message[sessionID] + const position = messageIndex.get(sessionID)?.get(messageID) + return position === undefined ? undefined : messages?.[position] + }, + sync(sessionID: string) { + return sync.run(`session.message:${sessionID}`, async () => { + const messages = (await api().message.list({ sessionID, limit: 200, order: "desc" })).data.toReversed() + messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) + setStore("session", "message", sessionID, reconcile(messages)) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session.message:${sessionID}`) + }, + }, + permission: { + list(sessionID: string) { + return store.session.permission[sessionID] + }, + sync(sessionID: string) { + return sync.run(`session.permission:${sessionID}`, async () => { + setStore("session", "permission", sessionID, await api().permission.list({ sessionID })) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session.permission:${sessionID}`) + }, + async reply(input: PermissionReplyInput) { + await api() + .permission.reply(input) + .catch((error: unknown) => { + if (!isPermissionNotFoundError(error)) throw error + }) + removePermission(input.sessionID, input.requestID) + }, + }, + form: { + list(sessionID: string, ref?: LocationRef) { + const forms = store.session.form[sessionID] + if (sessionID !== "global") return forms + if (!ref) return + const key = locationKey(ref) + return forms?.filter((form) => form.location && locationKey(form.location) === key) + }, + sync(sessionID: string, ref?: LocationRef) { + const key = `session.form:${sessionID}:${sessionID === "global" ? locationKey(ref ?? defaultLocation()) : ""}` + return sync.run(key, async () => { + if (sessionID === "global") { + const response = await api().form.request.list({ + location: locationQuery(ref ?? defaultLocation()), + }) + const location = { + directory: response.location.directory, + workspaceID: response.location.workspaceID, + } + const locationID = locationKey(location) + setStore("session", "form", sessionID, [ + ...(store.session.form[sessionID] ?? []).filter( + (form) => form.location && locationKey(form.location) !== locationID, + ), + ...response.data.filter((form) => form.sessionID === "global").map((form) => ({ ...form, location })), + ]) + return + } + setStore("session", "form", sessionID, await api().form.list({ sessionID })) + }) + }, + invalidate(sessionID: string, ref?: LocationRef) { + sync.invalidate( + `session.form:${sessionID}:${sessionID === "global" ? locationKey(ref ?? defaultLocation()) : ""}`, + ) + }, + }, + }, + project: { + list() { + return Object.values(store.project.info).toSorted((a, b) => b.time.updated - a.time.updated) + }, + get(projectID: string) { + return store.project.info[projectID] + }, + sync() { + return sync.run("project", async () => { + const projects = await api().project.list() + setStore("project", "info", reconcile(Object.fromEntries(projects.map((project) => [project.id, project])))) + }) + }, + invalidate() { + sync.invalidate("project") + }, + permission: { + list(projectID: string) { + return store.project.permission[projectID] + }, + sync(projectID: string) { + return sync.run(`project.permission:${projectID}`, async () => { + setStore("project", "permission", projectID, await api().permission.saved.list({ projectID })) + }) + }, + invalidate(projectID: string) { + sync.invalidate(`project.permission:${projectID}`) + }, + }, + }, + shell: { + list(location?: LocationRef) { + return Object.values(store.location[locationKey(location ?? defaultLocation())]?.shell ?? {}) + }, + listBySession(sessionID: string) { + return Object.values(store.location) + .flatMap((data) => Object.values(data.shell ?? {})) + .filter((shell) => shell.metadata.sessionID === sessionID) + }, + get(id: string) { + return Object.values(store.location) + .map((data) => data.shell?.[id]) + .find((shell) => shell !== undefined) + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.shell:${id}`, async () => { + const response = await api().shell.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { + ...store.location[key], + shell: Object.fromEntries( + response.data.map((info) => [ + info.id, + { + ...info, + location: { + directory: response.location.directory, + workspaceID: response.location.workspaceID, + }, + }, + ]), + ), + }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.shell:${locationKey(ref ?? defaultLocation())}`) + }, + }, + location: { + info(ref?: LocationRef) { + return store.location[locationKey(ref ?? defaultLocation())]?.info + }, + default() { + return defaultLocation() + }, + syncInfo(ref?: LocationRef) { + const current = ref ?? defaultLocation() + return sync.run(`location:${locationKey(current)}`, async () => { + const location = await api().location.get({ location: locationQuery(current) }) + const key = locationKey(location) + if (!store.location[key]) setStore("location", key, {}) + setStore("location", key, "info", location) + const requested = locationKey(current) + if (requested !== key) { + if (!store.location[requested]) setStore("location", requested, {}) + setStore("location", requested, "info", location) + } + if (!ref) { + setDefaultLocation({ directory: location.directory, workspaceID: location.workspaceID }) + } + }) + }, + async sync(ref?: LocationRef) { + await result.location.syncInfo(ref) + const requested = ref ?? defaultLocation() + const info = result.location.info(requested) + const location = info + ? { directory: info.directory, workspaceID: info.workspaceID } + : requested + await Promise.all([ + result.location.vcs.sync(location), + result.location.agent.sync(location), + result.location.command.sync(location), + result.location.integration.sync(location), + result.location.mcp.server.sync(location), + result.location.mcp.resource.sync(location), + result.location.model.sync(location), + result.location.provider.sync(location), + result.location.reference.sync(location), + result.location.skill.sync(location), + result.shell.sync(location), + result.session.form.sync("global", location), + ]) + }, + invalidate(ref?: LocationRef) { + const location = ref ?? defaultLocation() + sync.invalidate(`location:${locationKey(location)}`) + result.location.vcs.invalidate(location) + result.location.agent.invalidate(location) + result.location.command.invalidate(location) + result.location.integration.invalidate(location) + result.location.mcp.server.invalidate(location) + result.location.mcp.resource.invalidate(location) + result.location.model.invalidate(location) + result.location.provider.invalidate(location) + result.location.reference.invalidate(location) + result.location.skill.invalidate(location) + result.shell.invalidate(location) + result.session.form.invalidate("global", location) + }, + vcs: { + info(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.vcs + }, + sync(ref?: LocationRef) { + const location = ref ?? defaultLocation() + return sync.run(`location.vcs:${locationKey(location)}`, async () => { + const response = await api().vcs.get({ location: locationQuery(location) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], vcs: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.vcs:${locationKey(ref ?? defaultLocation())}`) + }, + }, + agent: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.agent + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.agent:${id}`, async () => { + const response = await api().agent.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], agent: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.agent:${locationKey(ref ?? defaultLocation())}`) + }, + }, + command: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.command + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.command:${id}`, async () => { + const response = await api().command.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], command: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.command:${locationKey(ref ?? defaultLocation())}`) + }, + }, + integration: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.integration + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.integration:${id}`, async () => { + const response = await api().integration.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], integration: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.integration:${locationKey(ref ?? defaultLocation())}`) + }, + }, + mcp: { + server: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.mcp?.server + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.mcp.server:${id}`, async () => { + const response = await api().mcp.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { + ...store.location[key], + mcp: { ...store.location[key]?.mcp, server: response.data }, + }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.mcp.server:${locationKey(ref ?? defaultLocation())}`) + }, + }, + resource: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.mcp?.resource + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.mcp.resource:${id}`, async () => { + const response = await api().mcp.resource.catalog({ + location: locationQuery(ref ?? defaultLocation()), + }) + const key = locationKey(response.location) + setStore("location", key, { + ...store.location[key], + mcp: { ...store.location[key]?.mcp, resource: response.data.resources }, + }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.mcp.resource:${locationKey(ref ?? defaultLocation())}`) + }, + }, + }, + model: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.model + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.model:${id}`, async () => { + const response = await api().model.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], model: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.model:${locationKey(ref ?? defaultLocation())}`) + }, + }, + provider: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.provider + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.provider:${id}`, async () => { + const response = await api().provider.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], provider: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.provider:${locationKey(ref ?? defaultLocation())}`) + }, + }, + reference: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.reference + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.reference:${id}`, async () => { + const response = await api().reference.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], reference: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.reference:${locationKey(ref ?? defaultLocation())}`) + }, + }, + websearch: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.websearch + }, + async refresh(ref?: LocationRef) { + const input = { location: locationQuery(ref ?? defaultLocation()) } + const providers = await api().websearch.providers(input) + const key = locationKey(providers.location) + setStore("location", key, { + ...store.location[key], + websearch: providers.data, + }) + }, + }, + skill: { + list(location?: LocationRef) { + return store.location[locationKey(location ?? defaultLocation())]?.skill + }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.skill:${id}`, async () => { + const response = await api().skill.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], skill: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.skill:${locationKey(ref ?? defaultLocation())}`) + }, + }, + }, + } + + createEffect(() => { + if (config.connection?.status() === "connected") return + sync.invalidate() + }) + + onCleanup( + config.event.listen(({ details }) => { + handleEvent(details) + }), + ) + + return result +} + +export type Data = ReturnType diff --git a/packages/client/src/solid/index.ts b/packages/client/src/solid/index.ts new file mode 100644 index 00000000000..9fd41baf55a --- /dev/null +++ b/packages/client/src/solid/index.ts @@ -0,0 +1,2 @@ +export * from "./data" +export * from "./connection" diff --git a/packages/client/test/solid-connection.test.ts b/packages/client/test/solid-connection.test.ts new file mode 100644 index 00000000000..c75d17ceae1 --- /dev/null +++ b/packages/client/test/solid-connection.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, test } from "bun:test" +import type { OpenCodeEvent } from "../src/promise" +import { coalesceClientEvents } from "../src/solid/connection" + +describe("coalesceClientEvents", () => { + const delta = (id: string, value: string, ordinal = 0) => + ({ + id, + created: 1, + type: "session.text.delta", + location: { directory: "/repo" }, + data: { sessionID: "ses", assistantMessageID: "msg", ordinal, delta: value }, + }) as OpenCodeEvent + + test("merges adjacent deltas for the same stream", () => { + const result = coalesceClientEvents([delta("evt_1", "hello "), delta("evt_2", "world")]) + expect(result).toHaveLength(1) + expect(result[0]).toMatchObject({ id: "evt_2", data: { delta: "hello world" } }) + }) + + test("coalesces tool input deltas by tool ID", () => { + const current = (eventID: string, id: string, value: string) => + ({ + id: eventID, + created: 1, + type: "session.tool.input.delta", + location: { directory: "/repo" }, + data: { sessionID: "ses", assistantMessageID: "msg", id, delta: value }, + }) as OpenCodeEvent + const result = coalesceClientEvents([ + current("evt_1", "call_1", "{"), + current("evt_2", "call_1", "}"), + current("evt_3", "call_2", "[]"), + ]) + expect(result).toHaveLength(2) + expect(result[0]).toMatchObject({ id: "evt_2", data: { id: "call_1", delta: "{}" } }) + expect(result[1]).toMatchObject({ id: "evt_3", data: { id: "call_2", delta: "[]" } }) + }) + + test("preserves boundaries between distinct delta streams", () => { + const events = [delta("evt_1", "a"), delta("evt_2", "b", 1), delta("evt_3", "c")] + expect(coalesceClientEvents(events).map((event) => event.id)).toEqual(["evt_1", "evt_2", "evt_3"]) + }) +}) diff --git a/packages/client/test/solid-data.test.ts b/packages/client/test/solid-data.test.ts new file mode 100644 index 00000000000..691e01216ed --- /dev/null +++ b/packages/client/test/solid-data.test.ts @@ -0,0 +1,86 @@ +import { expect, mock, test } from "bun:test" +import type { OpenCodeClient, OpenCodeEvent, Project, SessionInfo } from "../src/promise" +import { createServerData, type CreateServerDataInput } from "../src/solid/data" +import { createRoot } from "solid-js" + +const session = { + id: "ses_fork", + parentID: "ses_parent", + projectID: "pro_1", + title: "Fork", + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + time: { created: 1, updated: 1 }, + location: { directory: "/repo" }, +} as SessionInfo + +const project = { + id: "pro_1", + name: "Repo", + directory: "/repo", + canonical: "/repo", + vcs: "git", + sandboxes: [], + time: { created: 1, updated: 1 }, +} as Project + +test("refreshes forked sessions and worktree projects from events", async () => { + const listeners = new Set<(event: { name: OpenCodeEvent["type"]; details: OpenCodeEvent }) => void>() + const sessionGet = mock(async () => session) + const projectList = mock(async () => [project]) + const api = { + session: { get: sessionGet }, + project: { list: projectList }, + } as unknown as OpenCodeClient + const event = { + on: (() => () => {}) as CreateServerDataInput["event"]["on"], + listen(handler: (event: { name: OpenCodeEvent["type"]; details: OpenCodeEvent }) => void) { + listeners.add(handler) + return () => listeners.delete(handler) + }, + } + const emit = (details: OpenCodeEvent) => listeners.forEach((listener) => listener({ name: details.type, details })) + + await new Promise((resolve) => { + createRoot((dispose) => { + const data = createServerData({ api: () => api, directory: "/repo", event, connection: { status: () => "connected" } }) + emit({ type: "session.forked", data: { sessionID: session.id } } as unknown as OpenCodeEvent) + emit({ type: "worktree.updated", data: { projectID: project.id } } as unknown as OpenCodeEvent) + void Promise.all([sessionGet, projectList].map((fn) => fn.mock.results[0]?.value)).then(() => { + expect(data.session.get(session.id)).toEqual(session) + expect(data.project.get(project.id)).toEqual(project) + expect(sessionGet).toHaveBeenCalledTimes(1) + expect(projectList).toHaveBeenCalledTimes(1) + dispose() + resolve() + }) + }) + }) +}) + +test("resolves location info through the requested ref after canonicalization", async () => { + const requested = { directory: "/repo/../repo" } + const canonical = { + directory: "/repo", + project: { id: "pro_1", directory: "/repo", canonical: "/repo" }, + } + const api = { + location: { get: mock(async () => canonical) }, + } as unknown as OpenCodeClient + const event = { + on: (() => () => {}) as CreateServerDataInput["event"]["on"], + listen: () => () => {}, + } as CreateServerDataInput["event"] + + await new Promise((resolve) => { + createRoot((dispose) => { + const data = createServerData({ api: () => api, directory: requested.directory, event }) + void data.location.syncInfo(requested).then(() => { + expect(data.location.info(requested)).toEqual(canonical) + expect(data.location.info({ directory: canonical.directory })).toEqual(canonical) + dispose() + resolve() + }) + }) + }) +}) diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index 0aa48240f3f..bf49cdbe54a 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -1,177 +1,39 @@ import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client" +import { createClientConnection } from "@opencode-ai/client/solid" import { createGlobalEmitter } from "@solid-primitives/event-bus" -import { batch, onCleanup, onMount } from "solid-js" -import { createStore } from "solid-js/store" -import { errorMessage } from "../util/error" +import { onCleanup } from "solid-js" import { createSimpleContext } from "./helper" import { useLog } from "./log" -export type ClientConnectionStatus = "connected" | "connecting" | "reconnecting" -export type ClientConnectionEvent = { - readonly type: "client.connection" - readonly created: number - readonly data: { - readonly status: "connecting" | "connected" | "disconnected" | "reconnecting" - readonly attempt: number - readonly error?: string - } -} - type ManagedService = { reconnect: (signal: AbortSignal) => Promise<{ api: OpenCodeClient }> restart: () => Promise } type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract } -const connectTimeout = 2_000 -const connectionHistoryLimit = 50 -const eventFlushInterval = 10 export const { use: useClient, provider: ClientProvider } = createSimpleContext({ name: "Client", init: (props: { api: OpenCodeClient; service?: ManagedService }) => { const log = useLog({ component: "client" }) - const abort = new AbortController() - const history: ClientConnectionEvent[] = [] - let api = props.api + const service = props.service const events = createGlobalEmitter() - let pending: OpenCodeEvent[] = [] - let flushTimer: ReturnType | undefined - const [connection, setConnection] = createStore<{ - status: ClientConnectionStatus - attempt: number - error?: string - }>({ - status: "connecting", - attempt: 0, - }) - let stream: AbortController | undefined + let api = props.api - function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) { - history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } }) - if (history.length > connectionHistoryLimit) history.shift() - } - - function flushEvents() { - flushTimer = undefined - const queued = pending - pending = [] - batch(() => queued.forEach((event) => events.emit(event.type, event))) - } - - function emit(event: OpenCodeEvent) { - pending.push(event) - if (flushTimer) return - flushTimer = setTimeout(flushEvents, eventFlushInterval) - } - - async function connect(signal: AbortSignal, attempt: number) { - let connectedAt: number | undefined - - // Bound the initial handshake and tie this request to the stream lifetime. - const request = new AbortController() - const cancel = () => request.abort(signal.reason) - const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout) - signal.addEventListener("abort", cancel, { once: true }) - - try { - // Open the event stream and validate its initial handshake. - record(attempt === 0 ? "connecting" : "reconnecting", attempt) - log.info("event stream connecting", { attempt }) - - const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]() - const first = await iterator.next() - - if (signal.aborted) return { error: undefined, connectedAt } - if (first.done) { - const error = - request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected") - return { error, connectedAt } - } - if (first.value.type !== "server.connected") - return { error: new Error("Event stream did not start with server.connected"), connectedAt } - - // Publish the connected state before forwarding live events. - clearTimeout(timeout) - record("connected", attempt) - connectedAt = Date.now() - log.info("event stream connected") - emit(first.value) - setConnection({ status: "connected", attempt: 0, error: undefined }) - - // Forward events until the stream closes or this connection is cancelled. - while (!signal.aborted) { - const event = await iterator.next() - - if (signal.aborted) return { error: undefined, connectedAt } - if (event.done) return { error: new Error("Event stream disconnected"), connectedAt } - - if ("durable" in event.value) - log.debug("event", { - type: event.value.type, - aggregateID: event.value.durable.aggregateID, - seq: event.value.durable.seq, - }) - - emit(event.value) - } - - return { error: undefined, connectedAt } - } catch (error) { - return { error, connectedAt } - } finally { - request.abort() - clearTimeout(timeout) - signal.removeEventListener("abort", cancel) - } - } - - function start() { - stream?.abort() - const controller = new AbortController() - stream = controller - void (async () => { - let attempt = 0 - while (!abort.signal.aborted && !controller.signal.aborted) { - const result = await connect(controller.signal, attempt) - if (abort.signal.aborted || controller.signal.aborted) return - if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= 1_000) attempt = 0 - attempt += 1 - const message = errorMessage(result.error) - record("disconnected", attempt, message) - log.info("event stream disconnected", { - attempt, - error: message, - }) - setConnection({ status: "reconnecting", attempt, error: message }) - // Re-resolve the transport before retrying: the server may have - // moved (service restarted on a new port) or need starting. Static - // transports (--server, standalone) resolve to the same address. - if (props.service) { - const next = await props.service.reconnect(controller.signal).catch((error) => { - if (!controller.signal.aborted) - log.info("server resolution failed", { - attempt, - error: errorMessage(error), - }) - }) - if (abort.signal.aborted || controller.signal.aborted) return - if (next) { - api = next.api - if (attempt === 1) continue - } + const connection = createClientConnection(api, { + reconnect: service + ? async (signal) => { + api = (await service.reconnect(signal)).api + return api } - await wait(1_000, controller.signal) - } - })() - } + : undefined, + onEvent(event) { + events.emit(event.type, event) + }, + log, + }) - onMount(start) onCleanup(() => { - abort.abort() - stream?.abort() - if (flushTimer) clearTimeout(flushTimer) - pending = [] events.clear() }) @@ -183,35 +45,8 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( on: events.on, listen: events.listen, }, - connection: { - status() { - return connection.status - }, - attempt() { - return connection.attempt - }, - error() { - return connection.error - }, - internal: { - history() { - return history.slice() - }, - }, - }, - restart: props.service?.restart, + connection, + restart: service?.restart, } }, }) - -function wait(delay: number, signal: AbortSignal) { - return new Promise((resolve) => { - const timer = setTimeout(done, delay) - signal.addEventListener("abort", done, { once: true }) - function done() { - clearTimeout(timer) - signal.removeEventListener("abort", done) - resolve() - } - }) -} diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index dd1d19332b9..0affc5b0b62 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -1,1486 +1,22 @@ -// Client data layer: apply server events and cache API reads into a Solid store. -// Prefer straightforward projection. Do not add generation counters, stale-response -// merges, live/history overlays, or other race machinery here—last write wins. -// Reconnect invalidates cached reads; active UI owners decide what to sync again. - -import type { - AgentInfo, - CommandInfo, - FormInfo, - IntegrationInfo, - LocationRef, - LocationGetOutput, - McpResource, - McpServer, - ModelInfo, - PermissionSavedInfo, - PermissionRequest, - PermissionReplyInput, - Project, - ProviderInfo, - ReferenceInfo, - SessionMessageInfo, - SessionMessageAssistant, - SessionMessageAssistantReasoning, - SessionMessageAssistantText, - SessionMessageAssistantTool, - SessionInfo, - SessionInboxInfo, - ShellInfo, - SkillInfo, - VcsInfo, - OpenCodeEvent, - WebSearchProvider, -} from "@opencode-ai/client" -import { isPermissionNotFoundError } from "@opencode-ai/client" +import { createServerData } from "@opencode-ai/client/solid" import type { Plugin } from "@opencode-ai/plugin/tui" -import { createStore, produce, reconcile } from "solid-js/store" import { createSimpleContext } from "./helper" import { useClient } from "./client" -import { nonEmptyToolContent } from "../util/tool-display" -import type { SessionInbox } from "@opencode-ai/schema/session-inbox" -import { Worktree } from "@opencode-ai/schema/worktree" -import { createEffect, createSignal, onCleanup } from "solid-js" -export type DataSessionStatus = "idle" | "running" - -const messageIDFromEvent = (eventID: string) => eventID.replace(/^evt_/, "msg_") - -// Global MCP elicitations temporarily use "global" instead of a real session ID, so the -// server cannot recover their Location when settling them. Preserve the event Location -// until MCP elicitations carry session ownership. -export type FormWithLocation = FormInfo & { readonly location?: LocationRef } -type ShellWithLocation = ShellInfo & { readonly location: LocationRef } - -type LocationData = { - info?: LocationGetOutput - vcs?: VcsInfo - agent?: AgentInfo[] - command?: CommandInfo[] - integration?: IntegrationInfo[] - mcp?: { - server?: McpServer[] - resource?: McpResource[] - } - model?: ModelInfo[] - provider?: ProviderInfo[] - reference?: ReferenceInfo[] - websearch?: WebSearchProvider[] - // Currently running shell commands for this location, keyed by shell id. Entries are removed - // once the command exits or is deleted, so this only ever holds in-flight shells. - shell?: Record - skill?: SkillInfo[] -} - -type Store = { - session: { - info: Record - // Family index keyed by a family's root (or furthest-known-ancestor when the - // true root is not yet loaded). The value is a flat deduplicated list of every - // session ID in that family, including the key itself once its info arrives. - family: Record - active: Record - message: Record - pending: Record - input: Record - permission: Record - // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. - form: Record - } - project: { - info: Record - permission: Record - } - location: Record -} - -export function locationKey(location: LocationRef) { - return JSON.stringify([location.directory, location.workspaceID]) -} - -function locationQuery(ref?: LocationRef) { - return ref ? { directory: ref.directory, workspace: ref.workspaceID } : undefined -} - -function createSync() { - const state = new Map>() - return { - run(key: string, load: () => Promise) { - const active = state.get(key) - if (active === true) return Promise.resolve() - if (active) return active - const pending = load() - .then(() => { - if (state.get(key) === pending) state.set(key, true) - }) - .finally(() => { - if (state.get(key) === pending) state.delete(key) - }) - state.set(key, pending) - return pending - }, - complete(key: string) { - if (state.has(key)) return - state.set(key, true) - }, - invalidate(key?: string) { - if (key) { - state.delete(key) - return - } - state.clear() - }, - } -} +export { locationKey } from "@opencode-ai/client/solid" +export type { FormWithLocation } from "@opencode-ai/client/solid" export const { use: useData, provider: DataProvider } = createSimpleContext({ name: "Data", init: () => { - const [store, setStore] = createStore({ - session: { - info: {}, - family: {}, - active: {}, - message: {}, - pending: {}, - input: {}, - permission: {}, - form: {}, - }, - project: { - info: {}, - permission: {}, - }, - location: {}, - }) - const client = useClient() - const [defaultLocation, setDefaultLocation] = createSignal({ + const data = createServerData({ + api: () => client.api, + event: client.event, + connection: client.connection, directory: process.cwd(), }) - const messageIndex = new Map>() - const sync = createSync() - - function setSessionActive(sessionID: string, status: DataSessionStatus) { - setStore("session", "active", sessionID, status) - } - - function addPending(item: SessionInboxInfo) { - if (store.session.pending[item.sessionID]?.some((pending) => pending.id === item.id)) return - setStore("session", "pending", item.sessionID, [...(store.session.pending[item.sessionID] ?? []), item]) - } - - function removePending(sessionID: string, inboxID?: string) { - if (!inboxID) return - if (store.session.pending[sessionID]?.some((item) => item.id === inboxID)) - setStore( - "session", - "pending", - sessionID, - (store.session.pending[sessionID] ?? []).filter((item) => item.id !== inboxID), - ) - if (store.session.input[sessionID]?.includes(inboxID)) - setStore( - "session", - "input", - sessionID, - (store.session.input[sessionID] ?? []).filter((id) => id !== inboxID), - ) - } - - function removePermission(sessionID: string, requestID: string) { - const requests = store.session.permission[sessionID] - if (!requests?.some((request) => request.id === requestID)) return - setStore( - "session", - "permission", - sessionID, - requests.filter((request) => request.id !== requestID), - ) - } - - function updatePending(sessionID: string, inboxID: string, delivery: SessionInbox.Delivery) { - const index = store.session.pending[sessionID]?.findIndex((item) => item.id === inboxID) ?? -1 - const item = store.session.pending[sessionID]?.[index] - if (index < 0 || !item || item.delivery === delivery) return - setStore("session", "pending", sessionID, index, { ...item, delivery }) - } - - const message = { - update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { - setStore( - "session", - "message", - produce((draft) => { - fn((draft[sessionID] ??= []), index(sessionID)) - }), - ) - }, - append(messages: SessionMessageInfo[], index: Map, item: SessionMessageInfo) { - if (index.has(item.id)) return - index.set(item.id, messages.length) - messages.push(item) - }, - activeAssistant(messages: SessionMessageInfo[]) { - const item = messages.findLast((item) => item.type === "assistant" && !item.time.completed) - return item?.type === "assistant" ? item : undefined - }, - assistant(messages: SessionMessageInfo[], index: Map, messageID: string) { - const position = index.get(messageID) - const item = position === undefined ? undefined : messages[position] - return item?.type === "assistant" ? item : undefined - }, - shell(messages: SessionMessageInfo[], shellID: string) { - const item = messages.findLast((item) => item.type === "shell" && item.shellID === shellID) - return item?.type === "shell" ? item : undefined - }, - compaction(messages: SessionMessageInfo[]) { - const item = messages.findLast((item) => item.type === "compaction" && item.status === "running") - return item?.type === "compaction" ? item : undefined - }, - latestTool(assistant: SessionMessageAssistant | undefined, id?: string) { - return assistant?.content.findLast( - (item): item is SessionMessageAssistantTool => item.type === "tool" && (id === undefined || item.id === id), - ) - }, - latestText(assistant: SessionMessageAssistant | undefined) { - return assistant?.content.findLast((item): item is SessionMessageAssistantText => item.type === "text") - }, - latestReasoning(assistant: SessionMessageAssistant | undefined) { - return assistant?.content.findLast( - (item): item is SessionMessageAssistantReasoning => item.type === "reasoning" && !item.time?.completed, - ) - }, - reindex(messages: SessionMessageInfo[], index: Map, start: number) { - for (let position = start; position < messages.length; position++) { - const item = messages[position] - if (item) index.set(item.id, position) - } - }, - } - - function index(sessionID: string) { - const existing = messageIndex.get(sessionID) - if (existing) return existing - const created = new Map() - messageIndex.set(sessionID, created) - return created - } - - // Walk parentID upward through loaded session info to the family root. When a - // parent's info is missing, that missing ID is the furthest-known ancestor and - // is returned so orphan subtrees group under it until the parent arrives. A - // seen set guards against parent cycles, stopping at the last non-repeating - // ancestor. - function resolveRoot(sessionID: string) { - let current = sessionID - let parentID = store.session.info[sessionID]?.parentID - const seen = new Set([sessionID]) - while (parentID) { - if (seen.has(parentID)) break - seen.add(parentID) - current = parentID - parentID = store.session.info[parentID]?.parentID - } - return current - } - - // Register one session into the family index. Idempotent: syncing an - // existing session never duplicates its ID. When a tentative family keyed by - // sessionID exists (descendants arrived while sessionID's own info was - // absent) but sessionID turns out to have a parent, fold the orphan subtree - // into the resolved root's family and drop the tentative entry. - function registerSession(sessionID: string) { - const info = store.session.info[sessionID] - if (!info) return - const rootID = resolveRoot(sessionID) - setStore( - "session", - "family", - produce((draft) => { - if (sessionID !== rootID && draft[sessionID]) { - const members = (draft[rootID] ??= []) - for (const id of draft[sessionID]) { - if (!members.includes(id)) members.push(id) - } - delete draft[sessionID] - } - const family = (draft[rootID] ??= []) - if (!family.includes(sessionID)) family.push(sessionID) - }), - ) - } - - function removeSession(sessionID: string) { - messageIndex.delete(sessionID) - sync.invalidate(`session:${sessionID}`) - sync.invalidate(`session.pending:${sessionID}`) - sync.invalidate(`session.message:${sessionID}`) - sync.invalidate(`session.permission:${sessionID}`) - sync.invalidate(`session.form:${sessionID}:`) - setStore( - "session", - produce((draft) => { - delete draft.info[sessionID] - delete draft.active[sessionID] - delete draft.message[sessionID] - delete draft.pending[sessionID] - delete draft.input[sessionID] - delete draft.permission[sessionID] - delete draft.form[sessionID] - for (const [rootID, family] of Object.entries(draft.family)) { - const next = family.filter((id) => id !== sessionID) - if (next.length === 0) delete draft.family[rootID] - else draft.family[rootID] = next - } - }), - ) - } - - function handleEvent(event: OpenCodeEvent) { - switch (event.type) { - case "session.created": - result.session.invalidate(event.data.sessionID) - void result.session.sync(event.data.sessionID) - // Band-aid: a newly created session starts empty, so live events can be its source of truth. - // Fetching pending inputs and projected messages separately lets promotion move an input between snapshots, - // causing both requests to miss it and overwrite event-built state. Skip those racy initial reads until - // hydration can load pending and projected messages atomically. - sync.complete(`session.pending:${event.data.sessionID}`) - sync.complete(`session.message:${event.data.sessionID}`) - break - case "session.deleted": - removeSession(event.data.sessionID) - break - case "session.usage.updated": - if (store.session.info[event.data.sessionID]) - setStore("session", "info", event.data.sessionID, { - cost: event.data.cost, - tokens: event.data.tokens, - }) - break - case "catalog.updated": - result.location.model.invalidate(event.location) - result.location.provider.invalidate(event.location) - void Promise.all([result.location.model.sync(event.location), result.location.provider.sync(event.location)]) - break - case "agent.updated": - result.location.agent.invalidate(event.location) - void result.location.agent.sync(event.location) - break - case "command.updated": - result.location.command.invalidate(event.location) - void result.location.command.sync(event.location) - break - case "skill.updated": - result.location.skill.invalidate(event.location) - void result.location.skill.sync(event.location) - break - case "vcs.branch.updated": - setStore("location", locationKey(event.location ?? defaultLocation()), (data) => ({ - ...data, - vcs: { - branch: { - ...data?.vcs?.branch, - current: event.data.branch, - }, - }, - })) - break - case "session.agent.selected": { - const previous = store.session.info[event.data.sessionID]?.agent - if (store.session.info[event.data.sessionID]) - setStore("session", "info", event.data.sessionID, "agent", event.data.agent) - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "agent-switched", - agent: event.data.agent, - previous, - time: { created: event.created }, - }) - }) - break - } - case "session.model.selected": - if (store.session.info[event.data.sessionID]) - setStore("session", "info", event.data.sessionID, "model", event.data.model) - if (!store.session.message[event.data.sessionID]) break - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "model-switched", - model: event.data.model, - time: { created: event.created }, - }) - }) - void client.api.session - .message({ sessionID: event.data.sessionID, messageID: messageIDFromEvent(event.id) }) - .then((item) => { - message.update(event.data.sessionID, (draft, index) => { - const position = index.get(item.id) - if (position === undefined) return message.append(draft, index, item) - draft[position] = item - }) - }) - .catch((error) => console.error("Failed to load projected model switch message", error)) - break - case "session.renamed": - // Preserve the live title when it races the session's initial read. - void result.session.sync(event.data.sessionID).then(() => { - if (store.session.info[event.data.sessionID]) - setStore("session", "info", event.data.sessionID, "title", event.data.title) - }) - break - case "session.moved": { - const current = store.session.info[event.data.sessionID] - if (current) { - const previous = { - location: { ...current.location }, - projectID: current.projectID, - subpath: current.subpath, - } - setStore("session", "info", event.data.sessionID, "location", event.data.location) - if (event.data.projectID) - setStore("session", "info", event.data.sessionID, "projectID", event.data.projectID) - setStore("session", "info", event.data.sessionID, "subpath", event.data.subpath) - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "location-switched", - location: event.data.location, - projectID: event.data.projectID, - subpath: event.data.subpath, - previous, - time: { created: event.created }, - }) - }) - } - break - } - case "worktree.resolved": { - for (const [sessionID, info] of Object.entries(store.session.info)) { - const adopted = Worktree.adopt( - { projectID: info.projectID, directory: info.location.directory }, - event.data, - ) - if (!adopted) continue - setStore("session", "info", sessionID, "projectID", adopted.projectID) - setStore("session", "info", sessionID, "subpath", adopted.subpath) - } - break - } - case "session.inbox.delivered": { - const admitted = store.session.input[event.data.sessionID]?.includes(event.data.inboxID) ?? false - removePending(event.data.sessionID, event.data.inboxID) - message.update(event.data.sessionID, (draft, index) => { - const position = index.get(event.data.inboxID) - if (position === undefined) return - const existing = draft[position] - if (!existing || !admitted) return - existing.time.created = event.created - draft.splice(position, 1) - draft.push(existing) - message.reindex(draft, index, position) - }) - break - } - case "session.inbox.delivery.changed": - updatePending(event.data.sessionID, event.data.inboxID, event.data.delivery) - break - case "session.inbox.cancelled": { - removePending(event.data.sessionID, event.data.inboxID) - if (messageIndex.get(event.data.sessionID)?.has(event.data.inboxID)) - message.update(event.data.sessionID, (draft, index) => { - const position = index.get(event.data.inboxID) - if (position === undefined) return - draft.splice(position, 1) - index.delete(event.data.inboxID) - message.reindex(draft, index, position) - }) - break - } - case "session.inbox.enqueued": { - const item = event.data.item - addPending({ - id: event.data.inboxID, - sessionID: event.data.sessionID, - timeCreated: event.created, - ...item, - }) - if (!store.session.input[event.data.sessionID]?.includes(event.data.inboxID)) - setStore("session", "input", event.data.sessionID, [ - ...(store.session.input[event.data.sessionID] ?? []), - event.data.inboxID, - ]) - if (item.type !== "user" && item.type !== "synthetic") break - message.update(event.data.sessionID, (draft, index) => { - message.append( - draft, - index, - item.type === "user" - ? { - id: event.data.inboxID, - type: "user", - ...item.payload, - time: { created: event.created }, - } - : { - id: event.data.inboxID, - type: "synthetic", - ...item.payload, - time: { created: event.created }, - }, - ) - }) - break - } - case "session.instructions.updated": - // Mirror the projector: the initial baseline and empty-rendering deltas carry no text - // and produce no transcript message. - const updateText = event.data.text - if (updateText === undefined) break - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "system", - text: updateText, - description: `Instructions updated: ${Object.keys(event.data.delta).join(", ")}`, - metadata: event.metadata, - time: { created: event.created }, - }) - }) - break - case "session.synthetic": - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "synthetic", - text: event.data.text, - description: event.data.description, - metadata: event.data.metadata, - time: { created: event.created }, - }) - }) - break - case "session.shell.started": - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "shell", - shellID: event.data.shell.id, - command: event.data.shell.command, - status: event.data.shell.status, - exit: event.data.shell.exit, - metadata: event.metadata, - time: { created: event.created }, - }) - }) - break - case "session.shell.ended": - message.update(event.data.sessionID, (draft) => { - const match = message.shell(draft, event.data.shell.id) - if (!match) return - match.status = event.data.shell.status - match.exit = event.data.shell.exit - match.output = event.data.output - match.time.completed = event.created - }) - break - case "session.step.started": - message.update(event.data.sessionID, (draft, index) => { - const position = index.get(event.data.assistantMessageID) - const existing = position === undefined ? undefined : draft[position] - if (existing?.type === "assistant") { - existing.agent = event.data.agent - existing.model = event.data.model - existing.retry = undefined - existing.error = undefined - existing.finish = undefined - existing.time.completed = undefined - if (event.data.snapshot) existing.snapshot = { ...existing.snapshot, start: event.data.snapshot } - return - } - const currentAssistant = message.activeAssistant(draft) - if (currentAssistant) { - currentAssistant.retry = undefined - currentAssistant.time.completed = event.created - } - message.append(draft, index, { - id: event.data.assistantMessageID, - type: "assistant", - agent: event.data.agent, - model: event.data.model, - metadata: event.metadata, - content: [], - snapshot: event.data.snapshot ? { start: event.data.snapshot } : undefined, - time: { created: event.created }, - }) - }) - break - case "session.step.ended": { - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (!currentAssistant) return - currentAssistant.time.completed = event.created - currentAssistant.finish = event.data.finish - currentAssistant.cost = event.data.cost - currentAssistant.tokens = event.data.tokens - if (event.data.snapshot) - currentAssistant.snapshot = { ...currentAssistant.snapshot, end: event.data.snapshot } - }) - break - } - case "session.step.failed": - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (!currentAssistant) return - currentAssistant.time.completed = event.created - currentAssistant.finish = "error" - currentAssistant.error = event.data.error - currentAssistant.retry = undefined - if (event.data.cost !== undefined && event.data.tokens !== undefined) { - currentAssistant.cost = event.data.cost - currentAssistant.tokens = event.data.tokens - } - }) - break - case "session.text.started": - message.update(event.data.sessionID, (draft, index) => { - message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ - type: "text", - text: "", - }) - }) - break - case "session.text.delta": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestText(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) match.text += event.data.delta - }) - break - case "session.text.ended": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestText(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) match.text = event.data.text - }) - break - case "session.tool.input.started": - message.update(event.data.sessionID, (draft, index) => { - message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ - type: "tool", - id: event.data.id, - name: event.data.name, - time: { created: event.created }, - state: { status: "streaming", input: "" }, - }) - }) - break - case "session.tool.input.delta": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status === "streaming") match.state.input += event.data.delta - }) - break - case "session.tool.input.ended": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status === "streaming") match.state.input = event.data.text - }) - break - case "session.tool.called": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (!match) return - match.time.ran = event.created - match.executed = event.data.executed - match.providerState = event.data.state - match.state = { status: "running", input: event.data.input, metadata: {} } - }) - break - case "session.tool.progress": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status !== "running") return - match.state.metadata = event.data.metadata - }) - break - case "session.tool.success": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (match?.state.status !== "running") return - match.state = { - status: "completed", - input: match.state.input, - metadata: event.data.metadata, - content: [...event.data.content], - } - match.executed = event.data.executed || match.executed === true - match.providerResultState = event.data.resultState - match.time.completed = event.created - }) - break - case "session.tool.failed": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestTool( - message.assistant(draft, index, event.data.assistantMessageID), - event.data.id, - ) - if (!match || (match.state.status !== "streaming" && match.state.status !== "running")) return - match.state = { - status: "error", - error: event.data.error, - input: typeof match.state.input === "string" ? {} : match.state.input, - metadata: event.data.metadata, - content: event.data.content, - } - match.executed = event.data.executed || match.executed === true - match.providerResultState = event.data.resultState - match.time.completed = event.created - }) - break - case "session.reasoning.started": - message.update(event.data.sessionID, (draft, index) => { - message.assistant(draft, index, event.data.assistantMessageID)?.content.push({ - type: "reasoning", - text: "", - state: event.data.state, - time: { created: event.created }, - }) - }) - break - case "session.reasoning.delta": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestReasoning(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) match.text += event.data.delta - }) - break - case "session.reasoning.ended": - message.update(event.data.sessionID, (draft, index) => { - const match = message.latestReasoning(message.assistant(draft, index, event.data.assistantMessageID)) - if (match) { - match.text = event.data.text - match.time = { created: match.time?.created ?? event.created, completed: event.created } - if (event.data.state !== undefined) match.state = event.data.state - } - }) - break - case "session.retry.scheduled": - message.update(event.data.sessionID, (draft, index) => { - const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) - if (!currentAssistant) return - currentAssistant.retry = { - attempt: event.data.attempt, - at: event.data.at, - error: event.data.error, - } - }) - break - case "session.execution.started": - setSessionActive(event.data.sessionID, "running") - break - case "session.compaction.started": - if (event.data.inputID) removePending(event.data.sessionID, event.data.inputID) - message.update(event.data.sessionID, (draft, index) => { - message.append(draft, index, { - id: event.data.inputID ?? messageIDFromEvent(event.id), - type: "compaction", - status: "running", - reason: event.data.reason, - summary: "", - recent: event.data.recent ?? "", - time: { created: event.created }, - }) - }) - break - case "session.execution.succeeded": - case "session.execution.failed": - case "session.execution.interrupted": - setSessionActive(event.data.sessionID, "idle") - message.update(event.data.sessionID, (draft) => { - const currentAssistant = message.activeAssistant(draft) - if (currentAssistant) currentAssistant.retry = undefined - }) - break - case "session.revert.staged": - if (store.session.info[event.data.sessionID]) - setStore("session", "info", event.data.sessionID, "revert", event.data.revert) - break - case "session.revert.cleared": - if (store.session.info[event.data.sessionID]) - setStore("session", "info", event.data.sessionID, "revert", undefined) - break - case "session.revert.committed": - if (store.session.info[event.data.sessionID]) { - setStore("session", "info", event.data.sessionID, "revert", undefined) - } - setStore( - "session", - "input", - event.data.sessionID, - (store.session.input[event.data.sessionID] ?? []).filter((id) => id < event.data.to), - ) - message.update(event.data.sessionID, (draft, index) => { - const position = draft.findIndex((item) => item.id >= event.data.to) - if (position === -1) return - for (const item of draft.splice(position)) index.delete(item.id) - }) - break - case "session.compaction.delta": - message.update(event.data.sessionID, (draft) => { - const current = message.compaction(draft) - if (current?.status === "running") current.summary += event.data.text - }) - break - case "session.compaction.ended": - message.update(event.data.sessionID, (draft, index) => { - const position = draft.findLastIndex((item) => item.type === "compaction" && item.status === "running") - const current = draft[position] - if (current?.type === "compaction") { - Object.assign(current, { - status: "completed", - reason: event.data.reason, - summary: event.data.text, - recent: event.data.recent, - }) - return - } - message.append(draft, index, { - id: messageIDFromEvent(event.id), - type: "compaction", - status: "completed", - reason: event.data.reason, - summary: event.data.text, - recent: event.data.recent, - time: { created: event.created }, - }) - }) - break - case "session.compaction.failed": - if (event.data.inputID) removePending(event.data.sessionID, event.data.inputID) - message.update(event.data.sessionID, (draft, index) => { - const position = draft.findLastIndex((item) => item.type === "compaction" && item.status === "running") - const current = draft[position] - const failed: Extract = { - id: current?.id ?? event.data.inputID ?? messageIDFromEvent(event.id), - type: "compaction", - status: "failed", - reason: event.data.reason ?? "manual", - error: event.data.error ?? { - type: "compaction.failed", - message: "Compaction failed before recording an error", - }, - metadata: current?.type === "compaction" ? current.metadata : event.metadata, - time: current?.type === "compaction" ? current.time : { created: event.created }, - } - if (current?.type === "compaction") { - draft[position] = failed - return - } - message.append(draft, index, failed) - }) - break - case "permission.asked": - if (store.session.permission[event.data.sessionID]?.some((request) => request.id === event.data.id)) break - setStore("session", "permission", event.data.sessionID, [ - ...(store.session.permission[event.data.sessionID] ?? []), - event.data, - ]) - break - case "permission.replied": - removePermission(event.data.sessionID, event.data.requestID) - break - case "form.created": - if (store.session.form[event.data.form.sessionID]?.some((form) => form.id === event.data.form.id)) break - setStore("session", "form", event.data.form.sessionID, [ - ...(store.session.form[event.data.form.sessionID] ?? []), - event.data.form.sessionID === "global" ? { ...event.data.form, location: event.location } : event.data.form, - ]) - break - case "form.replied": - case "form.cancelled": - setStore( - "session", - "form", - event.data.sessionID, - (store.session.form[event.data.sessionID] ?? []).filter((form) => form.id !== event.data.id), - ) - break - case "shell.created": - setStore("location", locationKey(event.location ?? defaultLocation()), (data) => ({ - ...data, - shell: { - ...data?.shell, - [event.data.info.id]: { ...event.data.info, location: event.location ?? defaultLocation() }, - }, - })) - break - case "shell.exited": - case "shell.deleted": - if (event.location) { - setStore("location", locationKey(event.location), (data) => ({ - ...data, - shell: Object.fromEntries(Object.entries(data?.shell ?? {}).filter(([id]) => id !== event.data.id)), - })) - break - } - setStore( - "location", - produce((draft) => { - for (const data of Object.values(draft)) delete data.shell?.[event.data.id] - }), - ) - break - case "reference.updated": - result.location.reference.invalidate() - void result.location.reference.sync() - break - case "integration.updated": - result.location.integration.invalidate(event.location) - result.location.model.invalidate(event.location) - result.location.provider.invalidate(event.location) - void Promise.all([ - result.location.integration.sync(event.location), - result.location.model.sync(event.location), - result.location.provider.sync(event.location), - ]) - break - case "config.updated": - case "websearch.updated": - void result.location.websearch.refresh(event.location) - break - // Authenticating an MCP integration reconnects its server, which emits mcp.status.changed, - // so the mcp list syncs here rather than off integration.updated. - case "mcp.status.changed": - result.location.mcp.server.invalidate(event.location) - void result.location.mcp.server.sync(event.location) - break - case "mcp.resources.changed": - result.location.mcp.resource.invalidate(event.location) - void result.location.mcp.resource.sync(event.location) - break - } - } - - const result = { - on: client.event.on, - listen: client.event.listen, - session: { - list() { - return Object.values(store.session.info).toSorted((a, b) => b.time.updated - a.time.updated) - }, - get(sessionID: string) { - return store.session.info[sessionID] - }, - root(sessionID: string) { - return resolveRoot(sessionID) - }, - family(sessionID: string) { - return store.session.family[resolveRoot(sessionID)] ?? [] - }, - cost(sessionID: string) { - const session = store.session.info[sessionID] - if (!session) return 0 - if (session.parentID) return session.cost - return (store.session.family[sessionID] ?? [sessionID]).reduce( - (total, id) => total + (store.session.info[id]?.cost ?? 0), - 0, - ) - }, - status(sessionID: string) { - return store.session.active[sessionID] ?? "idle" - }, - input: { - list(sessionID: string) { - return store.session.input[sessionID] ?? [] - }, - has(sessionID: string, inboxID: string) { - return store.session.input[sessionID]?.includes(inboxID) ?? false - }, - }, - pending: { - list(sessionID: string) { - return store.session.pending[sessionID] ?? [] - }, - sync(sessionID: string) { - return sync.run(`session.pending:${sessionID}`, async () => { - const pending = await client.api.session.inbox.list({ sessionID }) - setStore("session", "pending", sessionID, reconcile(pending)) - setStore( - "session", - "input", - sessionID, - reconcile(pending.filter((item) => item.type !== "compaction").map((item) => item.id)), - ) - }) - }, - invalidate(sessionID: string) { - sync.invalidate(`session.pending:${sessionID}`) - }, - }, - sync(sessionID: string, options?: { children?: boolean }) { - return sync.run(options?.children ? `session.family:${sessionID}` : `session:${sessionID}`, async () => { - const [info, children] = await Promise.all([ - client.api.session.get({ sessionID }), - options?.children - ? client.api.session.list({ parentID: sessionID, order: "desc" }).then((response) => response.data) - : [], - ]) - const sessions = [info, ...children] - setStore( - "session", - "info", - produce((draft) => { - for (const session of sessions) draft[session.id] = session - }), - ) - for (const session of sessions) { - sync.complete(`session:${session.id}`) - registerSession(session.id) - } - }) - }, - invalidate(sessionID: string) { - sync.invalidate(`session:${sessionID}`) - }, - message: { - list(sessionID: string) { - return store.session.message[sessionID] ?? [] - }, - get(sessionID: string, messageID: string) { - const messages = store.session.message[sessionID] - const position = messageIndex.get(sessionID)?.get(messageID) - return position === undefined ? undefined : messages?.[position] - }, - sync(sessionID: string) { - return sync.run(`session.message:${sessionID}`, async () => { - const messages = ( - await client.api.message.list({ sessionID, limit: 200, order: "desc" }) - ).data.toReversed() - messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) - setStore("session", "message", sessionID, reconcile(messages)) - }) - }, - invalidate(sessionID: string) { - sync.invalidate(`session.message:${sessionID}`) - }, - }, - permission: { - list(sessionID: string) { - return store.session.permission[sessionID] - }, - sync(sessionID: string) { - return sync.run(`session.permission:${sessionID}`, async () => { - setStore("session", "permission", sessionID, await client.api.permission.list({ sessionID })) - }) - }, - invalidate(sessionID: string) { - sync.invalidate(`session.permission:${sessionID}`) - }, - async reply(input: PermissionReplyInput) { - await client.api.permission.reply(input).catch((error: unknown) => { - if (!isPermissionNotFoundError(error)) throw error - }) - removePermission(input.sessionID, input.requestID) - }, - }, - form: { - list(sessionID: string, ref?: LocationRef) { - const forms = store.session.form[sessionID] - if (sessionID !== "global") return forms - if (!ref) return - const key = locationKey(ref) - return forms?.filter((form) => form.location && locationKey(form.location) === key) - }, - sync(sessionID: string, ref?: LocationRef) { - const key = `session.form:${sessionID}:${sessionID === "global" ? locationKey(ref ?? defaultLocation()) : ""}` - return sync.run(key, async () => { - if (sessionID === "global") { - const response = await client.api.form.request.list({ - location: locationQuery(ref ?? defaultLocation()), - }) - const location = { - directory: response.location.directory, - workspaceID: response.location.workspaceID, - } - const locationID = locationKey(location) - setStore("session", "form", sessionID, [ - ...(store.session.form[sessionID] ?? []).filter( - (form) => form.location && locationKey(form.location) !== locationID, - ), - ...response.data.filter((form) => form.sessionID === "global").map((form) => ({ ...form, location })), - ]) - return - } - setStore("session", "form", sessionID, await client.api.form.list({ sessionID })) - }) - }, - invalidate(sessionID: string, ref?: LocationRef) { - sync.invalidate( - `session.form:${sessionID}:${sessionID === "global" ? locationKey(ref ?? defaultLocation()) : ""}`, - ) - }, - }, - }, - project: { - list() { - return Object.values(store.project.info).toSorted((a, b) => b.time.updated - a.time.updated) - }, - get(projectID: string) { - return store.project.info[projectID] - }, - sync() { - return sync.run("project", async () => { - const projects = await client.api.project.list() - setStore("project", "info", reconcile(Object.fromEntries(projects.map((project) => [project.id, project])))) - }) - }, - invalidate() { - sync.invalidate("project") - }, - permission: { - list(projectID: string) { - return store.project.permission[projectID] - }, - sync(projectID: string) { - return sync.run(`project.permission:${projectID}`, async () => { - setStore("project", "permission", projectID, await client.api.permission.saved.list({ projectID })) - }) - }, - invalidate(projectID: string) { - sync.invalidate(`project.permission:${projectID}`) - }, - }, - }, - shell: { - list(location?: LocationRef) { - return Object.values(store.location[locationKey(location ?? defaultLocation())]?.shell ?? {}) - }, - listBySession(sessionID: string) { - return Object.values(store.location) - .flatMap((data) => Object.values(data.shell ?? {})) - .filter((shell) => shell.metadata.sessionID === sessionID) - }, - get(id: string) { - return Object.values(store.location) - .map((data) => data.shell?.[id]) - .find((shell) => shell !== undefined) - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.shell:${id}`, async () => { - const response = await client.api.shell.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { - ...store.location[key], - shell: Object.fromEntries( - response.data.map((info) => [ - info.id, - { - ...info, - location: { - directory: response.location.directory, - workspaceID: response.location.workspaceID, - }, - }, - ]), - ), - }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.shell:${locationKey(ref ?? defaultLocation())}`) - }, - }, - location: { - info(ref?: LocationRef) { - return store.location[locationKey(ref ?? defaultLocation())]?.info - }, - default() { - return defaultLocation() - }, - syncInfo(ref?: LocationRef) { - const current = ref ?? defaultLocation() - return sync.run(`location:${locationKey(current)}`, async () => { - const location = await client.api.location.get({ location: locationQuery(current) }) - const key = locationKey(location) - if (!store.location[key]) setStore("location", key, {}) - setStore("location", key, "info", location) - if (!ref) { - setDefaultLocation({ directory: location.directory, workspaceID: location.workspaceID }) - } - }) - }, - async sync(ref?: LocationRef) { - await result.location.syncInfo(ref) - const location = ref ?? defaultLocation() - await Promise.all([ - result.location.vcs.sync(location), - result.location.agent.sync(location), - result.location.command.sync(location), - result.location.integration.sync(location), - result.location.mcp.server.sync(location), - result.location.mcp.resource.sync(location), - result.location.model.sync(location), - result.location.provider.sync(location), - result.location.reference.sync(location), - result.location.skill.sync(location), - result.shell.sync(location), - result.session.form.sync("global", location), - ]) - }, - invalidate(ref?: LocationRef) { - const location = ref ?? defaultLocation() - sync.invalidate(`location:${locationKey(location)}`) - result.location.vcs.invalidate(location) - result.location.agent.invalidate(location) - result.location.command.invalidate(location) - result.location.integration.invalidate(location) - result.location.mcp.server.invalidate(location) - result.location.mcp.resource.invalidate(location) - result.location.model.invalidate(location) - result.location.provider.invalidate(location) - result.location.reference.invalidate(location) - result.location.skill.invalidate(location) - result.shell.invalidate(location) - result.session.form.invalidate("global", location) - }, - vcs: { - info(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.vcs - }, - sync(ref?: LocationRef) { - const location = ref ?? defaultLocation() - return sync.run(`location.vcs:${locationKey(location)}`, async () => { - const response = await client.api.vcs.get({ location: locationQuery(location) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], vcs: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.vcs:${locationKey(ref ?? defaultLocation())}`) - }, - }, - agent: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.agent - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.agent:${id}`, async () => { - const response = await client.api.agent.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], agent: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.agent:${locationKey(ref ?? defaultLocation())}`) - }, - }, - command: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.command - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.command:${id}`, async () => { - const response = await client.api.command.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], command: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.command:${locationKey(ref ?? defaultLocation())}`) - }, - }, - integration: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.integration - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.integration:${id}`, async () => { - const response = await client.api.integration.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], integration: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.integration:${locationKey(ref ?? defaultLocation())}`) - }, - }, - mcp: { - server: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.mcp?.server - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.mcp.server:${id}`, async () => { - const response = await client.api.mcp.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { - ...store.location[key], - mcp: { ...store.location[key]?.mcp, server: response.data }, - }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.mcp.server:${locationKey(ref ?? defaultLocation())}`) - }, - }, - resource: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.mcp?.resource - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.mcp.resource:${id}`, async () => { - const response = await client.api.mcp.resource.catalog({ - location: locationQuery(ref ?? defaultLocation()), - }) - const key = locationKey(response.location) - setStore("location", key, { - ...store.location[key], - mcp: { ...store.location[key]?.mcp, resource: response.data.resources }, - }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.mcp.resource:${locationKey(ref ?? defaultLocation())}`) - }, - }, - }, - model: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.model - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.model:${id}`, async () => { - const response = await client.api.model.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], model: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.model:${locationKey(ref ?? defaultLocation())}`) - }, - }, - provider: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.provider - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.provider:${id}`, async () => { - const response = await client.api.provider.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], provider: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.provider:${locationKey(ref ?? defaultLocation())}`) - }, - }, - reference: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.reference - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.reference:${id}`, async () => { - const response = await client.api.reference.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], reference: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.reference:${locationKey(ref ?? defaultLocation())}`) - }, - }, - websearch: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.websearch - }, - async refresh(ref?: LocationRef) { - const input = { location: locationQuery(ref ?? defaultLocation()) } - const providers = await client.api.websearch.providers(input) - const key = locationKey(providers.location) - setStore("location", key, { - ...store.location[key], - websearch: providers.data, - }) - }, - }, - skill: { - list(location?: LocationRef) { - return store.location[locationKey(location ?? defaultLocation())]?.skill - }, - sync(ref?: LocationRef) { - const id = locationKey(ref ?? defaultLocation()) - return sync.run(`location.skill:${id}`, async () => { - const response = await client.api.skill.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(response.location) - setStore("location", key, { ...store.location[key], skill: response.data }) - }) - }, - invalidate(ref?: LocationRef) { - sync.invalidate(`location.skill:${locationKey(ref ?? defaultLocation())}`) - }, - }, - }, - } - result satisfies Plugin.Context["data"] - - createEffect(() => { - if (client.connection.status() === "connected") return - sync.invalidate() - }) - - onCleanup( - client.event.listen(({ details }) => { - if (details.type === "server.connected") { - void client.api.session - .active() - .then((active) => { - setStore( - "session", - "active", - reconcile(Object.fromEntries(Object.keys(active).map((sessionID) => [sessionID, "running" as const]))), - ) - }) - .catch(() => undefined) - void client.api.location - .get({ location: locationQuery(defaultLocation()) }) - .then((location) => { - const key = locationKey(location) - setStore("location", key, { ...store.location[key], info: location }) - }) - .catch((error) => console.error("Failed to preload location", error)) - void result.location.vcs.sync().catch((error) => console.error("Failed to preload VCS info", error)) - void result.project.sync().catch((error) => console.error("Failed to preload projects", error)) - return - } - handleEvent(details) - }), - ) - - return result + data satisfies Plugin.Context["data"] + return data }, }) diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 4f7a3b85211..db531173b9c 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -1873,15 +1873,14 @@ test("refreshes effective catalog data after catalog updates", async () => { test("refreshes agents after agent updates", async () => { const events = createEventStream() - let requests = 0 + let agentID = "build" const calls = createFetch((url) => { if (url.pathname !== "/api/agent") return - requests++ return json({ location: { directory, project: { id: "proj_test", directory } }, data: [ { - id: requests === 1 ? "build" : "reviewer", + id: agentID, request: { headers: {}, body: {} }, mode: "primary", hidden: false, @@ -1911,6 +1910,11 @@ test("refreshes agents after agent updates", async () => { try { await wait(() => data.location.agent.list()?.[0]?.id === "build") + await Bun.sleep(20) + events.emit({ id: "evt_agent_unlocated", created: 0, type: "agent.updated", data: {} }) + await Bun.sleep(20) + expect(data.location.agent.list()?.[0]?.id).toBe("build") + agentID = "reviewer" emitEvent(events, { id: "evt_agent", created: 0, type: "agent.updated", data: {} }) await wait(() => data.location.agent.list()?.[0]?.id === "reviewer") } finally { @@ -2801,7 +2805,7 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn await mounted const received: string[] = [] const unsubscribe = sync.listen((event) => received.push(event.name)) - emitEvent(events, { + events.emit({ id: "evt_admitted_1", created: 0, type: "session.inbox.enqueued", diff --git a/packages/tui/test/fixture/tui-client.ts b/packages/tui/test/fixture/tui-client.ts index 51f79a6a840..be4f8e7d2db 100644 --- a/packages/tui/test/fixture/tui-client.ts +++ b/packages/tui/test/fixture/tui-client.ts @@ -153,6 +153,11 @@ export function createFetch(override?: FetchHandler, events?: ReturnType