From a9bce9a18d6e8e819ba52101d4fd1323c0daf237 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 6 Nov 2026 23:10:29 +0700 Subject: [PATCH] Keep the chat still when an agent's turn ends (#8944) * Land a turn's reply, completed footer, and tool summary in one frame The stream handler closed turn liveness as soon as turn_completed arrived, while the final reply waited in the frame-coalesced reducer queue. For one commit the turn was finished with no reply: the footer vanished and the timeline jumped twice, and tool turns folded into their summary later. The viewed timeline owner now takes stream events whole: a turn boundary commits the queued content before liveness changes. Running and completed footers share one fixed content height so the phase switch moves nothing. * Document the turn-boundary commit and stop racing the re-pin in the turn-open test The turn-open test scrolled the received message into view with a programmatic scroll, which leaves the chat following its end. Pressing the turn summary swaps its icon, and that DOM change re-pins the chat to the end between pointerdown and pointerup. The summary had moved 14 px, which left the pointer 2 px inside the button; with the completed footer now 2 px taller, it moved 16 px and the click landed on the reply below. The message's fade and Show more are already on screen at the end, so the test no longer scrolls. --- docs/agent-stream-performance.md | 1 + .../app/e2e/browser/agent-turn-end.spec.ts | 200 ++++++++++++++++++ .../app/e2e/browser/turn-collapse.spec.ts | 1 - packages/app/src/agent-stream/turn-footer.tsx | 14 +- packages/app/src/contexts/session-context.tsx | 16 +- packages/app/src/runtime/host-runtime.ts | 2 + packages/app/src/timeline/replica.test.ts | 96 ++++++++- .../app/src/timeline/viewed-timeline-sync.ts | 16 +- 8 files changed, 319 insertions(+), 27 deletions(-) create mode 100644 packages/app/e2e/browser/agent-turn-end.spec.ts diff --git a/docs/agent-stream-performance.md b/docs/agent-stream-performance.md index fb022895bc..4c43718215 100644 --- a/docs/agent-stream-performance.md +++ b/docs/agent-stream-performance.md @@ -34,6 +34,7 @@ So arrival sets a _target_ and the reveal rate is derived from the backlog inste - **First sight of a text is revealed whole.** Only growth is paced. This is what makes history hydration, timeline replay, a virtualized row remounting on scroll, and an already-finished message all render complete on first paint without a special case for each. - **Leaving `phase: "streaming"` snaps the reveal.** A completed turn must never be left holding characters. `layoutStream` sets the phase, so anything outside the live head with an active turn is already complete. - **The reducer queue commits on a frame, with a timer as the ceiling.** A frame callback never fires in a hidden tab, so a timer races it and wins when nothing is painting — the store has to keep advancing either way. +- **A turn boundary commits the queue before turn liveness changes.** The final reply arrives a few milliseconds before `turn_completed`, so it is still queued when the turn ends. `applyStreamEvent` in `viewed-timeline-sync.ts` flushes the agent's queue first; otherwise the chat paints a finished turn with no reply, drops the footer, and jumps twice. - **A history row re-renders only when its item or layout item identity changes.** Every coalesced tick hands the list a new layout. Without a memo boundary each tick re-rendered every mounted row (measured on the inverted FlatList native used before LegendList: about 50 rows on a diff --git a/packages/app/e2e/browser/agent-turn-end.spec.ts b/packages/app/e2e/browser/agent-turn-end.spec.ts new file mode 100644 index 0000000000..523297729c --- /dev/null +++ b/packages/app/e2e/browser/agent-turn-end.spec.ts @@ -0,0 +1,200 @@ +import type { Page } from "@playwright/test"; +import { expect, test as baseTest } from "../support/fixtures"; +import { expectAgentIdle } from "../support/helpers/agent-stream"; +import { submitMessage } from "../support/helpers/composer"; +import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate"; +import { + openAgentRoute, + seedMockAgentWorkspace, + type MockAgentWorkspace, +} from "../support/helpers/mock-agent"; + +// A turn ends with its final reply, `turn_completed`, and the idle status snapshot arriving +// within a few milliseconds of each other. These tests hold all three, let the running +// turn paint, then release them together and sample every painted frame. + +interface Box { + top: number; + height: number; +} + +interface TurnEndFrame { + at: number; + running: boolean; + footer: Box | null; + userRow: Box | null; + replyVisible: boolean; +} + +declare global { + interface Window { + __turnEndFrames?: { active: boolean; frames: TurnEndFrame[] }; + } +} + +const test = baseTest.extend<{ agent: MockAgentWorkspace }>({ + agent: async ({ page: _page }, provide) => { + const agent = await seedMockAgentWorkspace({ + repoPrefix: "turn-end-", + title: "Turn end", + initialPrompt: "emit 4 tool agent stream updates", + }); + await agent.client.waitForFinish(agent.agentId, 15_000); + await provide(agent); + await agent.cleanup(); + }, +}); + +async function recordTurnEndFrames( + page: Page, + input: { prompt: string; reply: string | null }, +): Promise { + await page.evaluate(({ prompt, reply }) => { + const state = { active: true, frames: [] as TurnEndFrame[] }; + const isVisible = (element: Element | null | undefined) => Boolean(element?.checkVisibility()); + const boxOf = (element: Element | null | undefined): Box | null => { + if (!element || !isVisible(element)) return null; + const rect = element.getBoundingClientRect(); + return rect.height > 0 ? { top: rect.top, height: rect.height } : null; + }; + const sample = () => { + const viewport = Array.from( + document.querySelectorAll('[data-testid="agent-chat-scroll"]'), + ).find((candidate) => isVisible(candidate)); + const footers = Array.from(viewport?.querySelectorAll('[data-testid="turn-footer"]') ?? []) + .map(boxOf) + .filter((box): box is Box => box !== null) + .sort((left, right) => left.top - right.top); + const userRow = Array.from( + viewport?.querySelectorAll('[data-testid="user-message"]') ?? [], + ).find((candidate) => candidate.textContent?.includes(prompt)); + const replyRow = + reply === null + ? undefined + : Array.from(viewport?.querySelectorAll('[data-testid="assistant-message"]') ?? []).find( + (candidate) => candidate.textContent?.includes(reply), + ); + state.frames.push({ + at: performance.now(), + running: Boolean(viewport?.querySelector('[data-testid="turn-working-indicator"]')), + footer: footers.at(-1) ?? null, + userRow: boxOf(userRow), + replyVisible: boxOf(replyRow) !== null, + }); + if (state.active) requestAnimationFrame(() => setTimeout(sample, 0)); + }; + window.__turnEndFrames = state; + requestAnimationFrame(() => setTimeout(sample, 0)); + }, input); +} + +async function recordPaintsFor(page: Page, durationMs: number): Promise { + const until = await page.evaluate((duration) => performance.now() + duration, durationMs); + await expect + .poll(() => page.evaluate(() => window.__turnEndFrames?.frames.at(-1)?.at ?? 0)) + .toBeGreaterThanOrEqual(until); +} + +async function stopRecording(page: Page): Promise { + return page.evaluate(() => { + const state = window.__turnEndFrames; + if (!state) throw new Error("Turn end frames were not recorded"); + state.active = false; + return state.frames; + }); +} + +async function recordTurnEnd( + page: Page, + agent: MockAgentWorkspace, + input: { prompt: string; reply: string | null }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectAgentIdle(page); + + if (input.reply === null) { + gate.setAgentStreamItemSuppressed("assistant_message", true); + } else { + gate.holdNextAgentStreamItem("assistant_message"); + } + gate.holdNextAgentStreamEvent("turn_completed"); + // The daemon sends more than one idle snapshot; hold the first and drop the rest until release. + gate.holdNextAgentUpdate(agent.agentId, "idle"); + gate.setServerMessageSuppressed("agent_update", true); + await recordTurnEndFrames(page, input); + await submitMessage(page, input.prompt); + await Promise.all([ + input.reply === null ? Promise.resolve() : gate.waitForHeldAgentStreamItem("assistant_message"), + gate.waitForHeldAgentStreamEvent("turn_completed"), + gate.waitForHeldAgentUpdate(agent.agentId, "idle"), + ]); + await expect(page.getByTestId("turn-working-indicator")).toBeVisible(); + await recordPaintsFor(page, 200); + + if (input.reply !== null) gate.releaseHeldAgentStreamItem("assistant_message"); + gate.releaseHeldAgentStreamEvent("turn_completed"); + gate.releaseHeldAgentUpdate(agent.agentId, "idle"); + gate.setServerMessageSuppressed("agent_update", false); + await expectAgentIdle(page); + await recordPaintsFor(page, 300); + return stopRecording(page); +} + +function describeFrames(frames: TurnEndFrame[]): string { + return frames + .map( + (frame, index) => + `${index}: ${frame.running ? "running " : "complete"} footer=${ + frame.footer ? `${frame.footer.top.toFixed(1)}+${frame.footer.height.toFixed(1)}` : "none" + } prompt=${frame.userRow?.top.toFixed(1) ?? "none"} reply=${frame.replyVisible}`, + ) + .join("\n"); +} + +function expectSteadyTurnEnd(frames: TurnEndFrame[], input: { reply: string | null }): void { + const firstRunning = frames.findIndex((frame) => frame.running && frame.footer !== null); + const firstComplete = frames.findIndex((frame, index) => index > firstRunning && !frame.running); + const record = `frames:\n${describeFrames(frames)}`; + expect(firstRunning, record).toBeGreaterThanOrEqual(0); + expect(firstComplete, record).toBeGreaterThan(firstRunning); + + const transition = frames.slice(firstRunning); + const runningHeight = transition[0].footer?.height; + expect( + transition.map((frame) => frame.footer?.height ?? null), + `the footer must stay painted at one height\n${record}`, + ).toEqual(transition.map(() => runningHeight)); + if (input.reply !== null) { + expect( + frames.slice(firstComplete).every((frame) => frame.replyVisible), + `a completed turn must show its reply in the same frame\n${record}`, + ).toBe(true); + } + // Tool rows land while the turn runs; only the turn end itself must be one step. + const ending = frames.slice(firstComplete - 1); + const promptMoves = ending + .slice(1) + .filter((frame, index) => frame.userRow?.top !== ending[index].userRow?.top); + expect( + promptMoves.length, + `the prompt must move at most once as the turn ends\n${record}`, + ).toBeLessThanOrEqual(1); +} + +test.describe("Turn end", () => { + test("ends a plain reply without moving or dropping its footer", async ({ page, agent }) => { + const input = { prompt: "emit 1 coalesced agent stream updates", reply: "stress-update-0" }; + expectSteadyTurnEnd(await recordTurnEnd(page, agent, input), input); + }); + + test("ends a tool turn with its reply and completed footer together", async ({ page, agent }) => { + const input = { prompt: "emit 3 tool agent stream updates", reply: "Completed 3 commands" }; + expectSteadyTurnEnd(await recordTurnEnd(page, agent, input), input); + }); + + test("keeps the footer height when a turn ends on a tool call", async ({ page, agent }) => { + const input = { prompt: "emit 2 tool agent stream updates", reply: null }; + expectSteadyTurnEnd(await recordTurnEnd(page, agent, input), input); + }); +}); diff --git a/packages/app/e2e/browser/turn-collapse.spec.ts b/packages/app/e2e/browser/turn-collapse.spec.ts index a7138e0850..bcfca2f763 100644 --- a/packages/app/e2e/browser/turn-collapse.spec.ts +++ b/packages/app/e2e/browser/turn-collapse.spec.ts @@ -407,7 +407,6 @@ test("opening a turn keeps the received message above it clipped on every frame" .getByTestId("user-message") .filter({ has: page.getByRole("button", { name: "Sent by Reviewer", exact: true }) }); await expect(bubble.getByRole("button", { name: "Show more", exact: true })).toBeVisible(); - await bubble.scrollIntoViewIfNeeded(); await watchReceivedMessagePreview(page, "Sent by Reviewer"); await page diff --git a/packages/app/src/agent-stream/turn-footer.tsx b/packages/app/src/agent-stream/turn-footer.tsx index 5508803b5d..08fceb4005 100644 --- a/packages/app/src/agent-stream/turn-footer.tsx +++ b/packages/app/src/agent-stream/turn-footer.tsx @@ -23,6 +23,10 @@ import { useRetainedPanelActive } from "@/components/retained-panel"; const ThemedSyncedLoader = withUnistyles(SyncedLoader); const workingIndicatorColorMapping = (theme: Theme) => ({ color: theme.colors.foreground }); export const TURN_FOOTER_BOTTOM_SPACING = SPACING[8]; +// Every footer variant (running, completed with Copy, completed with only its duration) gets +// this content height, so a turn ending swaps the footer's controls without moving the rows +// above it. +const TURN_FOOTER_CONTENT_HEIGHT = 24; export type TurnContentStrategy = StreamStrategy; export type AssistantTurnForkHandler = (input: { @@ -208,7 +212,11 @@ function CompletedTurnFooter({ function TurnFooterRow({ children }: { children: ReactNode }) { const rowStyle = useMemo(() => [stylesheet.streamItemWrapper, stylesheet.turnFooterRow], []); - return {children}; + return ( + + {children} + + ); } const stylesheet = StyleSheet.create((theme) => ({ @@ -229,11 +237,11 @@ const stylesheet = StyleSheet.create((theme) => ({ flexDirection: "row", alignItems: "center", alignSelf: "flex-start", - minHeight: 24, + height: TURN_FOOTER_CONTENT_HEIGHT + TURN_FOOTER_BOTTOM_SPACING, paddingBottom: TURN_FOOTER_BOTTOM_SPACING, }, turnFooterContent: { - height: 24, + height: TURN_FOOTER_CONTENT_HEIGHT, flexDirection: "row", alignItems: "center", justifyContent: "flex-start", diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 89c32c0bd5..0b46f5f2de 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -12,7 +12,6 @@ import { refreshAgentInitializationTimeout, } from "@/hooks/use-agent-initialization"; import type { StreamItem } from "@/types/stream"; -import { deriveAgentStreamTurnLiveness } from "@/timeline/session-stream-reducers"; import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan"; import { requestTimelineReplacement } from "@/timeline/timeline-replacement"; import { type ViewedTimelineOwner } from "@/timeline/viewed-timeline-sync"; @@ -360,8 +359,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const onStream = (message: SessionOutboundMessage) => { if (message.type !== "agent_stream") return; const { agentId, event, timestamp, seq, epoch } = message.payload; - const parsedTimestamp = new Date(timestamp); - const streamEvent = event; if ( event.type === "turn_started" || event.type === "turn_completed" || @@ -370,18 +367,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider ) { voiceRuntime?.onTurnEvent(serverId, agentId, event.type); } - const turnLiveness = deriveAgentStreamTurnLiveness([ - { event: streamEvent, seq, epoch, timestamp: parsedTimestamp }, - ]); - if (turnLiveness.length > 0) { - getHostRuntimeStore().applyAgentTurnLiveness(serverId, agentId, turnLiveness); - } - sync.enqueueStreamEvent(agentId, { - event: streamEvent, - seq, - epoch, - timestamp: parsedTimestamp, - }); + sync.applyStreamEvent(agentId, { event, seq, epoch, timestamp: new Date(timestamp) }); // NOTE: We don't update lastActivityAt on every stream event to prevent // cascading rerenders. The agent_update handler updates agent.lastActivityAt diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 4d31258aaa..b6d10dc8c7 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -2852,6 +2852,8 @@ export class HostRuntimeStore { replica, replaceDemandedAgentIds: (agentIds) => directory.setAgentRouteDemand(agentIds), drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(serverId, agentId), + applyTurnLiveness: (agentId, transitions) => + directory.applyAgentTurnLiveness(agentId, transitions), ports, }); } diff --git a/packages/app/src/timeline/replica.test.ts b/packages/app/src/timeline/replica.test.ts index 859ae8e6b7..0ce9b8ca2b 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -11,6 +11,7 @@ import type { CachedTimeline } from "@/runtime/replica-cache"; import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store"; import { useCreateFlowStore, type PendingCreateAttempt } from "@/stores/create-flow-store"; import type { StreamItem } from "@/types/stream"; +import type { TurnLivenessTransition } from "./turn-liveness"; import { createTimelineReplica, createViewedTimelineOwner, @@ -40,7 +41,13 @@ function cachedTimeline(): CachedTimeline { }; } -function createOwner(storage: TimelineReplicaStorage): ViewedTimelineOwner { +function createOwner( + storage: TimelineReplicaStorage, + applyTurnLiveness: ( + agentId: string, + transitions: readonly TurnLivenessTransition[], + ) => void = () => undefined, +): ViewedTimelineOwner { const replica = createTimelineReplica({ serverId: SERVER_ID, storage, @@ -51,6 +58,7 @@ function createOwner(storage: TimelineReplicaStorage): ViewedTimelineOwner { replica, replaceDemandedAgentIds: () => undefined, drainQueuedAgentMessage: () => undefined, + applyTurnLiveness, ports: { observe: () => ({ ready: Promise.resolve(), release: async () => undefined }), readCursor: () => undefined, @@ -98,7 +106,7 @@ describe("viewed timeline persistence", () => { .status, ) .toBe("painted"); - owner.enqueueStreamEvent(AGENT_ID, { + owner.applyStreamEvent(AGENT_ID, { event: { type: "timeline", provider: "codex", @@ -147,7 +155,7 @@ describe("viewed timeline persistence", () => { .map((entry) => entry.text); }; expect.soft(replyTexts()).toEqual([complete]); - owner.enqueueStreamEvent(AGENT_ID, { + owner.applyStreamEvent(AGENT_ID, { event: { type: "turn_completed", provider: "codex" } as AgentStreamEventPayload, seq: 7, epoch: "epoch-1", @@ -281,7 +289,7 @@ describe("viewed timeline persistence", () => { ) .toMatchObject({ status: "painted" }); - first.enqueueStreamEvent(AGENT_ID, { + first.applyStreamEvent(AGENT_ID, { event: { type: "timeline", provider: "codex", @@ -460,7 +468,7 @@ describe("viewed timeline persistence", () => { readTimeline: async () => undefined, commitTimeline: (_serverId, _agentId, timeline) => commits.push(timeline), }); - owner.enqueueStreamEvent(AGENT_ID, { + owner.applyStreamEvent(AGENT_ID, { event: { type: "timeline", provider: "codex", @@ -525,7 +533,7 @@ describe("viewed timeline persistence", () => { [AGENT_ID, 9], ["agent-2", 4], ] as const) { - owner.enqueueStreamEvent(agentId, { + owner.applyStreamEvent(agentId, { event: { type: "timeline", provider: "codex", @@ -543,6 +551,81 @@ describe("viewed timeline persistence", () => { }); }); +describe("turn boundaries", () => { + it("commits a turn's final reply before the turn stops running", () => { + useSessionStore.getState().initializeSession(SERVER_ID, null); + applySynced(AGENT_ID, 8); + const committedKinds = () => { + const session = useSessionStore.getState().sessions[SERVER_ID]; + return [ + ...(session?.agentStreamTail.get(AGENT_ID) ?? []), + ...(session?.agentStreamHead.get(AGENT_ID) ?? []), + ] + .filter((entry) => entry.id !== `network-${AGENT_ID}`) + .map((entry) => entry.kind); + }; + const livenessChanges: { type: string; committed: string[] }[] = []; + const owner = createOwner( + { readTimeline: async () => undefined, commitTimeline: () => undefined }, + (_agentId, transitions) => { + for (const transition of transitions) { + livenessChanges.push({ type: transition.type, committed: committedKinds() }); + } + }, + ); + const at = (second: number) => new Date(`2026-08-26T10:00:0${second}.000Z`); + + // One frame's worth of live events: the frame flush that would commit them never runs. + owner.applyStreamEvent(AGENT_ID, { + event: { type: "turn_started", provider: "codex", turnId: "turn-1" }, + seq: undefined, + epoch: undefined, + timestamp: at(1), + }); + owner.applyStreamEvent(AGENT_ID, { + event: { + type: "timeline", + provider: "codex", + turnId: "turn-1", + item: { + type: "tool_call", + callId: "call-1", + name: "Bash", + status: "completed", + error: null, + detail: { type: "shell", command: "ls" }, + }, + } as AgentStreamEventPayload, + seq: 9, + epoch: "epoch-1", + timestamp: at(2), + }); + owner.applyStreamEvent(AGENT_ID, { + event: { + type: "timeline", + provider: "codex", + turnId: "turn-1", + item: { type: "assistant_message", text: "Done.", messageId: "reply-1" }, + } as AgentStreamEventPayload, + seq: 10, + epoch: "epoch-1", + timestamp: at(3), + }); + owner.applyStreamEvent(AGENT_ID, { + event: { type: "turn_completed", provider: "codex", turnId: "turn-1" }, + seq: undefined, + epoch: undefined, + timestamp: at(4), + }); + + expect(livenessChanges.at(-1)).toEqual({ + type: "stream_close", + committed: ["tool_call", "assistant_message"], + }); + owner.dispose(); + }); +}); + describe("create handoff lifetime", () => { const DRAFT_ID = "draft-1"; @@ -606,6 +689,7 @@ describe("create handoff lifetime", () => { replica, replaceDemandedAgentIds: () => undefined, drainQueuedAgentMessage: () => undefined, + applyTurnLiveness: () => undefined, ports: { observe: () => ({ ready: Promise.resolve(), release: async () => undefined }), readCursor: () => undefined, diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 67defcec04..0f5369c1c4 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -21,12 +21,14 @@ import { } from "@/utils/agent-initialization"; import { createSessionAgentStreamReducerQueue, + deriveAgentStreamTurnLiveness, type AgentStreamReducerEvent, type ProcessTimelineResponseOutput, processTimelineResponse, } from "./session-stream-reducers"; import { isTimelineResumeSnapshotAuthoritative } from "./timeline-sync-plan"; import { replaceWithCanonicalStream } from "@/types/stream"; +import type { TurnLivenessTransition } from "./turn-liveness"; export interface TimelineReplicaStorage { readTimeline(serverId: string, agentId: string): Promise; @@ -361,7 +363,12 @@ export type ViewedTimelineOwnerPorts = Omit< export interface ViewedTimelineOwner extends ViewedTimelineSync { applyTimelineResponse(payload: TimelineResponsePayload): void; - enqueueStreamEvent(agentId: string, event: AgentStreamReducerEvent): void; + /** + * Accepts one live stream event. Content is coalesced to the next frame; a turn boundary + * commits the turn's queued content before its liveness changes, so the view never shows + * a finished turn without its final reply. + */ + applyStreamEvent(agentId: string, event: AgentStreamReducerEvent): void; flushStreamAgent(agentId: string): void; } @@ -370,6 +377,7 @@ export function createViewedTimelineOwner(input: { replica: TimelineReplica; replaceDemandedAgentIds: (agentIds: string[]) => void; drainQueuedAgentMessage: (agentId: string) => void; + applyTurnLiveness: (agentId: string, transitions: readonly TurnLivenessTransition[]) => void; ports: ViewedTimelineOwnerPorts; }): ViewedTimelineOwner { const forcedTailReplacements = new Set(); @@ -409,8 +417,12 @@ export function createViewedTimelineOwner(input: { }); if (accepted) input.replica.timelineUpdated(payload.agentId); }, - enqueueStreamEvent(agentId, event) { + applyStreamEvent(agentId, event) { streamQueue.enqueue(agentId, event); + const turnLiveness = deriveAgentStreamTurnLiveness([event]); + if (turnLiveness.length === 0) return; + streamQueue.flushAgent(agentId); + input.applyTurnLiveness(agentId, turnLiveness); }, flushStreamAgent(agentId) { streamQueue.flushAgent(agentId);