fix(timeline): stabilize optimistic turn layout

Keep initial history loading dormant until user intent and preserve active-turn layout across delayed transport updates. Add painted-frame coverage for optimistic timeline transitions.
This commit is contained in:
Mohamed Boudra
2026-07-30 22:31:35 +02:00
parent 60c53973d8
commit 43c15f707d
22 changed files with 763 additions and 263 deletions
+499 -191
View File
@@ -4,24 +4,49 @@ import { expectAgentIdle } from "./helpers/agent-stream";
import { expectComposerVisible, submitMessage } from "./helpers/composer";
import { clickNewChat, gotoWorkspace } from "./helpers/launcher";
import { seedWorkspace } from "./helpers/seed-client";
import {
openAgentTimeline,
scrollTimelineToNewestLoadedEdge,
scrollTimelineUntilOlderHistoryIsReachable,
seedLongMockAgentTimeline,
} from "./helpers/timeline-pagination";
import {
openAgentRoute,
seedMockAgentWorkspace,
type MockAgentWorkspace,
} from "./helpers/mock-agent";
import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";
interface FrameRect {
top: number;
right: number;
bottom: number;
left: number;
width: number;
height: number;
}
interface ElementFrame {
mounted: boolean;
visible: boolean;
painted: boolean;
rect: FrameRect | null;
opacity: number | null;
backgroundColor: string | null;
}
interface ContentChildFrame extends ElementFrame {
key: string;
text: string;
}
interface TurnFrame {
scrollHeight: number;
workingVisible: boolean;
promptVisible: boolean;
promptTop: number | null;
assistantVisible: boolean;
at: number;
userRow: ElementFrame;
footerRow: ElementFrame;
spinner: ElementFrame;
interruptControl: ElementFrame;
composer: ElementFrame & { value: string | null };
contentChildren: ContentChildFrame[];
scroll: ElementFrame & {
scrollTop: number | null;
clientHeight: number | null;
scrollHeight: number | null;
};
}
declare global {
@@ -30,165 +55,485 @@ declare global {
}
}
const test = baseTest.extend<{ tenSecondAgent: MockAgentWorkspace }>({
tenSecondAgent: async ({ page: _page }, provide) => {
const test = baseTest.extend<{ streamingAgent: MockAgentWorkspace }>({
streamingAgent: async ({ page: _page }, provide) => {
const agent = await seedMockAgentWorkspace({
repoPrefix: "consecutive-ten-second-turns-",
title: "Consecutive ten-second turns",
model: "ten-second-stream",
initialPrompt: "emit 100 coalesced agent stream updates",
});
await agent.client.waitForFinish(agent.agentId, 15_000);
await provide(agent);
await agent.cleanup();
},
});
async function waitForTurnToComplete(page: Page, completedTurnCount: number): Promise<void> {
await expect(page.getByText("(end of synthetic stream)", { exact: true })).toHaveCount(
completedTurnCount + 1,
{ timeout: 30_000 },
);
await expectAgentIdle(page);
}
async function recordTurnFrames(page: Page, prompt: string): Promise<void> {
await page.evaluate((promptText) => {
const isVisible = (candidate: Element) => candidate.getBoundingClientRect().height > 0;
// The timeline does not exist yet when a new agent is being created, so it is
// resolved per frame rather than captured up front.
const findScroll = () =>
Array.from(document.querySelectorAll('[data-testid="agent-chat-scroll"]')).find(isVisible);
const state = { active: true, frames: [] as TurnFrame[] };
const emptyElement = (): ElementFrame => ({
mounted: false,
visible: false,
painted: false,
rect: null,
opacity: null,
backgroundColor: null,
});
const rectOf = (element: Element): FrameRect => {
const rect = element.getBoundingClientRect();
return {
top: rect.top,
right: rect.right,
bottom: rect.bottom,
left: rect.left,
width: rect.width,
height: rect.height,
};
};
const intersects = (first: FrameRect, second: FrameRect) =>
first.bottom > second.top &&
first.top < second.bottom &&
first.right > second.left &&
first.left < second.right;
const windowRect: FrameRect = {
top: 0,
right: innerWidth,
bottom: innerHeight,
left: 0,
width: innerWidth,
height: innerHeight,
};
const hasColor = (color: string) =>
color !== "" && color !== "none" && color !== "transparent" && color !== "rgba(0, 0, 0, 0)";
const isVisible = (element: Element, clip?: Element | null) => {
const rect = rectOf(element);
return (
element.isConnected &&
element.checkVisibility() &&
rect.width > 0 &&
rect.height > 0 &&
intersects(rect, windowRect) &&
(!clip || intersects(rect, rectOf(clip)))
);
};
const isPainted = (element: Element, clip?: Element | null) => {
if (!isVisible(element, clip)) return false;
return [element, ...Array.from(element.querySelectorAll("*"))].some((candidate) => {
if (!isVisible(candidate, clip)) return false;
const style = getComputedStyle(candidate);
if (Number(style.opacity) <= 0) return false;
if (hasColor(style.backgroundColor)) return true;
if (candidate.textContent?.trim() && hasColor(style.color)) return true;
if (candidate instanceof SVGElement) {
const fill = style.fill || candidate.getAttribute("fill") || "";
const stroke = style.stroke || candidate.getAttribute("stroke") || "";
return hasColor(fill) || hasColor(stroke);
}
return candidate instanceof HTMLImageElement || candidate instanceof HTMLCanvasElement;
});
};
const snapshot = (
element: Element | null | undefined,
clip?: Element | null,
painted?: boolean,
): ElementFrame => {
if (!element?.isConnected) return emptyElement();
const style = getComputedStyle(element);
return {
mounted: true,
visible: isVisible(element, clip),
painted: painted ?? isPainted(element, clip),
rect: rectOf(element),
opacity: Number(style.opacity),
backgroundColor: style.backgroundColor,
};
};
const unionRect = (elements: Element[]): FrameRect | null => {
if (elements.length === 0) return null;
const rects = elements.map(rectOf);
const top = Math.min(...rects.map((rect) => rect.top));
const right = Math.max(...rects.map((rect) => rect.right));
const bottom = Math.max(...rects.map((rect) => rect.bottom));
const left = Math.min(...rects.map((rect) => rect.left));
return { top, right, bottom, left, width: right - left, height: bottom - top };
};
const spinnerSnapshot = (
footer: Element | null | undefined,
clip: Element | null,
): ElementFrame => {
const dots = Array.from(footer?.querySelectorAll("*") ?? []).filter((candidate) =>
hasColor(getComputedStyle(candidate).backgroundColor),
);
if (dots.length === 0) return emptyElement();
const opacities = dots.map((dot) => Number(getComputedStyle(dot).opacity));
return {
mounted: true,
visible: dots.some((dot) => isVisible(dot, clip)),
painted: dots.some((dot, index) => opacities[index] > 0 && isVisible(dot, clip)),
rect: unionRect(dots),
opacity: Math.max(...opacities),
backgroundColor: getComputedStyle(dots[0]).backgroundColor,
};
};
const sample = () => {
const scroll = findScroll();
// Re-find the row every frame: reconciliation replaces the node, and a stale
// reference would keep reporting the position of something already detached.
const promptRow = Array.from(document.querySelectorAll('[data-testid="user-message"]')).find(
(candidate) => isVisible(candidate) && candidate.textContent?.includes(promptText),
const viewport = Array.from(
document.querySelectorAll('[data-testid="agent-chat-scroll"]'),
).find((candidate) => isVisible(candidate));
const promptRow = Array.from(
viewport?.querySelectorAll('[data-testid="user-message"]') ?? [],
).find((candidate) => candidate.textContent?.includes(promptText));
const footer = Array.from(
viewport?.querySelectorAll('[data-testid="turn-working-indicator"]') ?? [],
)[0];
const spinner = spinnerSnapshot(footer, viewport ?? null);
const footerRow = footer?.parentElement;
const composer = Array.from(
document.querySelectorAll('[aria-label="Message agent..."]'),
).find((candidate) => isVisible(candidate));
const composerRoot = composer?.closest('[data-testid="message-input-root"]');
const interrupt = Array.from(
composerRoot?.querySelectorAll('[role="button"][aria-label]') ?? [],
).find((candidate) =>
/stop agent|canceling agent/i.test(candidate.getAttribute("aria-label") ?? ""),
);
const scrollFrame = snapshot(viewport);
const contentChildren = Array.from(viewport?.firstElementChild?.children ?? []).map(
(child, index): ContentChildFrame =>
Object.assign(snapshot(child, viewport), {
key:
(child.querySelector('[data-testid="turn-working-indicator"]')
? "turn-footer"
: (child.getAttribute("data-history-row-id") ??
child.getAttribute("data-testid"))) ?? `child-${index}`,
text: child.textContent?.trim().slice(0, 120) ?? "",
}),
);
state.frames.push({
scrollHeight: scroll ? scroll.scrollHeight : 0,
workingVisible: Array.from(
document.querySelectorAll('[data-testid="turn-working-indicator"]'),
).some(isVisible),
promptVisible: Boolean(promptRow),
promptTop: promptRow ? promptRow.getBoundingClientRect().top : null,
assistantVisible: Array.from(
document.querySelectorAll('[data-testid="assistant-message"]'),
).some(isVisible),
at: performance.now(),
userRow: snapshot(promptRow, viewport),
footerRow: snapshot(footerRow, viewport, spinner.painted),
spinner,
interruptControl: snapshot(interrupt),
composer: {
...snapshot(composerRoot ?? composer),
value:
composer instanceof HTMLInputElement || composer instanceof HTMLTextAreaElement
? composer.value
: null,
},
contentChildren,
scroll: {
...scrollFrame,
scrollTop: viewport instanceof HTMLElement ? viewport.scrollTop : null,
clientHeight: viewport instanceof HTMLElement ? viewport.clientHeight : null,
scrollHeight: viewport instanceof HTMLElement ? viewport.scrollHeight : null,
},
});
if (state.active) requestAnimationFrame(sample);
if (state.active) scheduleAfterNextPaint();
};
const scheduleAfterNextPaint = () => {
requestAnimationFrame(() => setTimeout(sample, 0));
};
window.__consecutiveTurnFrames = state;
sample();
scheduleAfterNextPaint();
}, prompt);
}
async function expectUninterruptedTurn(page: Page): Promise<void> {
const result = await page.evaluate(() => {
async function waitForRecordedFrames(
page: Page,
predicate: "user-visible" | "turn-running",
count: number,
): Promise<void> {
await expect
.poll(() =>
page.evaluate(
(input) => {
const frames = window.__consecutiveTurnFrames?.frames ?? [];
return frames.filter((frame) => {
if (input.predicate === "user-visible") return frame.userRow.visible;
return (
frame.interruptControl.painted && frame.footerRow.painted && frame.spinner.painted
);
}).length;
},
{ predicate },
),
)
.toBeGreaterThanOrEqual(count);
}
async function recordPaintsFor(page: Page, durationMs: number): Promise<void> {
const until = await page.evaluate((duration) => performance.now() + duration, durationMs);
await expect
.poll(() => page.evaluate(() => window.__consecutiveTurnFrames?.frames.at(-1)?.at ?? 0))
.toBeGreaterThanOrEqual(until);
}
async function stopTurnFrameRecording(page: Page): Promise<TurnFrame[]> {
return page.evaluate(() => {
const state = window.__consecutiveTurnFrames;
if (!state) throw new Error("Turn frames were never recorded");
state.active = false;
const start = state.frames.findIndex((frame) => frame.promptVisible);
// The turn runs from the footer first appearing to it going away for good. Bounding
// it that way keeps this measuring one thing: the footer leaving the layout and
// coming back. How long the footer takes to appear after the row is a separate
// question and does not belong in this assertion.
const firstWorking = state.frames.findIndex((frame) => frame.workingVisible);
const lastWorking = state.frames.reduce(
(latest, frame, index) => (frame.workingVisible ? index : latest),
-1,
);
const turn = state.frames.slice(Math.max(start, firstWorking), lastWorking + 1);
const blinkAt = turn.findIndex((frame) => !frame.workingVisible);
let largestHeightDrop = 0;
let largestHeightDropAt = -1;
for (let index = 1; index < turn.length; index += 1) {
const drop = turn[index - 1].scrollHeight - turn[index].scrollHeight;
if (drop > largestHeightDrop) {
largestHeightDrop = drop;
largestHeightDropAt = index;
}
}
return {
start,
blinkAt,
largestHeightDrop,
turnFrameCount: turn.length,
largestHeightDropAt,
heightsAroundDrop: turn
.slice(Math.max(0, largestHeightDropAt - 6), largestHeightDropAt + 6)
.map((frame) => frame.scrollHeight)
.join(","),
};
return state.frames;
});
expect(result.start, "the submitted prompt never appeared").toBeGreaterThan(-1);
expect(
result.blinkAt,
`the working indicator blinked off at frame ${result.blinkAt} of ${result.turnFrameCount} and came back`,
).toBe(-1);
expect(
result.largestHeightDrop,
`the timeline lost ${result.largestHeightDrop}px of content mid-turn (first heights: ${result.heightsAroundDrop})`,
).toBe(0);
}
async function expectSubmittedPromptNeverMovedDown(page: Page): Promise<void> {
const result = await page.evaluate(() => {
const state = window.__consecutiveTurnFrames;
if (!state) throw new Error("Turn frames were never recorded");
state.active = false;
const start = state.frames.findIndex((frame) => frame.promptVisible);
const tops = state.frames
.slice(start)
.map((frame) => frame.promptTop)
.filter((top): top is number => top !== null);
let largestDownwardShift = 0;
let largestDownwardShiftAt = -1;
for (let index = 1; index < tops.length; index += 1) {
const shift = tops[index] - tops[index - 1];
if (shift > largestDownwardShift) {
largestDownwardShift = shift;
largestDownwardShiftAt = index;
}
function formatFrames(frames: TurnFrame[], centers: number | number[]): string {
const indexes = new Set<number>();
for (const center of Array.isArray(centers) ? centers : [centers]) {
for (let index = Math.max(0, center - 3); index <= center + 3; index += 1) {
if (index < frames.length) indexes.add(index);
}
return {
start,
largestDownwardShift,
topsAroundShift: tops
.slice(Math.max(0, largestDownwardShiftAt - 4), largestDownwardShiftAt + 4)
.map((top) => Math.round(top))
.join(","),
};
});
expect(result.start, "the submitted prompt never appeared").toBeGreaterThan(-1);
expect(
Math.round(result.largestDownwardShift),
`the submitted prompt was pushed down ${Math.round(result.largestDownwardShift)}px (tops: ${result.topsAroundShift})`,
).toBeLessThanOrEqual(1);
}
return Array.from(indexes)
.sort((left, right) => left - right)
.map((index) => `frame ${index}: ${JSON.stringify(frames[index])}`)
.join("\n");
}
test("keeps the first submitted prompt in place once the assistant starts streaming", async ({
page,
tenSecondAgent,
}) => {
await openAgentRoute(page, {
workspaceId: tenSecondAgent.workspaceId,
agentId: tenSecondAgent.agentId,
});
function expectRowContinuity(frames: TurnFrame[]): void {
const first = frames.findIndex((frame) => hasPaintedLayout(frame.userRow));
const baseline = frames[first]?.userRow.rect?.top;
const disappeared = frames.findIndex(
(frame, index) => index > first && !hasPaintedLayout(frame.userRow),
);
const reappeared = frames.findIndex(
(frame, index) => index > disappeared && disappeared >= 0 && hasPaintedLayout(frame.userRow),
);
const shifted = frames.findIndex(
(frame, index) =>
index > first &&
hasPaintedLayout(frame.userRow) &&
baseline !== undefined &&
Math.abs((frame.userRow.rect?.top ?? baseline) - baseline) > 1,
);
const violations = [
...(first < 0 ? ["submitted row never became visible"] : []),
...(disappeared >= 0
? [
`submitted row disappeared for ${Math.max(0, reappeared - disappeared)} recorded frames (${Math.max(0, (frames[reappeared]?.at ?? frames.at(-1)?.at ?? 0) - frames[disappeared].at).toFixed(1)}ms) before authoritative recovery`,
]
: []),
...(shifted >= 0
? [
`submitted row moved from ${baseline?.toFixed(1)} to ${frames[shifted].userRow.rect?.top.toFixed(1)}`,
]
: []),
];
let failure = Math.max(0, first);
if (shifted >= 0) failure = shifted;
if (disappeared >= 0) failure = disappeared;
expect(
violations,
`transition violations:\n${formatFrames(frames, [failure, Math.max(failure, reappeared)])}`,
).toEqual([]);
}
const prompt = "First prompt.";
const composer = page.getByRole("textbox", { name: "Message agent..." }).first();
await composer.fill(prompt);
interface FrameViolation {
frame: number;
reason: string;
}
function geometryChanged(
value: number | null | undefined,
expected: number | null | undefined,
): boolean {
if (value == null || expected == null) return true;
return Math.abs(value - expected) > 1;
}
function hasPaintedLayout(element: ElementFrame): boolean {
return [element.mounted, element.visible, element.painted].every(Boolean);
}
function violationsForChecks(
frame: number,
checks: Array<{ passes: boolean; reason: string }>,
): FrameViolation[] {
return checks.filter(({ passes }) => !passes).map(({ reason }) => ({ frame, reason }));
}
function collectElementViolations(
frame: TurnFrame,
baseline: TurnFrame | undefined,
index: number,
): FrameViolation[] {
return violationsForChecks(index, [
{
passes: hasPaintedLayout(frame.userRow),
reason: "submitted row left the painted layout",
},
{
passes: !geometryChanged(frame.userRow.rect?.top, baseline?.userRow.rect?.top),
reason: "submitted row moved vertically",
},
{
passes: [hasPaintedLayout(frame.footerRow), (frame.footerRow.rect?.height ?? 0) > 0].every(
Boolean,
),
reason: "footer row was absent from the painted layout",
},
{
passes: [
geometryChanged(frame.footerRow.rect?.top, baseline?.footerRow.rect?.top),
geometryChanged(frame.footerRow.rect?.height, baseline?.footerRow.rect?.height),
].every((hasChanged) => !hasChanged),
reason: "footer geometry moved",
},
{
passes: hasPaintedLayout(frame.spinner),
reason: "working spinner was not painted",
},
{
passes: [hasPaintedLayout(frame.composer), frame.composer.value === ""].every(Boolean),
reason: "composer was not painted and empty",
},
{
passes: [
geometryChanged(frame.composer.rect?.top, baseline?.composer.rect?.top),
geometryChanged(frame.composer.rect?.height, baseline?.composer.rect?.height),
].every((hasChanged) => !hasChanged),
reason: "composer geometry moved",
},
]);
}
function collectScrollViolations(
frame: TurnFrame,
baseline: TurnFrame | undefined,
index: number,
): FrameViolation[] {
return violationsForChecks(index, [
{
passes: [
frame.scroll.mounted,
frame.scroll.visible,
!geometryChanged(frame.scroll.rect?.top, baseline?.scroll.rect?.top),
!geometryChanged(frame.scroll.rect?.height, baseline?.scroll.rect?.height),
!geometryChanged(frame.scroll.clientHeight, baseline?.scroll.clientHeight),
].every(Boolean),
reason: "scroll viewport geometry moved",
},
{
passes: [
geometryChanged(frame.scroll.scrollTop, baseline?.scroll.scrollTop),
geometryChanged(frame.scroll.scrollHeight, baseline?.scroll.scrollHeight),
].every((hasChanged) => !hasChanged),
reason: "scroll content metrics moved",
},
]);
}
function collectContentViolations(
frame: TurnFrame,
stableContentChildren: Map<string, ContentChildFrame>,
index: number,
): FrameViolation[] {
return Array.from(stableContentChildren).flatMap(([key, expected]) => {
const child = frame.contentChildren.find((candidate) => candidate.key === key);
return violationsForChecks(index, [
{
passes: [
Boolean(child?.mounted),
!geometryChanged(child?.rect?.top, expected.rect?.top),
!geometryChanged(child?.rect?.bottom, expected.rect?.bottom),
!geometryChanged(child?.rect?.height, expected.rect?.height),
].every(Boolean),
reason: `content row ${key} geometry moved`,
},
]);
});
}
function expectAtomicIdleToRunningTransition(frames: TurnFrame[]): void {
const first = frames.findIndex((frame) => frame.userRow.visible);
const running = frames.findIndex(
(frame, index) =>
index >= first &&
frame.interruptControl.painted &&
frame.footerRow.painted &&
frame.spinner.painted,
);
const transition = frames.slice(first);
const baseline = transition[0];
const violations: FrameViolation[] = [];
const stableContentChildren = new Map(
baseline?.contentChildren
.filter(({ key }) => key !== "turn-footer")
.map((child) => [child.key, child]),
);
if (first < 0) violations.push({ frame: 0, reason: "submitted row never became visible" });
if (running < 0) violations.push({ frame: Math.max(first, 0), reason: "turn never visibly ran" });
for (const [offset, frame] of transition.entries()) {
const index = first + offset;
violations.push(...collectElementViolations(frame, baseline, index));
violations.push(...collectScrollViolations(frame, baseline, index));
violations.push(...collectContentViolations(frame, stableContentChildren, index));
}
for (let offset = 1; offset < transition.length; offset += 1) {
const before = transition[offset - 1];
const after = transition[offset];
const footerEnteredLayout = [
(before.footerRow.rect?.height ?? 0) <= 0,
(after.footerRow.rect?.height ?? 0) > 0,
].every(Boolean);
if (footerEnteredLayout) {
violations.push({
frame: first + offset,
reason: "footer height changed from zero to non-zero",
});
}
}
const failure = violations[0]?.frame ?? Math.max(first, 0);
expect(
violations,
`transition violations:\n${violations.map(({ frame, reason }) => `frame ${frame}: ${reason}`).join("\n")}\n\nframe record:\n${formatFrames(frames, [failure, running])}`,
).toEqual([]);
}
async function recordDelayedRunningTransition(
page: Page,
agent: MockAgentWorkspace,
): Promise<TurnFrame[]> {
const gate = await installDaemonWebSocketGate(page);
await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId });
await expectAgentIdle(page);
await expect(page.getByRole("button", { name: "Copy turn" })).toHaveCount(1);
const prompt = "Second prompt keeps streaming.";
gate.holdNextAgentUpdate(agent.agentId, "running");
gate.setMessageSubmissionDispositionStripped(true);
gate.setAgentStreamEventSuppressed("turn_started", true);
gate.setAgentStreamItemSuppressed("assistant_message", true);
await recordTurnFrames(page, prompt);
await composer.press("Enter");
await submitMessage(page, prompt);
await gate.waitForHeldServerMessage();
await expect(page.getByText(prompt, { exact: true })).toBeVisible();
await expect(page.getByTestId("assistant-message").first()).toBeVisible();
await waitForTurnToComplete(page, 0);
await expectSubmittedPromptNeverMovedDown(page);
});
let released = false;
try {
await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toBeVisible();
await expect(page.getByRole("textbox", { name: "Message agent..." }).first()).toHaveValue("");
await Promise.all([
gate.waitForAgentStreamItem("user_message"),
gate.waitForServerMessage("send_agent_message_response"),
]);
await recordPaintsFor(page, 80);
gate.releaseHeldServerMessage();
released = true;
await waitForRecordedFrames(page, "turn-running", 3);
const frames = await stopTurnFrameRecording(page);
gate.setMessageSubmissionDispositionStripped(false);
gate.setAgentStreamEventSuppressed("turn_started", false);
gate.setAgentStreamItemSuppressed("assistant_message", false);
return frames;
} finally {
if (!released) gate.releaseHeldServerMessage();
gate.setMessageSubmissionDispositionStripped(false);
gate.setAgentStreamEventSuppressed("turn_started", false);
gate.setAgentStreamItemSuppressed("assistant_message", false);
}
}
// The first prompt of a brand new agent is submitted before the agent exists, so the
// authoritative timeline arrives after the row is already on screen. That is the only
@@ -201,74 +546,37 @@ test("keeps the first prompt of a new agent in place through authoritative hydra
await page.addInitScript(() => {
localStorage.setItem(
"@paseo:create-agent-preferences",
JSON.stringify({ provider: "mock", providerPreferences: { mock: { mode: "load-test" } } }),
JSON.stringify({
provider: "mock",
providerPreferences: {
mock: { mode: "load-test", model: "ten-second-stream" },
},
}),
);
});
const gate = await installDaemonWebSocketGate(page);
await gotoWorkspace(page, workspace.workspaceId);
await clickNewChat(page);
await expectComposerVisible(page);
const prompt = "First prompt of a new agent.";
const composer = page.getByRole("textbox", { name: "Message agent..." }).first();
await composer.fill(prompt);
const prompt = "Delay synthetic user message by 300ms.";
await recordTurnFrames(page, prompt);
await composer.press("Enter");
await submitMessage(page, prompt);
await expect(
page.getByTestId("user-message").filter({ hasText: prompt }).first(),
).toBeVisible();
await expect(page.getByTestId("assistant-message").first()).toBeVisible({ timeout: 30_000 });
await expect(page.getByTestId("turn-working-elapsed")).toHaveCount(1);
await expectSubmittedPromptNeverMovedDown(page);
const submittedRow = page.getByTestId("user-message").filter({ hasText: prompt }).first();
await expect(submittedRow).toBeVisible();
await gate.waitForAgentStreamItem("user_message");
await recordPaintsFor(page, 80);
expectRowContinuity(await stopTurnFrameRecording(page));
} finally {
await workspace.cleanup();
}
});
test("keeps the turn footer and timeline height stable while a second turn runs", async ({
test("commits a follow-up prompt and running footer in one painted frame", async ({
page,
tenSecondAgent,
streamingAgent,
}) => {
await openAgentRoute(page, {
workspaceId: tenSecondAgent.workspaceId,
agentId: tenSecondAgent.agentId,
});
await submitMessage(page, "First prompt.");
await waitForTurnToComplete(page, 0);
const secondPrompt = "Second prompt.";
const composer = page.getByRole("textbox", { name: "Message agent..." }).first();
await composer.fill(secondPrompt);
await recordTurnFrames(page, secondPrompt);
await composer.press("Enter");
await expect(page.getByText(secondPrompt, { exact: true })).toBeVisible();
await waitForTurnToComplete(page, 1);
await expectUninterruptedTurn(page);
});
test("keeps the timeline height stable when submitting into a long timeline", async ({ page }) => {
const agent = await seedLongMockAgentTimeline({ turns: 40 });
try {
await openAgentTimeline(page, agent);
await expect(page.getByText(agent.newestPrompt, { exact: true })).toBeVisible();
// A real long chat is fully loaded and virtualized before the user types again.
await scrollTimelineUntilOlderHistoryIsReachable(page, agent.oldestPrompt);
await scrollTimelineToNewestLoadedEdge(page);
const prompt = "Stability probe into a long timeline.";
const composer = page.getByRole("textbox", { name: "Message agent..." }).first();
await composer.fill(prompt);
await recordTurnFrames(page, prompt);
await composer.press("Enter");
await expect(page.getByText(prompt, { exact: true }).last()).toBeVisible();
// Observe the whole turn: wait for it to start before waiting for it to end.
await expect(page.getByRole("button", { name: /stop|cancel/i }).first()).toBeVisible();
await expectAgentIdle(page, 60_000);
await expectUninterruptedTurn(page);
} finally {
await agent.cleanup();
}
const frames = await recordDelayedRunningTransition(page, streamingAgent);
expectAtomicIdleToRunningTransition(frames);
});
@@ -16,6 +16,7 @@ import {
rememberTimelinePromptPosition,
reloadAgentTimelineFromPersistedReplica,
scrollTimelineUntilOlderHistoryIsReachable,
scrollTimelineToOldestLoadedEdge,
scrollTimelineToNewestLoadedEdge,
seedLongMockAgentTimeline,
sendLiveTurnBeforeHydration,
@@ -143,6 +144,7 @@ test.describe("Agent timeline pagination", () => {
await makeLoadedTimelineFitViewport(page);
const history = await holdOlderHistoryPages(page, agent);
await openAgentTimeline(page, agent);
await scrollTimelineToOldestLoadedEdge(page);
await history.expectRequestedPages(1);
const loading = await rememberOlderHistoryLoadingOperation(page);
@@ -19,6 +19,11 @@ interface ClientRequest {
payload?: unknown;
}
interface HeldAgentUpdate {
agentId: string;
status: string;
}
function readSessionMessage(message: string | Buffer): ClientRequest | null {
if (typeof message !== "string") return null;
try {
@@ -122,15 +127,37 @@ function shouldSuppressServerMessage(input: {
message: ClientRequest | null;
messageTypes: ReadonlySet<string>;
agentStreamEventTypes: ReadonlySet<string>;
agentStreamItemTypes: ReadonlySet<string>;
suppressAgentStream: boolean;
}): boolean {
const messageType = typeof input.message?.type === "string" ? input.message.type : null;
if (messageType && input.messageTypes.has(messageType)) return true;
if (input.suppressAgentStream && messageType === "agent_stream") return true;
const itemType = readAgentStreamItemType(input.message);
if (itemType && input.agentStreamItemTypes.has(itemType)) return true;
const eventType = readAgentStreamEventType(input.message);
return Boolean(eventType && input.agentStreamEventTypes.has(eventType));
}
function shouldHoldServerMessage(
message: ClientRequest | null,
messageType: string | null,
agentUpdate: HeldAgentUpdate | null,
): boolean {
if (message?.type === messageType) return true;
if (message?.type !== "agent_update" || !agentUpdate) return false;
const payload = message.payload;
if (
!payload ||
typeof payload !== "object" ||
(payload as { kind?: unknown }).kind !== "upsert"
) {
return false;
}
const agent = (payload as { agent?: { id?: unknown; status?: unknown } }).agent;
return agent?.id === agentUpdate.agentId && agent.status === agentUpdate.status;
}
export async function installDaemonWebSocketGate(page: Page) {
let acceptingConnections = true;
let reconnectWithFreshClient = false;
@@ -142,10 +169,12 @@ export async function installDaemonWebSocketGate(page: Page) {
let heldClientRequest: { server: WebSocketRoute; message: string | Buffer } | null = null;
let resolveHeldClientRequest: (() => void) | null = null;
let heldServerMessageType: string | null = null;
let heldAgentUpdate: HeldAgentUpdate | null = null;
let heldServerMessage: { browser: WebSocketRoute; message: string | Buffer } | null = null;
let resolveHeldServerMessage: (() => void) | null = null;
const suppressedServerMessageTypes = new Set<string>();
const suppressedAgentStreamEventTypes = new Set<string>();
const suppressedAgentStreamItemTypes = new Set<string>();
const activeSockets = new Set<WebSocketRoute>();
let latestServer: WebSocketRoute | null = null;
const directoryStarts: DirectoryRequestStartCounts = {
@@ -237,7 +266,7 @@ export async function installDaemonWebSocketGate(page: Page) {
for (const resolve of serverMessageWaiters) resolve();
serverMessageWaiters.clear();
}
if (serverMessage?.type === heldServerMessageType) {
if (shouldHoldServerMessage(serverMessage, heldServerMessageType, heldAgentUpdate)) {
heldServerMessage = { browser: ws, message: outboundMessage };
resolveHeldServerMessage?.();
resolveHeldServerMessage = null;
@@ -248,6 +277,7 @@ export async function installDaemonWebSocketGate(page: Page) {
message: serverMessage,
messageTypes: suppressedServerMessageTypes,
agentStreamEventTypes: suppressedAgentStreamEventTypes,
agentStreamItemTypes: suppressedAgentStreamItemTypes,
suppressAgentStream,
})
)
@@ -296,6 +326,12 @@ export async function installDaemonWebSocketGate(page: Page) {
},
holdNextServerMessage(type: string): void {
heldServerMessageType = type;
heldAgentUpdate = null;
heldServerMessage = null;
},
holdNextAgentUpdate(agentId: string, status: string): void {
heldServerMessageType = null;
heldAgentUpdate = { agentId, status };
heldServerMessage = null;
},
waitForHeldServerMessage(): Promise<void> {
@@ -309,6 +345,7 @@ export async function installDaemonWebSocketGate(page: Page) {
heldServerMessage.browser.send(heldServerMessage.message);
heldServerMessage = null;
heldServerMessageType = null;
heldAgentUpdate = null;
},
requestTimelineTail(agentId: string): void {
if (!latestServer) throw new Error("No daemon WebSocket is connected");
@@ -381,6 +418,13 @@ export async function installDaemonWebSocketGate(page: Page) {
suppressedAgentStreamEventTypes.delete(type);
}
},
setAgentStreamItemSuppressed(type: string, suppressed: boolean): void {
if (suppressed) {
suppressedAgentStreamItemTypes.add(type);
} else {
suppressedAgentStreamItemTypes.delete(type);
}
},
setAssistantMessageIdsStripped(stripped: boolean): void {
stripAssistantMessageIds = stripped;
},
@@ -4,11 +4,16 @@ import {
createHistoryStartPaginationState,
evaluateHistoryStartPagination,
isHistoryStartLoadingOperation,
isHistoryStartSlotReserved,
rearmHistoryStartPagination,
settleHistoryStartPagination,
type HistoryStartPaginationInput,
} from "./history-start-pagination";
function createArmedHistoryStartPaginationState() {
return rearmHistoryStartPagination(createHistoryStartPaginationState());
}
const visibleHistoryStart: HistoryStartPaginationInput = {
distanceFromHistoryStart: 0,
hasOlderHistory: true,
@@ -18,9 +23,20 @@ const visibleHistoryStart: HistoryStartPaginationInput = {
};
describe("history start pagination", () => {
it("waits for user scroll intent before loading or reserving the loader slot", () => {
const dormant = createHistoryStartPaginationState();
const evaluated = evaluateHistoryStartPagination(dormant, visibleHistoryStart);
expect(evaluated).toEqual({ state: { status: "dormant" }, shouldLoad: false });
expect(isHistoryStartSlotReserved(evaluated.state, true)).toBe(false);
expect(isHistoryStartSlotReserved(rearmHistoryStartPagination(evaluated.state), true)).toBe(
true,
);
});
it("waits for anchored page geometry before authorizing another page", () => {
const first = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
const inFlight = evaluateHistoryStartPagination(first.state, {
@@ -43,7 +59,7 @@ describe("history start pagination", () => {
it("loads one page each time anchored geometry leaves and returns to history start", () => {
const first = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
const pageApplied = evaluateHistoryStartPagination(first.state, {
@@ -70,7 +86,7 @@ describe("history start pagination", () => {
it("continues the same loading operation when anchored geometry remains at history start", () => {
const first = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
const pageApplied = evaluateHistoryStartPagination(first.state, {
@@ -90,7 +106,7 @@ describe("history start pagination", () => {
it("latches a request that finishes without cursor progress", () => {
const first = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
const inFlight = evaluateHistoryStartPagination(first.state, {
@@ -115,7 +131,7 @@ describe("history start pagination", () => {
it("allows another user attempt after a request finishes without progress", () => {
const first = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
const inFlight = evaluateHistoryStartPagination(first.state, {
@@ -134,7 +150,7 @@ describe("history start pagination", () => {
it("latches an attempt that becomes invalid before entering flight", () => {
const requested = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
@@ -145,7 +161,7 @@ describe("history start pagination", () => {
it("does not mistake repeated edge observations for a finished request", () => {
const first = evaluateHistoryStartPagination(
createHistoryStartPaginationState(),
createArmedHistoryStartPaginationState(),
visibleHistoryStart,
);
const repeated = evaluateHistoryStartPagination(first.state, visibleHistoryStart);
@@ -155,7 +171,7 @@ describe("history start pagination", () => {
});
it("waits while history loading is unavailable or already active", () => {
const state = createHistoryStartPaginationState();
const state = createArmedHistoryStartPaginationState();
expect([
evaluateHistoryStartPagination(state, { ...visibleHistoryStart, isReady: false }).shouldLoad,
@@ -1,6 +1,7 @@
export const HISTORY_START_THRESHOLD_PX = 96;
export type HistoryStartPaginationState =
| { status: "dormant" }
| { status: "ready" }
| { status: "loading"; requestedProgressKey: string; requestObserved: boolean }
| { status: "settling"; loadedProgressKey: string }
@@ -20,7 +21,7 @@ export interface HistoryStartPaginationTransition {
}
export function createHistoryStartPaginationState(): HistoryStartPaginationState {
return { status: "ready" };
return { status: "dormant" };
}
export function isHistoryStartLoadingOperation(state: HistoryStartPaginationState): boolean {
@@ -30,7 +31,14 @@ export function isHistoryStartLoadingOperation(state: HistoryStartPaginationStat
export function rearmHistoryStartPagination(
state: HistoryStartPaginationState,
): HistoryStartPaginationState {
return state.status === "latched" ? { status: "ready" } : state;
return state.status === "dormant" || state.status === "latched" ? { status: "ready" } : state;
}
export function isHistoryStartSlotReserved(
state: HistoryStartPaginationState,
hasOlderHistory: boolean,
): boolean {
return state.status !== "dormant" && (hasOlderHistory || state.status !== "ready");
}
export function abandonHistoryStartPaginationRequest(
@@ -51,6 +59,9 @@ export function evaluateHistoryStartPagination(
state: HistoryStartPaginationState,
input: HistoryStartPaginationInput,
): HistoryStartPaginationTransition {
if (state.status === "dormant") {
return { state, shouldLoad: false };
}
if (state.status === "loading") {
if (input.progressKey !== null && input.progressKey !== state.requestedProgressKey) {
return {
+5 -5
View File
@@ -82,7 +82,7 @@ function strategyFor(platform: "web" | "android"): StreamStrategy {
function layoutFor(input: {
platform: "web" | "android";
agentStatus?: string;
isTurnActive?: boolean;
tail: StreamItem[];
head?: StreamItem[];
timingIds?: string[];
@@ -90,7 +90,7 @@ function layoutFor(input: {
const strategy = strategyFor(input.platform);
return layoutStream({
strategy,
agentStatus: input.agentStatus ?? "idle",
isTurnActive: input.isTurnActive ?? false,
history: orderTailForStreamRenderStrategy({
strategy,
streamItems: input.tail,
@@ -151,14 +151,14 @@ describe("layoutStream", () => {
const thirdBlock = assistantMessage("turn:block:2", 4, { groupId: "turn", index: 2 });
const splitLayout = layoutFor({
platform,
agentStatus: "running",
isTurnActive: true,
tail: [userMessage("u1", 1), firstBlock],
head: [secondBlock, thirdBlock],
timingIds: [firstBlock.id, secondBlock.id, thirdBlock.id],
});
const unsplitLayout = layoutFor({
platform,
agentStatus: "running",
isTurnActive: true,
tail: [userMessage("u1", 1), firstBlock, secondBlock, thirdBlock],
timingIds: [firstBlock.id, secondBlock.id, thirdBlock.id],
});
@@ -439,7 +439,7 @@ describe("layoutStream", () => {
const tool = toolCall("tool-1", 3);
const layout = layoutFor({
platform,
agentStatus: "running",
isTurnActive: true,
tail: [userMessage("u1", 1), assistant, tool],
timingIds: [assistant.id],
});
+2 -2
View File
@@ -36,7 +36,7 @@ export interface StreamLayout {
export interface StreamLayoutInput {
strategy: StreamStrategy;
agentStatus: string;
isTurnActive: boolean;
history: StreamItem[];
liveHead: StreamItem[];
timingByAssistantId: Map<string, TurnTiming>;
@@ -117,7 +117,7 @@ function findLatestAssistantInTurn(input: {
}
function resolveAuxiliaryTurnFooter(input: StreamLayoutInput): TurnFooterHost | null {
if (input.agentStatus === "running") {
if (input.isTurnActive) {
return null;
}
+8 -8
View File
@@ -35,7 +35,7 @@ describe("buildAgentStreamRenderModel", () => {
const head = [assistantMessage("live-a", 121)];
const model = buildAgentStreamRenderModel({
agentStatus: "running",
isTurnActive: true,
tail,
head,
platform: "web",
@@ -53,7 +53,7 @@ describe("buildAgentStreamRenderModel", () => {
const head = [assistantMessage("live-a", 3)];
const model = buildAgentStreamRenderModel({
agentStatus: "running",
isTurnActive: true,
tail,
head,
platform: "web",
@@ -71,14 +71,14 @@ describe("buildAgentStreamRenderModel", () => {
const secondHead = [assistantMessage("live-b", 4)];
const first = buildAgentStreamRenderModel({
agentStatus: "running",
isTurnActive: true,
tail,
head: firstHead,
platform: "native",
isMobileBreakpoint: false,
});
const second = buildAgentStreamRenderModel({
agentStatus: "running",
isTurnActive: true,
tail,
head: secondHead,
platform: "native",
@@ -95,7 +95,7 @@ describe("buildAgentStreamRenderModel", () => {
const head = [assistantMessage("live-a", 4)];
const model = buildAgentStreamRenderModel({
agentStatus: "running",
isTurnActive: true,
tail,
head,
platform: "web",
@@ -111,7 +111,7 @@ describe("buildAgentStreamRenderModel", () => {
const head = [assistantMessage("live-a", 4)];
const model = buildAgentStreamRenderModel({
agentStatus: "idle",
isTurnActive: false,
tail,
head,
platform: "web",
@@ -130,7 +130,7 @@ describe("buildAgentStreamRenderModel", () => {
const tail = [userMessage("u1", 1), assistantMessage("a1", 4)];
const model = buildAgentStreamRenderModel({
agentStatus: "idle",
isTurnActive: false,
tail,
head: [],
platform: "native",
@@ -149,7 +149,7 @@ describe("buildAgentStreamRenderModel", () => {
const tail = [userMessage("u1", 1), userMessage("u2", 4)];
const model = buildAgentStreamRenderModel({
agentStatus: "idle",
isTurnActive: false,
tail,
head: [],
platform: "web",
+10 -10
View File
@@ -35,7 +35,7 @@ export interface AgentStreamRenderModel {
}
export interface BuildAgentStreamRenderModelInput {
agentStatus: string;
isTurnActive: boolean;
tail: StreamItem[];
head: StreamItem[];
platform: "web" | "native";
@@ -56,7 +56,7 @@ const splitHistoryCache = new WeakMap<
>();
const turnTimingCache = new WeakMap<
StreamItem[],
WeakMap<StreamItem[], Map<string, StreamTurnTiming>>
WeakMap<StreamItem[], Map<boolean, StreamTurnTiming>>
>();
function getOrderedItems(params: {
@@ -131,7 +131,7 @@ function splitOrderedTail(params: {
}
function getTurnTiming(params: {
agentStatus: string;
isTurnActive: boolean;
tail: StreamItem[];
head: StreamItem[];
}): StreamTurnTiming {
@@ -140,17 +140,17 @@ function getTurnTiming(params: {
cachedByHead = new WeakMap();
turnTimingCache.set(params.tail, cachedByHead);
}
let cachedByStatus = cachedByHead.get(params.head);
if (!cachedByStatus) {
cachedByStatus = new Map();
cachedByHead.set(params.head, cachedByStatus);
let cachedByActivity = cachedByHead.get(params.head);
if (!cachedByActivity) {
cachedByActivity = new Map();
cachedByHead.set(params.head, cachedByActivity);
}
const cached = cachedByStatus.get(params.agentStatus);
const cached = cachedByActivity.get(params.isTurnActive);
if (cached) {
return cached;
}
const timing = deriveStreamTurnTiming(params);
cachedByStatus.set(params.agentStatus, timing);
cachedByActivity.set(params.isTurnActive, timing);
return timing;
}
@@ -188,7 +188,7 @@ export function buildAgentStreamRenderModel(
isMobileBreakpoint: input.isMobileBreakpoint,
});
const turnTiming = getTurnTiming({
agentStatus: input.agentStatus,
isTurnActive: input.isTurnActive,
tail: input.tail,
head: input.head,
});
@@ -36,6 +36,7 @@ import {
createHistoryStartPaginationState,
evaluateHistoryStartPagination,
isHistoryStartLoadingOperation,
isHistoryStartSlotReserved,
rearmHistoryStartPagination,
settleHistoryStartPagination,
type HistoryStartPaginationInput,
@@ -583,7 +584,7 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
const historyFooterContent = useMemo(() => {
const isLoadingOperation = isHistoryStartLoadingOperation(historyStartPaginationState);
if (!hasOlderHistory && !isLoadingOperation) {
if (!isHistoryStartSlotReserved(historyStartPaginationState, hasOlderHistory)) {
return null;
}
return (
@@ -21,6 +21,7 @@ import {
createHistoryStartPaginationState,
evaluateHistoryStartPagination,
isHistoryStartLoadingOperation,
isHistoryStartSlotReserved,
rearmHistoryStartPagination,
settleHistoryStartPagination,
type HistoryStartPaginationInput,
@@ -180,7 +181,6 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool
const [historyStartPaginationState, setHistoryStartPaginationState] = useState(
createHistoryStartPaginationState,
);
const [isHistoryStartSlotReserved, setIsHistoryStartSlotReserved] = useState(hasOlderHistory);
const historyStartPaginationStateRef = useRef(historyStartPaginationState);
const historyStartPrependAnchorRef = useRef<HistoryStartPrependAnchor | null>(null);
const historyStartPrependAnchorActiveRef = useRef(false);
@@ -783,14 +783,9 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool
const liveAuxiliary = useMemo(() => {
return renderLiveAuxiliary();
}, [renderLiveAuxiliary]);
useEffect(() => {
if (hasOlderHistory || isHistoryStartLoadingOperation(historyStartPaginationState)) {
setIsHistoryStartSlotReserved(true);
}
}, [hasOlderHistory, historyStartPaginationState]);
const historyStartSlot = useMemo(() => {
const isLoadingOperation = isHistoryStartLoadingOperation(historyStartPaginationState);
if (!isHistoryStartSlotReserved && !hasOlderHistory && !isLoadingOperation) {
if (!isHistoryStartSlotReserved(historyStartPaginationState, hasOlderHistory)) {
return null;
}
return (
@@ -803,7 +798,7 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool
) : null}
</div>
);
}, [hasOlderHistory, historyStartPaginationState, isHistoryStartSlotReserved]);
}, [hasOlderHistory, historyStartPaginationState]);
const shouldRenderEmpty =
!boundary.hasMountedHistory &&
!boundary.hasVirtualizedHistory &&
+17 -11
View File
@@ -242,6 +242,7 @@ export interface AgentStreamViewProps {
streamHead?: StreamItem[];
pendingPermissions: Map<string, PendingPermission>;
pendingMessageSubmissions?: readonly PendingMessageSubmission[];
hasActiveTurn?: boolean;
routeBottomAnchorRequest?: BottomAnchorRouteRequest | null;
isAuthoritativeHistoryReady?: boolean;
toast?: ToastApi | null;
@@ -332,6 +333,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
streamHead: providedStreamHead,
pendingPermissions,
pendingMessageSubmissions = EMPTY_PENDING_MESSAGE_SUBMISSIONS,
hasActiveTurn = false,
routeBottomAnchorRequest = null,
isAuthoritativeHistoryReady = true,
toast,
@@ -563,6 +565,11 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
}
const effectiveStreamItems = isActive ? streamItems : frozenStreamItemsRef.current;
const effectiveStreamHead = isActive ? streamHead : frozenStreamHeadRef.current;
const isTurnActive = [
context.status === "running",
pendingMessageSubmissions.length > 0,
hasActiveTurn,
].some(Boolean);
// Keep retained history outside the 48ms live-head flush path.
const preparedToolCallHistory = useMemo(
() => prepareToolCallHistory(toolCallDetailLevel, effectiveStreamItems),
@@ -575,12 +582,12 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
tail: effectiveStreamItems,
head: effectiveStreamHead ?? EMPTY_STREAM_HEAD,
preparedHistory: preparedToolCallHistory,
isTurnActive: context.status === "running",
isTurnActive,
}),
[
context.status,
effectiveStreamHead,
effectiveStreamItems,
isTurnActive,
preparedToolCallHistory,
toolCallDetailLevel,
],
@@ -588,27 +595,27 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
const baseRenderModel = useMemo(() => {
return buildAgentStreamRenderModel({
agentStatus: context.status,
isTurnActive,
tail: projectedToolCalls.tail,
head: projectedToolCalls.head,
platform: isWeb ? "web" : "native",
isMobileBreakpoint: isMobile,
});
}, [context.status, isMobile, projectedToolCalls.head, projectedToolCalls.tail]);
}, [isMobile, isTurnActive, projectedToolCalls.head, projectedToolCalls.tail]);
const streamLayout = useMemo(
() =>
layoutStream({
strategy: streamRenderStrategy,
agentStatus: context.status,
isTurnActive,
history: baseRenderModel.history,
liveHead: baseRenderModel.segments.liveHead,
timingByAssistantId: baseRenderModel.turnTiming.byAssistantId,
}),
[
context.status,
baseRenderModel.history,
baseRenderModel.segments.liveHead,
baseRenderModel.turnTiming.byAssistantId,
isTurnActive,
streamRenderStrategy,
],
);
@@ -892,8 +899,6 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
[pendingPermissions, agentId],
);
const showRunningTurnFooter =
context.status === "running" || pendingMessageSubmissions.length > 0;
const pendingPermissionsNode = useMemo(
() =>
renderPendingPermissionsNode({
@@ -904,9 +909,9 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
);
const turnFooterNode = useMemo(
() =>
showRunningTurnFooter || bottomTurnFooterHost ? (
isTurnActive || bottomTurnFooterHost ? (
<TurnFooter
isRunning={showRunningTurnFooter}
isRunning={isTurnActive}
inFlightTurnStartedAt={baseRenderModel.turnTiming.runningStartedAt}
host={bottomTurnFooterHost}
strategy={streamRenderStrategy}
@@ -917,7 +922,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
[
handleForkAssistantTurn,
readOnly,
showRunningTurnFooter,
isTurnActive,
baseRenderModel.turnTiming.runningStartedAt,
bottomTurnFooterHost,
streamRenderStrategy,
@@ -1175,6 +1180,7 @@ function agentStreamViewPropsEqual(
if (left.pendingMessageSubmissions !== right.pendingMessageSubmissions) {
reasons.push("pendingMessageSubmissions");
}
if (left.hasActiveTurn !== right.hasActiveTurn) reasons.push("hasActiveTurn");
if (
!bottomAnchorRouteRequestsEqual(left.routeBottomAnchorRequest, right.routeBottomAnchorRequest)
) {
@@ -52,8 +52,8 @@ export function acceptMessageSubmission(
);
if (index < 0) return submissions as MessageSubmissionRecord[];
// COMPAT(messageSubmissionDisposition): daemons before v0.2.3 omitted outOfBand.
// Their normal-send response follows the ordered running/canonical events, while an
// out-of-band response arrives with the agent still idle. Remove after 2027-01-27.
// An idle response cannot distinguish a normal send that has not reported running yet
// from an out-of-band send; the separate active-turn lease bridges that gap. Remove after 2027-01-27.
const legacyOutOfBand = outOfBand === undefined && !isAgentRunning;
if (
outOfBand === true ||
@@ -546,6 +546,18 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
}
}, [flushAgentLastActivity, serverId, isConnected, setInitializingAgents]);
useEffect(
() =>
client.subscribeConnectionStatus((connection) => {
if (connection.status === "connected") return;
const activeAgentTurns = useSessionStore.getState().sessions[serverId]?.activeAgentTurns;
for (const agentId of activeAgentTurns ?? []) {
setAgentStreamState(serverId, agentId, { turnActive: false });
}
}),
[client, serverId, setAgentStreamState],
);
const applyWorkspaceSetupProgress = useCallback(
(payload: WorkspaceSetupProgressPayload) => {
upsertWorkspaceSetupProgress({ serverId, payload });
+4
View File
@@ -1297,6 +1297,9 @@ const AgentStreamSection = memo(function AgentStreamSection({
? getActiveMessageSubmissions(state.sessions[serverId]?.messageSubmissions.get(agentId))
: EMPTY_MESSAGE_SUBMISSIONS,
);
const hasActiveTurn = useSessionStore((state) =>
agentId ? state.sessions[serverId]?.activeAgentTurns.has(agentId) === true : false,
);
const streamItems = streamItemsRaw ?? EMPTY_STREAM_ITEMS;
const pendingPermissionList = useStoreWithEqualityFn(
useSessionStore,
@@ -1337,6 +1340,7 @@ const AgentStreamSection = memo(function AgentStreamSection({
isAuthoritativeHistoryReady={hasAppliedAuthoritativeHistory}
toast={toast}
pendingMessageSubmissions={pendingMessageSubmissions}
hasActiveTurn={hasActiveTurn}
onOpenWorkspaceFile={onOpenWorkspaceFile}
/>
);
+51 -4
View File
@@ -404,6 +404,7 @@ export interface SessionState {
// Stream state (head/tail model)
agentStreamTail: Map<string, StreamItem[]>;
agentStreamHead: Map<string, StreamItem[]>;
activeAgentTurns: Set<string>;
messageSubmissions: Map<string, MessageSubmissionRecord[]>;
agentTimelineCursor: Map<string, AgentTimelineCursorState>;
agentTimelineHasOlder: Map<string, boolean>;
@@ -499,6 +500,7 @@ interface SessionStoreActions {
state: {
tail?: StreamItem[];
head?: StreamItem[];
turnActive?: boolean;
acknowledgedClientMessageIds?: readonly string[];
},
) => void;
@@ -657,6 +659,18 @@ function applyRunningAgentsToAcceptedSubmissions(input: {
return nextSubmissions;
}
function updateActiveAgentTurn(
activeAgentTurns: ReadonlySet<string>,
agentId: string,
active: boolean,
): Set<string> {
if (activeAgentTurns.has(agentId) === active) return activeAgentTurns as Set<string>;
const next = new Set(activeAgentTurns);
if (active) next.add(agentId);
else next.delete(agentId);
return next;
}
// Helper to create initial session state
function createInitialSessionState(
serverId: string,
@@ -678,6 +692,7 @@ function createInitialSessionState(
currentAssistantMessage: "",
agentStreamTail: new Map(),
agentStreamHead: new Map(),
activeAgentTurns: new Set(),
messageSubmissions: new Map(),
agentTimelineCursor: new Map(),
agentTimelineHasOlder: new Map(),
@@ -1100,6 +1115,7 @@ export const useSessionStore = create<SessionStore>()(
let nextHead = session.agentStreamHead;
let changedTail = false;
let changedHead = false;
let activeAgentTurns = session.activeAgentTurns;
if (state.tail !== undefined) {
const existingTail = session.agentStreamTail.get(agentId);
@@ -1126,6 +1142,10 @@ export const useSessionStore = create<SessionStore>()(
}
}
if (state.turnActive !== undefined) {
activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, state.turnActive);
}
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const observedSubmissions = observeMessageSubmissionCanonical(
currentSubmissions,
@@ -1133,7 +1153,12 @@ export const useSessionStore = create<SessionStore>()(
);
const changedSubmissions = observedSubmissions !== currentSubmissions;
if (!changedTail && !changedHead && !changedSubmissions) {
if (
!changedTail &&
!changedHead &&
!changedSubmissions &&
activeAgentTurns === session.activeAgentTurns
) {
return prev;
}
@@ -1155,6 +1180,7 @@ export const useSessionStore = create<SessionStore>()(
...session,
agentStreamTail: nextTail,
agentStreamHead: nextHead,
activeAgentTurns,
messageSubmissions,
},
},
@@ -1196,6 +1222,7 @@ export const useSessionStore = create<SessionStore>()(
stream.head === currentHead
? session.agentStreamHead
: new Map(session.agentStreamHead).set(agentId, stream.head),
activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true),
messageSubmissions,
},
},
@@ -1208,13 +1235,25 @@ export const useSessionStore = create<SessionStore>()(
const session = prev.sessions[serverId];
if (!session) return prev;
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const isAgentRunning = session.agents.get(agentId)?.status === "running";
const submissions = acceptMessageSubmission(
currentSubmissions,
clientMessageId,
session.agents.get(agentId)?.status === "running",
isAgentRunning,
outOfBand,
);
if (submissions === currentSubmissions) return prev;
let activeAgentTurns = session.activeAgentTurns;
if (
submissions !== currentSubmissions &&
!isAgentRunning &&
outOfBand === true &&
submissions.length === 0
) {
activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, false);
}
if (submissions === currentSubmissions && activeAgentTurns === session.activeAgentTurns) {
return prev;
}
const messageSubmissions = new Map(session.messageSubmissions);
if (submissions.length > 0) {
messageSubmissions.set(agentId, submissions);
@@ -1225,7 +1264,7 @@ export const useSessionStore = create<SessionStore>()(
...prev,
sessions: {
...prev.sessions,
[serverId]: { ...session, messageSubmissions },
[serverId]: { ...session, activeAgentTurns, messageSubmissions },
},
};
});
@@ -1256,6 +1295,12 @@ export const useSessionStore = create<SessionStore>()(
} else {
messageSubmissions.delete(agentId);
}
const activeAgentTurns =
outcome === "rejected" &&
result.submissions.length === 0 &&
session.agents.get(agentId)?.status !== "running"
? updateActiveAgentTurn(session.activeAgentTurns, agentId, false)
: session.activeAgentTurns;
return {
...prev,
sessions: {
@@ -1270,6 +1315,7 @@ export const useSessionStore = create<SessionStore>()(
stream.head === currentHead
? session.agentStreamHead
: new Map(session.agentStreamHead).set(agentId, stream.head),
activeAgentTurns,
messageSubmissions,
},
},
@@ -1313,6 +1359,7 @@ export const useSessionStore = create<SessionStore>()(
...session,
agentStreamTail: nextTail,
agentStreamHead: nextHead,
activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true),
},
},
};
@@ -8,6 +8,7 @@ import {
} from "@/types/stream";
import {
createAgentStreamReducerQueue,
deriveAgentStreamTurnActivity,
processTimelineResponse,
processAgentStreamEvent,
processAgentStreamEvents,
@@ -130,6 +131,31 @@ function makeSubmittedUserMessage(
});
}
describe("deriveAgentStreamTurnActivity", () => {
it("tracks start through every terminal turn event", () => {
const started = makeStreamReducerEvent(
{ type: "turn_started", provider: "claude" } as AgentStreamEventPayload,
1,
);
expect(deriveAgentStreamTurnActivity([started])).toBe(true);
for (const event of [
{ type: "turn_completed", provider: "claude" },
{ type: "turn_failed", provider: "claude", error: "failed" },
{ type: "turn_canceled", provider: "claude", reason: "canceled" },
] as AgentStreamEventPayload[]) {
expect(deriveAgentStreamTurnActivity([started, makeStreamReducerEvent(event, 2)])).toBe(
false,
);
}
});
it("leaves activity unchanged when a batch has no lifecycle event", () => {
expect(
deriveAgentStreamTurnActivity([makeStreamReducerEvent(makeAssistantTimelineEvent("hi"), 1)]),
).toBeUndefined();
});
});
function getAssistantTexts(items: StreamItem[]): string[] {
return items
.filter((item): item is Extract<StreamItem, { kind: "assistant_message" }> => {
@@ -1310,9 +1310,27 @@ export function createAgentStreamReducerQueue(
interface StreamStatePatch {
tail?: StreamItem[];
head?: StreamItem[];
turnActive?: boolean;
acknowledgedClientMessageIds?: readonly string[];
}
export function deriveAgentStreamTurnActivity(
events: readonly AgentStreamReducerEvent[],
): boolean | undefined {
let active: boolean | undefined;
for (const { event } of events) {
if (event.type === "turn_started") active = true;
else if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
active = false;
}
}
return active;
}
export interface CreateSessionAgentStreamReducerQueueInput {
serverId: string;
setAgentStreamState: (serverId: string, agentId: string, state: StreamStatePatch) => void;
@@ -1348,14 +1366,17 @@ export function createSessionAgentStreamReducerQueue(
};
},
commit: (agentId, result, events) => {
const turnActive = deriveAgentStreamTurnActivity(events);
if (
result.changedTail ||
result.changedHead ||
turnActive !== undefined ||
result.acknowledgedClientMessageIds.length > 0
) {
setAgentStreamState(serverId, agentId, {
...(result.changedTail ? { tail: result.tail } : {}),
...(result.changedHead ? { head: result.head } : {}),
...(turnActive === undefined ? {} : { turnActive }),
...(result.acknowledgedClientMessageIds.length > 0
? { acknowledgedClientMessageIds: result.acknowledgedClientMessageIds }
: {}),
+4 -4
View File
@@ -26,7 +26,7 @@ describe("deriveStreamTurnTiming", () => {
const submittedAt = new Date("2026-05-15T00:00:00.000Z");
const timing = deriveStreamTurnTiming({
agentStatus: "running",
isTurnActive: true,
tail: [],
head: [user("submitted", submittedAt)],
});
@@ -39,7 +39,7 @@ describe("deriveStreamTurnTiming", () => {
const secondUserAt = new Date("2026-05-15T00:01:00.000Z");
const timing = deriveStreamTurnTiming({
agentStatus: "running",
isTurnActive: true,
tail: [
user("u1", firstUserAt),
assistant("a1", new Date("2026-05-15T00:00:05.000Z")),
@@ -57,7 +57,7 @@ describe("deriveStreamTurnTiming", () => {
const assistantAt = new Date("2026-05-15T00:00:07.000Z");
const timing = deriveStreamTurnTiming({
agentStatus: "idle",
isTurnActive: false,
tail: [
user("u1", userAt),
assistant("a1", assistantAt),
@@ -79,7 +79,7 @@ describe("deriveStreamTurnTiming", () => {
const lastAssistantAt = new Date("2026-05-15T00:00:07.000Z");
const timing = deriveStreamTurnTiming({
agentStatus: "idle",
isTurnActive: false,
tail: [
user("u1", userAt),
assistant("a1", firstAssistantAt),
+3 -4
View File
@@ -12,7 +12,7 @@ export interface StreamTurnTiming {
}
export function deriveStreamTurnTiming(params: {
agentStatus: string;
isTurnActive: boolean;
tail: StreamItem[];
head: StreamItem[];
}): StreamTurnTiming {
@@ -59,9 +59,8 @@ export function deriveStreamTurnTiming(params: {
visitItem(item);
}
const isRunning = params.agentStatus === "running";
const runningStartedAt = isRunning ? currentUserAt : null;
if (params.agentStatus !== "running") {
const runningStartedAt = params.isTurnActive ? currentUserAt : null;
if (!params.isTurnActive) {
flushCompletedTurn();
}
@@ -114,6 +114,7 @@ export function removeAgentDirectoryReplica(serverId: string, agentId: string):
store.setAgentAuthoritativeHistoryApplied(serverId, agentId, false);
store.setAgentStreamTail(serverId, removeKey);
store.clearAgentStreamHead(serverId, agentId);
store.setAgentStreamState(serverId, agentId, { turnActive: false });
useSessionStore.setState((state) => {
if (!state.agentLastActivity.has(agentId)) return state;
const agentLastActivity = new Map(state.agentLastActivity);
@@ -243,6 +243,12 @@ function promptToText(prompt: AgentPromptInput): string {
.trim();
}
function parseUserMessageDelayMs(prompt: AgentPromptInput): number {
const match = /delay synthetic user message by (\d+)ms/i.exec(promptToText(prompt));
const delayMs = Number(match?.[1] ?? 0);
return Number.isSafeInteger(delayMs) ? Math.min(delayMs, 2_000) : 0;
}
function parseLargeAgentStreamPayloadPrompt(
prompt: AgentPromptInput,
): LargeAgentStreamPayloadRequest | null {
@@ -657,7 +663,7 @@ export class MockLoadTestAgentSession implements AgentSession {
};
this.activeTurn = turn;
const userMessageId = randomUUID();
setTimeout(() => {
const userMessageTimer = setTimeout(() => {
if (this.activeTurn?.turnId !== turnId) {
return;
}
@@ -672,7 +678,8 @@ export class MockLoadTestAgentSession implements AgentSession {
...(options?.clientMessageId ? { clientMessageId: options.clientMessageId } : {}),
},
});
}, 0);
}, parseUserMessageDelayMs(prompt));
userMessageTimer.unref?.();
const largePayload = parseLargeAgentStreamPayloadPrompt(prompt);
const stress = parseAgentStreamStressPrompt(prompt);