Fix history compatibility and turn presentation regressions (#8010)

* fix(client): preserve history loading on older hosts

Whole-turn paging is optional, but requiring its capability blocked existing history reads before any request was sent. Negotiate the paging enhancement at the client boundary while preserving item-based requests for older daemons.

* fix(app): preserve row ownership during late turn updates

Late lifecycle events retagged the newest row even when they updated an earlier call, causing the next update to append a duplicate tool ID. Assign turn membership only to rows written by the event before finalizing unrelated reasoning.

Co-authored-by: luke <luopeihuan@gmail.com>

* fix(app): preserve reasoning continuity and turn expansion controls

* fix(app): hide reasoning inside collapsed turns

---------

Co-authored-by: luke <luopeihuan@gmail.com>
This commit is contained in:
Mohamed Boudra
2026-10-27 15:49:42 +07:00
committed by GitHub
co-authored by luke
parent 8a542048b6
commit ef58ace424
16 changed files with 492 additions and 117 deletions
+2 -2
View File
@@ -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
+5 -4
View File
@@ -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
@@ -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 () => {
+121 -6
View File
@@ -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<void> {
@@ -59,6 +75,21 @@ async function groupedToolsInsideLatestTurn(page: Page): Promise<string[]> {
});
}
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();
}
});
@@ -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);
@@ -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,
@@ -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<StreamViewportHandle>(),
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 });
@@ -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<string>;
@@ -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));
};
+3 -14
View File
@@ -792,9 +792,6 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
[agentId, client, handleInlinePathPress, resolvedServerId, toast, workspaceRoot],
);
const isExpandedTurnItem = useStableEvent((id: string) =>
presentation.turns.nestedItemIds.has(id),
);
const renderThoughtItem = useCallback(
(layoutItem: StreamLayoutItem, item: Extract<StreamItem, { kind: "thought" }>) => {
return (
@@ -804,11 +801,11 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
text={item.text}
status={item.status}
isLastInSequence={layoutItem.isLastInToolSequence}
defaultExpanded={autoExpandReasoning || isExpandedTurnItem(item.id)}
defaultExpanded={autoExpandReasoning}
/>
);
},
[autoExpandReasoning, setInlineDetailsExpanded, isExpandedTurnItem],
[autoExpandReasoning, setInlineDetailsExpanded],
);
const renderSingleToolCallItem = useCallback(
@@ -838,8 +835,6 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
itemId={item.id}
onInlineDetailsExpandedChangeByItemId={setInlineDetailsExpanded}
toolName={data.name}
defaultExpanded={isExpandedTurnItem(item.id)}
forceInline={isExpandedTurnItem(item.id)}
error={data.error}
status={data.status}
detail={data.detail}
@@ -869,13 +864,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
/>
);
},
[
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
@@ -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 size={16} color={color} />,
[ProviderIcon],
);
const openSender = useCallback(() => {
if (host) navigateToAgent({ serverId: host, agentId: sender.id });
}, [host, sender.id]);
return (
<Button variant="ghost" size="sm" leftIcon={ProviderIcon} disabled={!host} onPress={openSender}>
<Button
variant="ghost"
size="sm"
leftIcon={renderProviderIcon}
disabled={!host}
onPress={openSender}
>
{t("turnSummary.sentBy", { name: agent?.title ?? sender.title ?? sender.id })}
</Button>
);
@@ -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] },
}));
@@ -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
// ---------------------------------------------------------------------------
@@ -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 {
+76
View File
@@ -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";
+40 -48
View File
@@ -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<AgentStreamEventPayload, { type: "timeline" }>,
): 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(
+71 -5
View File
@@ -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",
+7 -4
View File
@@ -686,6 +686,7 @@ export interface FetchAgentOptions {
}
type LegacyFetchAgentOptions = Omit<FetchAgentOptions, "agentId">;
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<FetchAgentTimelinePayload> {
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 } : {}),