From d6d377fbc6ad2584a57613f5a482cdef7d772f9f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 31 Jul 2026 14:35:26 +0200 Subject: [PATCH] fix(timeline): keep turn activity coherent Derive footer, tab, and composer activity from one lifecycle and submission projection. Idle directory transitions now request an authoritative resume so missed terminal events converge without treating directory delivery order as turn authority. --- docs/timeline-sync.md | 13 +-- .../app/e2e/agent-consecutive-turns.spec.ts | 71 ++++++++++++++- .../app/e2e/agent-message-submission.spec.ts | 88 ++++++++++++++++++- packages/app/e2e/helpers/agent-stream.ts | 11 +++ .../app/e2e/helpers/daemon-websocket-gate.ts | 39 ++++---- packages/app/src/agent-stream/view.tsx | 5 +- .../app/src/composer/draft/workspace-tab.tsx | 6 ++ packages/app/src/composer/index.tsx | 19 ++-- packages/app/src/contexts/session-context.tsx | 14 +++ packages/app/src/panels/agent-panel.tsx | 25 +++--- .../src/panels/provider-subagent-panel.tsx | 6 ++ .../src/runtime/directory-sync/index.test.ts | 2 +- .../app/src/runtime/directory-sync/index.ts | 4 +- packages/app/src/runtime/host-runtime.ts | 26 +++++- packages/app/src/stores/session-store.ts | 14 +++ .../timeline/session-stream-reducers.test.ts | 17 +++- .../src/timeline/session-stream-reducers.ts | 28 +++++- .../src/timeline/viewed-timeline-sync.test.ts | 33 +++++++ .../app/src/timeline/viewed-timeline-sync.ts | 35 +++++--- 19 files changed, 384 insertions(+), 72 deletions(-) diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index aae28cb279..74b19d18fc 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -63,7 +63,8 @@ Focus alone does not mutate timeline state; the tail response is compared with t authoritative range first. - The same epoch and `window.maxSeq` is an exact display no-op. The app advances synchronization - bookkeeping without replacing timeline arrays, preserving an upward-scrolled viewport. + bookkeeping and reconciles canonical submission identities without replacing timeline arrays, + preserving an upward-scrolled viewport. - When the page overlaps or is adjacent to the local end cursor, only projected items newer than that cursor are applied. Already-covered rows are not replayed. - A true middle gap, epoch change, or rewind atomically replaces stale canonical history with the @@ -162,10 +163,12 @@ Turn liveness has one client-side owner. `turn_started` opens the turn; ordered turn during hydration or after a selective stream subscription lapses, but an ordinary idle directory update does not close one or settle a submission because it can arrive before the terminal stream event. A completed timeline resume commits the response's coherent agent snapshot and closes any open -turn not backed by newer sequence evidence. Disconnect and replica removal are the destructive close -boundaries. Footer activity is the owned turn state unioned with unsettled submissions, and its running -start time comes from lifecycle/submission evidence rather than whichever timeline rows happen to be -mounted. +turn not backed by newer sequence evidence. An accepted running-to-idle directory transition requests +that resume for a viewed agent; it does not close liveness itself. This gives missed terminal events and +canonical prompt acknowledgements one convergence boundary without making directory delivery order an +authority for the footer. Disconnect and replica removal are the destructive close boundaries. Footer +activity is the owned turn state unioned with unsettled submissions, and its running start time comes +from lifecycle/submission evidence rather than whichever timeline rows happen to be mounted. Canonical submitted user rows carry the provider's `messageId` and Paseo's optional `clientMessageId`. The user-message producer reconciles them by `clientMessageId`, adds provider diff --git a/packages/app/e2e/agent-consecutive-turns.spec.ts b/packages/app/e2e/agent-consecutive-turns.spec.ts index f35b349c9b..ff0d1b94fc 100644 --- a/packages/app/e2e/agent-consecutive-turns.spec.ts +++ b/packages/app/e2e/agent-consecutive-turns.spec.ts @@ -45,6 +45,8 @@ interface TurnFrame { attachment: ElementFrame; footerRow: ElementFrame; spinner: ElementFrame; + agentTab: ElementFrame; + tabProgress: ElementFrame; interruptControl: ElementFrame; composer: ElementFrame & { value: string | null }; contentChildren: ContentChildFrame[]; @@ -199,6 +201,15 @@ async function recordTurnFrames(page: Page, prompt: string): Promise { }; const findImageAttachment = (row: Element | undefined) => row?.querySelector('[role="button"][aria-label="Open image attachment"]'); + const findAgentTabState = () => { + const agentTab = Array.from( + document.querySelectorAll('[data-testid^="workspace-tab-agent_"]'), + ).find((candidate) => isVisible(candidate)); + return { + agentTab, + tabProgress: agentTab?.querySelector('[role="progressbar"][aria-label="Agent running"]'), + }; + }; const sample = () => { const viewport = Array.from( document.querySelectorAll('[data-testid="agent-chat-scroll"]'), @@ -221,6 +232,7 @@ async function recordTurnFrames(page: Page, prompt: string): Promise { ).find((candidate) => /stop agent|canceling agent/i.test(candidate.getAttribute("aria-label") ?? ""), ); + const { agentTab, tabProgress } = findAgentTabState(); const scrollFrame = snapshot(viewport); const contentChildren = Array.from(viewport?.firstElementChild?.children ?? []).map( (child, index): ContentChildFrame => @@ -239,6 +251,8 @@ async function recordTurnFrames(page: Page, prompt: string): Promise { attachment: snapshot(attachment, viewport), footerRow: snapshot(footerRow, viewport, spinner.painted), spinner, + agentTab: snapshot(agentTab), + tabProgress: snapshot(tabProgress), interruptControl: snapshot(interrupt), composer: { ...snapshot(composerRoot ?? composer), @@ -278,7 +292,10 @@ async function waitForRecordedFrames( return frames.filter((frame) => { if (input.predicate === "user-visible") return frame.userRow.visible; return ( - frame.interruptControl.painted && frame.footerRow.painted && frame.spinner.painted + frame.interruptControl.painted && + frame.footerRow.painted && + frame.spinner.painted && + frame.tabProgress.painted ); }).length; }, @@ -415,6 +432,14 @@ function collectElementViolations( passes: hasPaintedLayout(frame.spinner), reason: "working spinner was not painted", }, + { + passes: hasPaintedLayout(frame.tabProgress), + reason: "selected tab running indicator was not painted", + }, + { + passes: hasPaintedLayout(frame.interruptControl), + reason: "interrupt control was not painted", + }, { passes: [hasPaintedLayout(frame.composer), frame.composer.value === ""].every(Boolean), reason: "composer was not painted and empty", @@ -483,7 +508,8 @@ function expectAtomicIdleToRunningTransition(frames: TurnFrame[]): void { index >= first && frame.interruptControl.painted && frame.footerRow.painted && - frame.spinner.painted, + frame.spinner.painted && + frame.tabProgress.painted, ); const transition = frames.slice(first); const baseline = transition[0]; @@ -522,6 +548,45 @@ function expectAtomicIdleToRunningTransition(frames: TurnFrame[]): void { ).toEqual([]); } +function expectAtomicFirstPromptTransition(frames: TurnFrame[]): void { + expectRowContinuity(frames); + const first = frames.findIndex((frame) => hasPaintedLayout(frame.userRow)); + const transition = frames.slice(first); + const violations: FrameViolation[] = []; + for (const [offset, frame] of transition.entries()) { + const index = first + offset; + violations.push( + ...violationsForChecks(index, [ + { + passes: hasPaintedLayout(frame.footerRow), + reason: "footer was not painted with the first prompt", + }, + { + passes: hasPaintedLayout(frame.spinner), + reason: "working spinner was not painted with the first prompt", + }, + { + passes: hasPaintedLayout(frame.composer) && frame.composer.value === "", + reason: "composer was not painted and empty with the first prompt", + }, + { + passes: !frame.agentTab.mounted || hasPaintedLayout(frame.tabProgress), + reason: "created agent tab appeared without running state", + }, + { + passes: !frame.agentTab.mounted || hasPaintedLayout(frame.interruptControl), + reason: "created agent appeared without its interrupt control", + }, + ]), + ); + } + 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)}`, + ).toEqual([]); +} + async function recordDelayedRunningTransition( page: Page, agent: MockAgentWorkspace, @@ -593,7 +658,7 @@ test("keeps the first prompt of a new agent in place through authoritative hydra await expect(submittedRow).toBeVisible(); await gate.waitForAgentStreamItem("user_message"); await recordPaintsFor(page, 80); - expectRowContinuity(await stopTurnFrameRecording(page)); + expectAtomicFirstPromptTransition(await stopTurnFrameRecording(page)); } finally { await workspace.cleanup(); } diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index 0026448eaf..53758e40c3 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -1,6 +1,11 @@ import type { Locator, Page } from "@playwright/test"; import { expect, test as baseTest } from "./fixtures"; -import { awaitToolCall, expectAgentIdle } from "./helpers/agent-stream"; +import { + awaitToolCall, + expectAgentIdle, + expectAgentSurfacesIdle, + expectRunningAgentChrome, +} from "./helpers/agent-stream"; import { gateNextAgentMessage } from "./helpers/agent-message-gate"; import { attachImageFromMenu, @@ -8,6 +13,7 @@ import { expectComposerEditable, expectAttachmentPill, expectComposerVisible, + cancelAgent, fillComposerDraft, sendDraftToQueue, startRunningMockAgent, @@ -664,6 +670,86 @@ async function completeDraftCreateSubmission( } test.describe("Agent message submission", () => { + test("shows every agent surface idle after interrupting a submitted turn", async ({ page }) => { + const title = "Interrupted submission"; + const agent = await seedMockAgentWorkspace({ + repoPrefix: "submission-interrupt-", + title, + model: "one-minute-stream", + }); + try { + await openAgentRoute(page, agent); + await expectComposerVisible(page); + await submitMessage(page, "Interrupt this submitted turn."); + await expectRunningAgentChrome(page, title); + + await cancelAgent(page); + + await expectAgentSurfacesIdle(page, title); + } finally { + await agent.cleanup(); + } + }); + + test("reconciles every agent surface when interrupt completion is missed", async ({ page }) => { + const gate = await installDaemonWebSocketGate(page); + const title = "Missed interrupt completion"; + const agent = await seedMockAgentWorkspace({ + repoPrefix: "submission-missed-interrupt-", + title, + model: "one-minute-stream", + }); + try { + await openAgentRoute(page, agent); + await expectComposerVisible(page); + await submitMessage(page, "Interrupt without delivering the terminal event."); + await expectRunningAgentChrome(page, title); + gate.setAgentStreamEventSuppressed("turn_canceled", true); + + await cancelAgent(page); + await gate.waitForAgentStreamEvent("turn_canceled"); + await agent.client.waitForFinish(agent.agentId, 30_000); + + await expectAgentSurfacesIdle(page, title); + } finally { + gate.setAgentStreamEventSuppressed("turn_canceled", false); + await agent.cleanup(); + } + }); + + test("reconciles an interrupted submission when live acknowledgements are missed", async ({ + page, + }) => { + const gate = await installDaemonWebSocketGate(page); + const title = "Missed submission acknowledgements"; + const prompt = "Interrupt without delivering live acknowledgements."; + const agent = await seedMockAgentWorkspace({ + repoPrefix: "submission-missed-acknowledgements-", + title, + model: "one-minute-stream", + }); + try { + await openAgentRoute(page, agent); + await expectComposerVisible(page); + gate.setAgentStreamItemSuppressed("user_message", true); + gate.setAgentStreamEventSuppressed("turn_canceled", true); + + await submitMessage(page, prompt); + await gate.waitForAgentStreamItem("user_message"); + await expectRunningAgentChrome(page, title); + await cancelAgent(page); + await gate.waitForAgentStreamEvent("turn_canceled"); + await agent.client.waitForFinish(agent.agentId, 30_000); + + await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toHaveCount(1); + await expectAgentSurfacesIdle(page, title); + } finally { + gate.setAgentStreamItemSuppressed("user_message", false); + gate.setAgentStreamEventSuppressed("turn_canceled", false); + await agent.cleanup(); + } + }); + test("keeps layout stable when submitting to an agent with existing history", async ({ page, submissionScenario, diff --git a/packages/app/e2e/helpers/agent-stream.ts b/packages/app/e2e/helpers/agent-stream.ts index 3fc23212b0..0654ff5a40 100644 --- a/packages/app/e2e/helpers/agent-stream.ts +++ b/packages/app/e2e/helpers/agent-stream.ts @@ -35,6 +35,17 @@ export async function expectRunningAgentChrome(page: Page, title: string): Promi await expectInlineWorkingIndicator(page); } +export async function expectAgentSurfacesIdle(page: Page, title: string): Promise { + const tab = page.getByRole("button", { name: title, exact: true }); + + await expect(tab).toBeVisible({ timeout: 30_000 }); + await expect(tab.getByRole("progressbar", { name: "Agent running" })).toHaveCount(0); + await expect(page.getByRole("button", { name: /stop agent|canceling agent/i })).toHaveCount(0); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + await expect(page.getByTestId("turn-working-elapsed")).toHaveCount(0); + await expectTurnCopyButton(page); +} + export async function expectTurnCopyButton(page: Page): Promise { await expect(page.getByRole("button", { name: "Copy turn" }).first()).toBeVisible({ timeout: 30_000, diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index 091d8f0b39..44d5a14ac1 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -184,9 +184,25 @@ export async function installDaemonWebSocketGate(page: Page) { const clientRequestCounts = new Map(); const timelineRequestCounts = new Map(); const serverMessageCounts = new Map(); + const agentStreamEventCounts = new Map(); const agentStreamItemCounts = new Map(); const serverMessageWaiters = new Set<() => void>(); + const recordServerMessage = (message: ClientRequest | null): void => { + const messageType = typeof message?.type === "string" ? message.type : null; + const itemType = readAgentStreamItemType(message); + const eventType = readAgentStreamEventType(message); + if (messageType) + serverMessageCounts.set(messageType, (serverMessageCounts.get(messageType) ?? 0) + 1); + if (itemType) + agentStreamItemCounts.set(itemType, (agentStreamItemCounts.get(itemType) ?? 0) + 1); + if (eventType) + agentStreamEventCounts.set(eventType, (agentStreamEventCounts.get(eventType) ?? 0) + 1); + if (!messageType && !itemType && !eventType) return; + for (const resolve of serverMessageWaiters) resolve(); + serverMessageWaiters.clear(); + }; + await page.routeWebSocket(daemonWsRoutePattern(), (ws) => { if (!acceptingConnections) { void ws.close({ code: 1008, reason: "Blocked by reconnect test." }); @@ -258,23 +274,7 @@ export async function installDaemonWebSocketGate(page: Page) { forceTimelineEpochReset && serverMessage?.type === "fetch_agent_timeline_response"; outboundMessage = forceTimelineReset(outboundMessage, shouldForceTimelineReset); if (shouldForceTimelineReset) forceTimelineEpochReset = false; - if (typeof serverMessage?.type === "string") { - serverMessageCounts.set( - serverMessage.type, - (serverMessageCounts.get(serverMessage.type) ?? 0) + 1, - ); - for (const resolve of serverMessageWaiters) resolve(); - serverMessageWaiters.clear(); - } - const agentStreamItemType = readAgentStreamItemType(serverMessage); - if (agentStreamItemType) { - agentStreamItemCounts.set( - agentStreamItemType, - (agentStreamItemCounts.get(agentStreamItemType) ?? 0) + 1, - ); - for (const resolve of serverMessageWaiters) resolve(); - serverMessageWaiters.clear(); - } + recordServerMessage(serverMessage); if (shouldHoldServerMessage(serverMessage, heldServerMessageType, heldAgentUpdate)) { heldServerMessage = { browser: ws, message: outboundMessage }; resolveHeldServerMessage?.(); @@ -472,5 +472,10 @@ export async function installDaemonWebSocketGate(page: Page) { await new Promise((resolve) => serverMessageWaiters.add(resolve)); } }, + async waitForAgentStreamEvent(type: string, count = 1): Promise { + while ((agentStreamEventCounts.get(type) ?? 0) < count) { + await new Promise((resolve) => serverMessageWaiters.add(resolve)); + } + }, }; } diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index c3ad6c2983..8639afbbcc 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -243,7 +243,7 @@ export interface AgentStreamViewProps { streamHead?: StreamItem[]; pendingPermissions: Map; pendingMessageSubmissions?: readonly PendingMessageSubmission[]; - turnPresentation?: TurnPresentation; + turnPresentation: TurnPresentation; routeBottomAnchorRequest?: BottomAnchorRouteRequest | null; isAuthoritativeHistoryReady?: boolean; toast?: ToastApi | null; @@ -271,7 +271,6 @@ const AGENT_CAPABILITY_FLAG_KEYS: (keyof AgentCapabilityFlags)[] = [ const EMPTY_STREAM_HEAD: StreamItem[] = []; const EMPTY_PENDING_MESSAGE_SUBMISSIONS: readonly PendingMessageSubmission[] = []; -const IDLE_TURN_PRESENTATION: TurnPresentation = { isActive: false, startedAt: null }; const GROUPED_TOOL_CALL_DETAIL_MAX_HEIGHT = 200; function buildChatHistoryAttachment(input: { @@ -335,7 +334,7 @@ const AgentStreamViewComponent = forwardRef resolveTurnPresentation(TURN_LIVENESS_IDLE, pendingMessageSubmissions), + [pendingMessageSubmissions], + ); useCommandCenterActions({ sourceId: `draft:${serverId}:${tabId}`, enabled: isPaneFocused && !isSubmitting, @@ -645,6 +650,7 @@ export function WorkspaceDraftAgentTab({ context={draftAgent} streamItems={submittedStreamItems} pendingMessageSubmissions={pendingMessageSubmissions} + turnPresentation={turnPresentation} pendingPermissions={EMPTY_PENDING_PERMISSIONS} onOpenWorkspaceFile={onOpenWorkspaceFile} /> diff --git a/packages/app/src/composer/index.tsx b/packages/app/src/composer/index.tsx index 12be754227..3333f740b2 100644 --- a/packages/app/src/composer/index.tsx +++ b/packages/app/src/composer/index.tsx @@ -41,7 +41,7 @@ import { } from "@/composer/agent-controls"; import { ContextWindowMeter } from "@/components/context-window-meter"; import { useImageAttachmentPicker } from "@/hooks/use-image-attachment-picker"; -import { useSessionStore } from "@/stores/session-store"; +import { selectAgentTurnPresentation, useSessionStore } from "@/stores/session-store"; import { useFilePicker } from "@/hooks/use-file-picker"; import { useFileDrop } from "@/components/file-drop/use-file-drop"; import type { DroppedItem } from "@/components/file-drop/types"; @@ -913,16 +913,14 @@ function ComposerCancelButton({ interface ComposerCancelButtonSlotProps extends ComposerCancelButtonProps { isAgentRunning: boolean; hasSendableContent: boolean; - isProcessing: boolean; } function ComposerCancelButtonSlot({ isAgentRunning, hasSendableContent, - isProcessing, ...rest }: ComposerCancelButtonSlotProps) { - if (!isAgentRunning || hasSendableContent || isProcessing) return null; + if (!isAgentRunning || hasSendableContent) return null; return ; } @@ -943,7 +941,6 @@ interface ComposerRightControlsSlotProps extends ComposerVoiceModeButtonProps { hasAgent: boolean; isAgentRunning: boolean; hasSendableContent: boolean; - isProcessing: boolean; isCompact: boolean; cancelButton: ReactElement; } @@ -953,7 +950,6 @@ function ComposerRightControlsSlot({ hasAgent, isAgentRunning, hasSendableContent, - isProcessing, isCompact, cancelButton, ...voiceProps @@ -961,7 +957,7 @@ function ComposerRightControlsSlot({ const hideVoiceForCompactInput = isCompact && hasSendableContent; const showVoiceModeButton = !isVoiceModeForAgent && hasAgent && !isAgentRunning && !hideVoiceForCompactInput; - const shouldShowCancelButton = isAgentRunning && !hasSendableContent && !isProcessing; + const shouldShowCancelButton = isAgentRunning && !hasSendableContent; if (!showVoiceModeButton && !shouldShowCancelButton) return null; return ( @@ -1300,7 +1296,10 @@ export function Composer({ onSubmitMessageRef.current = onSubmitMessage; }, [onSubmitMessage]); - const isAgentRunning = agentState.status === "running"; + const hasActiveTurn = useSessionStore( + (state) => selectAgentTurnPresentation(state.sessions[serverId], agentId).isActive, + ); + const isAgentRunning = agentState.status === "running" || hasActiveTurn; const hasAgent = agentState.status !== null; const queueWriter = useMemo( @@ -1697,7 +1696,6 @@ export function Composer({ 0) { + setAgentStreamState(serverId, agentId, { + acknowledgedClientMessageIds: result.acknowledgedClientMessageIds, + }); + } markAgentHistorySynchronized(serverId, agentId); } else { applyAgentTimelineResponseState(serverId, agentId, { @@ -664,6 +669,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider markAgentHistorySynchronized, recoverTimelineGap, serverId, + setAgentStreamState, setInitializingAgents, ], ); @@ -731,6 +737,14 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider viewedTimelineSyncRef.current?.setConnected(isConnected); }, [isConnected]); + useEffect( + () => + getHostRuntimeStore().subscribeAgentStoppedRunning(serverId, (agentId) => { + viewedTimelineSyncRef.current?.reconcileAgent(agentId); + }), + [serverId], + ); + // Daemon message handlers - directly update Zustand store useEffect(() => { const agentStreamReducerQueue = createSessionAgentStreamReducerQueue({ diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index c3d9cd762b..5e49bc32aa 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -26,7 +26,6 @@ import { useRetainedPanelActive } from "@/components/retained-panel"; import { SidebarCallout } from "@/components/sidebar-callout"; import { Composer } from "@/composer"; import { getActiveMessageSubmissions } from "@/composer/submission/model"; -import { resolveTurnPresentation, TURN_LIVENESS_IDLE } from "@/timeline/turn-liveness"; import { RewindComposerRestoreProvider } from "@/components/rewind/composer-restore"; import { getProviderIcon } from "@/components/provider-icons"; import { @@ -73,7 +72,12 @@ import { WorkspaceDraftAgentTab } from "@/composer/draft/workspace-tab"; import { useCreateFlowStore } from "@/stores/create-flow-store"; import { buildDraftStoreKey, generateDraftId } from "@/stores/draft-keys"; import { usePanelStore } from "@/stores/panel-store"; -import { selectAgentTimelineState, type Agent, useSessionStore } from "@/stores/session-store"; +import { + selectAgentTimelineState, + selectAgentTurnPresentation, + type Agent, + useSessionStore, +} from "@/stores/session-store"; import { useWorkspaceLayoutStore } from "@/stores/workspace-layout-store"; import { buildWorkspaceTabPersistenceKey } from "@/workspace-tabs/model"; import type { Theme } from "@/styles/theme"; @@ -318,6 +322,7 @@ function useAgentPanelDescriptor( pendingPermissionCount: agent?.pendingPermissions.length ?? 0, requiresAttention: agent?.requiresAttention ?? false, attentionReason: agent?.attentionReason ?? null, + isTurnActive: selectAgentTurnPresentation(session, target.agentId).isActive, }; }), ); @@ -333,7 +338,7 @@ function useAgentPanelDescriptor( icon, statusBucket: descriptorState.status ? deriveSidebarStateBucket({ - status: descriptorState.status, + status: descriptorState.isTurnActive ? "running" : descriptorState.status, pendingPermissionCount: descriptorState.pendingPermissionCount, requiresAttention: descriptorState.requiresAttention, attentionReason: descriptorState.attentionReason, @@ -1298,14 +1303,12 @@ const AgentStreamSection = memo(function AgentStreamSection({ ? getActiveMessageSubmissions(state.sessions[serverId]?.messageSubmissions.get(agentId)) : EMPTY_MESSAGE_SUBMISSIONS, ); - const turnLiveness = useSessionStore((state) => - agentId - ? (state.sessions[serverId]?.agentTurnLiveness.get(agentId) ?? TURN_LIVENESS_IDLE) - : TURN_LIVENESS_IDLE, - ); - const turnPresentation = useMemo( - () => resolveTurnPresentation(turnLiveness, pendingMessageSubmissions), - [pendingMessageSubmissions, turnLiveness], + const turnPresentation = useSessionStore( + useShallow((state) => + agentId + ? selectAgentTurnPresentation(state.sessions[serverId], agentId) + : { isActive: false, startedAt: null }, + ), ); const streamItems = streamItemsRaw ?? EMPTY_STREAM_ITEMS; const pendingPermissionList = useStoreWithEqualityFn( diff --git a/packages/app/src/panels/provider-subagent-panel.tsx b/packages/app/src/panels/provider-subagent-panel.tsx index 12558a2ed9..aae1e44907 100644 --- a/packages/app/src/panels/provider-subagent-panel.tsx +++ b/packages/app/src/panels/provider-subagent-panel.tsx @@ -20,6 +20,7 @@ import type { PendingPermission } from "@/types/shared"; import type { StreamItem } from "@/types/stream"; import { deriveSidebarStateBucket } from "@/utils/sidebar-agent-state"; import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import type { TurnPresentation } from "@/timeline/turn-liveness"; const EMPTY_PERMISSIONS = new Map(); const EMPTY_STREAM_ITEMS: StreamItem[] = []; @@ -158,6 +159,10 @@ function ProviderSubagentPanel() { }), [isLoadingOlder, loadOlder, progressKey, timeline?.hasOlder], ); + const turnPresentation = useMemo( + () => ({ isActive: descriptor?.status === "running", startedAt: null }), + [descriptor?.status], + ); if (serverInfo && !supported) { return ( @@ -175,6 +180,7 @@ function ProviderSubagentPanel() { context={streamContext} streamItems={timeline?.tail ?? EMPTY_STREAM_ITEMS} streamHead={timeline?.head ?? EMPTY_STREAM_ITEMS} + turnPresentation={turnPresentation} pendingPermissions={EMPTY_PERMISSIONS} isAuthoritativeHistoryReady onOpenWorkspaceFile={openFileInWorkspace} diff --git a/packages/app/src/runtime/directory-sync/index.test.ts b/packages/app/src/runtime/directory-sync/index.test.ts index 7279677139..221859c4b7 100644 --- a/packages/app/src/runtime/directory-sync/index.test.ts +++ b/packages/app/src/runtime/directory-sync/index.test.ts @@ -74,7 +74,7 @@ function createDirectory(serverId: string): { serverIds.add(serverId); const client = new FakeDirectoryClient(); const directory = new DirectorySync(serverId, { - drainQueuedAgentMessage: () => undefined, + onAgentStoppedRunning: () => undefined, markAgentLoading: () => undefined, markAgentReady: () => undefined, markAgentError: () => undefined, diff --git a/packages/app/src/runtime/directory-sync/index.ts b/packages/app/src/runtime/directory-sync/index.ts index 317eabf6af..4bd46bbf88 100644 --- a/packages/app/src/runtime/directory-sync/index.ts +++ b/packages/app/src/runtime/directory-sync/index.ts @@ -79,13 +79,13 @@ export class DirectorySync { constructor( private readonly serverId: string, private readonly callbacks: { - drainQueuedAgentMessage: (agentId: string) => void; + onAgentStoppedRunning: (agentId: string) => void; markAgentLoading: () => void; markAgentReady: () => void; markAgentError: (error: string) => void; }, ) { - this.agents = new AgentDirectoryReplica(serverId, callbacks.drainQueuedAgentMessage); + this.agents = new AgentDirectoryReplica(serverId, callbacks.onAgentStoppedRunning); this.workspaces = new WorkspaceDirectoryReplica(serverId); } diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index d77682a2f8..6d05aa0e7f 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -1340,6 +1340,7 @@ interface AgentDirectoryRefreshInput { export class HostRuntimeStore { private controllers = new Map(); private serverListeners = new Map void>>(); + private agentStoppedRunningListeners = new Map void>>(); private globalListeners = new Set<() => void>(); private hostListListeners = new Set<() => void>(); private version = 0; @@ -1584,11 +1585,12 @@ export class HostRuntimeStore { rekeyMap(this.controllers, oldServerId, newServerId); rekeyMap(this.lastConnectionStatusByServer, oldServerId, newServerId); rekeyMap(this.directoryBootstrapInFlight, oldServerId, newServerId); + rekeyMap(this.agentStoppedRunningListeners, oldServerId, newServerId); this.replicaCache.reconcileServerId(oldServerId, newServerId); this.directorySyncByServer.get(oldServerId)?.dispose(); this.directorySyncByServer.delete(oldServerId); const directory = new DirectorySync(newServerId, { - drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(newServerId, agentId), + onAgentStoppedRunning: (agentId) => this.onAgentStoppedRunning(newServerId, agentId), markAgentLoading: () => controller.markAgentDirectorySyncLoading(), markAgentReady: () => controller.markAgentDirectorySyncReady(), markAgentError: (error) => controller.markAgentDirectorySyncError(error), @@ -1925,8 +1927,7 @@ export class HostRuntimeStore { this.directorySyncByServer.set( host.serverId, new DirectorySync(host.serverId, { - drainQueuedAgentMessage: (agentId) => - this.drainQueuedAgentMessage(host.serverId, agentId), + onAgentStoppedRunning: (agentId) => this.onAgentStoppedRunning(host.serverId, agentId), markAgentLoading: () => controller.markAgentDirectorySyncLoading(), markAgentReady: () => controller.markAgentDirectorySyncReady(), markAgentError: (error) => controller.markAgentDirectorySyncError(error), @@ -2125,6 +2126,18 @@ export class HostRuntimeStore { }; } + subscribeAgentStoppedRunning(serverId: string, listener: (agentId: string) => void): () => void { + const listeners = this.agentStoppedRunningListeners.get(serverId) ?? new Set(); + listeners.add(listener); + this.agentStoppedRunningListeners.set(serverId, listeners); + return () => { + const current = this.agentStoppedRunningListeners.get(serverId); + if (!current) return; + current.delete(listener); + if (current.size === 0) this.agentStoppedRunningListeners.delete(serverId); + }; + } + subscribeAll(listener: () => void): () => void { this.globalListeners.add(listener); return () => { @@ -2233,6 +2246,13 @@ export class HostRuntimeStore { listener(); } } + + private onAgentStoppedRunning(serverId: string, agentId: string): void { + this.drainQueuedAgentMessage(serverId, agentId); + for (const listener of this.agentStoppedRunningListeners.get(serverId) ?? []) { + listener(agentId); + } + } } let singletonHostRuntimeStore: HostRuntimeStore | null = null; diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index c37c298eea..dc057d1b42 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -15,6 +15,7 @@ import { import { acceptMessageSubmission, beginMessageSubmission, + getActiveMessageSubmissions, observeMessageSubmissionCanonical, rejectMessageSubmission, type MessageSubmissionRecord, @@ -54,7 +55,10 @@ import { } from "@/utils/workspace-agent-activity"; import { applyTurnLivenessTransition, + resolveTurnPresentation, + TURN_LIVENESS_IDLE, type TurnLiveness, + type TurnPresentation, type TurnLivenessTransition, } from "@/timeline/turn-liveness"; @@ -377,6 +381,16 @@ export function selectAgentTimelineState( return items.length > 0 ? { status: "painted", items } : { status: "cold" }; } +export function selectAgentTurnPresentation( + session: SessionState | undefined, + agentId: string, +): TurnPresentation { + return resolveTurnPresentation( + session?.agentTurnLiveness.get(agentId) ?? TURN_LIVENESS_IDLE, + getActiveMessageSubmissions(session?.messageSubmissions.get(agentId)), + ); +} + export type WorkspaceRestoreStatus = "restoring" | "failed" | "needs-host-upgrade"; // Per-session state diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index d192280e56..6e96514a69 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -250,7 +250,8 @@ const baseStreamInput: ProcessAgentStreamEventInput = { describe("processTimelineResponse", () => { it("discards an unchanged resume tail without replacing timeline state", () => { - const currentTail = [makeAssistantItem("existing tail", "existing-tail")]; + const submitted = makeSubmittedUserMessage("local prompt", "client-message"); + const currentTail = [submitted, makeAssistantItem("existing tail", "existing-tail")]; const currentHead = [makeAssistantItem("existing head", "existing-head")]; const result = processTimelineResponse({ @@ -264,7 +265,18 @@ describe("processTimelineResponse", () => { window: { minSeq: 1, maxSeq: 40, nextSeq: 41 }, startCursor: { seq: 1 }, endCursor: { seq: 40 }, - entries: [makeTimelineEntry(1, "existing tail", "assistant_message", 40)], + entries: [ + { + ...makeTimelineEntry(1, "provider prompt", "user_message"), + item: { + type: "user_message", + text: "provider prompt", + messageId: "provider-message", + clientMessageId: "client-message", + }, + }, + makeTimelineEntry(2, "existing tail", "assistant_message", 40), + ], }, }); @@ -272,6 +284,7 @@ describe("processTimelineResponse", () => { expect(result.tail).toBe(currentTail); expect(result.head).toBe(currentHead); expect(result.cursorChanged).toBe(false); + expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]); expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]); }); diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 772c36b5d6..daa1871968 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -760,6 +760,27 @@ function applyAcceptedForwardTimelineUnits(params: { return { tail, head, acknowledgedClientMessageIds: [...acknowledgedClientMessageIds] }; } +function deriveCanonicalAcknowledgements(params: { + units: TimelineUnit[]; + epoch: string; + currentTail: StreamItem[]; + currentHead: StreamItem[]; +}): string[] { + let tail = params.currentTail; + let head = params.currentHead; + const acknowledged = new Set(); + for (const unit of params.units) { + if (unit.event.type !== "timeline" || unit.event.item.type !== "user_message") continue; + const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch }); + tail = applied.tail; + head = applied.head; + for (const clientMessageId of applied.acknowledgedClientMessageIds) { + acknowledged.add(clientMessageId); + } + } + return [...acknowledged]; +} + function applyTimelineIncrementalPath(args: { timelineUnits: TimelineUnit[]; payload: ProcessTimelineResponseInput["payload"]; @@ -949,7 +970,12 @@ export function processTimelineResponse( cursorChanged: false, older: "unchanged", sideEffects: [], - acknowledgedClientMessageIds: [], + acknowledgedClientMessageIds: deriveCanonicalAcknowledgements({ + units: timelineUnits, + epoch: payload.epoch, + currentTail, + currentHead, + }), }; } else if (replace || resumeTailPolicy.kind === "replace") { const isResumeReplacement = resumeTailPolicy.kind === "replace"; diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index edf5a7895f..85b52092e8 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -201,6 +201,39 @@ test("a gap absorbed by a running tail is recovered after the tail completes", a await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); }); +test("an idle reconciliation waits for an in-flight tail and fetches a fresh snapshot", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const initialTail = await world.nextFetch("agent-a"); + + world.sync.reconcileAgent("agent-a"); + + world.expectNoPendingFetch(); + initialTail.respond({ hasNewer: false }); + + const reconciliation = await world.nextFetch("agent-a"); + expect(reconciliation.request).toEqual({ + direction: "tail", + limit: 40, + projection: "projected", + }); + reconciliation.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); +}); + +test("does not reconcile an agent outside the viewed membership", () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + + world.sync.reconcileAgent("agent-a"); + + world.expectNoPendingFetch(); + world.expectNoPendingMembership(); +}); + test("unchanged visible-set publication does not cancel paged catch-up", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 1784083f3f..caa4c63d78 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -33,6 +33,7 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge { setActive(active: boolean): void; setConnected(connected: boolean): void; setDeliveryMode(mode: TimelineDeliveryMode): void; + reconcileAgent(agentId: string): void; recoverGap(agentId: string, cursor: { epoch: string; endSeq: number }): void; dispose(): void; } @@ -95,9 +96,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const sources = new Map(); const catchUps = new Map(); const catchUpGenerations = new Map(); - // Recovery owed but not runnable yet: disconnected, unacknowledged, or parked behind a tail. + // Authoritative fetch owed but not runnable yet: disconnected, unacknowledged, or parked. // Acknowledgement and tail completion are the only drain points. - const pendingGaps = new Map(); + const pendingCatchUps = new Map(); const lingeringRemovals = new Map void>(); const visibilityCatchUpPending = new Set(); const visibilityCatchUpErrors = new Set(); @@ -145,7 +146,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed catchUpGenerations.set(agentId, (catchUpGenerations.get(agentId) ?? 0) + 1); catchUps.get(agentId)?.cancelRetry?.(); catchUps.delete(agentId); - pendingGaps.delete(agentId); + pendingCatchUps.delete(agentId); }; const fetchUntilCurrent = async ( @@ -182,9 +183,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed throw new Error(`Timeline page for ${agentId} hasNewer without an end cursor`); } catchUps.set(agentId, { generation, status: "complete" }); - const pendingGap = pendingGaps.get(agentId); - if (pendingGap) { - startCatchUp(agentId, { request: pendingGap, supersede: true }); + const pendingCatchUp = pendingCatchUps.get(agentId); + if (pendingCatchUp) { + startCatchUp(agentId, { request: pendingCatchUp, supersede: true }); return; } setVisibilityCatchUpReady(agentId); @@ -211,14 +212,14 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed ) => { const { request, supersede = false } = options; if (!connected || !isDesired(agentId) || !isAcknowledged(agentId)) { - if (request) pendingGaps.set(agentId, request); + if (request) pendingCatchUps.set(agentId, request); return; } const nextRequest = request ?? planTimelineTailFetch(); const current = catchUps.get(agentId); const decision = decideCatchUp({ current, request: nextRequest, supersede }); if (decision === "keep-and-park") { - pendingGaps.set(agentId, nextRequest); + pendingCatchUps.set(agentId, nextRequest); return; } if (decision === "keep") { @@ -228,14 +229,17 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const generation = (catchUpGenerations.get(agentId) ?? 0) + 1; catchUpGenerations.set(agentId, generation); catchUps.set(agentId, { generation, status: "running", request: nextRequest }); - pendingGaps.delete(agentId); + pendingCatchUps.delete(agentId); void fetchUntilCurrent(agentId, generation, nextRequest); }; const startAcknowledgedCatchUps = () => { for (const agentId of acknowledged) { - const gap = pendingGaps.get(agentId); - startCatchUp(agentId, { request: gap, supersede: Boolean(gap) }); + const pendingCatchUp = pendingCatchUps.get(agentId); + startCatchUp(agentId, { + request: pendingCatchUp, + supersede: Boolean(pendingCatchUp), + }); } }; @@ -445,6 +449,15 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed if (deliveryMode === "selective" && connected) void reconcileMembership(); else if (connected) startAcknowledgedCatchUps(); }, + reconcileAgent(agentId) { + if (!isDesired(agentId)) return; + const request = planTimelineTailFetch(); + if (catchUps.get(agentId)?.status === "running") { + pendingCatchUps.set(agentId, request); + return; + } + startCatchUp(agentId, { request, supersede: true }); + }, recoverGap(agentId, cursor) { if (!isDesired(agentId)) return; startCatchUp(agentId, {