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, {