From 5945a8d429f4d4f67309befeee9d598242301dda Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 13:54:23 -0400 Subject: [PATCH] fix(tui): restore queued compaction indicator (#36440) --- packages/tui/src/context/data.tsx | 46 +++++++++- packages/tui/src/routes/session/index.tsx | 17 ++++ packages/tui/src/routes/session/rows.ts | 17 ++++ packages/tui/test/app-lifecycle.test.tsx | 1 + packages/tui/test/cli/tui/data.test.tsx | 101 ++++++++++++++++++++++ 5 files changed, 181 insertions(+), 1 deletion(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 03d109b2a1f..e71c3a6985e 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -63,6 +63,7 @@ type Data = { status: Record message: Record input: Record + compaction: Record permission: Record // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. form: Record @@ -91,6 +92,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ status: {}, message: {}, input: {}, + compaction: {}, permission: {}, form: {}, }, @@ -112,6 +114,21 @@ 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), + ) + } + const message = { update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { setStore( @@ -221,6 +238,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ delete draft.status[sessionID] delete draft.message[sessionID] delete draft.input[sessionID] + delete draft.compaction[sessionID] delete draft.permission[sessionID] delete draft.form[sessionID] for (const [rootID, family] of Object.entries(draft.family)) { @@ -611,8 +629,10 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setSessionStatus(event.data.sessionID, "running") break case "session.compaction.admitted": + addCompaction(event.data.sessionID, event.data.inputID) 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), @@ -689,6 +709,7 @@ 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] @@ -821,6 +842,24 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return store.session.input[sessionID]?.includes(inputID) ?? 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), + ), + ) + }, + }, async refresh(sessionID: string) { setStore("session", "info", sessionID, await sdk.api.session.get({ sessionID })) registerSession(sessionID) @@ -1098,9 +1137,14 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ 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) : [] connected = true refreshActive() - void Promise.allSettled([bootstrap(), ...messages.map(result.session.message.refresh)]) + void Promise.allSettled([ + bootstrap(), + ...messages.map(result.session.message.refresh), + ...compactions.map(result.session.compaction.refresh), + ]) return } handleEvent(details) diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 55ef1459baa..91234c3d154 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -1059,6 +1059,9 @@ function SessionRowView(props: { row: SessionRow; message: (messageID: string) = {(message) => } )} + + + {(row) => } @@ -1370,6 +1373,20 @@ function CompactionMessage(props: { message: Extract + + + ◇ + Compaction queued + + + + ) +} + function statusLabel(status: "added" | "modified" | "deleted") { if (status === "added") return "A" if (status === "deleted") return "D" diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index 9f03467bfb3..59393ff1437 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -10,6 +10,7 @@ export type PartRef = { export type SessionRow = | { type: "message"; messageID: string } + | { type: "compaction-queued"; inputID: string } | { type: "part"; ref: PartRef } | { type: "group" @@ -31,6 +32,14 @@ export function createSessionRows(sessionID: Accessor) { const boundary = revertBoundary() const rows = reduceSessionRows(boundary ? messages.filter((message) => message.id < boundary) : messages, inputs) partitionPending(rows, pendingPermissions()) + const position = rows.findIndex((row) => row.type === "message" && inputs.has(row.messageID)) + rows.splice( + position === -1 ? rows.length : position, + 0, + ...data.session.compaction + .list(sessionID()) + .map((inputID): SessionRow => ({ type: "compaction-queued", inputID })), + ) return rows } @@ -54,6 +63,7 @@ export function createSessionRows(sessionID: Accessor) { createEffect( on(sessionID, (id) => { setRows(reconcile(reduce())) + void data.session.compaction.refresh(id).catch(() => undefined) void data.session.message.refresh(id).then( () => { if (sessionID() !== id) return @@ -71,6 +81,13 @@ export function createSessionRows(sessionID: Accessor) { }), ) + createEffect( + on( + () => data.session.compaction.list(sessionID()).map((inputID) => inputID), + () => setRows(reconcile(reduce())), + ), + ) + createEffect( on( () => diff --git a/packages/tui/test/app-lifecycle.test.tsx b/packages/tui/test/app-lifecycle.test.tsx index c1530ee1894..48d4a48aba6 100644 --- a/packages/tui/test/app-lifecycle.test.tsx +++ b/packages/tui/test/app-lifecycle.test.tsx @@ -96,6 +96,7 @@ test("session lifecycle updates the terminal title and prints the epilogue after }) if (url.pathname === "/api/session/dummy") return json({ data: session }) if (url.pathname === "/api/session/dummy/message") return json({ data: [], cursor: {} }) + if (url.pathname === "/api/session/dummy/pending") return json({ data: [] }) if (url.pathname === "/api/session/dummy/permission") return json({ data: [] }) }, events) const server = Bun.serve({ port: 0, fetch: (request) => calls.fetch(request) }) diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 364f71e6de5..936e141b310 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -1047,6 +1047,14 @@ test("tracks session status from active sessions and execution events", async () await wait(() => data.session.status("session-retry") === "idle") expect(data.session.message.get("session-retry", "message-retry")).not.toHaveProperty("retry") + emitEvent(events, { + id: "evt_manual_compaction_admitted", + created: 0, + type: "session.compaction.admitted", + durable: durable("session-manual", 1), + data: { sessionID: "session-manual", inputID: "message-compaction" }, + }) + await wait(() => data.session.compaction.list("session-manual").includes("message-compaction")) emitEvent(events, { id: "evt_manual_compaction_started", created: 1, @@ -1064,6 +1072,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([]) const compactionRow = manualRows.find( (row) => row.type === "message" && row.messageID === "message-compaction", ) @@ -1137,6 +1146,98 @@ test("tracks session status from active sessions and execution events", async () } }) +test("restores queued compaction from durable pending input", async () => { + const events = createEventStream() + const sessionID = "session-compaction-queued" + let pending = [ + { + admittedSeq: 3, + id: "message-compaction-queued", + sessionID, + timeCreated: 1, + type: "compaction" as const, + }, + { + admittedSeq: 4, + id: "message-compaction-later", + sessionID, + timeCreated: 2, + type: "compaction" as const, + }, + ] + const calls = createFetch((url) => { + if (url.pathname !== `/api/session/${sessionID}/pending`) return + return json({ data: pending }) + }, events) + let data!: ReturnType + let rows!: ReturnType + + function Probe() { + data = useData() + rows = createSessionRows(() => sessionID) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await wait(() => data.session.compaction.list(sessionID).length === 2) + expect(data.session.compaction.list(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([ + { type: "compaction-queued", inputID: "message-compaction-queued" }, + { type: "compaction-queued", inputID: "message-compaction-later" }, + ]) + + emitEvent(events, { + id: "evt_compaction_started", + created: 2, + type: "session.compaction.started", + durable: durable(sessionID, 4), + data: { + sessionID, + reason: "manual", + recent: "", + inputID: "message-compaction-queued", + }, + }) + await wait(() => data.session.compaction.list(sessionID).length === 1) + expect(data.session.compaction.list(sessionID)).toEqual(["message-compaction-later"]) + + emitEvent(events, { + id: "evt_compaction_ended", + created: 3, + type: "session.compaction.ended", + durable: durable(sessionID, 5), + data: { sessionID, reason: "manual", text: "Summary", recent: "" }, + }) + expect(data.session.compaction.list(sessionID)).toEqual(["message-compaction-later"]) + + pending = [] + emitEvent(events, { + id: "evt_reconnected", + type: "server.connected", + data: {}, + }) + await wait(() => data.session.compaction.list(sessionID).length === 0) + } finally { + app.renderer.destroy() + } +}) + test("refreshes integrations after integration updates", async () => { const events = createEventStream() const requests = { integration: 0, model: 0, provider: 0 }