diff --git a/docs/protocol-compatibility.md b/docs/protocol-compatibility.md index e5ba1db2a3..f7244e67ba 100644 --- a/docs/protocol-compatibility.md +++ b/docs/protocol-compatibility.md @@ -2,7 +2,7 @@ The app and the daemon are separate products that ship separately. A user updates the app from an app store or a desktop auto-update; they update the daemon when they feel like it. Every combination happens in the wild: new app against an old daemon, old app against a new daemon, and both sides months apart. -In development both sides are always the same version, which is why this is the constraint contributors miss most often. +Development can have the same version drift: an app reload picks up new code while a long-running daemon keeps its loaded version. Check the connected daemon version when diagnosing behavior that crosses this boundary. Two contracts follow from it. @@ -38,7 +38,7 @@ The app checks for the capability and either runs the feature or tells the user - **No defensive branches spread through the feature.** Detection happens in one place, and everything downstream reads a clean shape. - Capability flags live in `features` on the `server_info` message (`packages/protocol/src/messages.ts`, the `server_info` schema). -Existing functionality keeps working across versions because of the protocol contract. Gating a new feature never substitutes for that. +Existing functionality keeps working across versions. Never gate an established request on a new enhancement: negotiate the optional enhancement at the client boundary and preserve the existing request for older hosts. Whole-turn pagination can require a capability; reading conversation history cannot. ## Client capability ownership diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 0dd2c2093f..812718be58 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -50,10 +50,11 @@ all currently available rows of an active run. There is no per-turn cap: a large exceed the target. Local history mounting uses the same boundaries so revealing older rows does not change the collapse state of a turn already displayed. -`wholeTurns` is optional for existing SDK callers. The app requires -`features.wholeTurnTimeline` once at the client fetch boundary; unsupported hosts receive an -update-host error. Forward catch-up stays incremental, but its reset snapshot honors the -whole-turn request. Source coverage still includes interleaved tool updates. +`wholeTurns` is a preference negotiated once at the client fetch boundary. Hosts advertising +`features.wholeTurnTimeline` receive it; other hosts receive the existing item-based request. +Reading history remains available on older hosts, whose pages can still split turns. Forward +catch-up stays incremental, but its reset snapshot honors a supported whole-turn request. +Source coverage still includes interleaved tool updates. Canonical IDs define runs. With missing IDs, human prompts, heartbeats and typed received agent messages each start a fallback turn. This deterministic fallback is shared by paging diff --git a/packages/app/e2e/browser/owned-subscriptions-old-daemon.spec.ts b/packages/app/e2e/browser/owned-subscriptions-old-daemon.spec.ts index 4b49d8105e..1fdef794a0 100644 --- a/packages/app/e2e/browser/owned-subscriptions-old-daemon.spec.ts +++ b/packages/app/e2e/browser/owned-subscriptions-old-daemon.spec.ts @@ -10,9 +10,12 @@ import { buildCreateAgentPreferences, buildSeededHost } from "../support/helpers import { startIsolatedHostDaemon } from "../support/helpers/isolated-host-daemon"; import type { MockAgentWorkspace } from "../support/helpers/mock-agent"; import { seedWorkspace } from "../support/helpers/seed-client"; -import { openAgentTimeline } from "../support/helpers/timeline-pagination"; +import { + openAgentTimeline, + expectTimelinePromptVisible, +} from "../support/helpers/timeline-pagination"; -test("requires a host update for whole-turn history while retaining old-daemon connections and terminals", async ({ +test("loads history across browser reconnection while retaining old-daemon connections and terminals", async ({ page, }) => { test.setTimeout(120_000); @@ -70,17 +73,14 @@ test("requires a host update for whole-turn history while retaining old-daemon c ); await openAgentTimeline(page, agent, serverId); - const updateMessage = page.getByText("Update the host to load complete conversation turns.", { - exact: true, - }); - await expect(updateMessage).toBeVisible(); + await expectTimelinePromptVisible(page, "emit 1 coalesced agent stream updates"); expect(observationSockets.size).toBe(1); - await test.step("retain one directory observation after reconnecting the old daemon", async () => { - await daemon.restart(); - await expect.poll(() => observationSockets.size, { timeout: 30_000 }).toBe(2); - await expect(updateMessage).toBeVisible(); - await agent.client.connect(); + await test.step("reconnect the browser and reload the old daemon history", async () => { + observationSockets.clear(); + await page.reload(); + await expectTimelinePromptVisible(page, "emit 1 coalesced agent stream updates"); + await expect.poll(() => observationSockets.size).toBe(1); }); await test.step("open a terminal and receive its output", async () => { diff --git a/packages/app/e2e/browser/turn-collapse.spec.ts b/packages/app/e2e/browser/turn-collapse.spec.ts index bef85080cf..c170386266 100644 --- a/packages/app/e2e/browser/turn-collapse.spec.ts +++ b/packages/app/e2e/browser/turn-collapse.spec.ts @@ -1,13 +1,29 @@ +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import path from "node:path"; import { holdOlderHistoryPages, userScrollsTimelineToHistoryStart, } from "../support/helpers/timeline-pagination"; import { getActiveTabTestId } from "../support/helpers/launcher"; -import type { Page } from "@playwright/test"; +import type { Page, Locator } from "@playwright/test"; import { test, expect } from "../support/fixtures"; import { expectComposerVisible } from "../support/helpers/composer"; import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent"; +test.use({ + e2eDaemonConfig: { + version: 1, + agents: { + providers: { + codex: { + enabled: true, + command: [process.execPath, path.resolve("e2e/fixtures/catalog-codex.mjs")], + }, + }, + }, + }, +}); + type ChatDisplay = "full" | "collapseToolCalls" | "collapseTurns"; async function useChatDisplay(page: Page, chatDisplay: ChatDisplay): Promise { @@ -59,6 +75,21 @@ async function groupedToolsInsideLatestTurn(page: Page): Promise { }); } +async function expectCompactMessageToggle(prompt: Locator, label: string) { + const button = prompt.getByRole("button", { name: label, exact: true }); + const text = prompt.getByText(label, { exact: true }); + await expect + .poll( + async () => { + const control = (await button.boundingBox())!; + const glyphs = (await text.boundingBox())!; + return control.height - glyphs.height; + }, + { message: "The inline toggle adds no vertical space beyond its text" }, + ) + .toBe(0); +} + async function expandAndCollapseLongPrompt(page: Page) { const prompt = page.getByTestId("user-message").last(); const before = await prompt.boundingBox(); @@ -67,6 +98,7 @@ async function expandAndCollapseLongPrompt(page: Page) { await expect .poll(async () => (await prompt.getByText("Show more", { exact: true }).boundingBox())!.x) .toBe(messageLeft); + await expectCompactMessageToggle(prompt, "Show more"); await prompt.getByRole("button", { name: "Show more", exact: true }).click(); await expect(prompt.getByRole("button", { name: "Show less", exact: true })).toBeVisible(); await expect @@ -75,6 +107,7 @@ async function expandAndCollapseLongPrompt(page: Page) { await expect .poll(async () => (await prompt.boundingBox())?.height ?? 0) .toBeGreaterThan(before!.height); + await expectCompactMessageToggle(prompt, "Show less"); await prompt.getByRole("button", { name: "Show less", exact: true }).click(); await expect.poll(async () => (await prompt.boundingBox())?.height ?? 0).toBe(before!.height); await prompt.scrollIntoViewIfNeeded(); @@ -128,18 +161,21 @@ test.describe("Turn collapse", () => { }); test("received messages show their sender and open that agent", async ({ page }) => { + await page.addInitScript(() => + localStorage.setItem("@paseo:app-settings", JSON.stringify({ theme: "dark" })), + ); const recipient = await seedMockAgentWorkspace({ repoPrefix: "sender-attribution-", title: "Implementation", }); try { const sender = await recipient.client.createAgent({ - provider: "mock", + provider: "codex", cwd: recipient.cwd, workspaceId: recipient.workspaceId, title: "Reviewer", - model: "e2e-fast-stream", - modeId: "load-test", + model: "gpt-6.1-sol", + modeId: "full-access", }); await recipient.client.sendAgentMessage( recipient.agentId, @@ -153,6 +189,10 @@ test("received messages show their sender and open that agent", async ({ page }) await expect(page.getByTestId("user-message").filter({ has: attribution })).toContainText( "I reviewed the turn-grouping changes.", ); + const labelColor = await attribution + .getByText("Sent by Reviewer", { exact: true }) + .evaluate((element) => getComputedStyle(element).color); + await expect(attribution.locator("svg")).toHaveCSS("fill", labelColor); await attribution.scrollIntoViewIfNeeded(); await page.screenshot({ path: "/tmp/sender-attribution.png" }); await attribution.click(); @@ -237,7 +277,7 @@ async function useThinkingDisplay( } for (const autoExpandReasoning of [false, true]) { - test(`reasoning follows Expand thinking (${autoExpandReasoning}) and opens fully with its turn`, async ({ + test(`reasoning follows Expand thinking (${autoExpandReasoning}) when opening its turn`, async ({ page, }) => { await page.setViewportSize({ width: 1280, height: 1000 }); @@ -261,15 +301,32 @@ for (const autoExpandReasoning of [false, true]) { await expectComposerVisible(page); const summary = page.getByTestId("turn-summary").last().getByRole("button"); await expect(summary).toBeVisible(); + await expect(reasoning).toHaveCount(0); await summary.click(); - await expect(reasoning).toBeVisible(); + await expect(reasoning).toHaveCount(autoExpandReasoning ? 1 : 0); await expect.poll(() => groupedToolsInsideLatestTurn(page)).toEqual([]); + await openOneToolOutput(page); + await summary.click(); + await expect(reasoning).toHaveCount(0); } finally { await agent.cleanup(); } }); } +async function openOneToolOutput(page: Page) { + const output = page.getByText(/\[burst\] tick 1 userIsAtBottom=true/).filter({ visible: true }); + await expect(output).toHaveCount(0); + const command = page + .getByTestId("tool-call-badge") + .filter({ hasText: "node scripts/simulate-stream-burst.mjs" }) + .first(); + await command.getByRole("button").first().click(); + await expect(output.first()).toBeVisible(); + await command.getByRole("button").first().click(); + await expect(output).toHaveCount(0); +} + async function expectThinkingPresentation(page: Page, autoExpandReasoning: boolean) { const reasoning = page .getByText(/Need to find the scroll container, the layout effect/) @@ -296,3 +353,61 @@ async function switchToCollapsedTurns(page: Page) { }); await page.reload(); } + +test("a scheduled heartbeat opens a separate turn without reopening the previous response", async ({ + page, +}) => { + test.setTimeout(90_000); + await useChatDisplay(page, "collapseTurns"); + const agent = await seedMockAgentWorkspace({ + repoPrefix: "heartbeat-turn-", + title: "Heartbeat turns", + model: "ten-second-stream", + initialPrompt: "Inspect the scroll container", + }); + const schedules = agent.client as unknown as Pick< + DaemonClient, + "scheduleCreate" | "scheduleRunOnce" | "scheduleDelete" + >; + let scheduleId: string | undefined; + try { + await agent.client.waitForFinish(agent.agentId, 30_000); + await openAgentRoute(page, agent); + await expectComposerVisible(page); + const summaries = page.getByTestId("turn-summary"); + await expect(summaries).toHaveCount(1); + const previousSummary = await summaries.first().innerText(); + const previousReplyId = await page + .getByTestId("assistant-message") + .last() + .evaluate((element) => element.closest("[data-message-id]")!.getAttribute("data-message-id")); + const created = await schedules.scheduleCreate({ + name: "Check the completed work", + prompt: "Inspect the scroll container again", + cadence: { type: "cron", expression: "0 0 1 1 *" }, + target: { type: "self", agentId: agent.agentId }, + runOnCreate: false, + }); + expect(created.error).toBeNull(); + scheduleId = created.schedule!.id; + const fired = await schedules.scheduleRunOnce({ id: scheduleId }); + expect(fired.error).toBeNull(); + await expect( + page.getByTestId("tool-call-badge").filter({ hasText: /^Heartbeat/ }), + ).toBeVisible(); + await expect(summaries.first()).toHaveText(previousSummary); + const previousReply = page + .locator(`[data-message-id="${previousReplyId}"]`) + .getByTestId("assistant-message"); + await expect(previousReply).toBeVisible(); + await expect(previousReply).toContainText("(end of synthetic stream)"); + await agent.client.waitForFinish(agent.agentId, 30_000); + await expect(summaries).toHaveCount(2); + await expect.poll(() => rowsBetweenLatestSummaryAndReply(page)).toBe(0); + await expect(page.getByRole("button", { name: /^Worked for/ })).toHaveCount(2); + await page.screenshot({ path: "/tmp/heartbeat-turn-footer.png" }); + } finally { + if (scheduleId) await schedules.scheduleDelete({ id: scheduleId }); + await agent.cleanup(); + } +}); diff --git a/packages/app/src/agent-stream/presentation.test.ts b/packages/app/src/agent-stream/presentation.test.ts index bba7267a5d..0cfc793228 100644 --- a/packages/app/src/agent-stream/presentation.test.ts +++ b/packages/app/src/agent-stream/presentation.test.ts @@ -707,7 +707,7 @@ it.each([false, true])( }; const present = createStreamPresentation(); const closed = present(input); - expect(closed.tail.some((item) => item.id === "thought")).toBe(autoExpandReasoning); + expect(closed.tail.some((item) => item.id === "thought")).toBe(false); const opened = present({ ...input, expandedTurnHostIds: new Set(["prompt"]) }); expect(opened.tail).toEqual(tail); expect(opened.turns.nestedItemIds.has("thought")).toBe(true); diff --git a/packages/app/src/agent-stream/presentation.ts b/packages/app/src/agent-stream/presentation.ts index ad5d362d38..5138cc1647 100644 --- a/packages/app/src/agent-stream/presentation.ts +++ b/packages/app/src/agent-stream/presentation.ts @@ -111,7 +111,6 @@ export function createStreamPresentation() { ) { collapsed = collapseTurns({ tail: projection.tail, - autoExpandReasoning: input.autoExpandReasoning, isLatestTurnOpen, expandedHostIds: input.expandedTurnHostIds, getGroupedCalls: (itemId) => projection.groupsByHostId.get(itemId)?.run.calls, diff --git a/packages/app/src/agent-stream/strategy-web.test.tsx b/packages/app/src/agent-stream/strategy-web.test.tsx index 1f61d0a14c..5ad0be03c2 100644 --- a/packages/app/src/agent-stream/strategy-web.test.tsx +++ b/packages/app/src/agent-stream/strategy-web.test.tsx @@ -7,7 +7,7 @@ import { act } from "react"; import { createRoot, type Root } from "react-dom/client"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { RetainedPanelActivity } from "@/components/retained-panel"; -import type { StreamItem } from "@/types/stream"; +import { reduceStreamUpdate, type StreamItem } from "@/types/stream"; import type { StreamRenderInput, StreamSegmentRenderers, StreamViewportHandle } from "./strategy"; import { createWebStreamStrategy } from "./strategy-web"; @@ -131,6 +131,79 @@ describe("createWebStreamStrategy", () => { vi.restoreAllMocks(); }); + it.each(["liveHead", "historyMounted"] as const)( + "keeps one rendered tool row after late prior-turn updates in %s", + (lane) => { + const strategy = createWebStreamStrategy({ isMobileBreakpoint: true }); + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + const input: StreamRenderInput = { + agentId: "agent", + segments: { historyVirtualized: [], historyMounted: [], liveHead: [] }, + boundary: { + hasVirtualizedHistory: false, + hasMountedHistory: lane === "historyMounted", + hasLiveHead: lane === "liveHead", + }, + renderers: createRenderers(() => {}), + listEmptyComponent: null, + viewportRef: React.createRef(), + routeBottomAnchorRequest: null, + isAuthoritativeHistoryReady: true, + onNearBottomChange: () => {}, + onNearHistoryStart: () => true, + isLoadingOlderHistory: false, + hasOlderHistory: false, + olderHistoryProgressKey: null, + scrollEnabled: true, + listStyle: null, + baseListContentContainerStyle: null, + forwardListContentContainerStyle: null, + contentMaxWidth: DEFAULT_CONTENT_MAX_WIDTH, + }; + let items: StreamItem[] = []; + const updates = [ + { callId: "background", turnId: "old", status: "running" }, + { callId: "exec", turnId: "current", status: "running" }, + { callId: "background", turnId: "old", status: "completed" }, + { callId: "exec", turnId: "current", status: "completed" }, + ] as const; + for (const [index, update] of updates.entries()) { + items = reduceStreamUpdate( + items, + { + type: "timeline", + provider: "codex", + turnId: update.turnId, + item: { + type: "tool_call", + callId: update.callId, + name: "exec_command", + status: update.status, + error: null, + detail: { type: "shell", command: update.callId }, + }, + }, + new Date(index), + ); + act(() => + root?.render( + strategy.render({ + ...input, + segments: { ...input.segments, [lane]: items }, + }), + ), + ); + } + const rows = [...container.querySelectorAll("[data-history-row-id]")]; + expect(rows.map((row) => row.getAttribute("data-history-row-id"))).toEqual([ + "agent_tool_turn:old/background", + "agent_tool_turn:current/exec", + ]); + }, + ); + it("mounts virtualized history without recursive row measurement updates", () => { const rowRenderCount = vi.fn(); const strategy = createWebStreamStrategy({ isMobileBreakpoint: true }); diff --git a/packages/app/src/agent-stream/turn-collapse/model.ts b/packages/app/src/agent-stream/turn-collapse/model.ts index c8c1617e24..20ab84a817 100644 --- a/packages/app/src/agent-stream/turn-collapse/model.ts +++ b/packages/app/src/agent-stream/turn-collapse/model.ts @@ -29,7 +29,6 @@ export interface TurnCollapse { export interface CollapseTurnsInput { tail: StreamItem[]; - autoExpandReasoning?: boolean; /** The latest turn is still producing output, so it stays unfolded. */ isLatestTurnOpen: boolean; expandedHostIds: ReadonlySet; @@ -58,12 +57,8 @@ function hasWork(summary: TurnSummary): boolean { return Object.keys(paseoActions).length > 0 || Object.values(counts).some((count) => count > 0); } -function staysVisibleWhenCollapsed(item: StreamItem, autoExpandReasoning = false): boolean { - return ( - isTurnTrigger(item) || - item.kind === "compaction" || - (autoExpandReasoning && item.kind === "thought") - ); +function staysVisibleWhenCollapsed(item: StreamItem): boolean { + return isTurnTrigger(item) || item.kind === "compaction"; } function findLastAssistantIndex(tail: StreamItem[], start: number, end: number): number { @@ -77,7 +72,7 @@ function findLastAssistantIndex(tail: StreamItem[], start: number, end: number): * Folds each answered turn into its user message: everything the agent did before its last * message collapses into one summary line, and that message stays in the chat. A turn folds * only once it is complete, its user message is loaded, and it sent messages or called tools - * before its reply. Reasoning stays visible when Expand thinking is enabled. + * before its reply. Reasoning is revealed only when the turn opens. */ export function collapseTurns(input: CollapseTurnsInput): TurnCollapse { const { tail } = input; @@ -121,9 +116,7 @@ export function collapseTurns(input: CollapseTurnsInput): TurnCollapse { if (expanded) nestedItemIds.add(item.id); } output.push( - ...(expanded - ? revealed - : folded.filter((item) => staysVisibleWhenCollapsed(item, input.autoExpandReasoning))), + ...(expanded ? revealed : folded.filter((item) => staysVisibleWhenCollapsed(item))), ); output.push(...tail.slice(replyIndex, runEnd)); }; diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 1cfddc8ecc..049f0d704b 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -792,9 +792,6 @@ const AgentStreamViewComponent = forwardRef - presentation.turns.nestedItemIds.has(id), - ); const renderThoughtItem = useCallback( (layoutItem: StreamLayoutItem, item: Extract) => { return ( @@ -804,11 +801,11 @@ const AgentStreamViewComponent = forwardRef ); }, - [autoExpandReasoning, setInlineDetailsExpanded, isExpandedTurnItem], + [autoExpandReasoning, setInlineDetailsExpanded], ); const renderSingleToolCallItem = useCallback( @@ -838,8 +835,6 @@ const AgentStreamViewComponent = forwardRef ); }, - [ - context.cwd, - resolvedServerId, - setInlineDetailsExpanded, - handleToolCallOpenFile, - isExpandedTurnItem, - ], + [context.cwd, resolvedServerId, setInlineDetailsExpanded, handleToolCallOpenFile], ); // Read through a stable event so live group updates do not change the renderer identity diff --git a/packages/app/src/components/user-message/index.tsx b/packages/app/src/components/user-message/index.tsx index 18a48798fb..d420ffdde5 100644 --- a/packages/app/src/components/user-message/index.tsx +++ b/packages/app/src/components/user-message/index.tsx @@ -61,11 +61,21 @@ export function MessageSender({ return session?.agents.get(sender.id) ?? session?.agentDetails.get(sender.id); }); const ProviderIcon = useProviderIcon(agent?.provider ?? "", host); + const renderProviderIcon = useCallback( + (color: string) => , + [ProviderIcon], + ); const openSender = useCallback(() => { if (host) navigateToAgent({ serverId: host, agentId: sender.id }); }, [host, sender.id]); return ( - ); @@ -74,6 +84,6 @@ export function MessageSender({ const styles = StyleSheet.create((theme) => ({ preview: { maxHeight: PREVIEW_HEIGHT, overflow: "hidden" }, body: { flexShrink: 0 }, - toggle: { alignSelf: "flex-start", paddingHorizontal: 0, borderWidth: 0 }, + toggle: { alignSelf: "flex-start", paddingHorizontal: 0, minHeight: 0, borderWidth: 0 }, fade: { position: "absolute", bottom: 0, left: 0, right: 0, height: theme.spacing[12] }, })); diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 9e57142846..0e401ad7a0 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -19,6 +19,66 @@ import { type TimelineCursor, } from "./session-stream-reducers"; +it("keeps streamed reasoning together while the initial history is loading", () => { + let head: StreamItem[] = []; + const chunks = ["Checking a long oper", "ation before", " the re", "sult arrives."]; + for (const [index, text] of chunks.entries()) { + const result = processAgentStreamEvent({ + event: { + type: "timeline", + provider: "codex", + turnId: "turn-18", + item: { type: "reasoning", text }, + }, + seq: index + 1, + epoch: "epoch-1", + currentTail: [], + currentHead: head, + currentCursor: undefined, + timestamp: new Date(1000 + index), + hasAuthoritativeBaseline: false, + }); + head = result.head; + expect(head).toHaveLength(1); + expect(head[0]).toMatchObject({ + kind: "thought", + text: chunks.slice(0, index + 1).join(""), + turnId: "turn-18", + }); + } +}); + +it.each([false, true])( + "keeps adjacent reasoning from different turns separate (history loaded: %s)", + (hasAuthoritativeBaseline) => { + let head: StreamItem[] = []; + let tail: StreamItem[] = []; + for (const [index, turnId] of ["turn-1", "turn-2"].entries()) { + const result = processAgentStreamEvent({ + event: { + type: "timeline", + provider: "codex", + turnId, + item: { type: "reasoning", text: turnId }, + }, + seq: index + 1, + epoch: "epoch-1", + currentTail: tail, + currentHead: head, + currentCursor: undefined, + timestamp: new Date(1000), + hasAuthoritativeBaseline, + }); + head = result.head; + tail = result.tail; + } + expect([...tail, ...head]).toMatchObject([ + { kind: "thought", text: "turn-1", turnId: "turn-1" }, + { kind: "thought", text: "turn-2", turnId: "turn-2" }, + ]); + }, +); + // --------------------------------------------------------------------------- // Test helpers // --------------------------------------------------------------------------- diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 34a461d893..a46ec8188f 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -1615,19 +1615,17 @@ export function processAgentStreamEvent( unmatchedUserMessageInsert: "head", }); } else { - const overlay = applyStreamEvent({ - tail: currentHead, - head: [], - event, - timestamp, + // Before history arrives, the overlay is one ordered stream. Reducing it + // directly keeps consecutive reasoning chunks in the same row. + const overlay = reduceStreamUpdate(currentHead, event, timestamp, { source: "live", timelineCursor, }); streamResult = { tail: currentTail, - head: [...overlay.tail, ...overlay.head], + head: overlay, changedTail: false, - changedHead: overlay.changedTail || overlay.changedHead, + changedHead: overlay !== currentHead, }; } } else { diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 62e26698db..376d04ea67 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -661,6 +661,82 @@ describe("stream reducer canonical tool calls", () => { expect(new Set(tools.map((tool) => tool.id)).size).toBe(2); }); + it.each(["live", "canonical"] as const)( + "keeps current tool identity during %s updates when an older turn reports late", + (source) => { + let state: StreamItem[] = []; + const events = [ + { callId: "background", turnId: "codex-turn-17", status: "running" as const }, + { callId: "exec-current", turnId: "codex-turn-18", status: "running" as const }, + { callId: "background", turnId: "codex-turn-17", status: "completed" as const }, + { callId: "exec-current", turnId: "codex-turn-18", status: "completed" as const }, + ]; + for (const [index, item] of events.entries()) { + state = reduceStreamUpdate( + state, + canonicalToolTimeline({ + provider: "codex", + name: "exec_command", + ...item, + }), + new Date(index * 1000), + { source }, + ); + } + expect( + state.filter(isAgentToolCallItem).map((item) => ({ + id: item.id, + turnId: item.turnId, + status: item.payload.data.status, + })), + ).toEqual([ + { + id: "agent_tool_turn:codex-turn-17/background", + turnId: "codex-turn-17", + status: "completed", + }, + { + id: "agent_tool_turn:codex-turn-18/exec-current", + turnId: "codex-turn-18", + status: "completed", + }, + ]); + }, + ); + + it("does not move reasoning into an older turn when a late tool update finalizes it", () => { + const background = canonicalToolTimeline({ + provider: "codex", + name: "exec_command", + callId: "background", + turnId: "old", + status: "running", + }); + let state = reduceStreamUpdate([], background, new Date(0)); + state = reduceStreamUpdate( + state, + { + type: "timeline", + provider: "codex", + turnId: "current", + item: { type: "reasoning", text: "Working on the current request" }, + }, + new Date(1), + ); + state = reduceStreamUpdate( + state, + canonicalToolTimeline({ + provider: "codex", + name: "exec_command", + callId: "background", + turnId: "old", + status: "completed", + }), + new Date(2), + ); + expect(state.at(-1)).toMatchObject({ kind: "thought", turnId: "current", status: "ready" }); + }); + it("merges lifecycle updates for the same call occurrence", () => { const callId = "tool-updated-within-turn"; const turnId = "autonomous-turn-1"; diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index b1d0903a92..0226993a8c 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -988,6 +988,7 @@ function appendThought( text: string, timestamp: Date, timelineCursor?: TimelinePosition, + turnId?: string, ): StreamItem[] { const { chunk, hasContent } = normalizeChunk(text); if (!chunk) { @@ -995,7 +996,7 @@ function appendThought( } const last = state[state.length - 1]; - if (last && last.kind === "thought") { + if (last && last.kind === "thought" && last.turnId === turnId) { const updated: ThoughtItem = { ...last, ...(timelineCursor ? { timelineCursor } : {}), @@ -1513,36 +1514,30 @@ function reduceTimelineEvent( const item = event.item; switch (item.type) { case "user_message": - return finalizeActiveThoughts( - appendUserMessage( - state, - item.text, - timestamp, - source, - item.messageId, - item.clientMessageId, - timelineCursor, - event.turnId, - ), + return appendUserMessage( + state, + item.text, + timestamp, + source, + item.messageId, + item.clientMessageId, + timelineCursor, + event.turnId, ); case "assistant_message": - return finalizeActiveThoughts( - appendAssistantMessage( - state, - item.text, - timestamp, - source, - item.messageId, - reservedItemIds, - timelineCursor, - ), + return appendAssistantMessage( + state, + item.text, + timestamp, + source, + item.messageId, + reservedItemIds, + timelineCursor, ); case "reasoning": - return appendThought(state, item.text, timestamp, timelineCursor); + return appendThought(state, item.text, timestamp, timelineCursor, event.turnId); case "tool_call": - return finalizeActiveThoughts( - reduceTimelineToolCall(state, event, item, timestamp, timelineCursor), - ); + return reduceTimelineToolCall(state, event, item, timestamp, timelineCursor); case "todo": { const items: TodoEntry[] = (item.items ?? []).map((todo) => ({ text: todo.text, @@ -1551,9 +1546,7 @@ function reduceTimelineEvent( status: todo.status, activeForm: todo.activeForm, })); - return finalizeActiveThoughts( - appendTodoList(state, event.provider, items, timestamp, timelineCursor), - ); + return appendTodoList(state, event.provider, items, timestamp, timelineCursor); } case "error": { const notification: NotificationItem = { @@ -1565,7 +1558,7 @@ function reduceTimelineEvent( level: "error", message: item.message ?? "Unknown error", }; - return finalizeActiveThoughts(appendNotification(state, notification)); + return appendNotification(state, notification); } case "notification": { const notification: NotificationItem = { @@ -1577,16 +1570,12 @@ function reduceTimelineEvent( level: item.level, message: item.message, }; - return finalizeActiveThoughts(appendNotification(state, notification)); + return appendNotification(state, notification); } case "compaction": - return finalizeActiveThoughts( - reduceTimelineCompaction(state, item, timestamp, timelineCursor), - ); + return reduceTimelineCompaction(state, item, timestamp, timelineCursor); case "plugin": - return finalizeActiveThoughts( - appendPluginTimelineItem(state, item, timestamp, timelineCursor), - ); + return appendPluginTimelineItem(state, item, timestamp, timelineCursor); default: return state; } @@ -1603,8 +1592,9 @@ export function reduceStreamUpdate( ): StreamItem[] { const source = options?.source ?? "live"; switch (event.type) { - case "timeline": - return applyTimelineTurnId( + case "timeline": { + const updated = applyTimelineTurnId( + state, reduceTimelineEvent( state, event, @@ -1615,6 +1605,8 @@ export function reduceStreamUpdate( ), event, ); + return event.item.type === "reasoning" ? updated : finalizeActiveThoughts(updated); + } case "thread_started": case "turn_started": case "turn_completed": @@ -1630,6 +1622,7 @@ export function reduceStreamUpdate( } function applyTimelineTurnId( + previous: StreamItem[], items: StreamItem[], event: Extract, ): StreamItem[] { @@ -1639,15 +1632,14 @@ function applyTimelineTurnId( return reconcileCanonicalUserTurnMembership(items, clientMessageId, event.turnId); } - if (!event.turnId || items.length === 0) return items; - const index = items.length - 1; - const last = items[index]; - if (!last || last.turnId === event.turnId) return items; - return [ - ...items.slice(0, index), - { ...last, turnId: event.turnId } as StreamItem, - ...items.slice(index + 1), - ]; + if (!event.turnId || items === previous) return items; + // Assign membership only to rows written by this event, before finalizing + // unrelated reasoning rows. Late lifecycle updates can replace an earlier row. + return items.map((item, index) => + item === previous[index] || item.turnId === event.turnId + ? item + : ({ ...item, turnId: event.turnId } as StreamItem), + ); } function reconcileCanonicalUserTurnMembership( diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index f3b483193c..2a3b409cee 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -1640,6 +1640,77 @@ test("preserves legacy fetchAgent id overload", async () => { await expect(responsePromise).rejects.toThrow("legacy fetch sentinel"); }); +test.each([undefined, false, true])( + "loads history with whole-turn support advertised as %s", + async (wholeTurnTimeline) => { + const transport = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "history-compatibility", + reconnect: { enabled: false }, + transportFactory: () => transport.transport, + }); + clients.push(client); + const connected = client.connect(); + transport.triggerOpen({ + features: wholeTurnTimeline === undefined ? {} : { wholeTurnTimeline }, + }); + await connected; + + const requests = [ + { direction: "tail" as const }, + { direction: "before" as const, cursor: { epoch: "epoch-1", seq: 50 } }, + { direction: "after" as const, cursor: { epoch: "epoch-1", seq: 50 } }, + { direction: "before" as const, cursor: { epoch: "epoch-1", seq: 80 }, mergeWindow: true }, + ]; + for (const [index, request] of requests.entries()) { + const requestId = `history-${index}`; + const result = client.fetchAgentTimeline("agent-1", { + ...request, + requestId, + wholeTurns: true, + limit: 40, + projection: "projected", + }); + // Attach immediately so the old rejection is captured by the regression. + const completed = expect(result).resolves.toMatchObject({ requestId, entries: [] }); + expect(parseSentFrame(transport.sent.at(-1))).toEqual({ + type: "fetch_agent_timeline_request", + agentId: "agent-1", + requestId, + ...request, + limit: 40, + projection: "projected", + ...(wholeTurnTimeline === true ? { wholeTurns: true } : {}), + }); + transport.triggerMessage( + wrapSessionMessage({ + type: "fetch_agent_timeline_response", + payload: { + requestId, + agentId: "agent-1", + agent: null, + direction: request.direction, + projection: "projected", + epoch: "epoch-1", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 0, maxSeq: 0, nextSeq: 1 }, + startCursor: null, + endCursor: null, + hasOlder: false, + hasNewer: false, + entries: [], + error: null, + }, + }), + ); + await completed; + } + }, +); + test("honors explicit fetchAgentTimeline timeout below the session RPC default", async () => { useHeartbeatClock(); const logger = createMockLogger(); @@ -1658,11 +1729,6 @@ test("honors explicit fetchAgentTimeline timeout below the session RPC default", mock.triggerOpen(); await connectPromise; - await expect(client.fetchAgentTimeline("agent-1", { wholeTurns: true })).rejects.toThrow( - "Update the host to load complete conversation turns.", - ); - expect(mock.sent).toHaveLength(0); - const responsePromise = client.fetchAgentTimeline("agent-1", { requestId: "req-timeline-1", direction: "tail", diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 8840aa1eb4..524c3e264a 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -686,6 +686,7 @@ export interface FetchAgentOptions { } type LegacyFetchAgentOptions = Omit; export interface FetchAgentTimelineOptions { + /** Prefer whole-turn pages when supported; older hosts retain item-based paging. */ wholeTurns?: boolean; direction?: FetchAgentTimelineDirection; cursor?: FetchAgentTimelineCursor; @@ -3549,13 +3550,15 @@ export class DaemonClient { agentId: string, options: FetchAgentTimelineOptions = {}, ): Promise { - if (options.wholeTurns && this.lastServerInfoMessage?.features?.wholeTurnTimeline !== true) { - throw new Error("Update the host to load complete conversation turns."); - } + // COMPAT(wholeTurnTimeline): added after v0.13.2 (#8010); remove after 2027-04-27 once the daemon floor supports whole-turn pages. + // Reading history predates this capability and must keep working on older hosts. + const supportsWholeTurns = this.lastServerInfoMessage?.features?.wholeTurnTimeline === true; const resolvedRequestId = this.createRequestId(options.requestId); const message = SessionInboundMessageSchema.parse({ type: "fetch_agent_timeline_request", - ...(options.wholeTurns !== undefined ? { wholeTurns: options.wholeTurns } : {}), + ...(supportsWholeTurns && options.wholeTurns !== undefined + ? { wholeTurns: options.wholeTurns } + : {}), agentId, requestId: resolvedRequestId, ...(options.direction ? { direction: options.direction } : {}),