diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index 8d924c9e9b5..cbb398a55d3 100644 --- a/packages/tui/src/app.tsx +++ b/packages/tui/src/app.tsx @@ -1146,11 +1146,8 @@ function App(props: { onSnapshot?: () => Promise; pluginHost: TuiPlugi return render({ params: route.data.data }) }) - // Suppress the full-screen reconnecting overlay for transient disconnects (initial startup, host - // reload, sub-second event-stream blips). After the first successful connect, show it only once the - // connection has been lost for a full second; before the first connect give a longer grace period so - // startup never flashes it, but a server that dies before ever connecting still surfaces instead of - // leaving a silent empty app. Hide it immediately the moment status leaves "connecting". + // Suppress the full-screen overlay for transient startup and event-stream retry states. + // Initial connection gets a longer grace period; retries surface more quickly. const [showReconnecting, setShowReconnecting] = createSignal(false) let reconnectTimer: ReturnType | undefined createEffect(() => { @@ -1158,7 +1155,8 @@ function App(props: { onSnapshot?: () => Promise; pluginHost: TuiPlugi clearTimeout(reconnectTimer) reconnectTimer = undefined } - if (sdk.connection.status() !== "connecting") { + const status = sdk.connection.status() + if (status === "connected") { setShowReconnecting(false) return } @@ -1167,7 +1165,7 @@ function App(props: { onSnapshot?: () => Promise; pluginHost: TuiPlugi reconnectTimer = undefined setShowReconnecting(true) }, - sdk.connection.connectedOnce() ? 1000 : 5000, + status === "reconnecting" ? 1000 : 5000, ).unref() }) onCleanup(() => { diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index da44a0f8e95..8ac669492a0 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -1,3 +1,8 @@ +// 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 { AgentInfo, CommandInfo, @@ -106,36 +111,12 @@ 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: SessionMessageInfo[], index: Map) => void) { setStore( @@ -237,8 +218,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function removeSession(sessionID: string) { - sessionRefreshApplied.set(sessionID, nextSessionRefresh(sessionID)) - sessionUsage.delete(sessionID) messageIndex.delete(sessionID) setStore( "session", @@ -267,7 +246,11 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ removeSession(event.data.sessionID) break case "session.usage.updated": - updateSessionUsage(event.data.sessionID, event.data.cost, event.data.tokens) + 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": void Promise.all([ @@ -807,20 +790,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const result = { on: sdk.event.on, listen: sdk.event.listen, - connection: { - status() { - return sdk.connection.status() - }, - attempt() { - return sdk.connection.attempt() - }, - error() { - return sdk.connection.error() - }, - connectedOnce() { - return sdk.connection.connectedOnce() - }, - }, session: { list() { return Object.values(store.session.info).toSorted((a, b) => b.time.updated - a.time.updated) @@ -846,17 +815,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }, }, async refresh(sessionID: string) { - 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, - ) + setStore("session", "info", sessionID, mutable(await sdk.api.session.get({ sessionID }))) registerSession(sessionID) }, message: { @@ -872,21 +831,11 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return position === undefined ? undefined : messages?.[position] }, async refresh(sessionID: string) { - const live = [...(store.session.message[sessionID] ?? [])] setStore("session", "message", sessionID, []) messageIndex.set(sessionID, new Map()) - const loaded = mutable( + const messages = 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) }, @@ -1031,8 +980,6 @@ 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({ @@ -1046,15 +993,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ "session", "info", produce((draft) => { - 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) draft[session.id] = mutable(session) }), ) for (const session of response.data) registerSession(session.id) @@ -1101,23 +1040,16 @@ 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) => { - if (generation !== connectionGeneration) return - const status: Record = Object.fromEntries( - Object.keys(active).map((sessionID) => [sessionID, "running" as const]), + setStore( + "session", + "status", + reconcile(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/context/sdk.tsx b/packages/tui/src/context/sdk.tsx index 0ad516ceacc..728ced3ed42 100644 --- a/packages/tui/src/context/sdk.tsx +++ b/packages/tui/src/context/sdk.tsx @@ -5,7 +5,7 @@ import { onCleanup, onMount } from "solid-js" import { createStore } from "solid-js/store" import { createSimpleContext } from "./helper" -export type SDKConnectionStatus = "connected" | "connecting" +export type SDKConnectionStatus = "connected" | "connecting" | "reconnecting" type SDKEventMap = { [Type in V2Event["type"]]: Extract } const connectTimeout = 2_000 @@ -27,11 +27,9 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ status: SDKConnectionStatus attempt: number error?: string - connectedOnce: boolean }>({ status: "connecting", attempt: 0, - connectedOnce: false, }) let stream: AbortController | undefined @@ -70,7 +68,7 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ clearTimeout(timeout) attempt = 0 events.emit(first.value.type, first.value) - setConnection({ status: "connected", attempt: 0, error: undefined, connectedOnce: true }) + setConnection({ status: "connected", attempt: 0, error: undefined }) connected() while (!abort.signal.aborted && !controller.signal.aborted) { const event = await iterator.next() @@ -98,7 +96,7 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ } } setConnection({ - status: "connecting", + status: "reconnecting", attempt, error: error instanceof Error ? error.message : String(error), }) @@ -136,9 +134,6 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ error() { return connection.error }, - connectedOnce() { - return connection.connectedOnce - }, }, reload: props.reload, } diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index c116f431787..e0612bb059c 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -6,7 +6,7 @@ import { SessionMessage } from "@opencode-ai/core/session/message" import { EventV2 } from "@opencode-ai/core/event" import { onMount } from "solid-js" import { ProjectProvider } from "../../../src/context/project" -import { SDKProvider } from "../../../src/context/sdk" +import { SDKProvider, useSDK } from "../../../src/context/sdk" import { DataProvider, useData } from "../../../src/context/data" import { createSessionRows, type SessionRow } from "../../../src/routes/session/rows" import { createApi, createClient, createEventStream, createFetch, directory, json } from "../../fixture/tui-sdk" @@ -114,24 +114,22 @@ test("refreshes resources into reactive getters", async () => { } }) -test("applies absolute usage events without losing full session updates", async () => { +test("applies absolute usage events to session info", 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") { - sessionsRequested = true - return new Promise((resolve) => { - resolveSessions = resolve + 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/${sessionID}`) { - return new Promise((resolve) => { - resolveSession.push(resolve) - }) - } }, events) let data!: ReturnType @@ -153,7 +151,7 @@ test("applies absolute usage events without losing full session updates", async )) try { - await wait(() => sessionsRequested) + await data.session.refresh(sessionID) emitEvent(events, { id: "evt_usage_2", created: 2, @@ -164,38 +162,6 @@ test("applies absolute usage events without losing full session updates", 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, @@ -204,7 +170,6 @@ test("applies absolute usage events without losing full session updates", async cache: { read: 1, write: 1 }, }) - const fullRefresh = data.session.refresh(sessionID) emitEvent(events, { id: "evt_usage_3", created: 3, @@ -216,57 +181,8 @@ test("applies absolute usage events without losing full session updates", async }, }) await wait(() => data.session.get(sessionID)?.cost === 1) - 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") + expect(data.session.get(sessionID)?.title).toBe("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, @@ -274,8 +190,7 @@ test("applies absolute usage events without losing full session updates", async durable: durable(sessionID, 9, 2), data: { sessionID }, }) - await Bun.sleep(20) - expect(data.session.get(sessionID)).toBeUndefined() + await wait(() => data.session.get(sessionID) === undefined) } finally { app.renderer.destroy() } @@ -579,9 +494,11 @@ test("reconnects the event stream and bootstraps fresh data", async () => { }) }, events) let data!: ReturnType + let sdk!: ReturnType function Probe() { data = useData() + sdk = useSDK() return } @@ -600,32 +517,24 @@ test("reconnects the event stream and bootstraps fresh data", async () => { try { await wait(() => data.location.model.list()?.[0]?.id === "model-1") await wait(() => data.session.status("session-stale") === "running") - expect(data.connection.status()).toBe("connected") - expect(data.connection.attempt()).toBe(0) + expect(sdk.connection.status()).toBe("connected") + expect(sdk.connection.attempt()).toBe(0) events.disconnect() - await wait(() => data.connection.status() === "connecting") - expect(data.connection.attempt()).toBe(1) - expect(data.connection.error()).toBe("Event stream disconnected") + await wait(() => sdk.connection.status() === "reconnecting") + expect(sdk.connection.attempt()).toBe(1) + expect(sdk.connection.error()).toBe("Event stream disconnected") - await wait(() => requests.active === 2 && data.connection.status() === "connected", 4000) - 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(() => requests.active === 2 && sdk.connection.status() === "connected", 4000) + resolveActive(json({ data: { "session-new": { type: "running" } } })) await wait(() => data.location.model.list()?.[0]?.id === "model-2", 4000) await wait(() => data.session.status("session-stale") === "idle") expect(data.session.status("session-new")).toBe("running") expect(requests.event).toBe(2) - expect(data.connection.status()).toBe("connected") - expect(data.connection.attempt()).toBe(0) - expect(data.connection.error()).toBeUndefined() + expect(sdk.connection.status()).toBe("connected") + expect(sdk.connection.attempt()).toBe(0) + expect(sdk.connection.error()).toBeUndefined() } finally { app.renderer.destroy() } @@ -766,7 +675,7 @@ test("removes committed revert messages from local state", async () => { } }) -test("connectedOnce is false until first connect and persists across disconnect", async () => { +test("distinguishes initial connection from reconnection", async () => { const encoder = new TextEncoder() let stream: ReadableStreamDefaultController | undefined const eventResponse = () => @@ -792,10 +701,10 @@ test("connectedOnce is false until first connect and persists across disconnect" const calls = createFetch((url) => { if (url.pathname === "/api/event") return eventResponse() }) - let data!: ReturnType + let sdk!: ReturnType function Probe() { - data = useData() + sdk = useSDK() return } @@ -813,16 +722,13 @@ test("connectedOnce is false until first connect and persists across disconnect" try { await wait(() => stream !== undefined) - expect(data.connection.status()).toBe("connecting") - expect(data.connection.connectedOnce()).toBe(false) + expect(sdk.connection.status()).toBe("connecting") connect() - await wait(() => data.connection.status() === "connected") - expect(data.connection.connectedOnce()).toBe(true) + await wait(() => sdk.connection.status() === "connected") disconnect() - await wait(() => data.connection.status() === "connecting") - expect(data.connection.connectedOnce()).toBe(true) + await wait(() => sdk.connection.status() === "reconnecting") } finally { app.renderer.destroy() } @@ -1462,9 +1368,11 @@ test("adds and dismisses permission requests from live events", async () => { const events = createEventStream() const calls = createFetch(undefined, events) let data!: ReturnType + let sdk!: ReturnType function Probe() { data = useData() + sdk = useSDK() return } @@ -1481,7 +1389,7 @@ test("adds and dismisses permission requests from live events", async () => { )) try { - await wait(() => data.connection.status() === "connected") + await wait(() => sdk.connection.status() === "connected") emitEvent(events, { id: "evt_permission_asked_1", created: 0, @@ -1580,9 +1488,11 @@ test("adds, dismisses, and refreshes form requests", async () => { return json({ data: [{ id: "frm_remote", sessionID: "ses_1", mode: "form", fields: [] }] }) }, events) let data!: ReturnType + let sdk!: ReturnType function Probe() { data = useData() + sdk = useSDK() return } @@ -1599,7 +1509,7 @@ test("adds, dismisses, and refreshes form requests", async () => { )) try { - await wait(() => data.connection.status() === "connected") + await wait(() => sdk.connection.status() === "connected") emitEvent(events, { id: "evt_form_created_1", created: 0,