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.
2026-10-26 03:03:04 +00:00 · 2026-07-31 14:35:26 +02:00
parent b69870892d
commit d6d377fbc6
19 changed files with 384 additions and 72 deletions
@@ -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
@@ -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<void> {
    };
    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<void> {
      ).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<void> {
        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();
  }
@@ -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,
@@ -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<void> {
+  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<void> {
  await expect(page.getByRole("button", { name: "Copy turn" }).first()).toBeVisible({
    timeout: 30_000,
@@ -184,9 +184,25 @@ export async function installDaemonWebSocketGate(page: Page) {
  const clientRequestCounts = new Map<string, number>();
  const timelineRequestCounts = new Map<string, number>();
  const serverMessageCounts = new Map<string, number>();
+  const agentStreamEventCounts = new Map<string, number>();
  const agentStreamItemCounts = new Map<string, number>();
  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<void>((resolve) => serverMessageWaiters.add(resolve));
      }
    },
+    async waitForAgentStreamEvent(type: string, count = 1): Promise<void> {
+      while ((agentStreamEventCounts.get(type) ?? 0) < count) {
+        await new Promise<void>((resolve) => serverMessageWaiters.add(resolve));
+      }
+    },
  };
 }
@@ -243,7 +243,7 @@ export interface AgentStreamViewProps {
  streamHead?: StreamItem[];
  pendingPermissions: Map<string, PendingPermission>;
  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<AgentStreamViewHandle, AgentStreamVi
      streamHead: providedStreamHead,
      pendingPermissions,
      pendingMessageSubmissions = EMPTY_PENDING_MESSAGE_SUBMISSIONS,
-      turnPresentation = IDLE_TURN_PRESENTATION,
+      turnPresentation,
      routeBottomAnchorRequest = null,
      isAuthoritativeHistoryReady = true,
      toast,
@@ -15,6 +15,7 @@ import { composerWorkspaceAttachment } from "@/composer/attachments/workspace";
 import { useAgentInputDraft } from "@/composer/draft/input-draft";
 import type { CreateAgentInitialValues } from "@/hooks/use-agent-form-state";
 import { useDraftAgentCreateFlow, type DraftCreateAttempt } from "@/composer/draft/create-flow";
+import { resolveTurnPresentation, TURN_LIVENESS_IDLE } from "@/timeline/turn-liveness";
 import { useHostRuntimeClient, useHostRuntimeIsConnected } from "@/runtime/host-runtime";
 import { buildWorkspaceDraftAgentConfig } from "@/screens/workspace/workspace-draft-agent-config";
 import { buildDraftStoreKey } from "@/stores/draft-keys";
@@ -542,6 +543,10 @@ export function WorkspaceDraftAgentTab({
      onCreated(result);
    },
  });
+  const turnPresentation = useMemo(
+    () => 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}
            />
@@ -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 <ComposerCancelButton {...rest} />;
 }

@@ -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 (
    <View style={styles.rightControls}>
@@ -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<QueueWriter>(
@@ -1697,7 +1696,6 @@ export function Composer({
      <ComposerCancelButtonSlot
        isAgentRunning={isAgentRunning}
        hasSendableContent={hasSendableContent}
-        isProcessing={isProcessing}
        buttonIconSize={buttonIconSize}
        cancelButtonStyle={cancelButtonStyle}
        handleCancelAgent={handleCancelAgent}
@@ -1716,7 +1714,6 @@ export function Composer({
      isAgentRunning,
      isCancellingAgent,
      isConnected,
-      isProcessing,
      t,
    ],
  );
@@ -1728,7 +1725,6 @@ export function Composer({
        hasAgent={hasAgent}
        isAgentRunning={isAgentRunning}
        hasSendableContent={hasSendableContent}
-        isProcessing={isProcessing}
        isCompact={isCompactLayout}
        buttonIconSize={buttonIconSize}
        handleToggleRealtimeVoice={handleToggleRealtimeVoice}
@@ -1749,7 +1745,6 @@ export function Composer({
      isAgentRunning,
      isConnected,
      isCompactLayout,
-      isProcessing,
      isVoiceModeForAgent,
      isVoiceSwitching,
      realtimeVoiceButtonStyle,
@@ -631,6 +631,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
      }

      if (result.commit === "discard") {
+        if (result.acknowledgedClientMessageIds.length > 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({
@@ -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(
@@ -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<string, PendingPermission>();
 const EMPTY_STREAM_ITEMS: StreamItem[] = [];
@@ -158,6 +159,10 @@ function ProviderSubagentPanel() {
    }),
    [isLoadingOlder, loadOlder, progressKey, timeline?.hasOlder],
  );
+  const turnPresentation = useMemo<TurnPresentation>(
+    () => ({ 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}
@@ -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,
@@ -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);
  }

@@ -1340,6 +1340,7 @@ interface AgentDirectoryRefreshInput {
 export class HostRuntimeStore {
  private controllers = new Map<string, HostRuntimeController>();
  private serverListeners = new Map<string, Set<() => void>>();
+  private agentStoppedRunningListeners = new Map<string, Set<(agentId: string) => 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;
@@ -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
@@ -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" }]);
  });

@@ -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<string>();
+  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";
@@ -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);
@@ -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<string, string[]>();
  const catchUps = new Map<string, CatchUpState>();
  const catchUpGenerations = new Map<string, number>();
-  // 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<string, ProjectedTimelineForwardFetchPlan>();
+  const pendingCatchUps = new Map<string, ProjectedTimelineForwardFetchPlan>();
  const lingeringRemovals = new Map<string, () => void>();
  const visibilityCatchUpPending = new Set<string>();
  const visibilityCatchUpErrors = new Set<string>();
@@ -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, {