From 999064d6a1bf41313d567eddf8a841e413df04b2 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 12:32:49 -0400 Subject: [PATCH 1/7] fix(tui): preserve admitted messages during hydration --- packages/tui/src/context/data.tsx | 21 +++++-- packages/tui/test/cli/tui/data.test.tsx | 80 +++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 5 deletions(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index e71c3a6985e2..3db6015a0e4c 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -1,7 +1,7 @@ // 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. +// Prefer straightforward projection. API reads replace cached state, except admitted +// inputs survive history refresh until the server projects them. Do not add generation +// counters or live/history overlays here. UI and the server own ordering concerns. import type { AgentInfo, @@ -877,9 +877,20 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return position === undefined ? undefined : messages?.[position] }, async refresh(sessionID: string) { + const pending = new Set(store.session.input[sessionID] ?? []) const messages = (await sdk.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)) + const projected = new Set(messages.map((message) => message.id)) + const next = [ + ...messages, + ...(store.session.message[sessionID] ?? []).filter( + (message) => + (message.type === "user" || message.type === "synthetic") && + (pending.has(message.id) || store.session.input[sessionID]?.includes(message.id)) && + !projected.has(message.id), + ), + ] + messageIndex.set(sessionID, new Map(next.map((message, index) => [message.id, index]))) + setStore("session", "message", sessionID, reconcile(next)) }, }, permission: { diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 936e141b3107..610c887d78b1 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -2210,6 +2210,86 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn } }) +test("preserves an admitted prompt when hydration races with promotion", async () => { + const events = createEventStream() + const sessionID = "session-hydration-race" + const messageID = "msg_initial_user" + let resolveMessages!: (response: Response) => void + let requestMessages!: () => void + const requested = new Promise((resolve) => { + requestMessages = resolve + }) + const calls = createFetch((url) => { + if (url.pathname !== `/api/session/${sessionID}/message`) return + requestMessages() + return new Promise((resolve) => { + resolveMessages = resolve + }) + }, events) + let data!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + data = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + emitEvent(events, { + id: "evt_initial_admitted", + created: 1, + type: "session.input.admitted", + durable: durable(sessionID), + data: { + sessionID, + inputID: messageID, + input: { type: "user", data: { text: "Initial prompt" }, delivery: "steer" }, + }, + }) + await wait(() => data.session.message.get(sessionID, messageID)?.type === "user") + + const refresh = data.session.message.refresh(sessionID) + await requested + emitEvent(events, { + id: "evt_initial_promoted", + created: 2, + type: "session.input.promoted", + durable: durable(sessionID, 1), + data: { sessionID, inputID: messageID }, + }) + await wait(() => data.session.input.list(sessionID).length === 0) + + resolveMessages(json({ data: [], cursor: {} })) + await refresh + + expect(data.session.message.get(sessionID, messageID)).toMatchObject({ + id: messageID, + type: "user", + text: "Initial prompt", + time: { created: 2 }, + }) + } finally { + app.renderer.destroy() + } +}) + test("projects live instruction updates with their message ID", async () => { const events = createEventStream() const calls = createFetch(undefined, events) From 6592b6c8da22baf0eadbece70dd96db17f31b3c6 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 12:41:56 -0400 Subject: [PATCH 2/7] refactor(tui): simplify hydration race fix --- packages/tui/src/context/data.tsx | 18 ++--- packages/tui/test/cli/tui/data.test.tsx | 102 +++--------------------- 2 files changed, 18 insertions(+), 102 deletions(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 3db6015a0e4c..a2d7c10db449 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -879,18 +879,12 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ async refresh(sessionID: string) { const pending = new Set(store.session.input[sessionID] ?? []) const messages = (await sdk.api.message.list({ sessionID, limit: 200, order: "desc" })).data.toReversed() - const projected = new Set(messages.map((message) => message.id)) - const next = [ - ...messages, - ...(store.session.message[sessionID] ?? []).filter( - (message) => - (message.type === "user" || message.type === "synthetic") && - (pending.has(message.id) || store.session.input[sessionID]?.includes(message.id)) && - !projected.has(message.id), - ), - ] - messageIndex.set(sessionID, new Map(next.map((message, index) => [message.id, index]))) - setStore("session", "message", sessionID, reconcile(next)) + const index = new Map(messages.map((message, index) => [message.id, index])) + store.session.message[sessionID] + ?.filter((message) => pending.has(message.id)) + .forEach((item) => message.append(messages, index, item)) + messageIndex.set(sessionID, index) + setStore("session", "message", sessionID, reconcile(messages)) }, }, permission: { diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 610c887d78b1..70a4c24f1986 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -2121,16 +2121,16 @@ test("settles pending tools when a live failure arrives", async () => { } }) -test("renders admitted prompts immediately and tracks them until promoted", async () => { +test("preserves admitted prompts when hydration races with promotion", async () => { const events = createEventStream() const sessionID = "session-1" const messageID = "msg_user_1" + const requested = Promise.withResolvers() + const response = Promise.withResolvers() const calls = createFetch((url) => { - if (url.pathname === `/api/session/${sessionID}/message`) - return json({ - data: [{ id: messageID, type: "user", text: "hello", time: { created: 0 } }], - cursor: {}, - }) + if (url.pathname !== `/api/session/${sessionID}/message`) return + requested.resolve() + return response.promise }, events) let sync!: ReturnType let ready!: () => void @@ -2177,12 +2177,11 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn expect(admitted?.metadata).toBeUndefined() expect(sync.session.input.list(sessionID)).toEqual([messageID]) - await sync.session.message.refresh(sessionID) - expect(sync.session.message.list(sessionID)?.[0]?.metadata).toBeUndefined() - + const refresh = sync.session.message.refresh(sessionID) + await requested.promise emitEvent(events, { id: "evt_prompted_1", - created: 0, + created: 1, type: "session.input.promoted", durable: durable(sessionID, 1), data: { @@ -2194,6 +2193,9 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn await wait(() => received.at(-1) === "session.input.promoted") expect(received.slice(-2)).toEqual(["session.input.admitted", "session.input.promoted"]) unsubscribe() + response.resolve(json({ data: [], cursor: {} })) + await refresh + const message = sync.session.message.list(sessionID)?.[0] expect(message?.type).toBe("user") if (message?.type !== "user") return @@ -2210,86 +2212,6 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn } }) -test("preserves an admitted prompt when hydration races with promotion", async () => { - const events = createEventStream() - const sessionID = "session-hydration-race" - const messageID = "msg_initial_user" - let resolveMessages!: (response: Response) => void - let requestMessages!: () => void - const requested = new Promise((resolve) => { - requestMessages = resolve - }) - const calls = createFetch((url) => { - if (url.pathname !== `/api/session/${sessionID}/message`) return - requestMessages() - return new Promise((resolve) => { - resolveMessages = resolve - }) - }, events) - let data!: ReturnType - let ready!: () => void - const mounted = new Promise((resolve) => { - ready = resolve - }) - - function Probe() { - data = useData() - onMount(ready) - return - } - - const app = await testRender(() => ( - - - - - - - - - - )) - - try { - await mounted - emitEvent(events, { - id: "evt_initial_admitted", - created: 1, - type: "session.input.admitted", - durable: durable(sessionID), - data: { - sessionID, - inputID: messageID, - input: { type: "user", data: { text: "Initial prompt" }, delivery: "steer" }, - }, - }) - await wait(() => data.session.message.get(sessionID, messageID)?.type === "user") - - const refresh = data.session.message.refresh(sessionID) - await requested - emitEvent(events, { - id: "evt_initial_promoted", - created: 2, - type: "session.input.promoted", - durable: durable(sessionID, 1), - data: { sessionID, inputID: messageID }, - }) - await wait(() => data.session.input.list(sessionID).length === 0) - - resolveMessages(json({ data: [], cursor: {} })) - await refresh - - expect(data.session.message.get(sessionID, messageID)).toMatchObject({ - id: messageID, - type: "user", - text: "Initial prompt", - time: { created: 2 }, - }) - } finally { - app.renderer.destroy() - } -}) - test("projects live instruction updates with their message ID", async () => { const events = createEventStream() const calls = createFetch(undefined, events) From c45b65c286a861e52f679e84775ad406537cdf7e Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 13:01:28 -0400 Subject: [PATCH 3/7] feat(tui): hydrate pending session work --- packages/tui/src/context/data.tsx | 57 +++++++++++++++++++++---- packages/tui/test/cli/tui/data.test.tsx | 31 ++++++++++++-- packages/tui/test/fixture/tui-sdk.ts | 1 + 3 files changed, 78 insertions(+), 11 deletions(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index a2d7c10db449..1f236abdf9b0 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -21,6 +21,7 @@ import type { SessionMessageAssistantText, SessionMessageAssistantTool, SessionInfo, + SessionPendingInfo, Shell, SkillInfo, } from "@opencode-ai/sdk/v2" @@ -161,6 +162,31 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const item = messages.findLast((item) => item.type === "compaction" && item.status === "running") return item?.type === "compaction" ? item : undefined }, + fromPending(item: SessionPendingInfo): SessionMessageInfo { + if (item.type === "user") + return { + id: item.id, + type: "user", + ...item.data, + time: { created: item.timeCreated }, + } + if (item.type === "synthetic") + return { + id: item.id, + type: "synthetic", + ...item.data, + time: { created: item.timeCreated }, + } + return { + id: item.id, + type: "compaction", + status: "running", + reason: "manual", + summary: "", + recent: "", + time: { created: item.timeCreated }, + } + }, latestTool(assistant: SessionMessageAssistant | undefined, callID?: string) { return assistant?.content.findLast( (item): item is SessionMessageAssistantTool => @@ -877,14 +903,29 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return position === undefined ? undefined : messages?.[position] }, async refresh(sessionID: string) { - const pending = new Set(store.session.input[sessionID] ?? []) - const messages = (await sdk.api.message.list({ sessionID, limit: 200, order: "desc" })).data.toReversed() - const index = new Map(messages.map((message, index) => [message.id, index])) - store.session.message[sessionID] - ?.filter((message) => pending.has(message.id)) - .forEach((item) => message.append(messages, index, item)) - messageIndex.set(sessionID, index) - setStore("session", "message", sessionID, reconcile(messages)) + const inputsBeforeRefresh = new Set(store.session.input[sessionID] ?? []) + const [projected, pending] = await Promise.all([ + sdk.api.message.list({ sessionID, limit: 200, order: "desc" }), + sdk.api.session.pending.list({ sessionID }), + ]) + const messages = projected.data.toReversed() + const projectedIDs = new Set(messages.map((message) => message.id)) + const pendingMessages = pending.filter((item) => !projectedIDs.has(item.id)).map(message.fromPending) + const next = [...messages, ...pendingMessages] + messageIndex.set(sessionID, new Map(next.map((message, index) => [message.id, index]))) + const inputsAfterRefresh = new Set(store.session.input[sessionID] ?? []) + setStore( + "session", + "input", + sessionID, + pendingMessages.flatMap((message) => + (message.type === "user" || message.type === "synthetic") && + (!inputsBeforeRefresh.has(message.id) || inputsAfterRefresh.has(message.id)) + ? [message.id] + : [], + ), + ) + setStore("session", "message", sessionID, reconcile(next)) }, }, permission: { diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 70a4c24f1986..33c945736ad1 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -2125,9 +2125,33 @@ test("preserves admitted prompts when hydration races with promotion", async () const events = createEventStream() const sessionID = "session-1" const messageID = "msg_user_1" + const queuedID = "msg_user_2" const requested = Promise.withResolvers() const response = Promise.withResolvers() const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/pending`) + return json({ + data: [ + { + admittedSeq: 0, + id: messageID, + sessionID, + timeCreated: 0, + type: "user", + data: { text: "hello" }, + delivery: "steer", + }, + { + admittedSeq: 1, + id: queuedID, + sessionID, + timeCreated: 1, + type: "user", + data: { text: "queued" }, + delivery: "queue", + }, + ], + }) if (url.pathname !== `/api/session/${sessionID}/message`) return requested.resolve() return response.promise @@ -2196,13 +2220,14 @@ test("preserves admitted prompts when hydration races with promotion", async () response.resolve(json({ data: [], cursor: {} })) await refresh - const message = sync.session.message.list(sessionID)?.[0] + const message = sync.session.message.get(sessionID, messageID) expect(message?.type).toBe("user") if (message?.type !== "user") return expect(message).toMatchObject({ id: messageID, text: "hello" }) expect(message.metadata).toBeUndefined() - expect(sync.session.input.list(sessionID)).toEqual([]) - expect(sync.session.message.ids(sessionID)).toEqual([messageID]) + expect(sync.session.input.list(sessionID)).toEqual([queuedID]) + expect(sync.session.message.ids(sessionID)).toEqual([messageID, queuedID]) + expect(sync.session.message.get(sessionID, queuedID)).toMatchObject({ id: queuedID, text: "queued" }) expect(sync.session.message.ids("missing")).toEqual([]) expect(sync.session.message.get(sessionID, messageID)).toBe(message) expect(sync.session.message.get(sessionID, "missing")).toBeUndefined() diff --git a/packages/tui/test/fixture/tui-sdk.ts b/packages/tui/test/fixture/tui-sdk.ts index 6708fd0fa4bc..60a4a432a853 100644 --- a/packages/tui/test/fixture/tui-sdk.ts +++ b/packages/tui/test/fixture/tui-sdk.ts @@ -105,6 +105,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType Date: Sat, 11 Jul 2026 13:13:34 -0400 Subject: [PATCH 4/7] refactor(tui): clarify pending reconciliation --- packages/tui/src/context/data.tsx | 38 +++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 1f236abdf9b0..672104863b80 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -83,6 +83,16 @@ function locationQuery(ref?: LocationRef) { return ref ? { directory: ref.directory, workspace: ref.workspaceID } : undefined } +// Apply local admissions and promotions that arrived while the server snapshot was loading. +function reconcilePendingInputs(baseline: string[], current: string[], server: string[]) { + const result = new Set(server) + const before = new Set(baseline) + const now = new Set(current) + baseline.filter((id) => !now.has(id)).forEach((id) => result.delete(id)) + current.filter((id) => !before.has(id)).forEach((id) => result.add(id)) + return [...result] +} + export const { use: useData, provider: DataProvider } = createSimpleContext({ name: "Data", init: () => { @@ -903,26 +913,30 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return position === undefined ? undefined : messages?.[position] }, async refresh(sessionID: string) { - const inputsBeforeRefresh = new Set(store.session.input[sessionID] ?? []) + const localInputs = store.session.input[sessionID] ?? [] const [projected, pending] = await Promise.all([ sdk.api.message.list({ sessionID, limit: 200, order: "desc" }), sdk.api.session.pending.list({ sessionID }), ]) - const messages = projected.data.toReversed() - const projectedIDs = new Set(messages.map((message) => message.id)) - const pendingMessages = pending.filter((item) => !projectedIDs.has(item.id)).map(message.fromPending) - const next = [...messages, ...pendingMessages] - messageIndex.set(sessionID, new Map(next.map((message, index) => [message.id, index]))) - const inputsAfterRefresh = new Set(store.session.input[sessionID] ?? []) + const pendingMessages = pending.map(message.fromPending) + const next = projected.data.toReversed() + const index = new Map(next.map((message, index) => [message.id, index])) + const localInputIDs = new Set(localInputs) + store.session.message[sessionID] + ?.filter((message) => localInputIDs.has(message.id)) + .forEach((item) => message.append(next, index, item)) + pendingMessages.forEach((item) => message.append(next, index, item)) + messageIndex.set(sessionID, index) setStore( "session", "input", sessionID, - pendingMessages.flatMap((message) => - (message.type === "user" || message.type === "synthetic") && - (!inputsBeforeRefresh.has(message.id) || inputsAfterRefresh.has(message.id)) - ? [message.id] - : [], + reconcilePendingInputs( + localInputs, + store.session.input[sessionID] ?? [], + pendingMessages.flatMap((message) => + message.type === "user" || message.type === "synthetic" ? [message.id] : [], + ), ), ) setStore("session", "message", sessionID, reconcile(next)) From 0b1662f228f3f41049575dfd086fd0f17b95b5bd Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 14:35:43 -0400 Subject: [PATCH 5/7] refactor(tui): model pending timeline inputs --- packages/tui/TIMELINE_PLAN.md | 242 +++++++++++++++ packages/tui/src/context/data.tsx | 293 ++++++++++-------- packages/tui/src/context/session-timeline.ts | 150 +++++++++ .../tui/src/routes/session/dialog-fork.tsx | 4 +- .../tui/src/routes/session/dialog-message.tsx | 2 +- packages/tui/src/routes/session/index.tsx | 12 +- packages/tui/src/routes/session/rows.ts | 49 +-- packages/tui/test/cli/tui/data.test.tsx | 218 +++++++++++-- .../tui/test/cli/tui/session-rows.test.ts | 87 +++++- .../tui/test/context/session-timeline.test.ts | 80 +++++ packages/tui/test/fixture/tui-sdk.ts | 1 + 11 files changed, 949 insertions(+), 189 deletions(-) create mode 100644 packages/tui/TIMELINE_PLAN.md create mode 100644 packages/tui/src/context/session-timeline.ts create mode 100644 packages/tui/test/context/session-timeline.test.ts diff --git a/packages/tui/TIMELINE_PLAN.md b/packages/tui/TIMELINE_PLAN.md new file mode 100644 index 000000000000..527825e77621 --- /dev/null +++ b/packages/tui/TIMELINE_PLAN.md @@ -0,0 +1,242 @@ +# Session Timeline Frontend Plan + +## Status + +This plan incorporates reviews of Solid identity, hydration concurrency, Session execution ordering, rendering, scroll behavior, and migration scope. The server contract remains unchanged: projected messages and pending inputs are separate reads, and the event stream has no replay cursor across reconnects. + +## Goals + +- Never overwrite an admitted prompt with a hydration snapshot. +- Keep one stable rendered identity from admission through promotion and projection. +- Keep projected history semantically distinct from pending frontend work. +- Make the unavoidable snapshot/live-event reconciliation narrow and explicit. +- Preserve targeted updates for streamed assistant content. +- Avoid changing unrelated message consumers in this PR. + +## Constraints Discovered In Review + +- Promotion atomically deletes pending state and inserts projected state. +- Reading pending before projected history guarantees an input admitted before the pending read appears in at least one response. +- An admission after the pending read may appear in neither response and must survive through its live event. +- The event stream does not replay events missed across a disconnect, so reconnect must perform authoritative reads. +- General event replay is unsafe. Text, reasoning, tool input, and compaction deltas are additive, while snapshots expose no event watermark proving which deltas they contain. +- Solid `` retains children by source-item identity. The existing `SessionRow` values lack a uniform `id`, so full row reduction does not express stable semantic identity. +- OpenTUI sticky scrolling follows the bottom but does not preserve a scrolled-up content anchor when rows are inserted above it. + +## Selected Architecture + +Keep the existing projected/live message cache. Add a lifecycle overlay containing only pending or locally promoted work. + +```ts +type TimelineInput = { + id: string + phase: "pending" | "promoted" + admittedSeq: number + delivery?: "steer" | "queue" + message?: SessionMessageInfo +} + +type Data = { + session: { + message: Record + input: Record + } +} +``` + +The overlay keeps pending content, phase, delivery, and ordering metadata together. It replaces the current parallel representation where content is inserted into `session.message` while only its ID is stored in `session.input`. + +Projected messages remain available through the existing projected cache. A small visible-timeline surface composes projected messages with the overlay for rendering: + +```ts +timeline.get(sessionID, messageID) +timeline.list(sessionID) +message.refresh(sessionID) +``` + +Existing `session.message` consumers remain unchanged unless they need pending work. The Session row renderer migrates to the visible timeline surface. + +## Lifecycle + +Phase precedence is monotone: + +```text +projected > promoted > pending +``` + +Content precedence is: + +```text +projected message > pending/admission content > unknown placeholder +``` + +Admission inserts one pending overlay entry. Promotion changes that entry to `promoted` without removing it. Observing the same ID in projected history removes the overlay entry because projected state is now authoritative. + +Promotion-before-admission delivery is tolerated defensively: promotion creates a non-rendered placeholder, and a later admission fills its content without downgrading its phase. + +## Hydration + +Hydration is sequential: + +```ts +const pending = await api.session.pending.list({ sessionID }) +const projected = await api.message.list({ sessionID, limit: 200, order: "desc" }) +``` + +For a promotion commit `X`, pending read `P`, and projected read `M`, `P < M` eliminates the unsafe `M < X < P` split snapshot. + +The proof is: + +- `X < P`: pending omits the input and projected contains it. +- `P < X < M`: pending and projected may both contain it; projected wins by ID. +- `M < X`: pending contains it, and a later promotion event changes its local phase. + +An input admitted after `P` and not promoted before `M` is absent from both snapshots. Its live admission operation must be preserved during installation. + +## Narrow Hydration Journal + +Only input topology operations are journaled: + +```ts +type TimelineOperation = + | { type: "admitted"; input: TimelineInput } + | { type: "promoted"; inputID: string; promotedSeq: number; created: number } + | { type: "reverted"; to: string } +``` + +All existing assistant, text, reasoning, tool, shell, and compaction streaming events continue to update the projected/live message cache exactly once. They are never replayed. + +Hydration performs: + +1. Open a per-Session lifecycle journal. +2. Continue applying admission and promotion immediately to the visible overlay while recording them. +3. Read pending inputs. +4. Read projected messages. +5. Build projected history and pending overlay from those snapshots. +6. Fold the lifecycle journal onto the overlay with an idempotent, monotone reducer. +7. Remove overlay entries whose IDs are present in projected history. +8. Install projected history and overlay together in one Solid batch. +9. Close the journal. + +If either request fails, do not install an authoritative empty collection. Preserve visible state and retry through the existing caller/reconnect path. + +## Refresh Coordination + +At most one refresh per Session and connection epoch may install. + +- Same-epoch callers join the current promise. +- Every `server.connected` increments a connection epoch. +- A result from an older epoch is discarded. +- Reconnect always starts or queues a refresh in the new epoch, even if an old refresh is still running. +- Different Sessions refresh independently. + +This prevents an old server response from winning after reconnect. + +## Ordering + +Visible ordering is derived rather than encoded by array mutation: + +1. Projected history in server order. +2. Promoted placeholders in promotion order. +3. Pending compaction barriers. +4. Pending steering inputs in `admittedSeq` order. +5. Pending queued inputs in `admittedSeq` order. + +This anticipates normal runner eligibility. Steers are consumed before queues, while compaction blocks both. A legitimate queue/steer interleaving selected by the runner is represented by durable promotion order once promoted. + +Compaction remains on its existing message lifecycle in the first implementation unless pending compaction coverage can be added without conflating its different event model with ordinary input admission. + +## Solid Identity + +The visible timeline feeds a row reducer whose output has stable IDs. Rendering reconciles those rows by ID before passing them to ``: + +```tsx +setRows(reconcile(reduce(), { key: "id" })) + + + {(row) => } + +``` + +The existing row reducer expands messages into message, part, exploration-group, and footer rows. Every `SessionRow` must gain a deterministic `id`: + +```ts +type SessionRow = + | { id: `message:${string}`; type: "message"; messageID: string } + | { id: `part:${string}:${string}`; type: "part"; ref: PartRef } + | { id: `assistant-footer:${string}`; type: "assistant-footer"; messageID: string } + | { id: `group:exploration:${string}`; type: "group"; /* ... */ } +``` + +Rows are installed through `reconcile(rows, { key: "id" })`. Exploration group identity is seeded from the first part that creates the group and remains fixed as refs move between pending and completed partitions. + +The outer OpenTUI row wrapper receives the same stable row ID so ownership and scroll anchoring are observable in tests. + +## Scroll Policy + +Stable identity prevents remounting but not geometric scroll movement. + +- Existing `stickyScroll` and `stickyStart="bottom"` continue to own bottom-following behavior. +- Ordinary admission/promotion updates should not run explicit anchoring. +- Explicit anchoring is deferred unless a deterministic regression shows hydration changes the viewed content while manually scrolled up. +- If required, capture the first visible stable row ID and viewport offset before snapshot installation, then restore its offset after OpenTUI layout. + +## Module Boundary + +Place overlay conversion, ordering, lifecycle reduction, journaling, and refresh coordination in a focused module: + +```text +context/data.tsx + projected/live message reducers + event routing + hydration and refresh coordination + | + v +context/session-timeline.ts + pending adapters + lifecycle overlay operations + visible timeline composition + deterministic ordering +``` + +Do not move general message-event reduction into this module. + +## Implementation Sequence + +1. Add pure lifecycle adapter, reducer, and ordering functions with deterministic unit tests. +2. Add the overlay state and visible timeline accessors behind the existing data context. +3. Change admission and promotion handlers to mutate the overlay while preserving existing projected/live behavior where required during migration. +4. Implement sequential hydration, the narrow journal, and same-epoch refresh coalescing. +5. Change `createSessionRows` and `SessionRowView` lookup to consume the visible timeline. +6. Add stable IDs to all `SessionRow` variants and keyed row reconciliation. +7. Migrate only pending-sensitive route logic; leave projected-only consumers on `session.message`. +8. Remove the old pending-ID reconciliation helper and mixed pending insertion from projected hydration. +9. Run focused data/row tests, TUI typecheck, and the full TUI suite. + +## Required Regressions + +- Pending input known only through the pending endpoint. +- Promotion before pending read. +- Promotion between pending and projected reads. +- Promotion after projected read. +- Admission after pending read. +- Admission and promotion during hydration. +- Duplicate or inverted admission/promotion delivery. +- Projected representation wins over the same pending/promoted ID. +- Streaming text delta during hydration is applied exactly once. +- Concurrent same-Session refresh callers join. +- Different Sessions refresh concurrently. +- Old-epoch refresh cannot install after reconnect. +- Failed pending or projected read does not erase visible state. +- Pending-to-promoted-to-projected retains one stable row ID. +- Unaffected neighboring rows retain identity across full row reduction. +- Steer and queue ordering follows the selected policy. + +## Explicit Non-Goals For This PR + +- Replay general Session events. +- Redesign the 200-message pagination behavior. +- Change server APIs or event schemas. +- Introduce Effect into the TUI data layer. +- Add scroll anchoring without a failing regression. +- Migrate undo, fork, transcript, usage, and other projected-history consumers unless existing behavior requires pending visibility. diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 672104863b80..3587eac7c168 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -21,7 +21,6 @@ import type { SessionMessageAssistantText, SessionMessageAssistantTool, SessionInfo, - SessionPendingInfo, Shell, SkillInfo, } from "@opencode-ai/sdk/v2" @@ -29,7 +28,15 @@ import type { OpenCodeEvent } from "@opencode-ai/client/promise" import { createStore, produce, reconcile } from "solid-js/store" import { createSimpleContext } from "./helper" import { useSDK } from "./sdk" -import { createSignal, onCleanup } from "solid-js" +import { batch, createSignal, onCleanup } from "solid-js" +import { + applyTimelineOperations, + fromAdmission, + fromPending, + visibleMessages, + type SessionTimelineInput, + type SessionTimelineOperation, +} from "./session-timeline" export type DataSessionStatus = "idle" | "running" @@ -63,7 +70,7 @@ type Data = { family: Record status: Record message: Record - input: Record + input: Record compaction: Record permission: Record // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. @@ -83,16 +90,6 @@ function locationQuery(ref?: LocationRef) { return ref ? { directory: ref.directory, workspace: ref.workspaceID } : undefined } -// Apply local admissions and promotions that arrived while the server snapshot was loading. -function reconcilePendingInputs(baseline: string[], current: string[], server: string[]) { - const result = new Set(server) - const before = new Set(baseline) - const now = new Set(current) - baseline.filter((id) => !now.has(id)).forEach((id) => result.delete(id)) - current.filter((id) => !before.has(id)).forEach((id) => result.add(id)) - return [...result] -} - export const { use: useData, provider: DataProvider } = createSimpleContext({ name: "Data", init: () => { @@ -118,8 +115,15 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ directory: process.cwd(), }) const messageIndex = new Map>() + const timelineHydrated = new Set() + const sessionGenerations = new Map() + const timelineRefreshes = new Map< + string, + { epoch: number; generation: number; operations: SessionTimelineOperation[]; promise: Promise } + >() let bootstrapping: Promise | undefined let connected = false + let connectionEpoch = 0 function setSessionStatus(sessionID: string, status: DataSessionStatus) { setStore("session", "status", sessionID, status) @@ -140,6 +144,28 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ ) } + function applyTimelineOperation(sessionID: string, operation: SessionTimelineOperation) { + const refresh = timelineRefreshes.get(sessionID) + if ( + refresh?.epoch === connectionEpoch && + refresh.generation === (sessionGenerations.get(sessionID) ?? 0) + ) + refresh.operations.push(operation) + setStore( + "session", + "input", + sessionID, + reconcile( + applyTimelineOperations( + store.session.input[sessionID] ?? [], + [operation], + new Set((store.session.message[sessionID] ?? []).map((message) => message.id)), + ), + { key: "id" }, + ), + ) + } + const message = { update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { setStore( @@ -172,31 +198,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const item = messages.findLast((item) => item.type === "compaction" && item.status === "running") return item?.type === "compaction" ? item : undefined }, - fromPending(item: SessionPendingInfo): SessionMessageInfo { - if (item.type === "user") - return { - id: item.id, - type: "user", - ...item.data, - time: { created: item.timeCreated }, - } - if (item.type === "synthetic") - return { - id: item.id, - type: "synthetic", - ...item.data, - time: { created: item.timeCreated }, - } - return { - id: item.id, - type: "compaction", - status: "running", - reason: "manual", - summary: "", - recent: "", - time: { created: item.timeCreated }, - } - }, latestTool(assistant: SessionMessageAssistant | undefined, callID?: string) { return assistant?.content.findLast( (item): item is SessionMessageAssistantTool => @@ -266,7 +267,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function removeSession(sessionID: string) { + sessionGenerations.set(sessionID, (sessionGenerations.get(sessionID) ?? 0) + 1) messageIndex.delete(sessionID) + timelineHydrated.delete(sessionID) setStore( "session", produce((draft) => { @@ -362,51 +365,45 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } break case "session.input.promoted": { - message.update(event.data.sessionID, (draft, index) => { - const position = index.get(event.data.inputID) - if (position === undefined) return - const existing = draft[position] - if (!existing || !store.session.input[event.data.sessionID]?.includes(event.data.inputID)) return - existing.time.created = event.created - draft.splice(position, 1) - draft.push(existing) - index.clear() - draft.forEach((message, indexValue) => index.set(message.id, indexValue)) + applyTimelineOperation(event.data.sessionID, { + type: "promoted", + inputID: event.data.inputID, + promotedSeq: event.durable.seq, + created: event.created, }) - setStore( - "session", - "input", - event.data.sessionID, - (store.session.input[event.data.sessionID] ?? []).filter((id) => id !== event.data.inputID), - ) + const epoch = connectionEpoch + const generation = sessionGenerations.get(event.data.sessionID) ?? 0 + void sdk.api.session + .message({ sessionID: event.data.sessionID, messageID: event.data.inputID }) + .then((item) => { + if (epoch !== connectionEpoch || generation !== (sessionGenerations.get(event.data.sessionID) ?? 0)) return + 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 + }) + setStore( + "session", + "input", + event.data.sessionID, + (inputs) => inputs?.filter((input) => input.id !== item.id) ?? [], + ) + }) + .catch(() => undefined) break } - case "session.input.admitted": - if (!store.session.input[event.data.sessionID]?.includes(event.data.inputID)) - setStore("session", "input", event.data.sessionID, [ - ...(store.session.input[event.data.sessionID] ?? []), - event.data.inputID, - ]) - message.update(event.data.sessionID, (draft, index) => { - message.append( - draft, - index, - event.data.input.type === "user" - ? { - id: event.data.inputID, - type: "user", - ...event.data.input.data, - time: { created: event.created }, - } - : { - id: event.data.inputID, - type: "synthetic", - ...event.data.input.data, - time: { created: event.created }, - }, - ) + case "session.input.admitted": { + applyTimelineOperation(event.data.sessionID, { + type: "admitted", + input: fromAdmission({ + id: event.data.inputID, + admittedSeq: event.durable.seq, + timeCreated: event.created, + input: event.data.input, + }), }) break + } case "session.instructions.updated": message.update(event.data.sessionID, (draft, index) => { message.append(draft, index, { @@ -699,15 +696,14 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setStore("session", "info", event.data.sessionID, "revert", undefined) break case "session.revert.committed": + sessionGenerations.set( + event.data.sessionID, + (sessionGenerations.get(event.data.sessionID) ?? 0) + 1, + ) 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), - ) + applyTimelineOperation(event.data.sessionID, { type: "reverted", to: event.data.to }) message.update(event.data.sessionID, (draft, index) => { const position = draft.findIndex((item) => item.id >= event.data.to) if (position === -1) return @@ -842,6 +838,74 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } } + function refreshTimeline(sessionID: string): Promise { + const existing = timelineRefreshes.get(sessionID) + const generation = sessionGenerations.get(sessionID) ?? 0 + if (existing?.epoch === connectionEpoch && existing.generation === generation) return existing.promise + + const epoch = connectionEpoch + const operations: SessionTimelineOperation[] = [] + const baseline = new Set( + timelineHydrated.has(sessionID) ? (store.session.message[sessionID] ?? []).map((message) => message.id) : [], + ) + const promise = (async () => { + const pending = await sdk.api.session.pending.list({ sessionID }) + if (generation !== (sessionGenerations.get(sessionID) ?? 0)) return + if (epoch !== connectionEpoch) return refreshTimeline(sessionID) + const projected = await sdk.api.message.list({ sessionID, limit: 200, order: "desc" }) + if (generation !== (sessionGenerations.get(sessionID) ?? 0)) return + if (epoch !== connectionEpoch) return refreshTimeline(sessionID) + + const messages: SessionMessageInfo[] = projected.data.toReversed() + const messagePositions = new Map(messages.map((message, index) => [message.id, index])) + store.session.message[sessionID] + ?.filter((message) => !baseline.has(message.id)) + .forEach((message) => { + const position = messagePositions.get(message.id) + if (position === undefined) { + messagePositions.set(message.id, messages.length) + messages.push(message) + return + } + messages[position] = message + }) + const inputs = applyTimelineOperations( + pending.filter((input) => input.type !== "compaction").map(fromPending), + operations, + new Set(messages.map((message) => message.id)), + ) + batch(() => { + messageIndex.set(sessionID, messagePositions) + setStore("session", "message", sessionID, reconcile(messages, { key: "id" })) + setStore("session", "input", sessionID, reconcile(inputs, { key: "id" })) + setStore( + "session", + "compaction", + sessionID, + reconcile(pending.filter((input) => input.type === "compaction").map((input) => input.id)), + ) + timelineHydrated.add(sessionID) + }) + })() + timelineRefreshes.set(sessionID, { epoch, generation, operations, promise }) + const cleanup = () => { + if (timelineRefreshes.get(sessionID)?.promise === promise) timelineRefreshes.delete(sessionID) + } + void promise.then(cleanup, cleanup) + return promise + } + + function timelineList(sessionID: string) { + return visibleMessages(store.session.message[sessionID] ?? [], store.session.input[sessionID] ?? []) + } + + function timelineGet(sessionID: string, messageID: string) { + const messages = store.session.message[sessionID] + const position = messageIndex.get(sessionID)?.get(messageID) + if (position !== undefined) return messages?.[position] + return store.session.input[sessionID]?.find((input) => input.id === messageID)?.message + } + const result = { on: sdk.event.on, listen: sdk.event.listen, @@ -872,10 +936,12 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }, input: { list(sessionID: string) { - return store.session.input[sessionID] ?? [] + return (store.session.input[sessionID] ?? []) + .filter((input) => input.phase === "pending" && input.message?.type !== "compaction") + .map((input) => input.id) }, has(sessionID: string, inputID: string) { - return store.session.input[sessionID]?.includes(inputID) ?? false + return store.session.input[sessionID]?.some((input) => input.id === inputID && input.phase === "pending") ?? false }, }, compaction: { @@ -912,34 +978,16 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const position = messageIndex.get(sessionID)?.get(messageID) return position === undefined ? undefined : messages?.[position] }, - async refresh(sessionID: string) { - const localInputs = store.session.input[sessionID] ?? [] - const [projected, pending] = await Promise.all([ - sdk.api.message.list({ sessionID, limit: 200, order: "desc" }), - sdk.api.session.pending.list({ sessionID }), - ]) - const pendingMessages = pending.map(message.fromPending) - const next = projected.data.toReversed() - const index = new Map(next.map((message, index) => [message.id, index])) - const localInputIDs = new Set(localInputs) - store.session.message[sessionID] - ?.filter((message) => localInputIDs.has(message.id)) - .forEach((item) => message.append(next, index, item)) - pendingMessages.forEach((item) => message.append(next, index, item)) - messageIndex.set(sessionID, index) - setStore( - "session", - "input", - sessionID, - reconcilePendingInputs( - localInputs, - store.session.input[sessionID] ?? [], - pendingMessages.flatMap((message) => - message.type === "user" || message.type === "synthetic" ? [message.id] : [], - ), - ), - ) - setStore("session", "message", sessionID, reconcile(next)) + refresh(sessionID: string) { + return refreshTimeline(sessionID) + }, + }, + timeline: { + list(sessionID: string) { + return timelineList(sessionID) + }, + get(sessionID: string, messageID: string) { + return timelineGet(sessionID, messageID) }, }, permission: { @@ -1196,15 +1244,18 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ onCleanup( sdk.event.listen(({ details }) => { if (details.type === "server.connected") { - const messages = connected ? Object.keys(store.session.message) : [] - const compactions = connected ? Object.keys(store.session.compaction) : [] + connectionEpoch += 1 + const sessions = connected + ? new Set([ + ...Object.keys(store.session.message), + ...Object.keys(store.session.input), + ...Object.keys(store.session.compaction), + ...timelineRefreshes.keys(), + ]) + : [] connected = true refreshActive() - void Promise.allSettled([ - bootstrap(), - ...messages.map(result.session.message.refresh), - ...compactions.map(result.session.compaction.refresh), - ]) + void Promise.allSettled([bootstrap(), ...Array.from(sessions).map(result.session.message.refresh)]) return } handleEvent(details) diff --git a/packages/tui/src/context/session-timeline.ts b/packages/tui/src/context/session-timeline.ts new file mode 100644 index 000000000000..83f236bdf67b --- /dev/null +++ b/packages/tui/src/context/session-timeline.ts @@ -0,0 +1,150 @@ +import type { SessionMessageInfo, SessionPendingInfo } from "@opencode-ai/sdk/v2" + +export type SessionTimelineInput = { + id: string + phase: "pending" | "promoted" + admittedSeq: number + promotedSeq?: number + delivery?: "steer" | "queue" + message?: SessionMessageInfo +} + +export type SessionTimelineOperation = + | { type: "admitted"; input: SessionTimelineInput } + | { type: "promoted"; inputID: string; promotedSeq: number; created: number } + | { type: "reverted"; to: string } + +export function fromAdmission(input: { + id: string + admittedSeq: number + timeCreated: number + input: + | { type: "user"; data: Extract["data"]; delivery: "steer" | "queue" } + | { + type: "synthetic" + data: Extract["data"] + delivery: "steer" | "queue" + } +}): SessionTimelineInput { + return { + id: input.id, + phase: "pending", + admittedSeq: input.admittedSeq, + delivery: input.input.delivery, + message: { + id: input.id, + type: input.input.type, + ...input.input.data, + time: { created: input.timeCreated }, + }, + } +} + +export function fromPending(item: SessionPendingInfo): SessionTimelineInput { + if (item.type === "user") + return { + id: item.id, + phase: "pending", + admittedSeq: item.admittedSeq, + delivery: item.delivery, + message: { + id: item.id, + type: "user", + ...item.data, + time: { created: item.timeCreated }, + }, + } + if (item.type === "synthetic") + return { + id: item.id, + phase: "pending", + admittedSeq: item.admittedSeq, + delivery: item.delivery, + message: { + id: item.id, + type: "synthetic", + ...item.data, + time: { created: item.timeCreated }, + }, + } + return { + id: item.id, + phase: "pending", + admittedSeq: item.admittedSeq, + message: { + id: item.id, + type: "compaction", + status: "running", + reason: "manual", + summary: "", + recent: "", + time: { created: item.timeCreated }, + }, + } +} + +export function applyTimelineOperations( + inputs: SessionTimelineInput[], + operations: SessionTimelineOperation[], + projectedIDs = new Set(), +) { + const result = new Map(inputs.filter((input) => !projectedIDs.has(input.id)).map((input) => [input.id, input])) + operations.forEach((operation) => { + if (operation.type === "reverted") { + result.forEach((_, id) => { + if (id >= operation.to) result.delete(id) + }) + return + } + if (projectedIDs.has(operation.type === "admitted" ? operation.input.id : operation.inputID)) return + if (operation.type === "admitted") { + const existing = result.get(operation.input.id) + result.set( + operation.input.id, + existing?.phase === "promoted" + ? { ...operation.input, phase: "promoted", promotedSeq: existing.promotedSeq } + : (existing ?? operation.input), + ) + return + } + const existing = result.get(operation.inputID) + result.set(operation.inputID, { + ...existing, + id: operation.inputID, + phase: "promoted", + admittedSeq: existing?.admittedSeq ?? operation.promotedSeq, + promotedSeq: operation.promotedSeq, + message: existing?.message + ? { + ...existing.message, + time: { ...existing.message.time, created: operation.created }, + } + : undefined, + }) + }) + return orderInputs([...result.values()]) +} + +export function visibleMessages(projected: SessionMessageInfo[], inputs: SessionTimelineInput[]) { + if (inputs.length === 0) return projected + const ids = new Set(projected.map((message) => message.id)) + return [ + ...projected, + ...inputs.filter((input) => !ids.has(input.id)).flatMap((input) => (input.message ? [input.message] : [])), + ] +} + +function orderInputs(inputs: SessionTimelineInput[]) { + const bucket = (input: SessionTimelineInput) => { + if (input.phase === "promoted") return 0 + if (input.message?.type === "compaction") return 1 + if (input.delivery === "steer") return 2 + return 3 + } + return inputs.toSorted( + (a, b) => + bucket(a) - bucket(b) || + (a.phase === "promoted" ? (a.promotedSeq ?? a.admittedSeq) - (b.promotedSeq ?? b.admittedSeq) : 0) || + a.admittedSeq - b.admittedSeq, + ) +} diff --git a/packages/tui/src/routes/session/dialog-fork.tsx b/packages/tui/src/routes/session/dialog-fork.tsx index ed2a5c7f4fdf..07ba5abbf329 100644 --- a/packages/tui/src/routes/session/dialog-fork.tsx +++ b/packages/tui/src/routes/session/dialog-fork.tsx @@ -25,7 +25,7 @@ export function DialogFork(props: { sessionID: string; messageID?: string; onMov return undefined }) if (!result) return dialog.clear() - const message = messageID ? data.session.message.get(props.sessionID, messageID) : undefined + const message = messageID ? data.session.timeline.get(props.sessionID, messageID) : undefined route.navigate({ sessionID: result.id, type: "session", @@ -59,7 +59,7 @@ export function DialogFork(props: { sessionID: string; messageID?: string; onMov value: undefined, onSelect: () => fork(), }, - ...data.session.message + ...data.session.timeline .list(props.sessionID) .filter((message) => message.type === "user") .toReversed() diff --git a/packages/tui/src/routes/session/dialog-message.tsx b/packages/tui/src/routes/session/dialog-message.tsx index db64c7b5da28..4b85d95b5145 100644 --- a/packages/tui/src/routes/session/dialog-message.tsx +++ b/packages/tui/src/routes/session/dialog-message.tsx @@ -12,7 +12,7 @@ export function DialogMessage(props: { messageID: string; sessionID: string }) { const clipboard = useClipboard() const toast = useToast() const sdk = useSDK() - const message = createMemo(() => data.session.message.get(props.sessionID, props.messageID)) + const message = createMemo(() => data.session.timeline.get(props.sessionID, props.messageID)) return ( data.session.get(route.sessionID)) - const messageIDs = createMemo(() => data.session.message.ids(route.sessionID)) - const sessionMessages = () => - messageIDs().flatMap((id) => { - const message = data.session.message.get(route.sessionID, id) - return message ? [message] : [] - }) + const sessionMessages = createMemo(() => data.session.timeline.list(route.sessionID)) const location = createMemo(() => session()?.location) createEffect(() => { @@ -962,7 +957,7 @@ export function Session() { {(row) => ( data.session.message.get(route.sessionID, messageID)} + message={(messageID) => data.session.timeline.get(route.sessionID, messageID)} /> )} @@ -1053,7 +1048,7 @@ export function Session() { function SessionRowView(props: { row: SessionRow; message: (messageID: string) => SessionMessageInfo | undefined }) { return ( - + {(row) => ( @@ -1516,7 +1511,6 @@ function UserMessage(props: { message: SessionMessageUser }) { return ( ) { const data = useData() @@ -27,7 +28,7 @@ export function createSessionRows(sessionID: Accessor) { const revertBoundary = () => data.session.get(sessionID())?.revert?.messageID function reduce() { - const messages = data.session.message.list(sessionID()) + const messages = data.session.timeline.list(sessionID()) const inputs = new Set(data.session.input.list(sessionID())) const boundary = revertBoundary() const rows = reduceSessionRows(boundary ? messages.filter((message) => message.id < boundary) : messages, inputs) @@ -38,7 +39,7 @@ export function createSessionRows(sessionID: Accessor) { 0, ...data.session.compaction .list(sessionID()) - .map((inputID): SessionRow => ({ type: "compaction-queued", inputID })), + .map((inputID): SessionRow => ({ id: `compaction-queued:${inputID}`, type: "compaction-queued", inputID })), ) return rows } @@ -62,12 +63,11 @@ export function createSessionRows(sessionID: Accessor) { createEffect( on(sessionID, (id) => { - setRows(reconcile(reduce())) - void data.session.compaction.refresh(id).catch(() => undefined) + setRows(reconcile(reduce(), { key: "id" })) void data.session.message.refresh(id).then( () => { if (sessionID() !== id) return - setRows(reconcile(reduce())) + setRows(reconcile(reduce(), { key: "id" })) }, () => undefined, ) @@ -77,21 +77,21 @@ export function createSessionRows(sessionID: Accessor) { // Re-reduce when the revert boundary changes (stage/clear/commit). createEffect( on(revertBoundary, () => { - setRows(reconcile(reduce())) + setRows(reconcile(reduce(), { key: "id" })) }), ) createEffect( on( () => data.session.compaction.list(sessionID()).map((inputID) => inputID), - () => setRows(reconcile(reduce())), + () => setRows(reconcile(reduce(), { key: "id" })), ), ) createEffect( on( () => - data.session.message.list(sessionID()).flatMap((message) => + data.session.timeline.list(sessionID()).flatMap((message) => message.type === "user" || message.type === "synthetic" ? [ { @@ -109,7 +109,7 @@ export function createSessionRows(sessionID: Accessor) { ] : [], ), - () => setRows(reconcile(reduce())), + () => setRows(reconcile(reduce(), { key: "id" })), ), ) @@ -118,11 +118,11 @@ export function createSessionRows(sessionID: Accessor) { produce((draft) => { if (draft.some((row) => row.type === "message" && row.messageID === messageID)) return const pending = isPending(messageID) - const message = data.session.message.get(sessionID(), messageID) + const message = data.session.timeline.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 }) + draft.splice(index, 0, { id: messageID, type: "message", messageID }) }), ) @@ -139,6 +139,7 @@ export function createSessionRows(sessionID: Accessor) { } completePrevious(draft, index) draft.splice(index, 0, { + id: rowID(ref), type: "group", kind: "exploration", refs: [ref], @@ -148,7 +149,7 @@ export function createSessionRows(sessionID: Accessor) { return } completePrevious(draft, index) - draft.splice(index, 0, { type: "part", ref }) + draft.splice(index, 0, { id: rowID(ref), type: "part", ref }) }), ) @@ -158,7 +159,7 @@ export function createSessionRows(sessionID: Accessor) { if (draft.some((row) => row.type === "assistant-footer" && row.messageID === messageID)) return const index = queuedStart(draft) completePrevious(draft, index) - draft.splice(index, 0, { type: "assistant-footer", messageID }) + draft.splice(index, 0, { id: `assistant-footer:${messageID}`, type: "assistant-footer", messageID }) }), ) @@ -171,7 +172,7 @@ export function createSessionRows(sessionID: Accessor) { ) const isPending = (messageID: string) => { - const message = data.session.message.get(sessionID(), messageID) + const message = data.session.timeline.get(sessionID(), messageID) if (message?.type === "user" || message?.type === "synthetic") return data.session.input.has(sessionID(), messageID) return message?.type === "compaction" && message.status === "running" } @@ -261,7 +262,7 @@ export function reduceSessionRows(messages: SessionMessageInfo[], inputs = new S if (message.type !== "assistant") { if (message.type === "synthetic" && !message.description?.trim()) return rows if (!pending.has(message.id)) completePrevious(rows) - rows.push({ type: "message", messageID: message.id }) + rows.push({ id: message.id, type: "message", messageID: message.id }) return rows } const ordinals = { text: 0, reasoning: 0 } @@ -272,7 +273,7 @@ export function reduceSessionRows(messages: SessionMessageInfo[], inputs = new S }) if ((message.finish && !["tool-calls", "unknown"].includes(message.finish)) || message.error || message.retry) { completePrevious(rows) - rows.push({ type: "assistant-footer", messageID: message.id }) + rows.push({ id: `assistant-footer:${message.id}`, type: "assistant-footer", messageID: message.id }) } return rows }, []) @@ -296,12 +297,16 @@ function append(rows: SessionRow[], ref: PartRef, part: SessionMessageAssistant[ return } completePrevious(rows) - rows.push({ type: "group", kind: "exploration", refs: [ref], pending: [], completed: false }) + rows.push({ id: rowID(ref), type: "group", kind: "exploration", refs: [ref], pending: [], completed: false }) return } } completePrevious(rows) - rows.push({ type: "part", ref }) + rows.push({ id: rowID(ref), type: "part", ref }) +} + +function rowID(ref: PartRef) { + return `part:${ref.messageID}:${ref.partID}` } function completePrevious(rows: SessionRow[], index = rows.length) { diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 33c945736ad1..424f2063f457 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -647,7 +647,7 @@ test("completes exploration when a queued prompt is promoted", async () => { data: { sessionID, inputID: "message-user" }, }) await wait(() => rows.find((row) => row.type === "group")?.completed === true) - expect(rows.at(-1)).toEqual({ type: "message", messageID: "message-user" }) + expect(rows.at(-1)).toMatchObject({ id: "message-user", type: "message", messageID: "message-user" }) } finally { app.renderer.destroy() } @@ -656,8 +656,27 @@ test("completes exploration when a queued prompt is promoted", async () => { test("removes committed revert messages from local state", async () => { const events = createEventStream() const sessionID = "session-revert" + const messageResponse = Promise.withResolvers() + const messageRequested = Promise.withResolvers() + let hydrating = false const calls = createFetch((url) => { - if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + if (url.pathname === `/api/session/${sessionID}/pending` && hydrating) + return json({ + data: ["msg_001", "msg_002", "msg_003"].map((id, admittedSeq) => ({ + admittedSeq, + id, + sessionID, + timeCreated: admittedSeq, + type: "user", + data: { text: id }, + delivery: "steer", + })), + }) + if (url.pathname === `/api/session/${sessionID}/message`) { + if (!hydrating) return json({ data: [], cursor: {} }) + messageRequested.resolve() + return messageResponse.promise + } }, events) let data!: ReturnType @@ -688,7 +707,10 @@ test("removes committed revert messages from local state", async () => { data: { sessionID, inputID, input: { type: "user", data: { text: inputID }, delivery: "steer" } }, }) } - await wait(() => data.session.message.ids(sessionID).length === 3) + await wait(() => data.session.timeline.list(sessionID).length === 3) + hydrating = true + const refresh = data.session.message.refresh(sessionID) + await messageRequested.promise emitEvent(events, { id: EventV2.ID.create(), @@ -697,11 +719,21 @@ test("removes committed revert messages from local state", async () => { durable: durable(sessionID, 3), data: { sessionID, to: "msg_002" }, }) + messageResponse.resolve( + json({ + data: [ + { id: "msg_003", type: "user", text: "msg_003", time: { created: 2 } }, + { id: "msg_002", type: "user", text: "msg_002", time: { created: 1 } }, + ], + cursor: {}, + }), + ) + await refresh - await wait(() => data.session.message.ids(sessionID).length === 1) - expect(data.session.message.ids(sessionID)).toEqual(["msg_001"]) - expect(data.session.message.get(sessionID, "msg_002")).toBeUndefined() - expect(data.session.message.get(sessionID, "msg_003")).toBeUndefined() + await wait(() => data.session.timeline.list(sessionID).length === 1) + expect(data.session.timeline.list(sessionID).map((message) => message.id)).toEqual(["msg_001"]) + expect(data.session.timeline.get(sessionID, "msg_002")).toBeUndefined() + expect(data.session.timeline.get(sessionID, "msg_003")).toBeUndefined() } finally { app.renderer.destroy() } @@ -1088,7 +1120,7 @@ test("tracks session status from active sessions and execution events", async () return message?.type === "compaction" && message.status === "completed" }) expect(manualRows.filter((row) => row.type === "message")).toEqual([ - { type: "message", messageID: "message-compaction" }, + { id: "message-compaction", type: "message", messageID: "message-compaction" }, ]) expect(manualRows.find((row) => row.type === "message" && row.messageID === "message-compaction")).toBe( compactionRow, @@ -1198,8 +1230,16 @@ test("restores queued compaction from durable pending input", async () => { ]) await wait(() => rows.filter((row) => row.type === "compaction-queued").length === 2) expect(rows.filter((row) => row.type === "compaction-queued")).toEqual([ - { type: "compaction-queued", inputID: "message-compaction-queued" }, - { type: "compaction-queued", inputID: "message-compaction-later" }, + { + id: "compaction-queued:message-compaction-queued", + type: "compaction-queued", + inputID: "message-compaction-queued", + }, + { + id: "compaction-queued:message-compaction-later", + type: "compaction-queued", + inputID: "message-compaction-later", + }, ]) emitEvent(events, { @@ -2195,8 +2235,8 @@ test("preserves admitted prompts when hydration races with promotion", async () input: { type: "user", data: { text: "hello" }, delivery: "steer" }, }, }) - await wait(() => sync.session.message.list(sessionID)?.length === 1) - const admitted = sync.session.message.list(sessionID)?.[0] + await wait(() => sync.session.timeline.list(sessionID).length === 1) + const admitted = sync.session.timeline.list(sessionID)[0] expect(admitted).toMatchObject({ id: messageID, type: "user", text: "hello" }) expect(admitted?.metadata).toBeUndefined() expect(sync.session.input.list(sessionID)).toEqual([messageID]) @@ -2220,23 +2260,165 @@ test("preserves admitted prompts when hydration races with promotion", async () response.resolve(json({ data: [], cursor: {} })) await refresh - const message = sync.session.message.get(sessionID, messageID) + const message = sync.session.timeline.get(sessionID, messageID) expect(message?.type).toBe("user") if (message?.type !== "user") return expect(message).toMatchObject({ id: messageID, text: "hello" }) expect(message.metadata).toBeUndefined() expect(sync.session.input.list(sessionID)).toEqual([queuedID]) - expect(sync.session.message.ids(sessionID)).toEqual([messageID, queuedID]) - expect(sync.session.message.get(sessionID, queuedID)).toMatchObject({ id: queuedID, text: "queued" }) - expect(sync.session.message.ids("missing")).toEqual([]) - expect(sync.session.message.get(sessionID, messageID)).toBe(message) - expect(sync.session.message.get(sessionID, "missing")).toBeUndefined() + expect(sync.session.timeline.list(sessionID).map((message) => message.id)).toEqual([messageID, queuedID]) + expect(sync.session.timeline.get(sessionID, queuedID)).toMatchObject({ id: queuedID, text: "queued" }) + expect(sync.session.timeline.list("missing")).toEqual([]) + expect(sync.session.timeline.get(sessionID, messageID)).toBe(message) + expect(sync.session.timeline.get(sessionID, "missing")).toBeUndefined() expect(received).toHaveLength(3) } finally { app.renderer.destroy() } }) +test("journals admissions after the pending snapshot and joins concurrent refreshes", async () => { + const events = createEventStream() + const sessionID = "session-journal" + const messageID = "msg_late" + const pendingResponse = Promise.withResolvers() + const messageResponse = Promise.withResolvers() + const messageRequested = Promise.withResolvers() + let pendingCalls = 0 + let messageCalls = 0 + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/pending`) { + pendingCalls += 1 + return pendingResponse.promise + } + if (url.pathname === `/api/session/${sessionID}/message`) { + messageCalls += 1 + messageRequested.resolve() + return messageResponse.promise + } + }, events) + let data!: ReturnType + let sdk!: ReturnType + + function Probe() { + data = useData() + sdk = useSDK() + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await wait(() => sdk.connection.status() === "connected") + const first = data.session.message.refresh(sessionID) + const second = data.session.message.refresh(sessionID) + expect(second).toBe(first) + expect(pendingCalls).toBe(1) + expect(messageCalls).toBe(0) + + pendingResponse.resolve(json({ data: [] })) + await messageRequested.promise + emitEvent(events, { + id: "evt_late_admission", + created: 1, + type: "session.input.admitted", + durable: durable(sessionID), + data: { + sessionID, + inputID: messageID, + input: { type: "user", data: { text: "late" }, delivery: "steer" }, + }, + }) + messageResponse.resolve(json({ data: [], cursor: {} })) + await Promise.all([first, second]) + + expect(messageCalls).toBe(1) + expect(data.session.timeline.list(sessionID).map((message) => message.id)).toEqual([messageID]) + expect(data.session.timeline.get(sessionID, messageID)).toMatchObject({ text: "late" }) + expect(data.session.input.list(sessionID)).toEqual([messageID]) + } finally { + app.renderer.destroy() + } +}) + +test("refreshes overlay-only sessions after reconnect", async () => { + const events = createEventStream() + const sessionID = "session-overlay" + const messageID = "msg_overlay" + let pendingCalls = 0 + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/pending`) { + pendingCalls += 1 + return json({ + data: [ + { + admittedSeq: 0, + id: messageID, + sessionID, + timeCreated: 0, + type: "user", + data: { text: "overlay" }, + delivery: "steer", + }, + ], + }) + } + if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + }, events) + let data!: ReturnType + let sdk!: ReturnType + + function Probe() { + data = useData() + sdk = useSDK() + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await wait(() => sdk.connection.status() === "connected") + emitEvent(events, { + id: "evt_overlay_admitted", + created: 0, + type: "session.input.admitted", + durable: durable(sessionID), + data: { + sessionID, + inputID: messageID, + input: { type: "user", data: { text: "overlay" }, delivery: "steer" }, + }, + }) + await wait(() => data.session.timeline.list(sessionID).some((message) => message.id === messageID)) + + emitEvent(events, { id: "evt_reconnected", type: "server.connected", data: {} }) + await wait(() => pendingCalls === 1) + expect(data.session.timeline.list(sessionID).map((message) => message.id)).toEqual([messageID]) + } finally { + app.renderer.destroy() + } +}) + test("projects live instruction updates with their message ID", async () => { const events = createEventStream() const calls = createFetch(undefined, events) diff --git a/packages/tui/test/cli/tui/session-rows.test.ts b/packages/tui/test/cli/tui/session-rows.test.ts index c43a6faf9dff..e7f9a2dfe988 100644 --- a/packages/tui/test/cli/tui/session-rows.test.ts +++ b/packages/tui/test/cli/tui/session-rows.test.ts @@ -1,5 +1,6 @@ import { expect, test } from "bun:test" import type { SessionMessageAssistant, SessionMessageInfo } from "@opencode-ai/sdk/v2" +import { createStore, reconcile } from "solid-js/store" import { reduceSessionRows } from "../../../src/routes/session/rows" test("groups exploration parts across assistant messages until a delimiter", () => { @@ -17,9 +18,10 @@ test("groups exploration parts across assistant messages until a delimiter", () ] expect(reduceSessionRows(messages)).toEqual([ - { type: "message", messageID: "user-1" }, - { type: "part", ref: { messageID: "assistant-1", partID: "text:0" } }, + { id: "user-1", type: "message", messageID: "user-1" }, + { id: "part:assistant-1:text:0", type: "part", ref: { messageID: "assistant-1", partID: "text:0" } }, { + id: "part:assistant-1:read-1", type: "group", kind: "exploration", pending: [], @@ -30,7 +32,7 @@ test("groups exploration parts across assistant messages until a delimiter", () { messageID: "assistant-2", partID: "grep-1" }, ], }, - { type: "part", ref: { messageID: "assistant-2", partID: "text:0" } }, + { id: "part:assistant-2:text:0", type: "part", ref: { messageID: "assistant-2", partID: "text:0" } }, ]) }) @@ -45,14 +47,16 @@ test("keeps non-exploration tools as individual part rows", () => { expect(reduceSessionRows(messages)).toEqual([ { + id: "part:assistant-1:read-1", type: "group", kind: "exploration", pending: [], completed: true, refs: [{ messageID: "assistant-1", partID: "read-1" }], }, - { type: "part", ref: { messageID: "assistant-1", partID: "bash-1" } }, + { id: "part:assistant-1:bash-1", type: "part", ref: { messageID: "assistant-1", partID: "bash-1" } }, { + id: "part:assistant-1:grep-1", type: "group", kind: "exploration", pending: [], @@ -73,10 +77,18 @@ test("assigns stable kind ordinals within an assistant message", () => { ] expect(reduceSessionRows(messages)).toEqual([ - { type: "part", ref: { messageID: "assistant-1", partID: "text:0" } }, - { type: "part", ref: { messageID: "assistant-1", partID: "reasoning:0" } }, - { type: "part", ref: { messageID: "assistant-1", partID: "text:1" } }, - { type: "part", ref: { messageID: "assistant-1", partID: "reasoning:1" } }, + { id: "part:assistant-1:text:0", type: "part", ref: { messageID: "assistant-1", partID: "text:0" } }, + { + id: "part:assistant-1:reasoning:0", + type: "part", + ref: { messageID: "assistant-1", partID: "reasoning:0" }, + }, + { id: "part:assistant-1:text:1", type: "part", ref: { messageID: "assistant-1", partID: "text:1" } }, + { + id: "part:assistant-1:reasoning:1", + type: "part", + ref: { messageID: "assistant-1", partID: "reasoning:1" }, + }, ]) }) @@ -93,8 +105,13 @@ test("groups across empty assistant reasoning parts", () => { ] expect(reduceSessionRows(messages)).toEqual([ - { type: "part", ref: { messageID: "assistant-1", partID: "reasoning:0" } }, { + id: "part:assistant-1:reasoning:0", + type: "part", + ref: { messageID: "assistant-1", partID: "reasoning:0" }, + }, + { + id: "part:assistant-1:read-1", type: "group", kind: "exploration", pending: [], @@ -120,21 +137,23 @@ test("completes exploration groups when another row follows", () => { expect(reduceSessionRows(messages)).toEqual([ { + id: "part:assistant-1:read-1", type: "group", kind: "exploration", pending: [], completed: true, refs: [{ messageID: "assistant-1", partID: "read-1" }], }, - { type: "message", messageID: "user-1" }, + { id: "user-1", type: "message", messageID: "user-1" }, { + id: "part:assistant-2:grep-1", type: "group", kind: "exploration", pending: [], completed: true, refs: [{ messageID: "assistant-2", partID: "grep-1" }], }, - { type: "assistant-footer", messageID: "assistant-2" }, + { id: "assistant-footer:assistant-2", type: "assistant-footer", messageID: "assistant-2" }, ]) }) @@ -152,6 +171,7 @@ test("hides synthetic messages without descriptions", () => { expect(reduceSessionRows(messages)).toEqual([ { + id: "part:assistant-1:read-1", type: "group", kind: "exploration", pending: [], @@ -179,14 +199,16 @@ test("renders synthetic messages with descriptions", () => { expect(reduceSessionRows(messages)).toEqual([ { + id: "part:assistant-1:read-1", type: "group", kind: "exploration", pending: [], completed: true, refs: [{ messageID: "assistant-1", partID: "read-1" }], }, - { type: "message", messageID: "synthetic-1" }, + { id: "synthetic-1", type: "message", messageID: "synthetic-1" }, { + id: "part:assistant-2:grep-1", type: "group", kind: "exploration", pending: [], @@ -204,7 +226,9 @@ test("renders a footer for a pre-output retry assistant after replay", () => { error: { type: "provider.transport", message: "Disconnected" }, } - expect(reduceSessionRows([message])).toEqual([{ type: "assistant-footer", messageID: "assistant-retry" }]) + expect(reduceSessionRows([message])).toEqual([ + { id: "assistant-footer:assistant-retry", type: "assistant-footer", messageID: "assistant-retry" }, + ]) }) test("places a running compaction barrier before every queued user message", () => { @@ -229,12 +253,43 @@ test("places a running compaction barrier before every queued user message", () ] expect(reduceSessionRows(messages, new Set(["user-before", "user-after"]))).toEqual([ - { type: "message", messageID: "compaction" }, - { type: "message", messageID: "user-before" }, - { type: "message", messageID: "user-after" }, + { id: "compaction", type: "message", messageID: "compaction" }, + { id: "user-before", type: "message", messageID: "user-before" }, + { id: "user-after", type: "message", messageID: "user-after" }, + ]) +}) + +test("assigns stable IDs to every row", () => { + const message = assistant("assistant", [ + { type: "text", text: "Hello" }, + { type: "tool", id: "read", name: "read", state: pending(), time: { created: 1 } }, + ]) + message.finish = "stop" + + expect(reduceSessionRows([message]).map((row) => row.id)).toEqual([ + "part:assistant:text:0", + "part:assistant:read", + "assistant-footer:assistant", ]) }) +test("keyed reconciliation preserves exploration row ownership", () => { + const first = assistant("assistant", [ + { type: "tool", id: "read", name: "read", state: pending(), time: { created: 1 } }, + ]) + const [rows, setRows] = createStore(reduceSessionRows([first])) + const group = rows[0] + const second = assistant("assistant", [ + { type: "tool", id: "read", name: "read", state: pending(), time: { created: 1 } }, + { type: "tool", id: "glob", name: "glob", state: pending(), time: { created: 2 } }, + ]) + + setRows(reconcile(reduceSessionRows([second]), { key: "id" })) + + expect(rows[0]).toBe(group) + expect(rows[0]?.type === "group" && rows[0].refs).toHaveLength(2) +}) + function assistant(id: string, content: SessionMessageAssistant["content"]): SessionMessageAssistant { return { type: "assistant", diff --git a/packages/tui/test/context/session-timeline.test.ts b/packages/tui/test/context/session-timeline.test.ts new file mode 100644 index 000000000000..44cc09d949b1 --- /dev/null +++ b/packages/tui/test/context/session-timeline.test.ts @@ -0,0 +1,80 @@ +import { expect, test } from "bun:test" +import type { SessionMessageInfo, SessionPendingInfo } from "@opencode-ai/sdk/v2" +import { + applyTimelineOperations, + fromPending, + visibleMessages, + type SessionTimelineInput, +} from "../../src/context/session-timeline" + +test("orders promoted work, compaction, steers, then queued inputs", () => { + const inputs = [ + pending("queue", 0, "queue"), + pending("steer-2", 3, "steer"), + fromPending({ admittedSeq: 2, id: "compaction", sessionID: "session", timeCreated: 2, type: "compaction" }), + pending("steer-1", 1, "steer"), + ] + + const result = applyTimelineOperations(inputs, [ + { type: "promoted", inputID: "steer-2", promotedSeq: 4, created: 4 }, + ]) + + expect(result.map((input) => input.id)).toEqual(["steer-2", "compaction", "steer-1", "queue"]) +}) + +test("replays an admission after the pending snapshot", () => { + const input = pending("late", 2, "steer") + expect(applyTimelineOperations([], [{ type: "admitted", input }])).toEqual([input]) +}) + +test("does not let late admission downgrade a promotion", () => { + const input = pending("input", 1, "steer") + const result = applyTimelineOperations([], [ + { type: "admitted", input }, + { type: "promoted", inputID: input.id, promotedSeq: 2, created: 2 }, + { type: "admitted", input }, + ]) + + expect(result).toMatchObject([{ id: input.id, phase: "promoted", promotedSeq: 2 }]) +}) + +test("retains promotion state until a later admission provides content", () => { + const input = pending("input", 1, "steer") + const result = applyTimelineOperations([], [ + { type: "promoted", inputID: input.id, promotedSeq: 2, created: 2 }, + { type: "admitted", input }, + ]) + + expect(result).toMatchObject([{ id: input.id, phase: "promoted", promotedSeq: 2, message: { text: "input" } }]) +}) + +test("applies committed reverts after a stale pending snapshot", () => { + const inputs = [pending("msg_001", 1, "steer"), pending("msg_002", 2, "queue")] + expect(applyTimelineOperations(inputs, [{ type: "reverted", to: "msg_002" }]).map((input) => input.id)).toEqual([ + "msg_001", + ]) +}) + +test("projected messages replace pending and promoted representations", () => { + const projected: SessionMessageInfo[] = [{ id: "input", type: "user", text: "hello", time: { created: 2 } }] + const inputs = applyTimelineOperations( + [pending("input", 1, "steer")], + [{ type: "promoted", inputID: "input", promotedSeq: 2, created: 2 }], + new Set(["input"]), + ) + + expect(inputs).toEqual([]) + expect(visibleMessages(projected, [pending("input", 1, "steer")])).toEqual(projected) +}) + +function pending(id: string, admittedSeq: number, delivery: "steer" | "queue"): SessionTimelineInput { + return fromPending({ + admittedSeq, + id, + sessionID: "session", + timeCreated: admittedSeq, + type: "user", + data: { text: id }, + delivery, + } satisfies SessionPendingInfo) +} diff --git a/packages/tui/test/fixture/tui-sdk.ts b/packages/tui/test/fixture/tui-sdk.ts index 60a4a432a853..282e33f0caed 100644 --- a/packages/tui/test/fixture/tui-sdk.ts +++ b/packages/tui/test/fixture/tui-sdk.ts @@ -106,6 +106,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType Date: Sat, 11 Jul 2026 14:57:38 -0400 Subject: [PATCH 6/7] refactor(tui): unify pending session work --- packages/tui/TIMELINE_PLAN.md | 30 ++-- packages/tui/src/context/data.tsx | 166 ++++++++---------- packages/tui/src/context/session-timeline.ts | 100 ++++++----- packages/tui/src/routes/session/rows.ts | 8 +- packages/tui/test/cli/tui/data.test.tsx | 23 +-- .../tui/test/context/session-timeline.test.ts | 14 +- 6 files changed, 177 insertions(+), 164 deletions(-) diff --git a/packages/tui/TIMELINE_PLAN.md b/packages/tui/TIMELINE_PLAN.md index 527825e77621..fe0b8301b356 100644 --- a/packages/tui/TIMELINE_PLAN.md +++ b/packages/tui/TIMELINE_PLAN.md @@ -28,23 +28,31 @@ This plan incorporates reviews of Solid identity, hydration concurrency, Session Keep the existing projected/live message cache. Add a lifecycle overlay containing only pending or locally promoted work. ```ts -type TimelineInput = { - id: string - phase: "pending" | "promoted" - admittedSeq: number - delivery?: "steer" | "queue" - message?: SessionMessageInfo -} +type SessionPendingWork = + | { + kind: "message" + id: string + phase: "pending" | "promoted" + admittedSeq: number + delivery: "steer" | "queue" + message?: SessionMessageInfo + } + | { + kind: "compaction" + id: string + phase: "pending" + admittedSeq: number + } type Data = { session: { message: Record - input: Record + pending: Record } } ``` -The overlay keeps pending content, phase, delivery, and ordering metadata together. It replaces the current parallel representation where content is inserted into `session.message` while only its ID is stored in `session.input`. +The overlay keeps every kind of pending Session work together with its phase and ordering metadata. It replaces both the parallel message-plus-ID representation and the separate queued-compaction store. Projected messages remain available through the existing projected cache. A small visible-timeline surface composes projected messages with the overlay for rendering: @@ -99,7 +107,7 @@ Only input topology operations are journaled: ```ts type TimelineOperation = - | { type: "admitted"; input: TimelineInput } + | { type: "admitted"; work: SessionPendingWork } | { type: "promoted"; inputID: string; promotedSeq: number; created: number } | { type: "reverted"; to: string } ``` @@ -144,7 +152,7 @@ Visible ordering is derived rather than encoded by array mutation: This anticipates normal runner eligibility. Steers are consumed before queues, while compaction blocks both. A legitimate queue/steer interleaving selected by the runner is represented by durable promotion order once promoted. -Compaction remains on its existing message lifecycle in the first implementation unless pending compaction coverage can be added without conflating its different event model with ordinary input admission. +Queued compaction remains a distinct overlay variant because it has no message payload or delivery mode. Starting compaction removes that pending variant and projects a running compaction message with the same stable input ID. ## Solid Identity diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 3587eac7c168..4ed0327ee18f 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -33,8 +33,9 @@ import { applyTimelineOperations, fromAdmission, fromPending, + pendingCompactions, visibleMessages, - type SessionTimelineInput, + type SessionPendingWork, type SessionTimelineOperation, } from "./session-timeline" @@ -70,8 +71,7 @@ type Data = { family: Record status: Record message: Record - input: Record - compaction: Record + pending: Record permission: Record // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. form: Record @@ -99,8 +99,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ family: {}, status: {}, message: {}, - input: {}, - compaction: {}, + pending: {}, permission: {}, form: {}, }, @@ -129,21 +128,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setStore("session", "status", sessionID, status) } - function addCompaction(sessionID: string, inputID: string) { - if (store.session.compaction[sessionID]?.includes(inputID)) return - setStore("session", "compaction", sessionID, [...(store.session.compaction[sessionID] ?? []), inputID]) - } - - function removeCompaction(sessionID: string, inputID?: string) { - if (!inputID || !store.session.compaction[sessionID]?.includes(inputID)) return - setStore( - "session", - "compaction", - sessionID, - store.session.compaction[sessionID].filter((id) => id !== inputID), - ) - } - function applyTimelineOperation(sessionID: string, operation: SessionTimelineOperation) { const refresh = timelineRefreshes.get(sessionID) if ( @@ -153,11 +137,11 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ refresh.operations.push(operation) setStore( "session", - "input", + "pending", sessionID, reconcile( applyTimelineOperations( - store.session.input[sessionID] ?? [], + store.session.pending[sessionID] ?? [], [operation], new Set((store.session.message[sessionID] ?? []).map((message) => message.id)), ), @@ -276,8 +260,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ delete draft.info[sessionID] delete draft.status[sessionID] delete draft.message[sessionID] - delete draft.input[sessionID] - delete draft.compaction[sessionID] + delete draft.pending[sessionID] delete draft.permission[sessionID] delete draft.form[sessionID] for (const [rootID, family] of Object.entries(draft.family)) { @@ -384,9 +367,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) setStore( "session", - "input", + "pending", event.data.sessionID, - (inputs) => inputs?.filter((input) => input.id !== item.id) ?? [], + (pending) => pending?.filter((work) => work.id !== item.id) ?? [], ) }) .catch(() => undefined) @@ -395,7 +378,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ case "session.input.admitted": { applyTimelineOperation(event.data.sessionID, { type: "admitted", - input: fromAdmission({ + work: fromAdmission({ id: event.data.inputID, admittedSeq: event.durable.seq, timeCreated: event.created, @@ -662,19 +645,30 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setSessionStatus(event.data.sessionID, "running") break case "session.compaction.admitted": - addCompaction(event.data.sessionID, event.data.inputID) + applyTimelineOperation(event.data.sessionID, { + type: "admitted", + work: { + kind: "compaction", + id: event.data.inputID, + phase: "pending", + admittedSeq: event.durable.seq, + }, + }) break case "session.compaction.started": - removeCompaction(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 }, + batch(() => { + if (event.data.inputID) + applyTimelineOperation(event.data.sessionID, { type: "removed", inputID: 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 @@ -741,27 +735,30 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) break case "session.compaction.failed": - removeCompaction(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) + batch(() => { + if (event.data.inputID) + applyTimelineOperation(event.data.sessionID, { type: "removed", inputID: 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.v2.asked": @@ -869,21 +866,15 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } messages[position] = message }) - const inputs = applyTimelineOperations( - pending.filter((input) => input.type !== "compaction").map(fromPending), + const pendingWork = applyTimelineOperations( + pending.map(fromPending), operations, new Set(messages.map((message) => message.id)), ) batch(() => { messageIndex.set(sessionID, messagePositions) setStore("session", "message", sessionID, reconcile(messages, { key: "id" })) - setStore("session", "input", sessionID, reconcile(inputs, { key: "id" })) - setStore( - "session", - "compaction", - sessionID, - reconcile(pending.filter((input) => input.type === "compaction").map((input) => input.id)), - ) + setStore("session", "pending", sessionID, reconcile(pendingWork, { key: "id" })) timelineHydrated.add(sessionID) }) })() @@ -896,14 +887,15 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function timelineList(sessionID: string) { - return visibleMessages(store.session.message[sessionID] ?? [], store.session.input[sessionID] ?? []) + return visibleMessages(store.session.message[sessionID] ?? [], store.session.pending[sessionID] ?? []) } function timelineGet(sessionID: string, messageID: string) { const messages = store.session.message[sessionID] const position = messageIndex.get(sessionID)?.get(messageID) if (position !== undefined) return messages?.[position] - return store.session.input[sessionID]?.find((input) => input.id === messageID)?.message + const work = store.session.pending[sessionID]?.find((work) => work.id === messageID) + return work?.kind === "message" ? work.message : undefined } const result = { @@ -936,29 +928,15 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }, input: { list(sessionID: string) { - return (store.session.input[sessionID] ?? []) - .filter((input) => input.phase === "pending" && input.message?.type !== "compaction") - .map((input) => input.id) + return (store.session.pending[sessionID] ?? []) + .filter((work) => work.kind === "message" && work.phase === "pending") + .map((work) => work.id) }, has(sessionID: string, inputID: string) { - return store.session.input[sessionID]?.some((input) => input.id === inputID && input.phase === "pending") ?? false - }, - }, - compaction: { - list(sessionID: string) { - return store.session.compaction[sessionID] ?? [] - }, - async refresh(sessionID: string) { - if (!store.session.compaction[sessionID]) setStore("session", "compaction", sessionID, []) - setStore( - "session", - "compaction", - sessionID, - reconcile( - (await sdk.api.session.pending.list({ sessionID })) - .filter((item) => item.type === "compaction") - .map((item) => item.id), - ), + return ( + store.session.pending[sessionID]?.some( + (work) => work.kind === "message" && work.id === inputID && work.phase === "pending", + ) ?? false ) }, }, @@ -989,6 +967,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ get(sessionID: string, messageID: string) { return timelineGet(sessionID, messageID) }, + compactions(sessionID: string) { + return pendingCompactions(store.session.pending[sessionID] ?? []) + }, }, permission: { list(sessionID: string) { @@ -1248,8 +1229,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const sessions = connected ? new Set([ ...Object.keys(store.session.message), - ...Object.keys(store.session.input), - ...Object.keys(store.session.compaction), + ...Object.keys(store.session.pending), ...timelineRefreshes.keys(), ]) : [] diff --git a/packages/tui/src/context/session-timeline.ts b/packages/tui/src/context/session-timeline.ts index 83f236bdf67b..ebace06746a9 100644 --- a/packages/tui/src/context/session-timeline.ts +++ b/packages/tui/src/context/session-timeline.ts @@ -1,17 +1,26 @@ import type { SessionMessageInfo, SessionPendingInfo } from "@opencode-ai/sdk/v2" -export type SessionTimelineInput = { - id: string - phase: "pending" | "promoted" - admittedSeq: number - promotedSeq?: number - delivery?: "steer" | "queue" - message?: SessionMessageInfo -} +export type SessionPendingWork = + | { + kind: "message" + id: string + phase: "pending" | "promoted" + admittedSeq: number + promotedSeq?: number + delivery: "steer" | "queue" + message?: SessionMessageInfo + } + | { + kind: "compaction" + id: string + phase: "pending" + admittedSeq: number + } export type SessionTimelineOperation = - | { type: "admitted"; input: SessionTimelineInput } + | { type: "admitted"; work: SessionPendingWork } | { type: "promoted"; inputID: string; promotedSeq: number; created: number } + | { type: "removed"; inputID: string } | { type: "reverted"; to: string } export function fromAdmission(input: { @@ -25,8 +34,9 @@ export function fromAdmission(input: { data: Extract["data"] delivery: "steer" | "queue" } -}): SessionTimelineInput { +}): SessionPendingWork { return { + kind: "message", id: input.id, phase: "pending", admittedSeq: input.admittedSeq, @@ -40,9 +50,10 @@ export function fromAdmission(input: { } } -export function fromPending(item: SessionPendingInfo): SessionTimelineInput { +export function fromPending(item: SessionPendingInfo): SessionPendingWork { if (item.type === "user") return { + kind: "message", id: item.id, phase: "pending", admittedSeq: item.admittedSeq, @@ -56,6 +67,7 @@ export function fromPending(item: SessionPendingInfo): SessionTimelineInput { } if (item.type === "synthetic") return { + kind: "message", id: item.id, phase: "pending", admittedSeq: item.admittedSeq, @@ -68,27 +80,19 @@ export function fromPending(item: SessionPendingInfo): SessionTimelineInput { }, } return { + kind: "compaction", id: item.id, phase: "pending", admittedSeq: item.admittedSeq, - message: { - id: item.id, - type: "compaction", - status: "running", - reason: "manual", - summary: "", - recent: "", - time: { created: item.timeCreated }, - }, } } export function applyTimelineOperations( - inputs: SessionTimelineInput[], + pending: SessionPendingWork[], operations: SessionTimelineOperation[], projectedIDs = new Set(), ) { - const result = new Map(inputs.filter((input) => !projectedIDs.has(input.id)).map((input) => [input.id, input])) + const result = new Map(pending.filter((work) => !projectedIDs.has(work.id)).map((work) => [work.id, work])) operations.forEach((operation) => { if (operation.type === "reverted") { result.forEach((_, id) => { @@ -96,25 +100,35 @@ export function applyTimelineOperations( }) return } - if (projectedIDs.has(operation.type === "admitted" ? operation.input.id : operation.inputID)) return + if (operation.type === "removed") { + result.delete(operation.inputID) + return + } + if (projectedIDs.has(operation.type === "admitted" ? operation.work.id : operation.inputID)) return if (operation.type === "admitted") { - const existing = result.get(operation.input.id) + const existing = result.get(operation.work.id) + if (operation.work.kind === "compaction") { + result.set(operation.work.id, existing ?? operation.work) + return + } result.set( - operation.input.id, - existing?.phase === "promoted" - ? { ...operation.input, phase: "promoted", promotedSeq: existing.promotedSeq } - : (existing ?? operation.input), + operation.work.id, + existing?.kind === "message" && existing.phase === "promoted" + ? { ...operation.work, phase: "promoted", promotedSeq: existing.promotedSeq } + : (existing ?? operation.work), ) return } const existing = result.get(operation.inputID) result.set(operation.inputID, { - ...existing, + ...(existing?.kind === "message" ? existing : undefined), + kind: "message", id: operation.inputID, phase: "promoted", admittedSeq: existing?.admittedSeq ?? operation.promotedSeq, promotedSeq: operation.promotedSeq, - message: existing?.message + delivery: existing?.kind === "message" ? existing.delivery : "steer", + message: existing?.kind === "message" && existing.message ? { ...existing.message, time: { ...existing.message.time, created: operation.created }, @@ -125,26 +139,32 @@ export function applyTimelineOperations( return orderInputs([...result.values()]) } -export function visibleMessages(projected: SessionMessageInfo[], inputs: SessionTimelineInput[]) { - if (inputs.length === 0) return projected +export function visibleMessages(projected: SessionMessageInfo[], pending: SessionPendingWork[]) { + if (pending.length === 0) return projected const ids = new Set(projected.map((message) => message.id)) return [ ...projected, - ...inputs.filter((input) => !ids.has(input.id)).flatMap((input) => (input.message ? [input.message] : [])), + ...pending.flatMap((work) => (work.kind === "message" && !ids.has(work.id) && work.message ? [work.message] : [])), ] } -function orderInputs(inputs: SessionTimelineInput[]) { - const bucket = (input: SessionTimelineInput) => { - if (input.phase === "promoted") return 0 - if (input.message?.type === "compaction") return 1 - if (input.delivery === "steer") return 2 +export function pendingCompactions(pending: SessionPendingWork[]) { + return pending.flatMap((work) => (work.kind === "compaction" ? [work.id] : [])) +} + +function orderInputs(pending: SessionPendingWork[]) { + const bucket = (work: SessionPendingWork) => { + if (work.kind === "message" && work.phase === "promoted") return 0 + if (work.kind === "compaction") return 1 + if (work.delivery === "steer") return 2 return 3 } - return inputs.toSorted( + return pending.toSorted( (a, b) => bucket(a) - bucket(b) || - (a.phase === "promoted" ? (a.promotedSeq ?? a.admittedSeq) - (b.promotedSeq ?? b.admittedSeq) : 0) || + (a.kind === "message" && a.phase === "promoted" && b.kind === "message" + ? (a.promotedSeq ?? a.admittedSeq) - (b.promotedSeq ?? b.admittedSeq) + : 0) || a.admittedSeq - b.admittedSeq, ) } diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index 6c4c87fa322a..e567bc51c790 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -37,9 +37,9 @@ export function createSessionRows(sessionID: Accessor) { rows.splice( position === -1 ? rows.length : position, 0, - ...data.session.compaction - .list(sessionID()) - .map((inputID): SessionRow => ({ id: `compaction-queued:${inputID}`, type: "compaction-queued", inputID })), + ...data.session.timeline + .compactions(sessionID()) + .map((inputID): SessionRow => ({ id: inputID, type: "compaction-queued", inputID })), ) return rows } @@ -83,7 +83,7 @@ export function createSessionRows(sessionID: Accessor) { createEffect( on( - () => data.session.compaction.list(sessionID()).map((inputID) => inputID), + () => data.session.timeline.compactions(sessionID()), () => setRows(reconcile(reduce(), { key: "id" })), ), ) diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 424f2063f457..117631a7ebf8 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -1086,7 +1086,7 @@ test("tracks session status from active sessions and execution events", async () durable: durable("session-manual", 1), data: { sessionID: "session-manual", inputID: "message-compaction" }, }) - await wait(() => data.session.compaction.list("session-manual").includes("message-compaction")) + await wait(() => data.session.timeline.compactions("session-manual").includes("message-compaction")) emitEvent(events, { id: "evt_manual_compaction_started", created: 1, @@ -1104,7 +1104,7 @@ test("tracks session status from active sessions and execution events", async () const message = data.session.message.get("session-manual", "message-compaction") return message?.type === "compaction" && message.status === "running" && message.summary === "Streamed summary" }) - expect(data.session.compaction.list("session-manual")).toEqual([]) + expect(data.session.timeline.compactions("session-manual")).toEqual([]) const compactionRow = manualRows.find( (row) => row.type === "message" && row.messageID === "message-compaction", ) @@ -1223,24 +1223,25 @@ test("restores queued compaction from durable pending input", async () => { )) try { - await wait(() => data.session.compaction.list(sessionID).length === 2) - expect(data.session.compaction.list(sessionID)).toEqual([ + await wait(() => data.session.timeline.compactions(sessionID).length === 2) + expect(data.session.timeline.compactions(sessionID)).toEqual([ "message-compaction-queued", "message-compaction-later", ]) await wait(() => rows.filter((row) => row.type === "compaction-queued").length === 2) expect(rows.filter((row) => row.type === "compaction-queued")).toEqual([ { - id: "compaction-queued:message-compaction-queued", + id: "message-compaction-queued", type: "compaction-queued", inputID: "message-compaction-queued", }, { - id: "compaction-queued:message-compaction-later", + id: "message-compaction-later", type: "compaction-queued", inputID: "message-compaction-later", }, ]) + const queuedRow = rows.find((row) => row.id === "message-compaction-queued") emitEvent(events, { id: "evt_compaction_started", @@ -1254,8 +1255,10 @@ test("restores queued compaction from durable pending input", async () => { inputID: "message-compaction-queued", }, }) - await wait(() => data.session.compaction.list(sessionID).length === 1) - expect(data.session.compaction.list(sessionID)).toEqual(["message-compaction-later"]) + await wait(() => data.session.timeline.compactions(sessionID).length === 1) + expect(data.session.timeline.compactions(sessionID)).toEqual(["message-compaction-later"]) + await wait(() => rows.some((row) => row.type === "message" && row.messageID === "message-compaction-queued")) + expect(rows.find((row) => row.id === "message-compaction-queued")).toBe(queuedRow) emitEvent(events, { id: "evt_compaction_ended", @@ -1264,7 +1267,7 @@ test("restores queued compaction from durable pending input", async () => { durable: durable(sessionID, 5), data: { sessionID, reason: "manual", text: "Summary", recent: "" }, }) - expect(data.session.compaction.list(sessionID)).toEqual(["message-compaction-later"]) + expect(data.session.timeline.compactions(sessionID)).toEqual(["message-compaction-later"]) pending = [] emitEvent(events, { @@ -1272,7 +1275,7 @@ test("restores queued compaction from durable pending input", async () => { type: "server.connected", data: {}, }) - await wait(() => data.session.compaction.list(sessionID).length === 0) + await wait(() => data.session.timeline.compactions(sessionID).length === 0) } finally { app.renderer.destroy() } diff --git a/packages/tui/test/context/session-timeline.test.ts b/packages/tui/test/context/session-timeline.test.ts index 44cc09d949b1..7c3ee8e256ee 100644 --- a/packages/tui/test/context/session-timeline.test.ts +++ b/packages/tui/test/context/session-timeline.test.ts @@ -3,8 +3,9 @@ import type { SessionMessageInfo, SessionPendingInfo } from "@opencode-ai/sdk/v2 import { applyTimelineOperations, fromPending, + pendingCompactions, visibleMessages, - type SessionTimelineInput, + type SessionPendingWork, } from "../../src/context/session-timeline" test("orders promoted work, compaction, steers, then queued inputs", () => { @@ -20,19 +21,20 @@ test("orders promoted work, compaction, steers, then queued inputs", () => { ]) expect(result.map((input) => input.id)).toEqual(["steer-2", "compaction", "steer-1", "queue"]) + expect(pendingCompactions(result)).toEqual(["compaction"]) }) test("replays an admission after the pending snapshot", () => { const input = pending("late", 2, "steer") - expect(applyTimelineOperations([], [{ type: "admitted", input }])).toEqual([input]) + expect(applyTimelineOperations([], [{ type: "admitted", work: input }])).toEqual([input]) }) test("does not let late admission downgrade a promotion", () => { const input = pending("input", 1, "steer") const result = applyTimelineOperations([], [ - { type: "admitted", input }, + { type: "admitted", work: input }, { type: "promoted", inputID: input.id, promotedSeq: 2, created: 2 }, - { type: "admitted", input }, + { type: "admitted", work: input }, ]) expect(result).toMatchObject([{ id: input.id, phase: "promoted", promotedSeq: 2 }]) @@ -42,7 +44,7 @@ test("retains promotion state until a later admission provides content", () => { const input = pending("input", 1, "steer") const result = applyTimelineOperations([], [ { type: "promoted", inputID: input.id, promotedSeq: 2, created: 2 }, - { type: "admitted", input }, + { type: "admitted", work: input }, ]) expect(result).toMatchObject([{ id: input.id, phase: "promoted", promotedSeq: 2, message: { text: "input" } }]) @@ -67,7 +69,7 @@ test("projected messages replace pending and promoted representations", () => { expect(visibleMessages(projected, [pending("input", 1, "steer")])).toEqual(projected) }) -function pending(id: string, admittedSeq: number, delivery: "steer" | "queue"): SessionTimelineInput { +function pending(id: string, admittedSeq: number, delivery: "steer" | "queue"): SessionPendingWork { return fromPending({ admittedSeq, id, From 37c017d23edf83098ef480d21d6da1c7e82e9f8f Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 15:09:52 -0400 Subject: [PATCH 7/7] refactor(tui): simplify timeline synchronization --- packages/tui/TIMELINE_PLAN.md | 244 ++++-------------- packages/tui/src/context/data.tsx | 98 +++---- packages/tui/src/context/session-timeline.ts | 92 +++---- packages/tui/src/routes/session/rows.ts | 47 ++-- .../tui/test/context/session-timeline.test.ts | 8 +- 5 files changed, 145 insertions(+), 344 deletions(-) diff --git a/packages/tui/TIMELINE_PLAN.md b/packages/tui/TIMELINE_PLAN.md index fe0b8301b356..ea98205a06c1 100644 --- a/packages/tui/TIMELINE_PLAN.md +++ b/packages/tui/TIMELINE_PLAN.md @@ -1,250 +1,118 @@ -# Session Timeline Frontend Plan +# Session Timeline Frontend Design -## Status +## Problem -This plan incorporates reviews of Solid identity, hydration concurrency, Session execution ordering, rendering, scroll behavior, and migration scope. The server contract remains unchanged: projected messages and pending inputs are separate reads, and the event stream has no replay cursor across reconnects. +The server exposes projected messages and pending work separately while live events may arrive during either read. The TUI must keep admitted work visible through hydration, promotion, reconnect, revert, and compaction without treating pending records as projected history. -## Goals +The server contract remains unchanged. -- Never overwrite an admitted prompt with a hydration snapshot. -- Keep one stable rendered identity from admission through promotion and projection. -- Keep projected history semantically distinct from pending frontend work. -- Make the unavoidable snapshot/live-event reconciliation narrow and explicit. -- Preserve targeted updates for streamed assistant content. -- Avoid changing unrelated message consumers in this PR. +## Model -## Constraints Discovered In Review - -- Promotion atomically deletes pending state and inserts projected state. -- Reading pending before projected history guarantees an input admitted before the pending read appears in at least one response. -- An admission after the pending read may appear in neither response and must survive through its live event. -- The event stream does not replay events missed across a disconnect, so reconnect must perform authoritative reads. -- General event replay is unsafe. Text, reasoning, tool input, and compaction deltas are additive, while snapshots expose no event watermark proving which deltas they contain. -- Solid `` retains children by source-item identity. The existing `SessionRow` values lack a uniform `id`, so full row reduction does not express stable semantic identity. -- OpenTUI sticky scrolling follows the bottom but does not preserve a scrolled-up content anchor when rows are inserted above it. - -## Selected Architecture - -Keep the existing projected/live message cache. Add a lifecycle overlay containing only pending or locally promoted work. +Projected history stays in the message cache. Everything not yet projected lives in one overlay: ```ts -type SessionPendingWork = +type SessionTimelineWork = | { - kind: "message" + kind: "input" id: string - phase: "pending" | "promoted" admittedSeq: number delivery: "steer" | "queue" + message: SessionMessageInfo + } + | { + kind: "promoted" + id: string + promotedSeq: number message?: SessionMessageInfo } | { kind: "compaction" id: string - phase: "pending" admittedSeq: number } - -type Data = { - session: { - message: Record - pending: Record - } -} -``` - -The overlay keeps every kind of pending Session work together with its phase and ordering metadata. It replaces both the parallel message-plus-ID representation and the separate queued-compaction store. - -Projected messages remain available through the existing projected cache. A small visible-timeline surface composes projected messages with the overlay for rendering: - -```ts -timeline.get(sessionID, messageID) -timeline.list(sessionID) -message.refresh(sessionID) ``` -Existing `session.message` consumers remain unchanged unless they need pending work. The Session row renderer migrates to the visible timeline surface. +The variants prevent fabricated state: -## Lifecycle +- Admitted input has content, delivery, and admission order. +- Promotion-before-admission needs only its ID and promotion order. +- Queued compaction has no message payload or delivery mode. -Phase precedence is monotone: - -```text -projected > promoted > pending -``` - -Content precedence is: - -```text -projected message > pending/admission content > unknown placeholder -``` - -Admission inserts one pending overlay entry. Promotion changes that entry to `promoted` without removing it. Observing the same ID in projected history removes the overlay entry because projected state is now authoritative. - -Promotion-before-admission delivery is tolerated defensively: promotion creates a non-rendered placeholder, and a later admission fills its content without downgrading its phase. +The visible timeline is projected messages followed by non-projected overlay messages. Queued compaction is exposed as its own row kind. ## Hydration -Hydration is sequential: +Hydration reads pending work before projected history: ```ts const pending = await api.session.pending.list({ sessionID }) -const projected = await api.message.list({ sessionID, limit: 200, order: "desc" }) +const projected = await api.message.list({ sessionID }) ``` -For a promotion commit `X`, pending read `P`, and projected read `M`, `P < M` eliminates the unsafe `M < X < P` split snapshot. - -The proof is: +Promotion atomically removes the pending record and inserts the projected message. Given pending read `P`, projected read `M`, and promotion `X`, with `P < M`: -- `X < P`: pending omits the input and projected contains it. -- `P < X < M`: pending and projected may both contain it; projected wins by ID. -- `M < X`: pending contains it, and a later promotion event changes its local phase. +- `X < P`: projected contains the input. +- `P < X < M`: projected wins by ID. +- `M < X`: pending contains the input and the promotion event retains it in the overlay. -An input admitted after `P` and not promoted before `M` is absent from both snapshots. Its live admission operation must be preserved during installation. +Concurrent reads permit the unsafe order `M < X < P`, where both snapshots omit the input. Sequential reads eliminate it. -## Narrow Hydration Journal +## Live Operations -Only input topology operations are journaled: +Only topology operations are journaled during hydration: ```ts -type TimelineOperation = - | { type: "admitted"; work: SessionPendingWork } +type SessionTimelineOperation = + | { type: "admitted"; work: SessionAdmittedWork } | { type: "promoted"; inputID: string; promotedSeq: number; created: number } + | { type: "removed"; inputID: string } | { type: "reverted"; to: string } ``` -All existing assistant, text, reasoning, tool, shell, and compaction streaming events continue to update the projected/live message cache exactly once. They are never replayed. +The journal is folded onto the two server snapshots before installation. Projected IDs always win. -Hydration performs: +Text, reasoning, tool, shell, and compaction deltas are not replayed. They are additive, and projected responses expose no event watermark proving whether a delta is already included. -1. Open a per-Session lifecycle journal. -2. Continue applying admission and promotion immediately to the visible overlay while recording them. -3. Read pending inputs. -4. Read projected messages. -5. Build projected history and pending overlay from those snapshots. -6. Fold the lifecycle journal onto the overlay with an idempotent, monotone reducer. -7. Remove overlay entries whose IDs are present in projected history. -8. Install projected history and overlay together in one Solid batch. -9. Close the journal. +## Refresh Ownership -If either request fails, do not install an authoritative empty collection. Preserve visible state and retry through the existing caller/reconnect path. +One refresh record per Session owns its promise, identity token, and topology journal. -## Refresh Coordination +- Same-Session callers join the active promise. +- Reconnect clears active records and starts replacements for every loaded, pending-only, or refreshing Session. +- Invalidated callers join the replacement refresh when one exists. +- Delete and committed revert remove the active record, preventing stale installation. +- Failed reads install nothing, preserving visible state. -At most one refresh per Session and connection epoch may install. - -- Same-epoch callers join the current promise. -- Every `server.connected` increments a connection epoch. -- A result from an older epoch is discarded. -- Reconnect always starts or queues a refresh in the new epoch, even if an old refresh is still running. -- Different Sessions refresh independently. - -This prevents an old server response from winning after reconnect. +No generation counters or durable client-side event log are required. ## Ordering -Visible ordering is derived rather than encoded by array mutation: +Overlay work is ordered by execution eligibility: -1. Projected history in server order. -2. Promoted placeholders in promotion order. -3. Pending compaction barriers. -4. Pending steering inputs in `admittedSeq` order. -5. Pending queued inputs in `admittedSeq` order. +1. Promoted inputs awaiting projection. +2. Queued compaction barriers. +3. Steering inputs by `admittedSeq`. +4. Queued inputs by `admittedSeq`. -This anticipates normal runner eligibility. Steers are consumed before queues, while compaction blocks both. A legitimate queue/steer interleaving selected by the runner is represented by durable promotion order once promoted. - -Queued compaction remains a distinct overlay variant because it has no message payload or delivery mode. Starting compaction removes that pending variant and projects a running compaction message with the same stable input ID. +Projected history always precedes the overlay. ## Solid Identity -The visible timeline feeds a row reducer whose output has stable IDs. Rendering reconciles those rows by ID before passing them to ``: - -```tsx -setRows(reconcile(reduce(), { key: "id" })) - - - {(row) => } - -``` - -The existing row reducer expands messages into message, part, exploration-group, and footer rows. Every `SessionRow` must gain a deterministic `id`: +Every `SessionRow` has a stable semantic ID and rows are installed with: ```ts -type SessionRow = - | { id: `message:${string}`; type: "message"; messageID: string } - | { id: `part:${string}:${string}`; type: "part"; ref: PartRef } - | { id: `assistant-footer:${string}`; type: "assistant-footer"; messageID: string } - | { id: `group:exploration:${string}`; type: "group"; /* ... */ } +setRows(reconcile(next, { key: "id" })) ``` -Rows are installed through `reconcile(rows, { key: "id" })`. Exploration group identity is seeded from the first part that creates the group and remains fixed as refs move between pending and completed partitions. - -The outer OpenTUI row wrapper receives the same stable row ID so ownership and scroll anchoring are observable in tests. +The same durable input ID identifies: -## Scroll Policy +- pending, promoted, and projected prompt rows; +- queued and running compaction rows. -Stable identity prevents remounting but not geometric scroll movement. +Solid therefore moves or updates the existing row owner instead of remounting it. Existing sticky-bottom behavior remains responsible for following output; this change adds no scroll-position policy. -- Existing `stickyScroll` and `stickyStart="bottom"` continue to own bottom-following behavior. -- Ordinary admission/promotion updates should not run explicit anchoring. -- Explicit anchoring is deferred unless a deterministic regression shows hydration changes the viewed content while manually scrolled up. -- If required, capture the first visible stable row ID and viewport offset before snapshot installation, then restore its offset after OpenTUI layout. +## Boundary -## Module Boundary - -Place overlay conversion, ordering, lifecycle reduction, journaling, and refresh coordination in a focused module: - -```text -context/data.tsx - projected/live message reducers - event routing - hydration and refresh coordination - | - v -context/session-timeline.ts - pending adapters - lifecycle overlay operations - visible timeline composition - deterministic ordering -``` +`session.message` remains projected-only. `session.timeline` composes projected history with the overlay for rendering and pending-aware interactions. The public `session.input` queries are compatibility views over admitted input variants. -Do not move general message-event reduction into this module. - -## Implementation Sequence - -1. Add pure lifecycle adapter, reducer, and ordering functions with deterministic unit tests. -2. Add the overlay state and visible timeline accessors behind the existing data context. -3. Change admission and promotion handlers to mutate the overlay while preserving existing projected/live behavior where required during migration. -4. Implement sequential hydration, the narrow journal, and same-epoch refresh coalescing. -5. Change `createSessionRows` and `SessionRowView` lookup to consume the visible timeline. -6. Add stable IDs to all `SessionRow` variants and keyed row reconciliation. -7. Migrate only pending-sensitive route logic; leave projected-only consumers on `session.message`. -8. Remove the old pending-ID reconciliation helper and mixed pending insertion from projected hydration. -9. Run focused data/row tests, TUI typecheck, and the full TUI suite. - -## Required Regressions - -- Pending input known only through the pending endpoint. -- Promotion before pending read. -- Promotion between pending and projected reads. -- Promotion after projected read. -- Admission after pending read. -- Admission and promotion during hydration. -- Duplicate or inverted admission/promotion delivery. -- Projected representation wins over the same pending/promoted ID. -- Streaming text delta during hydration is applied exactly once. -- Concurrent same-Session refresh callers join. -- Different Sessions refresh concurrently. -- Old-epoch refresh cannot install after reconnect. -- Failed pending or projected read does not erase visible state. -- Pending-to-promoted-to-projected retains one stable row ID. -- Unaffected neighboring rows retain identity across full row reduction. -- Steer and queue ordering follows the selected policy. - -## Explicit Non-Goals For This PR - -- Replay general Session events. -- Redesign the 200-message pagination behavior. -- Change server APIs or event schemas. -- Introduce Effect into the TUI data layer. -- Add scroll anchoring without a failing regression. -- Migrate undo, fork, transcript, usage, and other projected-history consumers unless existing behavior requires pending visibility. +This design guarantees pending-work visibility under the existing atomic-promotion, ordered-live-connection, and reconnect contracts. Recovering arbitrary silently dropped streaming events would require a server snapshot cursor or replay API. diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 4ed0327ee18f..cb0d3d07d0e9 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -31,11 +31,10 @@ import { useSDK } from "./sdk" import { batch, createSignal, onCleanup } from "solid-js" import { applyTimelineOperations, - fromAdmission, fromPending, pendingCompactions, visibleMessages, - type SessionPendingWork, + type SessionTimelineWork, type SessionTimelineOperation, } from "./session-timeline" @@ -71,7 +70,7 @@ type Data = { family: Record status: Record message: Record - pending: Record + pending: Record permission: Record // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. form: Record @@ -115,14 +114,12 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) const messageIndex = new Map>() const timelineHydrated = new Set() - const sessionGenerations = new Map() const timelineRefreshes = new Map< string, - { epoch: number; generation: number; operations: SessionTimelineOperation[]; promise: Promise } + { token: object; operations: SessionTimelineOperation[]; promise: Promise } >() let bootstrapping: Promise | undefined let connected = false - let connectionEpoch = 0 function setSessionStatus(sessionID: string, status: DataSessionStatus) { setStore("session", "status", sessionID, status) @@ -130,11 +127,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ function applyTimelineOperation(sessionID: string, operation: SessionTimelineOperation) { const refresh = timelineRefreshes.get(sessionID) - if ( - refresh?.epoch === connectionEpoch && - refresh.generation === (sessionGenerations.get(sessionID) ?? 0) - ) - refresh.operations.push(operation) + if (refresh) refresh.operations.push(operation) setStore( "session", "pending", @@ -165,13 +158,16 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ index.set(item.id, messages.length) messages.push(item) }, + get(messages: SessionMessageInfo[] | undefined, index: Map | undefined, messageID: string) { + const position = index?.get(messageID) + return position === undefined ? undefined : messages?.[position] + }, 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] + const item = message.get(messages, index, messageID) return item?.type === "assistant" ? item : undefined }, shell(messages: SessionMessageInfo[], shellID: string) { @@ -251,7 +247,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function removeSession(sessionID: string) { - sessionGenerations.set(sessionID, (sessionGenerations.get(sessionID) ?? 0) + 1) + timelineRefreshes.delete(sessionID) messageIndex.delete(sessionID) timelineHydrated.delete(sessionID) setStore( @@ -354,35 +350,17 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ promotedSeq: event.durable.seq, created: event.created, }) - const epoch = connectionEpoch - const generation = sessionGenerations.get(event.data.sessionID) ?? 0 - void sdk.api.session - .message({ sessionID: event.data.sessionID, messageID: event.data.inputID }) - .then((item) => { - if (epoch !== connectionEpoch || generation !== (sessionGenerations.get(event.data.sessionID) ?? 0)) return - 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 - }) - setStore( - "session", - "pending", - event.data.sessionID, - (pending) => pending?.filter((work) => work.id !== item.id) ?? [], - ) - }) - .catch(() => undefined) break } case "session.input.admitted": { applyTimelineOperation(event.data.sessionID, { type: "admitted", - work: fromAdmission({ + work: fromPending({ id: event.data.inputID, + sessionID: event.data.sessionID, admittedSeq: event.durable.seq, timeCreated: event.created, - input: event.data.input, + ...event.data.input, }), }) break @@ -650,7 +628,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ work: { kind: "compaction", id: event.data.inputID, - phase: "pending", admittedSeq: event.durable.seq, }, }) @@ -690,10 +667,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setStore("session", "info", event.data.sessionID, "revert", undefined) break case "session.revert.committed": - sessionGenerations.set( - event.data.sessionID, - (sessionGenerations.get(event.data.sessionID) ?? 0) + 1, - ) + timelineRefreshes.delete(event.data.sessionID) if (store.session.info[event.data.sessionID]) { setStore("session", "info", event.data.sessionID, "revert", undefined) } @@ -837,21 +811,18 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ function refreshTimeline(sessionID: string): Promise { const existing = timelineRefreshes.get(sessionID) - const generation = sessionGenerations.get(sessionID) ?? 0 - if (existing?.epoch === connectionEpoch && existing.generation === generation) return existing.promise + if (existing) return existing.promise - const epoch = connectionEpoch const operations: SessionTimelineOperation[] = [] + const token = {} const baseline = new Set( timelineHydrated.has(sessionID) ? (store.session.message[sessionID] ?? []).map((message) => message.id) : [], ) const promise = (async () => { const pending = await sdk.api.session.pending.list({ sessionID }) - if (generation !== (sessionGenerations.get(sessionID) ?? 0)) return - if (epoch !== connectionEpoch) return refreshTimeline(sessionID) + if (timelineRefreshes.get(sessionID)?.token !== token) return timelineRefreshes.get(sessionID)?.promise const projected = await sdk.api.message.list({ sessionID, limit: 200, order: "desc" }) - if (generation !== (sessionGenerations.get(sessionID) ?? 0)) return - if (epoch !== connectionEpoch) return refreshTimeline(sessionID) + if (timelineRefreshes.get(sessionID)?.token !== token) return timelineRefreshes.get(sessionID)?.promise const messages: SessionMessageInfo[] = projected.data.toReversed() const messagePositions = new Map(messages.map((message, index) => [message.id, index])) @@ -878,9 +849,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ timelineHydrated.add(sessionID) }) })() - timelineRefreshes.set(sessionID, { epoch, generation, operations, promise }) + timelineRefreshes.set(sessionID, { token, operations, promise }) const cleanup = () => { - if (timelineRefreshes.get(sessionID)?.promise === promise) timelineRefreshes.delete(sessionID) + if (timelineRefreshes.get(sessionID)?.token === token) timelineRefreshes.delete(sessionID) } void promise.then(cleanup, cleanup) return promise @@ -891,11 +862,10 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } function timelineGet(sessionID: string, messageID: string) { - const messages = store.session.message[sessionID] - const position = messageIndex.get(sessionID)?.get(messageID) - if (position !== undefined) return messages?.[position] + const projected = message.get(store.session.message[sessionID], messageIndex.get(sessionID), messageID) + if (projected) return projected const work = store.session.pending[sessionID]?.find((work) => work.id === messageID) - return work?.kind === "message" ? work.message : undefined + return work?.kind === "input" || work?.kind === "promoted" ? work.message : undefined } const result = { @@ -929,13 +899,13 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ input: { list(sessionID: string) { return (store.session.pending[sessionID] ?? []) - .filter((work) => work.kind === "message" && work.phase === "pending") + .filter((work) => work.kind === "input") .map((work) => work.id) }, has(sessionID: string, inputID: string) { return ( store.session.pending[sessionID]?.some( - (work) => work.kind === "message" && work.id === inputID && work.phase === "pending", + (work) => work.kind === "input" && work.id === inputID, ) ?? false ) }, @@ -952,9 +922,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ 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] + return message.get(store.session.message[sessionID], messageIndex.get(sessionID), messageID) }, refresh(sessionID: string) { return refreshTimeline(sessionID) @@ -1225,14 +1193,12 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ onCleanup( sdk.event.listen(({ details }) => { if (details.type === "server.connected") { - connectionEpoch += 1 - const sessions = connected - ? new Set([ - ...Object.keys(store.session.message), - ...Object.keys(store.session.pending), - ...timelineRefreshes.keys(), - ]) - : [] + const sessions = new Set([ + ...(connected ? Object.keys(store.session.message) : []), + ...(connected ? Object.keys(store.session.pending) : []), + ...timelineRefreshes.keys(), + ]) + timelineRefreshes.clear() connected = true refreshActive() void Promise.allSettled([bootstrap(), ...Array.from(sessions).map(result.session.message.refresh)]) diff --git a/packages/tui/src/context/session-timeline.ts b/packages/tui/src/context/session-timeline.ts index ebace06746a9..024581fd1468 100644 --- a/packages/tui/src/context/session-timeline.ts +++ b/packages/tui/src/context/session-timeline.ts @@ -1,61 +1,38 @@ import type { SessionMessageInfo, SessionPendingInfo } from "@opencode-ai/sdk/v2" -export type SessionPendingWork = +export type SessionTimelineWork = | { - kind: "message" + kind: "input" id: string - phase: "pending" | "promoted" admittedSeq: number - promotedSeq?: number delivery: "steer" | "queue" + message: SessionMessageInfo + } + | { + kind: "promoted" + id: string + promotedSeq: number message?: SessionMessageInfo } | { kind: "compaction" id: string - phase: "pending" admittedSeq: number } +export type SessionAdmittedWork = Exclude + export type SessionTimelineOperation = - | { type: "admitted"; work: SessionPendingWork } + | { type: "admitted"; work: SessionAdmittedWork } | { type: "promoted"; inputID: string; promotedSeq: number; created: number } | { type: "removed"; inputID: string } | { type: "reverted"; to: string } -export function fromAdmission(input: { - id: string - admittedSeq: number - timeCreated: number - input: - | { type: "user"; data: Extract["data"]; delivery: "steer" | "queue" } - | { - type: "synthetic" - data: Extract["data"] - delivery: "steer" | "queue" - } -}): SessionPendingWork { - return { - kind: "message", - id: input.id, - phase: "pending", - admittedSeq: input.admittedSeq, - delivery: input.input.delivery, - message: { - id: input.id, - type: input.input.type, - ...input.input.data, - time: { created: input.timeCreated }, - }, - } -} - -export function fromPending(item: SessionPendingInfo): SessionPendingWork { +export function fromPending(item: SessionPendingInfo): SessionAdmittedWork { if (item.type === "user") return { - kind: "message", + kind: "input", id: item.id, - phase: "pending", admittedSeq: item.admittedSeq, delivery: item.delivery, message: { @@ -67,9 +44,8 @@ export function fromPending(item: SessionPendingInfo): SessionPendingWork { } if (item.type === "synthetic") return { - kind: "message", + kind: "input", id: item.id, - phase: "pending", admittedSeq: item.admittedSeq, delivery: item.delivery, message: { @@ -82,13 +58,12 @@ export function fromPending(item: SessionPendingInfo): SessionPendingWork { return { kind: "compaction", id: item.id, - phase: "pending", admittedSeq: item.admittedSeq, } } export function applyTimelineOperations( - pending: SessionPendingWork[], + pending: SessionTimelineWork[], operations: SessionTimelineOperation[], projectedIDs = new Set(), ) { @@ -113,22 +88,18 @@ export function applyTimelineOperations( } result.set( operation.work.id, - existing?.kind === "message" && existing.phase === "promoted" - ? { ...operation.work, phase: "promoted", promotedSeq: existing.promotedSeq } + existing?.kind === "promoted" + ? { ...existing, message: existing.message ?? operation.work.message } : (existing ?? operation.work), ) return } const existing = result.get(operation.inputID) result.set(operation.inputID, { - ...(existing?.kind === "message" ? existing : undefined), - kind: "message", + kind: "promoted", id: operation.inputID, - phase: "promoted", - admittedSeq: existing?.admittedSeq ?? operation.promotedSeq, promotedSeq: operation.promotedSeq, - delivery: existing?.kind === "message" ? existing.delivery : "steer", - message: existing?.kind === "message" && existing.message + message: existing?.kind === "input" ? { ...existing.message, time: { ...existing.message.time, created: operation.created }, @@ -139,32 +110,29 @@ export function applyTimelineOperations( return orderInputs([...result.values()]) } -export function visibleMessages(projected: SessionMessageInfo[], pending: SessionPendingWork[]) { +export function visibleMessages(projected: SessionMessageInfo[], pending: SessionTimelineWork[]) { if (pending.length === 0) return projected const ids = new Set(projected.map((message) => message.id)) return [ ...projected, - ...pending.flatMap((work) => (work.kind === "message" && !ids.has(work.id) && work.message ? [work.message] : [])), + ...pending.flatMap((work) => + work.kind !== "compaction" && !ids.has(work.id) && work.message ? [work.message] : [], + ), ] } -export function pendingCompactions(pending: SessionPendingWork[]) { +export function pendingCompactions(pending: SessionTimelineWork[]) { return pending.flatMap((work) => (work.kind === "compaction" ? [work.id] : [])) } -function orderInputs(pending: SessionPendingWork[]) { - const bucket = (work: SessionPendingWork) => { - if (work.kind === "message" && work.phase === "promoted") return 0 +function orderInputs(pending: SessionTimelineWork[]) { + const bucket = (work: SessionTimelineWork) => { + if (work.kind === "promoted") return 0 if (work.kind === "compaction") return 1 if (work.delivery === "steer") return 2 return 3 } - return pending.toSorted( - (a, b) => - bucket(a) - bucket(b) || - (a.kind === "message" && a.phase === "promoted" && b.kind === "message" - ? (a.promotedSeq ?? a.admittedSeq) - (b.promotedSeq ?? b.admittedSeq) - : 0) || - a.admittedSeq - b.admittedSeq, - ) + const sequence = (work: SessionTimelineWork) => + work.kind === "promoted" ? work.promotedSeq : work.admittedSeq + return pending.toSorted((a, b) => bucket(a) - bucket(b) || sequence(a) - sequence(b)) } diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index e567bc51c790..6f74d14f321f 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -52,6 +52,10 @@ export function createSessionRows(sessionID: Accessor) { ) } + function replaceRows() { + setRows(reconcile(reduce(), { key: "id" })) + } + createEffect(() => { const pending = pendingPermissions() setRows( @@ -63,11 +67,10 @@ export function createSessionRows(sessionID: Accessor) { createEffect( on(sessionID, (id) => { - setRows(reconcile(reduce(), { key: "id" })) + replaceRows() void data.session.message.refresh(id).then( () => { - if (sessionID() !== id) return - setRows(reconcile(reduce(), { key: "id" })) + if (sessionID() === id) replaceRows() }, () => undefined, ) @@ -77,39 +80,35 @@ export function createSessionRows(sessionID: Accessor) { // Re-reduce when the revert boundary changes (stage/clear/commit). createEffect( on(revertBoundary, () => { - setRows(reconcile(reduce(), { key: "id" })) + replaceRows() }), ) - createEffect( - on( - () => data.session.timeline.compactions(sessionID()), - () => setRows(reconcile(reduce(), { key: "id" })), - ), - ) - createEffect( on( () => - data.session.timeline.list(sessionID()).flatMap((message) => - message.type === "user" || message.type === "synthetic" - ? [ - { - id: message.id, - created: message.time.created, - input: data.session.input.has(sessionID(), message.id), - }, - ] - : message.type === "compaction" + [ + ...data.session.timeline.list(sessionID()).flatMap((message) => + message.type === "user" || message.type === "synthetic" ? [ { id: message.id, created: message.time.created, + input: data.session.input.has(sessionID(), message.id), }, ] - : [], - ), - () => setRows(reconcile(reduce(), { key: "id" })), + : message.type === "compaction" + ? [ + { + id: message.id, + created: message.time.created, + }, + ] + : [], + ), + ...data.session.timeline.compactions(sessionID()).map((id) => ({ id, pending: true })), + ], + replaceRows, ), ) diff --git a/packages/tui/test/context/session-timeline.test.ts b/packages/tui/test/context/session-timeline.test.ts index 7c3ee8e256ee..641635c8fcf9 100644 --- a/packages/tui/test/context/session-timeline.test.ts +++ b/packages/tui/test/context/session-timeline.test.ts @@ -5,7 +5,7 @@ import { fromPending, pendingCompactions, visibleMessages, - type SessionPendingWork, + type SessionAdmittedWork, } from "../../src/context/session-timeline" test("orders promoted work, compaction, steers, then queued inputs", () => { @@ -37,7 +37,7 @@ test("does not let late admission downgrade a promotion", () => { { type: "admitted", work: input }, ]) - expect(result).toMatchObject([{ id: input.id, phase: "promoted", promotedSeq: 2 }]) + expect(result).toMatchObject([{ id: input.id, kind: "promoted", promotedSeq: 2 }]) }) test("retains promotion state until a later admission provides content", () => { @@ -47,7 +47,7 @@ test("retains promotion state until a later admission provides content", () => { { type: "admitted", work: input }, ]) - expect(result).toMatchObject([{ id: input.id, phase: "promoted", promotedSeq: 2, message: { text: "input" } }]) + expect(result).toMatchObject([{ id: input.id, kind: "promoted", promotedSeq: 2, message: { text: "input" } }]) }) test("applies committed reverts after a stale pending snapshot", () => { @@ -69,7 +69,7 @@ test("projected messages replace pending and promoted representations", () => { expect(visibleMessages(projected, [pending("input", 1, "steer")])).toEqual(projected) }) -function pending(id: string, admittedSeq: number, delivery: "steer" | "queue"): SessionPendingWork { +function pending(id: string, admittedSeq: number, delivery: "steer" | "queue"): SessionAdmittedWork { return fromPending({ admittedSeq, id,