From bf15c97e4b5723f54d5434469cbec2d3cc95b7fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=F0=9D=93=9B=F0=9D=93=B2=F0=9D=93=BD=F0=9D=93=BD?= =?UTF-8?q?=F0=9D=93=B5=F0=9D=93=AE=20=F0=9D=93=95=F0=9D=93=BB=F0=9D=93=AA?= =?UTF-8?q?=F0=9D=93=B7=F0=9D=93=B4?= Date: Wed, 8 Jul 2026 13:45:20 +0000 Subject: [PATCH] Revert "fix(tui): paginate session history" This reverts commit cc29f86cdc1371ec8255d68f750a732eb1f6cfd5. --- packages/tui/src/context/data.tsx | 306 +++++++++--------- packages/tui/src/routes/session/index.tsx | 111 ++----- packages/tui/src/routes/session/rows.ts | 126 +++++--- packages/tui/test/cli/tui/data.test.tsx | 205 ++++++------ .../tui/test/cli/tui/session-rows.test.ts | 22 +- 5 files changed, 385 insertions(+), 385 deletions(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index adefd3487f..da44a0f8e9 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -1,77 +1,60 @@ -// 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 may re-bootstrap; that is enough. UI and the server own ordering concerns. - import type { - AgentV2Info, - CommandV2Info, + AgentInfo, + CommandInfo, FormFormInfo, FormUrlInfo, IntegrationInfo, LocationRef, McpServer, - ModelV2Info, + ModelInfo, PermissionSavedInfo, PermissionV2Request, ProviderV2Info, ReferenceInfo, - SessionMessage, + SessionMessageInfo, SessionMessageAssistant, SessionMessageAssistantReasoning, SessionMessageAssistantText, SessionMessageAssistantTool, - SessionV2Info, + SessionInfo, Shell, - SkillV2Info, + SkillInfo, V2Event, } from "@opencode-ai/sdk/v2" import { createStore, produce, reconcile } from "solid-js/store" import { createSimpleContext } from "./helper" import { useSDK } from "./sdk" -import { batch, createSignal, onCleanup } from "solid-js" +import { createSignal, onCleanup } from "solid-js" export type DataSessionStatus = "idle" | "running" const messageIDFromEvent = (eventID: string) => eventID.replace(/^evt_/, "msg_") -const MESSAGE_PAGE_SIZE = 25 export type FormInfo = FormFormInfo | FormUrlInfo -// Per-session message timeline plus older-history paging. `items` is ascending; -// `cursor` is the opaque server cursor for the next older page after a desc first load. -type SessionMessages = { - items: SessionMessage[] - cursor?: string - complete: boolean - loading: boolean -} - type LocationData = { - agent?: AgentV2Info[] - command?: CommandV2Info[] + agent?: AgentInfo[] + command?: CommandInfo[] integration?: IntegrationInfo[] mcp?: McpServer[] - model?: ModelV2Info[] + model?: ModelInfo[] provider?: ProviderV2Info[] reference?: ReferenceInfo[] // 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?: SkillV2Info[] + skill?: SkillInfo[] } type Data = { session: { - info: Record + 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 status: Record - compaction: Partial> - compactionReason: Partial> - message: Record + message: Record input: Record permission: Record // Pending forms keyed by session ID. @@ -83,10 +66,6 @@ type Data = { location: Record } -function emptyMessages(): SessionMessages { - return { items: [], complete: false, loading: false } -} - function locationKey(location: LocationRef) { return JSON.stringify([location.directory, location.workspaceID]) } @@ -111,8 +90,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ info: {}, family: {}, status: {}, - compaction: {}, - compactionReason: {}, message: {}, input: {}, permission: {}, @@ -129,44 +106,66 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ directory: process.cwd(), }) const messageIndex = new Map>() + const sessionRefreshGeneration = new Map() + const sessionRefreshApplied = new Map() + const sessionUsage = new Map() + let connectionGeneration = 0 + let statusChanges: Set | undefined let bootstrapping: Promise | undefined function setSessionStatus(sessionID: string, status: DataSessionStatus) { + statusChanges?.add(sessionID) setStore("session", "status", sessionID, status) } + function nextSessionRefresh(sessionID: string) { + const generation = (sessionRefreshGeneration.get(sessionID) ?? 0) + 1 + sessionRefreshGeneration.set(sessionID, generation) + return generation + } + + function applySessionRefresh(sessionID: string, generation: number) { + if ((sessionRefreshApplied.get(sessionID) ?? 0) > generation) return false + sessionRefreshApplied.set(sessionID, generation) + return true + } + + function updateSessionUsage(sessionID: string, cost: number, tokens: SessionInfo["tokens"]) { + sessionUsage.set(sessionID, { generation: (sessionUsage.get(sessionID)?.generation ?? 0) + 1, cost, tokens }) + if (!store.session.info[sessionID]) return + setStore("session", "info", sessionID, { cost, tokens }) + } + const message = { - update(sessionID: string, fn: (messages: SessionMessage[], index: Map) => void) { + update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { setStore( "session", "message", produce((draft) => { - fn((draft[sessionID] ??= emptyMessages()).items, index(sessionID)) + fn((draft[sessionID] ??= []), index(sessionID)) }), ) }, - append(messages: SessionMessage[], index: Map, item: SessionMessage) { + append(messages: SessionMessageInfo[], index: Map, item: SessionMessageInfo) { if (index.has(item.id)) return index.set(item.id, messages.length) messages.push(item) }, - activeAssistant(messages: SessionMessage[]) { + activeAssistant(messages: SessionMessageInfo[]) { const item = messages.findLast((item) => item.type === "assistant" && !item.time.completed) return item?.type === "assistant" ? item : undefined }, - assistant(messages: SessionMessage[], index: Map, messageID: string) { + 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: SessionMessage[], shellID: string) { - const item = messages.findLast((item) => item.type === "shell" && item.shell.id === shellID) + shell(messages: SessionMessageInfo[], shellID: string) { + const item = messages.findLast((item) => item.type === "shell" && item.shellID === shellID) return item?.type === "shell" ? item : undefined }, - compaction(messages: SessionMessage[]) { - const item = messages.findLast( - (item) => item.type === "compaction" && (item.status === "queued" || item.status === "running"), - ) + compaction(messages: SessionMessageInfo[]) { + const item = messages.findLast((item) => item.type === "compaction" && item.status === "running") return item?.type === "compaction" ? item : undefined }, latestTool(assistant: SessionMessageAssistant | undefined, callID?: string) { @@ -238,13 +237,14 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function removeSession(sessionID: string) { + sessionRefreshApplied.set(sessionID, nextSessionRefresh(sessionID)) + sessionUsage.delete(sessionID) messageIndex.delete(sessionID) setStore( "session", produce((draft) => { delete draft.info[sessionID] delete draft.status[sessionID] - delete draft.compaction[sessionID] delete draft.message[sessionID] delete draft.input[sessionID] delete draft.permission[sessionID] @@ -267,11 +267,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ 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, - }) + updateSessionUsage(event.data.sessionID, event.data.cost, event.data.tokens) break case "catalog.updated": void Promise.all([ @@ -378,6 +374,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ id: messageIDFromEvent(event.id), type: "system", text: event.data.text, + metadata: event.metadata, time: { created: event.created }, }) }) @@ -387,9 +384,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ message.append(draft, index, { id: messageIDFromEvent(event.id), type: "synthetic", - sessionID: event.data.sessionID, text: event.data.text, description: event.data.description, + metadata: event.data.metadata, time: { created: event.created }, }) }) @@ -399,7 +396,11 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ message.append(draft, index, { id: messageIDFromEvent(event.id), type: "shell", - shell: event.data.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 }, }) }) @@ -408,7 +409,8 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ message.update(event.data.sessionID, (draft) => { const match = message.shell(draft, event.data.shell.id) if (!match) return - match.shell = event.data.shell + match.status = event.data.shell.status + match.exit = event.data.shell.exit match.output = event.data.output match.time.completed = event.created }) @@ -437,6 +439,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ 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 }, @@ -497,7 +500,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ id: event.data.callID, name: event.data.name, time: { created: event.created }, - state: { status: "pending", input: "" }, + state: { status: "streaming", input: "" }, }) }) break @@ -507,7 +510,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ message.assistant(draft, index, event.data.assistantMessageID), event.data.callID, ) - if (match?.state.status === "pending") match.state.input += event.data.delta + if (match?.state.status === "streaming") match.state.input += event.data.delta }) break case "session.tool.input.ended": @@ -516,7 +519,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ message.assistant(draft, index, event.data.assistantMessageID), event.data.callID, ) - if (match?.state.status === "pending") match.state.input = event.data.text + if (match?.state.status === "streaming") match.state.input = event.data.text }) break case "session.tool.called": @@ -568,7 +571,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ message.assistant(draft, index, event.data.assistantMessageID), event.data.callID, ) - if (!match || (match.state.status !== "pending" && match.state.status !== "running")) return + if (!match || (match.state.status !== "streaming" && match.state.status !== "running")) return match.state = { status: "error", error: event.data.error, @@ -623,29 +626,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setSessionStatus(event.data.sessionID, "running") break case "session.compaction.admitted": - message.update(event.data.sessionID, (draft, index) => { - if (message.compaction(draft)) return - message.append(draft, index, { - id: event.data.inputID, - type: "compaction", - status: "queued", - reason: "manual", - summary: "", - recent: "", - time: { created: event.created }, - }) - }) break case "session.compaction.started": - setStore("session", "compaction", event.data.sessionID, "") - setStore("session", "compactionReason", event.data.sessionID, event.data.reason) message.update(event.data.sessionID, (draft, index) => { - const current = message.compaction(draft) - if (current) { - current.status = "running" - current.reason = event.data.reason - return - } message.append(draft, index, { id: event.data.inputID ?? messageIDFromEvent(event.id), type: "compaction", @@ -661,10 +644,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ case "session.execution.failed": case "session.execution.interrupted": setSessionStatus(event.data.sessionID, "idle") - if (store.session.compaction[event.data.sessionID] !== undefined) - setStore("session", "compaction", event.data.sessionID, undefined) - if (store.session.compactionReason[event.data.sessionID] !== undefined) - setStore("session", "compactionReason", event.data.sessionID, undefined) message.update(event.data.sessionID, (draft) => { const currentAssistant = message.activeAssistant(draft) if (currentAssistant) currentAssistant.retry = undefined @@ -695,22 +674,26 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) break case "session.compaction.delta": - setStore("session", "compaction", event.data.sessionID, (text) => (text ?? "") + event.data.text) message.update(event.data.sessionID, (draft) => { const current = message.compaction(draft) - if (current) current.summary += event.data.text + if (current?.status === "running") current.summary += event.data.text }) break case "session.compaction.ended": - setStore("session", "compaction", event.data.sessionID, undefined) - setStore("session", "compactionReason", event.data.sessionID, undefined) message.update(event.data.sessionID, (draft, index) => { - const current = message.compaction(draft) - if (current) { - current.status = "completed" - current.reason = event.data.reason - current.summary = event.data.text - current.recent = event.data.recent + const position = draft.findLastIndex((item) => item.type === "compaction" && item.status === "running") + const current = draft[position] + if (current?.type === "compaction") { + draft[position] = { + id: current.id, + type: "compaction", + status: "completed", + reason: event.data.reason, + summary: event.data.text, + recent: event.data.recent, + metadata: current.metadata, + time: current.time, + } return } message.append(draft, index, { @@ -725,11 +708,26 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) break case "session.compaction.failed": - setStore("session", "compaction", event.data.sessionID, undefined) - setStore("session", "compactionReason", event.data.sessionID, undefined) - message.update(event.data.sessionID, (draft) => { - const current = message.compaction(draft) - if (current) current.status = "failed" + 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.v2.asked": @@ -847,73 +845,50 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return store.session.input[sessionID]?.includes(inputID) ?? false }, }, - compaction(sessionID: string) { - return store.session.compaction[sessionID] - }, async refresh(sessionID: string) { - setStore("session", "info", sessionID, mutable(await sdk.api.session.get({ sessionID }))) + const generation = nextSessionRefresh(sessionID) + const usageGeneration = sessionUsage.get(sessionID)?.generation ?? 0 + const info = mutable(await sdk.api.session.get({ sessionID })) + if (!applySessionRefresh(sessionID, generation)) return + const usage = sessionUsage.get(sessionID) + setStore( + "session", + "info", + sessionID, + usage && usage.generation !== usageGeneration ? { ...info, cost: usage.cost, tokens: usage.tokens } : info, + ) registerSession(sessionID) }, message: { ids(sessionID: string) { - return (store.session.message[sessionID]?.items ?? []).map((message) => message.id) + return (store.session.message[sessionID] ?? []).map((message) => message.id) }, list(sessionID: string) { - return store.session.message[sessionID]?.items ?? [] + return store.session.message[sessionID] ?? [] }, get(sessionID: string, messageID: string) { - const messages = store.session.message[sessionID]?.items + const messages = store.session.message[sessionID] const position = messageIndex.get(sessionID)?.get(messageID) return position === undefined ? undefined : messages?.[position] }, - cursor(sessionID: string) { - return store.session.message[sessionID]?.cursor - }, - complete(sessionID: string) { - return store.session.message[sessionID]?.complete ?? false - }, - loading(sessionID: string) { - return store.session.message[sessionID]?.loading ?? false - }, async refresh(sessionID: string) { - setStore("session", "message", sessionID, { ...emptyMessages(), loading: true }) + const live = [...(store.session.message[sessionID] ?? [])] + setStore("session", "message", sessionID, []) messageIndex.set(sessionID, new Map()) - const response = await sdk.api.message.list({ sessionID, limit: MESSAGE_PAGE_SIZE, order: "desc" }) - const items = mutable(response.data).toReversed() - messageIndex.set(sessionID, new Map(items.map((message, index) => [message.id, index]))) - setStore("session", "message", sessionID, { - items, - cursor: response.cursor.next ?? undefined, - complete: response.data.length < MESSAGE_PAGE_SIZE, - loading: false, - }) - const running = items.find((message) => message.type === "compaction" && message.status === "running") - setStore("session", "compaction", sessionID, running?.type === "compaction" ? running.summary : undefined) - setStore( - "session", - "compactionReason", - sessionID, - running?.type === "compaction" ? running.reason : undefined, - ) - }, - async more(sessionID: string) { - const current = store.session.message[sessionID] - if (!current || current.loading || current.complete || !current.cursor) return - const cursor = current.cursor - setStore("session", "message", sessionID, "loading", true) - const response = await sdk.api.message.list({ sessionID, limit: MESSAGE_PAGE_SIZE, cursor }) - const older = mutable(response.data).toReversed() - const prepend = older.filter((item) => !messageIndex.get(sessionID)?.has(item.id)) - const items = [...prepend, ...current.items] - messageIndex.set(sessionID, new Map(items.map((item, position) => [item.id, position]))) - batch(() => { - setStore("session", "message", sessionID, "items", items) - setStore("session", "message", sessionID, { - cursor: response.cursor.next ?? undefined, - complete: response.data.length < MESSAGE_PAGE_SIZE, - loading: false, - }) - }) + const loaded = mutable( + (await sdk.api.message.list({ sessionID, limit: 200, order: "desc" })).data, + ).toReversed() + const loadedIDs = new Set(loaded.map((message) => message.id)) + const liveByID = new Map(live.map((message) => [message.id, message])) + const messages = [ + ...loaded.map((message) => { + if (message.type === "user") return message + return liveByID.get(message.id) ?? message + }), + ...live.filter((message) => !loadedIDs.has(message.id)), + ].toSorted((a, b) => a.time.created - b.time.created) + messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) + setStore("session", "message", sessionID, messages) }, }, permission: { @@ -1056,6 +1031,8 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ async function bootstrap() { if (bootstrapping) return bootstrapping + const generation = new Map(sessionRefreshApplied) + const usageGeneration = new Map(Array.from(sessionUsage, ([id, usage]) => [id, usage.generation])) bootstrapping = Promise.allSettled([ sdk.api.session .list({ @@ -1069,7 +1046,15 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ "session", "info", produce((draft) => { - for (const session of response.data) draft[session.id] = mutable(session) + for (const session of response.data) { + if ((sessionRefreshApplied.get(session.id) ?? 0) !== (generation.get(session.id) ?? 0)) continue + const usage = sessionUsage.get(session.id) + draft[session.id] = mutable( + usage && usage.generation !== (usageGeneration.get(session.id) ?? 0) + ? { ...session, cost: usage.cost, tokens: usage.tokens } + : session, + ) + } }), ) for (const session of response.data) registerSession(session.id) @@ -1116,16 +1101,23 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function refreshActive() { + const generation = ++connectionGeneration + const changed = new Set() + statusChanges = changed void sdk.api.session .active() .then((active) => { - setStore( - "session", - "status", - reconcile(Object.fromEntries(Object.keys(active).map((sessionID) => [sessionID, "running" as const]))), + if (generation !== connectionGeneration) return + const status: Record = Object.fromEntries( + Object.keys(active).map((sessionID) => [sessionID, "running" as const]), ) + for (const sessionID of changed) status[sessionID] = store.session.status[sessionID] + setStore("session", "status", reconcile(status)) }) .catch(() => undefined) + .finally(() => { + if (statusChanges === changed) statusChanges = undefined + }) } onCleanup( diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 1c02329d62..e37fae9249 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -198,16 +198,6 @@ export function Session() { ?.id }) - // Admitted inputs and in-flight manual compaction sit after history, not in rows. - const pendingMessages = createMemo(() => { - const boundary = session()?.revert?.messageID - return messages().filter((message) => { - if (boundary && message.id >= boundary) return false - if (data.session.input.has(route.sessionID, message.id)) return true - return message.type === "compaction" && (message.status === "queued" || message.status === "running") - }) - }) - const lastAssistant = createMemo(() => { return messages().findLast((x) => x.type === "assistant") }) @@ -251,44 +241,37 @@ export function Session() { }), ) - createEffect( - on( - () => route.sessionID, - (sessionID) => { - void (async () => { - if (data.session.message.list(sessionID).length === 0) { - await Promise.all([ - data.session.refresh(sessionID), - data.session.message.refresh(sessionID), - data.session.permission.refresh(sessionID), - data.session.form.refresh(sessionID), - ]) - } - const info = data.session.get(sessionID) - if (!info) { - toast.show({ - message: `Session not found: ${sessionID}`, - variant: "error", - duration: 5000, - }) - navigate({ type: "home" }) - return - } - project.workspace.set(info.location.workspaceID) - editor.reconnect(info.location.directory) - if (route.sessionID === sessionID && scroll) scroll.scrollBy(100_000) - })().catch((error) => { - if (route.sessionID !== sessionID) return - toast.show({ - message: errorMessage(error), - variant: "error", - duration: 5000, - }) - navigate({ type: "home" }) + createEffect(() => { + const sessionID = route.sessionID + void (async () => { + await Promise.all([ + data.session.refresh(sessionID), + data.session.permission.refresh(sessionID), + data.session.form.refresh(sessionID), + ]) + const info = data.session.get(sessionID) + if (!info) { + toast.show({ + message: `Session not found: ${sessionID}`, + variant: "error", + duration: 5000, }) - }, - ), - ) + navigate({ type: "home" }) + return + } + project.workspace.set(info.location.workspaceID) + editor.reconnect(info.location.directory) + if (route.sessionID === sessionID && scroll) scroll.scrollBy(100_000) + })().catch((error) => { + if (route.sessionID !== sessionID) return + toast.show({ + message: errorMessage(error), + variant: "error", + duration: 5000, + }) + navigate({ type: "home" }) + }) + }) let seeded = false let scroll: ScrollBoxRenderable @@ -344,14 +327,12 @@ export function Session() { if (!targetID) { scroll.scrollBy(direction === "next" ? scroll.height : -scroll.height) - if (direction === "prev") loadOlder() dialog.clear() return } const child = scroll.getChildren().find((c) => c.id === targetID) if (child) scroll.scrollBy(child.y - scroll.y - 1) - if (direction === "prev") loadOlder() dialog.clear() } @@ -362,24 +343,6 @@ export function Session() { }, 50) } - let loadingOlder = false - function loadOlder() { - if (loadingOlder || scroll.scrollTop > 2) return - loadingOlder = true - const before = scroll.scrollHeight - void data.session.message.more(route.sessionID).then( - () => { - setTimeout(() => { - if (!scroll.isDestroyed) scroll.scrollBy(scroll.scrollHeight - before) - loadingOlder = false - }, 50) - }, - () => { - loadingOlder = false - }, - ) - } - const sessionCommandList = createMemo(() => [ { title: "Share session", @@ -585,7 +548,6 @@ export function Session() { hidden: true, run: () => { scroll.scrollBy(-scroll.height / 2) - loadOlder() dialog.clear() }, }, @@ -606,7 +568,6 @@ export function Session() { hidden: true, run: () => { scroll.scrollBy(-1) - loadOlder() dialog.clear() }, }, @@ -627,7 +588,6 @@ export function Session() { hidden: true, run: () => { scroll.scrollBy(-scroll.height / 4) - loadOlder() dialog.clear() }, }, @@ -648,7 +608,6 @@ export function Session() { hidden: true, run: () => { scroll.scrollTo(0) - loadOlder() dialog.clear() }, }, @@ -958,9 +917,6 @@ export function Session() { stickyStart="bottom" flexGrow={1} scrollAcceleration={scrollAcceleration()} - onMouseScroll={(event) => { - if (event.scroll?.direction === "up") void loadOlder() - }} > {(row) => ( @@ -970,13 +926,6 @@ export function Session() { /> )} - - {(message) => ( - - - - )} - ) { const [rows, setRows] = createStore([]) const revertBoundary = () => data.session.get(sessionID())?.revert?.messageID - function pendingIDs() { - const inputs = data.session.input.list(sessionID()) - const pending = new Set(inputs) - for (const message of data.session.message.list(sessionID())) { - if (message.type === "compaction" && (message.status === "queued" || message.status === "running")) - pending.add(message.id) - } - return pending - } - function reduce() { const messages = data.session.message.list(sessionID()) + const inputs = new Set(data.session.input.list(sessionID())) const boundary = revertBoundary() - const visible = boundary ? messages.filter((message) => message.id < boundary) : messages - const pending = pendingIDs() - const rows = reduceSessionRows(visible.filter((message) => !pending.has(message.id))) + const rows = reduceSessionRows(boundary ? messages.filter((message) => message.id < boundary) : messages, inputs) partitionPending(rows, pendingPermissions()) return rows } @@ -63,30 +52,48 @@ export function createSessionRows(sessionID: Accessor) { }) createEffect( - on(sessionID, () => { - setRows(reduce()) + on(sessionID, (id) => { + setRows(reconcile(reduce())) + void data.session.message.refresh(id).then( + () => { + if (sessionID() !== id) return + setRows(reconcile(reduce())) + }, + () => undefined, + ) }), ) + // Re-reduce when the revert boundary changes (stage/clear/commit). createEffect( on(revertBoundary, () => { - setRows(reduce()) + setRows(reconcile(reduce())) }), ) - // Pending inputs and compaction leaving the pending set change history membership. createEffect( on( - () => { - const messages = data.session.message.list(sessionID()) - const pending = data.session.input.list(sessionID()).join("\0") - const compaction = messages - .filter((message) => message.type === "compaction") - .map((message) => `${message.id}:${message.status}`) - .join("\0") - return `${pending}\u0001${compaction}` - }, - () => setRows(reduce()), + () => + data.session.message.list(sessionID()).flatMap((message) => + message.type === "user" + ? [ + { + id: message.id, + created: message.time.created, + input: data.session.input.has(sessionID(), message.id), + }, + ] + : message.type === "compaction" + ? [ + { + id: message.id, + created: message.time.created, + input: message.status === "running", + }, + ] + : [], + ), + () => setRows(reconcile(reduce())), ), ) @@ -94,9 +101,12 @@ export function createSessionRows(sessionID: Accessor) { setRows( produce((draft) => { if (draft.some((row) => row.type === "message" && row.messageID === messageID)) return - if (pendingIDs().has(messageID)) return - completePrevious(draft) - draft.push({ type: "message", messageID }) + const pending = isPending(messageID) + const message = data.session.message.get(sessionID(), messageID) + const index = + message?.type === "compaction" && pending ? queuedStart(draft) : pending ? draft.length : queuedStart(draft) + if (!pending) completePrevious(draft, index) + draft.splice(index, 0, { type: "message", messageID }) }), ) @@ -104,14 +114,15 @@ export function createSessionRows(sessionID: Accessor) { setRows( produce((draft) => { if (hasPart(draft, ref)) return + const index = queuedStart(draft) if (name && exploration(name)) { - const previous = draft.at(-1) + const previous = draft[index - 1] if (previous?.type === "group" && previous.kind === "exploration") { previous.refs.push(ref) return } - completePrevious(draft) - draft.push({ + completePrevious(draft, index) + draft.splice(index, 0, { type: "group", kind: "exploration", refs: [ref], @@ -120,8 +131,8 @@ export function createSessionRows(sessionID: Accessor) { }) return } - completePrevious(draft) - draft.push({ type: "part", ref }) + completePrevious(draft, index) + draft.splice(index, 0, { type: "part", ref }) }), ) @@ -129,8 +140,9 @@ export function createSessionRows(sessionID: Accessor) { setRows( produce((draft) => { if (draft.some((row) => row.type === "assistant-footer" && row.messageID === messageID)) return - completePrevious(draft) - draft.push({ type: "assistant-footer", messageID }) + const index = queuedStart(draft) + completePrevious(draft, index) + draft.splice(index, 0, { type: "assistant-footer", messageID }) }), ) @@ -142,13 +154,26 @@ export function createSessionRows(sessionID: Accessor) { }), ) + const isPending = (messageID: string) => { + const message = data.session.message.get(sessionID(), messageID) + if (message?.type === "user") return data.session.input.has(sessionID(), messageID) + return message?.type === "compaction" && message.status === "running" + } + + const queuedStart = (rows: SessionRow[]) => { + const index = rows.findIndex((row) => row.type === "message" && isPending(row.messageID)) + return index === -1 ? rows.length : index + } + const message = (event: { id: string; data: { sessionID: string } }) => { if (event.data.sessionID === sessionID()) appendMessage(event.id.replace(/^evt_/, "msg_")) } + const input = (event: { data: { sessionID: string; inputID: string } }) => { + if (event.data.sessionID === sessionID()) appendMessage(event.data.inputID) + } const subscriptions = [ - data.on("session.prompt.promoted", (event) => { - if (event.data.sessionID === sessionID()) appendMessage(event.data.inputID) - }), + data.on("session.prompt.admitted", input), + data.on("session.compaction.started", message), data.on("session.instructions.updated", message), data.on("session.synthetic", (event) => { if (event.data.sessionID === sessionID() && event.data.description?.trim()) @@ -157,7 +182,9 @@ export function createSessionRows(sessionID: Accessor) { data.on("session.shell.started", message), data.on("session.agent.selected", message), data.on("session.model.selected", message), - + data.on("session.compaction.ended", (event) => { + if (event.data.reason !== "manual") message(event) + }), data.on("session.text.delta", (event) => { if (event.data.sessionID === sessionID()) appendPart({ messageID: event.data.assistantMessageID, partID: `text:${event.data.ordinal}` }) @@ -197,11 +224,18 @@ export function createSessionRows(sessionID: Accessor) { return rows } -export function reduceSessionRows(messages: SessionMessage[]) { - return messages.reduce((rows, message) => { +export function reduceSessionRows(messages: SessionMessageInfo[], inputs = new Set()) { + const isInput = (message: SessionMessageInfo) => inputs.has(message.id) + const pendingCompactions = messages.filter((message) => message.type === "compaction" && message.status === "running") + const pending = new Set([...pendingCompactions.map((message) => message.id), ...inputs]) + return [ + ...messages.filter((message) => !pending.has(message.id)), + ...pendingCompactions, + ...messages.filter(isInput), + ].reduce((rows, message) => { if (message.type !== "assistant") { if (message.type === "synthetic" && !message.description?.trim()) return rows - completePrevious(rows) + if (!pending.has(message.id)) completePrevious(rows) rows.push({ type: "message", messageID: message.id }) return rows } diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index b17e6eea57..c116f43178 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -114,92 +114,24 @@ test("refreshes resources into reactive getters", async () => { } }) -test("pages older messages through nested message state", async () => { - const events = createEventStream() - const sessionID = "ses_message_page" - const pages: Array<{ limit?: string | null; order?: string | null; cursor?: string | null }> = [] - // Full first page (desc) so complete stays false until a short older page arrives. - const first = Array.from({ length: 50 }, (_, index) => { - const n = 51 - index - return { id: `msg_${n}`, type: "user" as const, text: String(n), time: { created: n } } - }) - const calls = createFetch((url) => { - if (url.pathname !== `/api/session/${sessionID}/message`) return - pages.push({ - limit: url.searchParams.get("limit"), - order: url.searchParams.get("order"), - cursor: url.searchParams.get("cursor"), - }) - if (!url.searchParams.get("cursor")) - return json({ - data: first, - cursor: { next: "cursor-older" }, - }) - return json({ - data: [{ id: "msg_1", type: "user", text: "one", time: { created: 1 } }], - cursor: {}, - }) - }, events) - let data!: ReturnType - - function Probe() { - data = useData() - return - } - - const app = await testRender(() => ( - - - - - - - - - - )) - - try { - await data.session.message.refresh(sessionID) - expect(pages).toEqual([{ limit: "50", order: "desc", cursor: null }]) - expect(data.session.message.ids(sessionID)).toEqual(first.toReversed().map((message) => message.id)) - expect(data.session.message.cursor(sessionID)).toBe("cursor-older") - expect(data.session.message.complete(sessionID)).toBe(false) - expect(data.session.message.loading(sessionID)).toBe(false) - - await data.session.message.more(sessionID) - expect(pages).toEqual([ - { limit: "50", order: "desc", cursor: null }, - { limit: "50", order: null, cursor: "cursor-older" }, - ]) - expect(data.session.message.ids(sessionID)[0]).toBe("msg_1") - expect(data.session.message.ids(sessionID)).toHaveLength(51) - expect(data.session.message.cursor(sessionID)).toBeUndefined() - expect(data.session.message.complete(sessionID)).toBe(true) - - await data.session.message.more(sessionID) - expect(pages).toHaveLength(2) - } finally { - app.renderer.destroy() - } -}) - -test("applies absolute usage events to session info", async () => { +test("applies absolute usage events without losing full session updates", async () => { const events = createEventStream() const sessionID = "ses_usage_refresh" + let resolveSessions!: (response: Response) => void + const resolveSession: Array<(response: Response) => void> = [] + let sessionsRequested = false const calls = createFetch((url) => { - if (url.pathname === `/api/session/${sessionID}`) - return json({ - data: { - id: sessionID, - projectID: "proj_test", - cost: 0, - tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, - time: { created: 0, updated: 0 }, - title: "Usage", - location: { directory }, - }, + if (url.pathname === "/api/session") { + sessionsRequested = true + return new Promise((resolve) => { + resolveSessions = resolve }) + } + if (url.pathname === `/api/session/${sessionID}`) { + return new Promise((resolve) => { + resolveSession.push(resolve) + }) + } }, events) let data!: ReturnType @@ -221,7 +153,7 @@ test("applies absolute usage events to session info", async () => { )) try { - await data.session.refresh(sessionID) + await wait(() => sessionsRequested) emitEvent(events, { id: "evt_usage_2", created: 2, @@ -232,6 +164,38 @@ test("applies absolute usage events to session info", async () => { tokens: { input: 5, output: 2, reasoning: 1, cache: { read: 1, write: 1 } }, }, }) + const initialRefresh = data.session.refresh(sessionID) + await wait(() => resolveSession.length === 1) + resolveSessions( + json({ + data: [ + { + id: sessionID, + projectID: "proj_test", + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + time: { created: 0, updated: 0 }, + title: "Stale usage", + location: { directory }, + }, + ], + cursor: {}, + }), + ) + resolveSession[0]( + json({ + data: { + id: sessionID, + projectID: "proj_test", + cost: 0.5, + tokens: { input: 5, output: 2, reasoning: 1, cache: { read: 1, write: 1 } }, + time: { created: 0, updated: 0 }, + title: "Current usage", + location: { directory }, + }, + }), + ) + await initialRefresh await wait(() => data.session.get(sessionID)?.cost === 0.5) expect(data.session.get(sessionID)?.tokens).toEqual({ input: 5, @@ -240,6 +204,7 @@ test("applies absolute usage events to session info", async () => { cache: { read: 1, write: 1 }, }) + const fullRefresh = data.session.refresh(sessionID) emitEvent(events, { id: "evt_usage_3", created: 3, @@ -251,8 +216,57 @@ test("applies absolute usage events to session info", async () => { }, }) await wait(() => data.session.get(sessionID)?.cost === 1) - expect(data.session.get(sessionID)?.title).toBe("Usage") + resolveSession[1]( + json({ + data: { + id: sessionID, + projectID: "proj_test", + cost: 0.75, + tokens: { input: 8, output: 3, reasoning: 1, cache: { read: 1, write: 1 } }, + time: { created: 0, updated: 0 }, + title: "Older usage", + location: { directory }, + }, + }), + ) + await fullRefresh + await Bun.sleep(20) + expect(data.session.get(sessionID)?.cost).toBe(1) + expect(data.session.get(sessionID)?.title).toBe("Older usage") + emitEvent(events, { + id: "evt_usage_6", + created: 6, + type: "session.usage.updated", + data: { + sessionID, + cost: 1.25, + tokens: { input: 12, output: 5, reasoning: 1, cache: { read: 1, write: 1 } }, + }, + }) + emitEvent(events, { + id: "evt_usage_7", + created: 7, + type: "session.usage.updated", + data: { + sessionID, + cost: 1.25, + tokens: { input: 12, output: 5, reasoning: 1, cache: { read: 1, write: 1 } }, + }, + }) + await wait(() => data.session.get(sessionID)?.cost === 1.25) + expect(data.session.get(sessionID)?.title).toBe("Older usage") + + emitEvent(events, { + id: "evt_usage_8", + created: 8, + type: "session.usage.updated", + data: { + sessionID, + cost: 1.5, + tokens: { input: 14, output: 6, reasoning: 1, cache: { read: 1, write: 1 } }, + }, + }) emitEvent(events, { id: "evt_usage_deleted", created: 9, @@ -260,7 +274,8 @@ test("applies absolute usage events to session info", async () => { durable: durable(sessionID, 9, 2), data: { sessionID }, }) - await wait(() => data.session.get(sessionID) === undefined) + await Bun.sleep(20) + expect(data.session.get(sessionID)).toBeUndefined() } finally { app.renderer.destroy() } @@ -594,7 +609,15 @@ test("reconnects the event stream and bootstraps fresh data", async () => { expect(data.connection.error()).toBe("Event stream disconnected") await wait(() => requests.active === 2 && data.connection.status() === "connected", 4000) - resolveActive(json({ data: { "session-new": { type: "running" } } })) + emitEvent(events, { + id: "evt_execution_started_after_reconnect", + created: 1, + type: "session.execution.started", + durable: durable("session-new"), + data: { sessionID: "session-new" }, + }) + await wait(() => data.session.status("session-new") === "running") + resolveActive(json({ data: {} })) await wait(() => data.location.model.list()?.[0]?.id === "model-2", 4000) await wait(() => data.session.status("session-stale") === "idle") @@ -608,18 +631,16 @@ test("reconnects the event stream and bootstraps fresh data", async () => { } }) -test("keeps pending prompts out of history rows until promoted", async () => { +test("completes exploration when a queued prompt is promoted", async () => { const events = createEventStream() const sessionID = "session-promotion" const calls = createFetch((url) => { if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) }, events) let rows!: ReturnType - let data!: ReturnType function Probe() { rows = createSessionRows(() => sessionID) - data = useData() return } @@ -674,8 +695,7 @@ test("keeps pending prompts out of history rows until promoted", async () => { delivery: "steer", }, }) - await wait(() => data.session.input.has(sessionID, "message-user")) - expect(rows.some((row) => row.type === "message" && row.messageID === "message-user")).toBe(false) + await wait(() => rows.at(-1)?.type === "message") expect(rows.find((row) => row.type === "group")?.completed).toBe(false) emitEvent(events, { @@ -685,8 +705,7 @@ test("keeps pending prompts out of history rows until promoted", async () => { durable: durable(sessionID, 3), data: { sessionID, inputID: "message-user" }, }) - await wait(() => rows.some((row) => row.type === "message" && row.messageID === "message-user")) - expect(data.session.input.has(sessionID, "message-user")).toBe(false) + await wait(() => rows.find((row) => row.type === "group")?.completed === true) expect(rows.at(-1)).toEqual({ type: "message", messageID: "message-user" }) } finally { app.renderer.destroy() diff --git a/packages/tui/test/cli/tui/session-rows.test.ts b/packages/tui/test/cli/tui/session-rows.test.ts index 1def9caa55..c43a6faf9d 100644 --- a/packages/tui/test/cli/tui/session-rows.test.ts +++ b/packages/tui/test/cli/tui/session-rows.test.ts @@ -207,25 +207,31 @@ test("renders a footer for a pre-output retry assistant after replay", () => { expect(reduceSessionRows([message])).toEqual([{ type: "assistant-footer", messageID: "assistant-retry" }]) }) -test("history reduce keeps chronological order without pending reordering", () => { +test("places a running compaction barrier before every queued user message", () => { + const queued = (id: string, text: string, created: number): SessionMessageInfo => ({ + type: "user", + id, + text, + time: { created }, + }) const messages: SessionMessageInfo[] = [ - { type: "user", id: "user-1", text: "Before", time: { created: 1 } }, + queued("user-before", "Before", 1), { type: "compaction", id: "compaction", - status: "completed", + status: "running", reason: "manual", - summary: "done", + summary: "", recent: "", time: { created: 2 }, }, - { type: "user", id: "user-2", text: "After", time: { created: 3 } }, + queued("user-after", "After", 3), ] - expect(reduceSessionRows(messages)).toEqual([ - { type: "message", messageID: "user-1" }, + expect(reduceSessionRows(messages, new Set(["user-before", "user-after"]))).toEqual([ { type: "message", messageID: "compaction" }, - { type: "message", messageID: "user-2" }, + { type: "message", messageID: "user-before" }, + { type: "message", messageID: "user-after" }, ]) })