diff --git a/packages/app/e2e/agent-consecutive-turns.spec.ts b/packages/app/e2e/agent-consecutive-turns.spec.ts index 3e6bbca13c..2ef5ccc40d 100644 --- a/packages/app/e2e/agent-consecutive-turns.spec.ts +++ b/packages/app/e2e/agent-consecutive-turns.spec.ts @@ -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 { - 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 { 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 { - const result = await page.evaluate(() => { +async function waitForRecordedFrames( + page: Page, + predicate: "user-visible" | "turn-running", + count: number, +): Promise { + 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 { + 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 { + 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 { - 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(); + 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, + 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 { + 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); }); diff --git a/packages/app/e2e/agent-timeline-pagination.spec.ts b/packages/app/e2e/agent-timeline-pagination.spec.ts index b833ef7abd..dda0dc3a99 100644 --- a/packages/app/e2e/agent-timeline-pagination.spec.ts +++ b/packages/app/e2e/agent-timeline-pagination.spec.ts @@ -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); diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index c1e57f7adc..783228e9c3 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -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; agentStreamEventTypes: ReadonlySet; + agentStreamItemTypes: ReadonlySet; 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(); const suppressedAgentStreamEventTypes = new Set(); + const suppressedAgentStreamItemTypes = new Set(); const activeSockets = new Set(); 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 { @@ -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; }, diff --git a/packages/app/src/agent-stream/history-start-pagination.test.ts b/packages/app/src/agent-stream/history-start-pagination.test.ts index af45b48f74..3a23fd18b3 100644 --- a/packages/app/src/agent-stream/history-start-pagination.test.ts +++ b/packages/app/src/agent-stream/history-start-pagination.test.ts @@ -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, diff --git a/packages/app/src/agent-stream/history-start-pagination.ts b/packages/app/src/agent-stream/history-start-pagination.ts index 317753b1ab..8edbbe475a 100644 --- a/packages/app/src/agent-stream/history-start-pagination.ts +++ b/packages/app/src/agent-stream/history-start-pagination.ts @@ -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 { diff --git a/packages/app/src/agent-stream/layout.test.ts b/packages/app/src/agent-stream/layout.test.ts index b996543be5..2fce5c4379 100644 --- a/packages/app/src/agent-stream/layout.test.ts +++ b/packages/app/src/agent-stream/layout.test.ts @@ -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], }); diff --git a/packages/app/src/agent-stream/layout.ts b/packages/app/src/agent-stream/layout.ts index f276db3825..9249f81592 100644 --- a/packages/app/src/agent-stream/layout.ts +++ b/packages/app/src/agent-stream/layout.ts @@ -36,7 +36,7 @@ export interface StreamLayout { export interface StreamLayoutInput { strategy: StreamStrategy; - agentStatus: string; + isTurnActive: boolean; history: StreamItem[]; liveHead: StreamItem[]; timingByAssistantId: Map; @@ -117,7 +117,7 @@ function findLatestAssistantInTurn(input: { } function resolveAuxiliaryTurnFooter(input: StreamLayoutInput): TurnFooterHost | null { - if (input.agentStatus === "running") { + if (input.isTurnActive) { return null; } diff --git a/packages/app/src/agent-stream/model.test.ts b/packages/app/src/agent-stream/model.test.ts index 806a501d68..030ea505b4 100644 --- a/packages/app/src/agent-stream/model.test.ts +++ b/packages/app/src/agent-stream/model.test.ts @@ -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", diff --git a/packages/app/src/agent-stream/model.ts b/packages/app/src/agent-stream/model.ts index fe92e63fc0..21bcae50a6 100644 --- a/packages/app/src/agent-stream/model.ts +++ b/packages/app/src/agent-stream/model.ts @@ -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> + WeakMap> >(); 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, }); diff --git a/packages/app/src/agent-stream/strategy-native.tsx b/packages/app/src/agent-stream/strategy-native.tsx index cb802918c8..1d0a7c17fa 100644 --- a/packages/app/src/agent-stream/strategy-native.tsx +++ b/packages/app/src/agent-stream/strategy-native.tsx @@ -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 ( diff --git a/packages/app/src/agent-stream/strategy-web.tsx b/packages/app/src/agent-stream/strategy-web.tsx index 72c872d684..3cd6cf88fa 100644 --- a/packages/app/src/agent-stream/strategy-web.tsx +++ b/packages/app/src/agent-stream/strategy-web.tsx @@ -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(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} ); - }, [hasOlderHistory, historyStartPaginationState, isHistoryStartSlotReserved]); + }, [hasOlderHistory, historyStartPaginationState]); const shouldRenderEmpty = !boundary.hasMountedHistory && !boundary.hasVirtualizedHistory && diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 63308e56b3..30d71e4bb0 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -242,6 +242,7 @@ export interface AgentStreamViewProps { streamHead?: StreamItem[]; pendingPermissions: Map; pendingMessageSubmissions?: readonly PendingMessageSubmission[]; + hasActiveTurn?: boolean; routeBottomAnchorRequest?: BottomAnchorRouteRequest | null; isAuthoritativeHistoryReady?: boolean; toast?: ToastApi | null; @@ -332,6 +333,7 @@ const AgentStreamViewComponent = forwardRef 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 { 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 0; const pendingPermissionsNode = useMemo( () => renderPendingPermissionsNode({ @@ -904,9 +909,9 @@ const AgentStreamViewComponent = forwardRef - showRunningTurnFooter || bottomTurnFooterHost ? ( + isTurnActive || bottomTurnFooterHost ? ( + 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 }); diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index dbc2ab46fb..9d529ba82c 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -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} /> ); diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 1d59abdfeb..97bbf4726c 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -404,6 +404,7 @@ export interface SessionState { // Stream state (head/tail model) agentStreamTail: Map; agentStreamHead: Map; + activeAgentTurns: Set; messageSubmissions: Map; agentTimelineCursor: Map; agentTimelineHasOlder: Map; @@ -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, + agentId: string, + active: boolean, +): Set { + if (activeAgentTurns.has(agentId) === active) return activeAgentTurns as Set; + 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()( 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()( } } + 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()( ); 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()( ...session, agentStreamTail: nextTail, agentStreamHead: nextHead, + activeAgentTurns, messageSubmissions, }, }, @@ -1196,6 +1222,7 @@ export const useSessionStore = create()( 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()( 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()( ...prev, sessions: { ...prev.sessions, - [serverId]: { ...session, messageSubmissions }, + [serverId]: { ...session, activeAgentTurns, messageSubmissions }, }, }; }); @@ -1256,6 +1295,12 @@ export const useSessionStore = create()( } 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()( stream.head === currentHead ? session.agentStreamHead : new Map(session.agentStreamHead).set(agentId, stream.head), + activeAgentTurns, messageSubmissions, }, }, @@ -1313,6 +1359,7 @@ export const useSessionStore = create()( ...session, agentStreamTail: nextTail, agentStreamHead: nextHead, + activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true), }, }, }; diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 79c392a240..369824b1af 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -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 => { diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 1cd26517f9..34e8b9014c 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -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 } : {}), diff --git a/packages/app/src/timeline/turn-time.test.ts b/packages/app/src/timeline/turn-time.test.ts index 3945edbcb7..47387ed42e 100644 --- a/packages/app/src/timeline/turn-time.test.ts +++ b/packages/app/src/timeline/turn-time.test.ts @@ -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), diff --git a/packages/app/src/timeline/turn-time.ts b/packages/app/src/timeline/turn-time.ts index 7ebd77c214..0c0af024c2 100644 --- a/packages/app/src/timeline/turn-time.ts +++ b/packages/app/src/timeline/turn-time.ts @@ -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(); } diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index 24e14c4208..a8acbefd4d 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -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); diff --git a/packages/server/src/server/agent/providers/mock-load-test-agent.ts b/packages/server/src/server/agent/providers/mock-load-test-agent.ts index 76f9848d8f..b30042b6cc 100644 --- a/packages/server/src/server/agent/providers/mock-load-test-agent.ts +++ b/packages/server/src/server/agent/providers/mock-load-test-agent.ts @@ -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);