diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index e79b9dd4eb..aae28cb279 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -157,11 +157,15 @@ row is recorded before handler output. The app creates two-phase submission reco this capability. Older hosts keep the shipped untracked optimistic-row behavior and roll that row back on RPC rejection. -Turn liveness is the union of an unsettled submission and an open stream turn. `turn_started` opens the -turn; ordered `turn_completed`, `turn_failed`, and `turn_canceled` events close it. A running directory -transition may seed or raise an open turn during hydration or after a selective stream subscription -lapses, but directory status never closes one or settles a submission. Disconnect and replica removal -are the other destructive close boundaries. +Turn liveness has one client-side owner. `turn_started` opens the turn; ordered `turn_completed`, +`turn_failed`, and `turn_canceled` events close it. A running directory transition may seed an open +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. 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/viewed-agent-timelines.spec.ts b/packages/app/e2e/viewed-agent-timelines.spec.ts index a16723aec4..35a4e24e1d 100644 --- a/packages/app/e2e/viewed-agent-timelines.spec.ts +++ b/packages/app/e2e/viewed-agent-timelines.spec.ts @@ -6,6 +6,11 @@ import { getServerId } from "./helpers/server-id"; import { observeTimelineSubscriptions } from "./helpers/timeline-delivery"; import { waitForWorkspaceTabsVisible } from "./helpers/workspace-tabs"; import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate"; +import { + expectAgentIdle, + expectInlineWorkingIndicator, + expectTurnCopyButton, +} from "./helpers/agent-stream"; import { expectReconnectingToastGone, expectReconnectingToastVisible, @@ -19,19 +24,21 @@ interface ViewedTimelineScenario { cleanup(): Promise; } -async function seedViewedTimelineScenario(): Promise { +async function seedViewedTimelineScenario( + options: { firstAgentModel?: string } = {}, +): Promise { const workspace = await seedWorkspace({ repoPrefix: "viewed-timelines-" }); - const createAgent = (title: string) => + const createAgent = (title: string, model = "ten-second-stream") => workspace.client.createAgent({ provider: "mock", cwd: workspace.repoPath, workspaceId: workspace.workspaceId, title, modeId: "load-test", - model: "ten-second-stream", + model, }); const [firstAgent, secondAgent] = await Promise.all([ - createAgent("First viewed chat"), + createAgent("First viewed chat", options.firstAgentModel), createAgent("Second viewed chat"), ]); return { @@ -71,7 +78,46 @@ async function commitMessage(scenario: ViewedTimelineScenario, agentId: string, expect(finish.status).toBe("idle"); } +async function startVisibleTurn( + page: Page, + scenario: ViewedTimelineScenario, + prompt: string, +): Promise { + await scenario.client.sendAgentMessage(scenario.firstAgentId, prompt); + await expect(page.getByText(prompt, { exact: true })).toBeVisible(); + await expectInlineWorkingIndicator(page); +} + +async function expectAgentConsistentlyIdle(page: Page, title: string): Promise { + const tab = page.getByRole("button", { name: title, exact: true }); + await expect(tab.locator('[data-status-bucket="running"]')).toHaveCount(0); + await expectAgentIdle(page); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + await expectTurnCopyButton(page); +} + test.describe("Viewed agent timelines", () => { + test("a turn that finishes while unsubscribed reopens with consistently idle chrome", async ({ + page, + }) => { + test.setTimeout(150_000); + const subscriptions = observeTimelineSubscriptions(page); + const scenario = await seedViewedTimelineScenario({ firstAgentModel: "one-minute-stream" }); + try { + await openAgent(page, scenario, scenario.firstAgentId); + await startVisibleTurn(page, scenario, "Finish after this chat becomes unsubscribed."); + await selectAgent(page, "Second viewed chat"); + await subscriptions.waitForSubscribedAgents([scenario.secondAgentId], { timeout: 45_000 }); + const finish = await scenario.client.waitForFinish(scenario.firstAgentId, 90_000); + expect(finish.status).toBe("idle"); + + await selectAgent(page, "First viewed chat"); + await expectAgentConsistentlyIdle(page, "First viewed chat"); + } finally { + await scenario.cleanup(); + } + }); + test("an unsubscribed hidden chat catches up when shown", async ({ page }) => { test.setTimeout(90_000); const subscriptions = observeTimelineSubscriptions(page); diff --git a/packages/app/src/agent-stream/model.test.ts b/packages/app/src/agent-stream/model.test.ts index 030ea505b4..c55c51ddaf 100644 --- a/packages/app/src/agent-stream/model.test.ts +++ b/packages/app/src/agent-stream/model.test.ts @@ -36,6 +36,7 @@ describe("buildAgentStreamRenderModel", () => { const model = buildAgentStreamRenderModel({ isTurnActive: true, + activeTurnStartedAt: tail.at(-2)?.timestamp ?? null, tail, head, platform: "web", @@ -54,6 +55,7 @@ describe("buildAgentStreamRenderModel", () => { const model = buildAgentStreamRenderModel({ isTurnActive: true, + activeTurnStartedAt: tail[0]?.timestamp ?? null, tail, head, platform: "web", @@ -72,6 +74,7 @@ describe("buildAgentStreamRenderModel", () => { const first = buildAgentStreamRenderModel({ isTurnActive: true, + activeTurnStartedAt: tail[0]?.timestamp ?? null, tail, head: firstHead, platform: "native", @@ -79,6 +82,7 @@ describe("buildAgentStreamRenderModel", () => { }); const second = buildAgentStreamRenderModel({ isTurnActive: true, + activeTurnStartedAt: tail[0]?.timestamp ?? null, tail, head: secondHead, platform: "native", @@ -96,6 +100,7 @@ describe("buildAgentStreamRenderModel", () => { const model = buildAgentStreamRenderModel({ isTurnActive: true, + activeTurnStartedAt: tail[0]?.timestamp ?? null, tail, head, platform: "web", @@ -112,6 +117,7 @@ describe("buildAgentStreamRenderModel", () => { const model = buildAgentStreamRenderModel({ isTurnActive: false, + activeTurnStartedAt: null, tail, head, platform: "web", @@ -131,6 +137,7 @@ describe("buildAgentStreamRenderModel", () => { const model = buildAgentStreamRenderModel({ isTurnActive: false, + activeTurnStartedAt: null, tail, head: [], platform: "native", @@ -150,6 +157,7 @@ describe("buildAgentStreamRenderModel", () => { const model = buildAgentStreamRenderModel({ isTurnActive: false, + activeTurnStartedAt: null, tail, head: [], platform: "web", diff --git a/packages/app/src/agent-stream/model.ts b/packages/app/src/agent-stream/model.ts index 21bcae50a6..f7a3854972 100644 --- a/packages/app/src/agent-stream/model.ts +++ b/packages/app/src/agent-stream/model.ts @@ -36,6 +36,7 @@ export interface AgentStreamRenderModel { export interface BuildAgentStreamRenderModelInput { isTurnActive: boolean; + activeTurnStartedAt: Date | null; tail: StreamItem[]; head: StreamItem[]; platform: "web" | "native"; @@ -56,7 +57,7 @@ const splitHistoryCache = new WeakMap< >(); const turnTimingCache = new WeakMap< StreamItem[], - WeakMap> + WeakMap> >(); function getOrderedItems(params: { @@ -132,6 +133,7 @@ function splitOrderedTail(params: { function getTurnTiming(params: { isTurnActive: boolean; + activeTurnStartedAt: Date | null; tail: StreamItem[]; head: StreamItem[]; }): StreamTurnTiming { @@ -145,12 +147,13 @@ function getTurnTiming(params: { cachedByActivity = new Map(); cachedByHead.set(params.head, cachedByActivity); } - const cached = cachedByActivity.get(params.isTurnActive); + const activityKey = `${params.isTurnActive}:${params.activeTurnStartedAt?.getTime() ?? "none"}`; + const cached = cachedByActivity.get(activityKey); if (cached) { return cached; } const timing = deriveStreamTurnTiming(params); - cachedByActivity.set(params.isTurnActive, timing); + cachedByActivity.set(activityKey, timing); return timing; } @@ -189,6 +192,7 @@ export function buildAgentStreamRenderModel( }); const turnTiming = getTurnTiming({ isTurnActive: input.isTurnActive, + activeTurnStartedAt: input.activeTurnStartedAt, tail: input.tail, head: input.head, }); diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 9deb5f7d2d..c3ad6c2983 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -42,6 +42,7 @@ import { import { PlanCard } from "@/components/plan-card"; import type { StreamItem } from "@/types/stream"; import type { PendingMessageSubmission } from "@/composer/submission/model"; +import type { TurnPresentation } from "@/timeline/turn-liveness"; import type { PendingPermission } from "@/types/shared"; import type { AgentCapabilityFlags, @@ -242,7 +243,7 @@ export interface AgentStreamViewProps { streamHead?: StreamItem[]; pendingPermissions: Map; pendingMessageSubmissions?: readonly PendingMessageSubmission[]; - hasActiveTurn?: boolean; + turnPresentation?: TurnPresentation; routeBottomAnchorRequest?: BottomAnchorRouteRequest | null; isAuthoritativeHistoryReady?: boolean; toast?: ToastApi | null; @@ -270,6 +271,7 @@ 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: { @@ -333,7 +335,7 @@ const AgentStreamViewComponent = forwardRef 0 || hasActiveTurn; + const isTurnActive = turnPresentation.isActive; // Keep retained history outside the 48ms live-head flush path. const preparedToolCallHistory = useMemo( () => prepareToolCallHistory(toolCallDetailLevel, effectiveStreamItems), @@ -592,12 +594,19 @@ const AgentStreamViewComponent = forwardRef { return buildAgentStreamRenderModel({ isTurnActive, + activeTurnStartedAt: turnPresentation.startedAt, tail: projectedToolCalls.tail, head: projectedToolCalls.head, platform: isWeb ? "web" : "native", isMobileBreakpoint: isMobile, }); - }, [isMobile, isTurnActive, projectedToolCalls.head, projectedToolCalls.tail]); + }, [ + isMobile, + isTurnActive, + projectedToolCalls.head, + projectedToolCalls.tail, + turnPresentation.startedAt, + ]); const streamLayout = useMemo( () => layoutStream({ @@ -1176,7 +1185,7 @@ function agentStreamViewPropsEqual( if (left.pendingMessageSubmissions !== right.pendingMessageSubmissions) { reasons.push("pendingMessageSubmissions"); } - if (left.hasActiveTurn !== right.hasActiveTurn) reasons.push("hasActiveTurn"); + if (left.turnPresentation !== right.turnPresentation) reasons.push("turnPresentation"); if ( !bottomAnchorRouteRequestsEqual(left.routeBottomAnchorRequest, right.routeBottomAnchorRequest) ) { diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 43d82e5c27..c8d0afdcf6 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -19,7 +19,7 @@ import { type TimelineReducerSideEffect, } from "@/timeline/session-stream-reducers"; import { useCreateFlowStore } from "@/stores/create-flow-store"; -import { isTimelineCatchUpComplete } from "@/timeline/timeline-sync-plan"; +import { isTimelineResumeSnapshotAuthoritative } from "@/timeline/timeline-sync-plan"; import { createViewedTimelineSync, type TimelineDeliveryMode, @@ -347,6 +347,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const setAgentStreamTail = useSessionStore((state) => state.setAgentStreamTail); const setAgentStreamHead = useSessionStore((state) => state.setAgentStreamHead); const setAgentStreamState = useSessionStore((state) => state.setAgentStreamState); + const applyAgentTurnLiveness = useSessionStore((state) => state.applyAgentTurnLiveness); + const clearAgentTurnLiveness = useSessionStore((state) => state.clearAgentTurnLiveness); const clearAgentStreamHead = useSessionStore((state) => state.clearAgentStreamHead); const setAgentTimelineCursor = useSessionStore((state) => state.setAgentTimelineCursor); const setInitializingAgents = useSessionStore((state) => state.setInitializingAgents); @@ -553,12 +555,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider () => client.subscribeConnectionStatus((connection) => { if (connection.status === "connected") return; - const activeAgentTurns = useSessionStore.getState().sessions[serverId]?.activeAgentTurns; - for (const agentId of activeAgentTurns ?? []) { - setAgentStreamState(serverId, agentId, { turnActive: false }); - } + clearAgentTurnLiveness(serverId); }), - [client, serverId, setAgentStreamState], + [clearAgentTurnLiveness, client, serverId], ); const applyWorkspaceSetupProgress = useCallback( @@ -577,13 +576,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider ) => { const agentId = payload.agentId; const initKey = getInitKey(serverId, agentId); - const catchUpComplete = isTimelineCatchUpComplete({ + const shouldMarkAuthoritativeHistoryApplied = isTimelineResumeSnapshotAuthoritative({ direction: payload.direction, hasNewer: payload.hasNewer, error: payload.error, }); - const shouldMarkAuthoritativeHistoryApplied = - payload.direction === "tail" || (payload.direction === "after" && catchUpComplete); // Read current store state const session = useSessionStore.getState().sessions[serverId]; @@ -622,6 +619,17 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider return; } + if (shouldMarkAuthoritativeHistoryApplied && payload.agent) { + applyAgentTurnLiveness(serverId, agentId, { + type: "resume_snapshot", + status: payload.agent.status, + startedAt: payload.agent.lastUserMessageAt + ? new Date(payload.agent.lastUserMessageAt) + : null, + coverage: { epoch: payload.epoch, seq: payload.window.maxSeq }, + }); + } + if (result.commit === "discard") { markAgentHistorySynchronized(serverId, agentId); } else { @@ -652,6 +660,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider }, [ applyAgentTimelineResponseState, + applyAgentTurnLiveness, markAgentHistorySynchronized, recoverTimelineGap, serverId, @@ -727,6 +736,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const agentStreamReducerQueue = createSessionAgentStreamReducerQueue({ serverId, setAgentStreamState, + applyAgentTurnLiveness, setAgentTimelineCursor, recoverTimelineGap, }); @@ -1088,6 +1098,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setAgentStreamTail, setAgentStreamHead, setAgentStreamState, + applyAgentTurnLiveness, clearAgentStreamHead, setAgentTimelineCursor, setInitializingAgents, diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index 9d529ba82c..c3d9cd762b 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -26,6 +26,7 @@ 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 { @@ -1297,8 +1298,14 @@ const AgentStreamSection = memo(function AgentStreamSection({ ? getActiveMessageSubmissions(state.sessions[serverId]?.messageSubmissions.get(agentId)) : EMPTY_MESSAGE_SUBMISSIONS, ); - const hasActiveTurn = useSessionStore((state) => - agentId ? state.sessions[serverId]?.activeAgentTurns.has(agentId) === true : false, + const 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 streamItems = streamItemsRaw ?? EMPTY_STREAM_ITEMS; const pendingPermissionList = useStoreWithEqualityFn( @@ -1340,7 +1347,7 @@ const AgentStreamSection = memo(function AgentStreamSection({ isAuthoritativeHistoryReady={hasAppliedAuthoritativeHistory} toast={toast} pendingMessageSubmissions={pendingMessageSubmissions} - hasActiveTurn={hasActiveTurn} + turnPresentation={turnPresentation} onOpenWorkspaceFile={onOpenWorkspaceFile} /> ); diff --git a/packages/app/src/stores/session-store.test.ts b/packages/app/src/stores/session-store.test.ts index 0f7647057d..6816ba41af 100644 --- a/packages/app/src/stores/session-store.test.ts +++ b/packages/app/src/stores/session-store.test.ts @@ -7,7 +7,6 @@ import { normalizeWorkspaceDescriptor, selectAgentTimelineState, useSessionStore, - type Agent, type WorkspaceDescriptor, } from "./session-store"; import type { StreamItem } from "../types/stream"; @@ -42,37 +41,6 @@ function initializeTestSession(): void { useSessionStore.getState().initializeSession("test-server", null as unknown as DaemonClient); } -function createAgent(status: Agent["status"]): Agent { - const timestamp = new Date("2026-07-30T10:00:00.000Z"); - return { - serverId: "test-server", - id: "agent-1", - provider: "codex", - status, - createdAt: timestamp, - updatedAt: timestamp, - lastUserMessageAt: null, - lastActivityAt: timestamp, - capabilities: { - supportsStreaming: true, - supportsSessionPersistence: true, - supportsDynamicModes: true, - supportsMcpServers: true, - supportsReasoningStream: true, - supportsToolInvocations: true, - }, - currentModeId: null, - availableModes: [], - pendingPermissions: [], - persistence: null, - title: "Agent", - cwd: "/repo", - model: null, - parentAgentId: null, - labels: {}, - }; -} - function getTestSessionReferences() { const state = useSessionStore.getState(); const session = state.sessions["test-server"]; @@ -165,24 +133,26 @@ describe("agent timeline state", () => { }); }); - it("lets a later running directory transition raise but never lower an open turn", () => { + it("stores turn liveness transitions without duplicating their policy", () => { initializeTestSession(); const store = useSessionStore.getState(); - store.setAgents("test-server", new Map([["agent-1", createAgent("idle")]])); - store.applyAgentTimelineResponseState("test-server", "agent-1", { - items: [], - head: [], - range: null, - older: "none", - synchronized: true, - acknowledgedClientMessageIds: [], + store.applyAgentTurnLiveness("test-server", "agent-1", { + type: "directory_running", + startedAt: null, + }); + expect(store.getSession("test-server")?.agentTurnLiveness.get("agent-1")).toEqual({ + phase: "open", + startedAt: null, + evidence: null, }); - store.setAgents("test-server", new Map([["agent-1", createAgent("running")]])); - expect(store.getSession("test-server")?.activeAgentTurns.has("agent-1")).toBe(true); - - store.setAgents("test-server", new Map([["agent-1", createAgent("idle")]])); - expect(store.getSession("test-server")?.activeAgentTurns.has("agent-1")).toBe(true); + store.applyAgentTurnLiveness("test-server", "agent-1", { + type: "resume_snapshot", + status: "idle", + startedAt: null, + coverage: { epoch: "epoch-1", seq: 40 }, + }); + expect(store.getSession("test-server")?.agentTurnLiveness.has("agent-1")).toBe(false); }); }); diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index e55908b33e..c37c298eea 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -52,6 +52,11 @@ import { buildWorkspaceAgentActivityIndex, type WorkspaceAgentActivity, } from "@/utils/workspace-agent-activity"; +import { + applyTurnLivenessTransition, + type TurnLiveness, + type TurnLivenessTransition, +} from "@/timeline/turn-liveness"; // Re-export types that were in session-context export type MessageEntry = @@ -404,7 +409,7 @@ export interface SessionState { // Stream state (head/tail model) agentStreamTail: Map; agentStreamHead: Map; - activeAgentTurns: Set; + agentTurnLiveness: Map; messageSubmissions: Map; agentTimelineCursor: Map; agentTimelineHasOlder: Map; @@ -500,10 +505,15 @@ interface SessionStoreActions { state: { tail?: StreamItem[]; head?: StreamItem[]; - turnActive?: boolean; acknowledgedClientMessageIds?: readonly string[]; }, ) => void; + applyAgentTurnLiveness: ( + serverId: string, + agentId: string, + transition: TurnLivenessTransition, + ) => void; + clearAgentTurnLiveness: (serverId: string) => void; beginAgentMessageSubmission: ( serverId: string, agentId: string, @@ -638,18 +648,6 @@ type SessionStore = SessionStoreState & SessionStoreActions; const agentLastActivityCoalescer = createAgentLastActivityCoalescer(); -function updateActiveAgentTurn( - activeAgentTurns: ReadonlySet, - agentId: string, - active: boolean, -): Set { - if (activeAgentTurns.has(agentId) === active) return activeAgentTurns as Set; - const next = new Set(activeAgentTurns); - if (active) next.add(agentId); - else next.delete(agentId); - return next; -} - // Helper to create initial session state function createInitialSessionState( serverId: string, @@ -671,7 +669,7 @@ function createInitialSessionState( currentAssistantMessage: "", agentStreamTail: new Map(), agentStreamHead: new Map(), - activeAgentTurns: new Set(), + agentTurnLiveness: new Map(), messageSubmissions: new Map(), agentTimelineCursor: new Map(), agentTimelineHasOlder: new Map(), @@ -1094,7 +1092,6 @@ export const useSessionStore = create()( let nextHead = session.agentStreamHead; let changedTail = false; let changedHead = false; - let activeAgentTurns = session.activeAgentTurns; if (state.tail !== undefined) { const existingTail = session.agentStreamTail.get(agentId); @@ -1121,10 +1118,6 @@ export const useSessionStore = create()( } } - if (state.turnActive !== undefined) { - activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, state.turnActive); - } - const currentSubmissions = session.messageSubmissions.get(agentId) ?? []; const observedSubmissions = observeMessageSubmissionCanonical( currentSubmissions, @@ -1132,12 +1125,7 @@ export const useSessionStore = create()( ); const changedSubmissions = observedSubmissions !== currentSubmissions; - if ( - !changedTail && - !changedHead && - !changedSubmissions && - activeAgentTurns === session.activeAgentTurns - ) { + if (!changedTail && !changedHead && !changedSubmissions) { return prev; } @@ -1159,7 +1147,6 @@ export const useSessionStore = create()( ...session, agentStreamTail: nextTail, agentStreamHead: nextHead, - activeAgentTurns, messageSubmissions, }, }, @@ -1167,6 +1154,40 @@ export const useSessionStore = create()( }); }, + applyAgentTurnLiveness: (serverId, agentId, transition) => { + set((prev) => { + const session = prev.sessions[serverId]; + if (!session) return prev; + const agentTurnLiveness = applyTurnLivenessTransition( + session.agentTurnLiveness, + agentId, + transition, + ); + if (agentTurnLiveness === session.agentTurnLiveness) return prev; + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { ...session, agentTurnLiveness }, + }, + }; + }); + }, + + clearAgentTurnLiveness: (serverId) => { + set((prev) => { + const session = prev.sessions[serverId]; + if (!session || session.agentTurnLiveness.size === 0) return prev; + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { ...session, agentTurnLiveness: new Map() }, + }, + }; + }); + }, + beginAgentMessageSubmission: (serverId, agentId, message) => { set((prev) => { const session = prev.sessions[serverId]; @@ -1591,14 +1612,7 @@ export const useSessionStore = create()( return prev; } const nextAgents = typeof agents === "function" ? agents(session.agents) : agents; - let activeAgentTurns = session.activeAgentTurns; - for (const [agentId, agent] of nextAgents) { - const previousAgent = session.agents.get(agentId); - if (previousAgent?.status !== "running" && agent.status === "running") { - activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, true); - } - } - if (session.agents === nextAgents && activeAgentTurns === session.activeAgentTurns) { + if (session.agents === nextAgents) { return prev; } return { @@ -1608,7 +1622,6 @@ export const useSessionStore = create()( [serverId]: { ...session, agents: nextAgents, - activeAgentTurns, workspaceAgentActivity: buildWorkspaceAgentActivityIndex( nextAgents, session.workspaceAgentActivity, diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index f8744a6ee6..d192280e56 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -8,7 +8,7 @@ import { } from "@/types/stream"; import { createAgentStreamReducerQueue, - deriveAgentStreamTurnActivity, + deriveAgentStreamTurnLiveness, processTimelineResponse, processAgentStreamEvent, processAgentStreamEvents, @@ -131,40 +131,60 @@ function makeSubmittedUserMessage( }); } -describe("deriveAgentStreamTurnActivity", () => { +describe("deriveAgentStreamTurnLiveness", () => { it("tracks start through every terminal turn event", () => { const started = makeStreamReducerEvent( { type: "turn_started", provider: "claude" } as AgentStreamEventPayload, 1, ); - expect(deriveAgentStreamTurnActivity([started])).toBe(true); + expect(deriveAgentStreamTurnLiveness([started])).toEqual({ + type: "stream_open", + startedAt: started.timestamp, + evidence: { epoch: "epoch-1", seq: 1 }, + }); for (const event of [ { type: "turn_completed", provider: "claude" }, { type: "turn_failed", provider: "claude", error: "failed" }, { type: "turn_canceled", provider: "claude", reason: "canceled" }, ] as AgentStreamEventPayload[]) { - expect(deriveAgentStreamTurnActivity([started, makeStreamReducerEvent(event, 2)])).toBe( - false, - ); + expect(deriveAgentStreamTurnLiveness([started, makeStreamReducerEvent(event, 2)])).toEqual({ + type: "stream_close", + }); } }); it("keeps an open turn without stream lifecycle and closes it on an ordered terminal event", () => { - let hasOpenTurn = true; - hasOpenTurn = - deriveAgentStreamTurnActivity([ + expect( + deriveAgentStreamTurnLiveness([ makeStreamReducerEvent(makeAssistantTimelineEvent("still working"), 1), - ]) ?? hasOpenTurn; - expect(hasOpenTurn).toBe(true); + ]), + ).toBeUndefined(); - hasOpenTurn = - deriveAgentStreamTurnActivity([ + expect( + deriveAgentStreamTurnLiveness([ makeStreamReducerEvent( { type: "turn_completed", provider: "claude" } as AgentStreamEventPayload, 2, ), - ]) ?? hasOpenTurn; - expect(hasOpenTurn).toBe(false); + ]), + ).toEqual({ type: "stream_close" }); + }); + + it("restarts timing when a terminal event and the next start share one reducer batch", () => { + const completed = makeStreamReducerEvent( + { type: "turn_completed", provider: "claude" } as AgentStreamEventPayload, + 2, + ); + const restarted = makeStreamReducerEvent( + { type: "turn_started", provider: "claude" } as AgentStreamEventPayload, + 3, + ); + + expect(deriveAgentStreamTurnLiveness([completed, restarted])).toEqual({ + type: "stream_restart", + startedAt: restarted.timestamp, + evidence: { epoch: "epoch-1", seq: 3 }, + }); }); }); diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 2179ad637b..772c36b5d6 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -1,6 +1,7 @@ import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages"; import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store"; import type { AssistantMessageItem, StreamItem } from "@/types/stream"; +import type { TurnLivenessTransition } from "@/timeline/turn-liveness"; import { applyStreamEvent, flushHeadToTail, @@ -1385,30 +1386,47 @@ export function createAgentStreamReducerQueue( interface StreamStatePatch { tail?: StreamItem[]; head?: StreamItem[]; - turnActive?: boolean; acknowledgedClientMessageIds?: readonly string[]; } -export function deriveAgentStreamTurnActivity( +export function deriveAgentStreamTurnLiveness( events: readonly AgentStreamReducerEvent[], -): boolean | undefined { - let active: boolean | undefined; - for (const { event } of events) { - if (event.type === "turn_started") active = true; - else if ( +): TurnLivenessTransition | undefined { + let transition: TurnLivenessTransition | undefined; + let shouldRestart = false; + let restartStartedAt: Date | null = null; + for (const { event, timestamp, seq, epoch } of events) { + if (event.type === "turn_started") { + if (shouldRestart) restartStartedAt ??= timestamp; + transition = + typeof seq === "number" && typeof epoch === "string" + ? { + type: restartStartedAt ? "stream_restart" : "stream_open", + startedAt: restartStartedAt ?? timestamp, + evidence: { epoch, seq }, + } + : { type: "directory_running", startedAt: timestamp }; + } else if ( event.type === "turn_completed" || event.type === "turn_failed" || event.type === "turn_canceled" ) { - active = false; + transition = { type: "stream_close" }; + shouldRestart = true; + restartStartedAt = null; } } - return active; + return transition; } export interface CreateSessionAgentStreamReducerQueueInput { serverId: string; setAgentStreamState: (serverId: string, agentId: string, state: StreamStatePatch) => void; + applyAgentTurnLiveness: ( + serverId: string, + agentId: string, + transition: TurnLivenessTransition, + ) => void; setAgentTimelineCursor: ( serverId: string, state: (prev: Map) => Map, @@ -1427,7 +1445,13 @@ function cancelAgentStreamReducerFlush(id: number) { export function createSessionAgentStreamReducerQueue( input: CreateSessionAgentStreamReducerQueueInput, ): AgentStreamReducerQueue { - const { serverId, setAgentStreamState, setAgentTimelineCursor, recoverTimelineGap } = input; + const { + serverId, + setAgentStreamState, + applyAgentTurnLiveness, + setAgentTimelineCursor, + recoverTimelineGap, + } = input; return createAgentStreamReducerQueue({ getSnapshot: (agentId) => { @@ -1441,22 +1465,23 @@ export function createSessionAgentStreamReducerQueue( }; }, commit: (agentId, result, events) => { - const turnActive = deriveAgentStreamTurnActivity(events); + const turnLiveness = deriveAgentStreamTurnLiveness(events); if ( result.changedTail || result.changedHead || - turnActive !== undefined || result.acknowledgedClientMessageIds.length > 0 ) { setAgentStreamState(serverId, agentId, { ...(result.changedTail ? { tail: result.tail } : {}), ...(result.changedHead ? { head: result.head } : {}), - ...(turnActive === undefined ? {} : { turnActive }), ...(result.acknowledgedClientMessageIds.length > 0 ? { acknowledgedClientMessageIds: result.acknowledgedClientMessageIds } : {}), }); } + if (turnLiveness) { + applyAgentTurnLiveness(serverId, agentId, turnLiveness); + } if (result.cursorChanged && result.cursor) { const nextCursor = result.cursor; diff --git a/packages/app/src/timeline/timeline-sync-plan.test.ts b/packages/app/src/timeline/timeline-sync-plan.test.ts index aa038c4240..5d75754e68 100644 --- a/packages/app/src/timeline/timeline-sync-plan.test.ts +++ b/packages/app/src/timeline/timeline-sync-plan.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "vitest"; import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; import { isTimelineCatchUpComplete, + isTimelineResumeSnapshotAuthoritative, planTimelineOlderFetch, planTimelineTailFetch, } from "./timeline-sync-plan"; @@ -33,4 +34,28 @@ describe("timeline sync planning", () => { true, ); }); + + test("only tail and completed forward pages carry an authoritative resume snapshot", () => { + expect( + isTimelineResumeSnapshotAuthoritative({ + direction: "tail", + hasNewer: false, + error: null, + }), + ).toBe(true); + expect( + isTimelineResumeSnapshotAuthoritative({ + direction: "after", + hasNewer: false, + error: null, + }), + ).toBe(true); + expect( + isTimelineResumeSnapshotAuthoritative({ + direction: "before", + hasNewer: false, + error: null, + }), + ).toBe(false); + }); }); diff --git a/packages/app/src/timeline/timeline-sync-plan.ts b/packages/app/src/timeline/timeline-sync-plan.ts index 69b590b985..95b7e859bf 100644 --- a/packages/app/src/timeline/timeline-sync-plan.ts +++ b/packages/app/src/timeline/timeline-sync-plan.ts @@ -71,3 +71,12 @@ export function isTimelineCatchUpComplete(input: { return input.direction !== "after" || !input.hasNewer; } + +export function isTimelineResumeSnapshotAuthoritative(input: { + direction: "tail" | "before" | "after"; + hasNewer: boolean; + error: string | null; +}): boolean { + if (input.error || input.direction === "before") return false; + return input.direction === "tail" || !input.hasNewer; +} diff --git a/packages/app/src/timeline/turn-liveness.test.ts b/packages/app/src/timeline/turn-liveness.test.ts new file mode 100644 index 0000000000..2bb4aee47c --- /dev/null +++ b/packages/app/src/timeline/turn-liveness.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it } from "vitest"; +import { reduceTurnLiveness, resolveTurnPresentation, TURN_LIVENESS_IDLE } from "./turn-liveness"; + +describe("reduceTurnLiveness", () => { + it("closes a directory-seeded turn when a completed resume snapshot is idle", () => { + const startedAt = new Date("2026-07-31T10:00:00.000Z"); + const opened = reduceTurnLiveness(TURN_LIVENESS_IDLE, { + type: "directory_running", + startedAt, + }); + + const resumed = reduceTurnLiveness(opened, { + type: "resume_snapshot", + status: "idle", + startedAt, + coverage: { epoch: "epoch-1", seq: 40 }, + }); + + expect(resumed).toEqual(TURN_LIVENESS_IDLE); + }); + + it("does not let an older resume snapshot close a newer live turn", () => { + const startedAt = new Date("2026-07-31T10:01:00.000Z"); + const liveTurn = reduceTurnLiveness(TURN_LIVENESS_IDLE, { + type: "stream_open", + startedAt, + evidence: { epoch: "epoch-1", seq: 41 }, + }); + + const resumed = reduceTurnLiveness(liveTurn, { + type: "resume_snapshot", + status: "idle", + startedAt: null, + coverage: { epoch: "epoch-1", seq: 40 }, + }); + + expect(resumed).toEqual({ + phase: "open", + startedAt, + evidence: { epoch: "epoch-1", seq: 41 }, + }); + }); + + it("uses submission evidence when the rendered tail does not contain the prompt", () => { + const submittedAt = new Date("2026-07-31T10:02:00.000Z"); + + expect(resolveTurnPresentation(TURN_LIVENESS_IDLE, [{ submittedAt }])).toEqual({ + isActive: true, + startedAt: submittedAt, + }); + }); + + it("fills a missing directory start time from a running resume snapshot", () => { + const snapshotStartedAt = new Date("2026-07-31T10:03:00.000Z"); + const opened = reduceTurnLiveness(TURN_LIVENESS_IDLE, { + type: "directory_running", + startedAt: null, + }); + + expect( + reduceTurnLiveness(opened, { + type: "resume_snapshot", + status: "running", + startedAt: snapshotStartedAt, + coverage: { epoch: "epoch-1", seq: 40 }, + }), + ).toEqual({ + phase: "open", + startedAt: snapshotStartedAt, + evidence: { epoch: "epoch-1", seq: 40 }, + }); + }); + + it("does not downgrade newer live evidence from an older running snapshot", () => { + const startedAt = new Date("2026-07-31T10:04:00.000Z"); + const liveTurn = reduceTurnLiveness(TURN_LIVENESS_IDLE, { + type: "stream_open", + startedAt, + evidence: { epoch: "epoch-1", seq: 41 }, + }); + + expect( + reduceTurnLiveness(liveTurn, { + type: "resume_snapshot", + status: "running", + startedAt, + coverage: { epoch: "epoch-1", seq: 40 }, + }), + ).toBe(liveTurn); + }); + + it("starts a new clock when consecutive turns settle in one stream batch", () => { + const firstStartedAt = new Date("2026-07-31T10:05:00.000Z"); + const secondStartedAt = new Date("2026-07-31T10:06:00.000Z"); + const firstTurn = reduceTurnLiveness(TURN_LIVENESS_IDLE, { + type: "stream_open", + startedAt: firstStartedAt, + evidence: { epoch: "epoch-1", seq: 41 }, + }); + + expect( + reduceTurnLiveness(firstTurn, { + type: "stream_restart", + startedAt: secondStartedAt, + evidence: { epoch: "epoch-1", seq: 43 }, + }), + ).toEqual({ + phase: "open", + startedAt: secondStartedAt, + evidence: { epoch: "epoch-1", seq: 43 }, + }); + }); +}); diff --git a/packages/app/src/timeline/turn-liveness.ts b/packages/app/src/timeline/turn-liveness.ts new file mode 100644 index 0000000000..8fb1835af2 --- /dev/null +++ b/packages/app/src/timeline/turn-liveness.ts @@ -0,0 +1,138 @@ +import type { AgentLifecycleStatus } from "@getpaseo/protocol/agent-lifecycle"; + +export interface TurnLivenessEvidence { + epoch: string; + seq: number; +} + +export type TurnLiveness = + | { phase: "idle" } + | { + phase: "open"; + startedAt: Date | null; + evidence: TurnLivenessEvidence | null; + }; + +export type TurnLivenessTransition = + | { type: "directory_running"; startedAt: Date | null } + | { + type: "stream_open"; + startedAt: Date; + evidence: TurnLivenessEvidence; + } + | { + type: "stream_restart"; + startedAt: Date; + evidence: TurnLivenessEvidence; + } + | { type: "stream_close" } + | { type: "destructive_close" } + | { + type: "resume_snapshot"; + status: AgentLifecycleStatus; + startedAt: Date | null; + coverage: TurnLivenessEvidence; + }; + +export const TURN_LIVENESS_IDLE: TurnLiveness = { phase: "idle" }; + +export function readTurnLiveness( + turns: ReadonlyMap, + agentId: string, +): TurnLiveness { + return turns.get(agentId) ?? TURN_LIVENESS_IDLE; +} + +export function applyTurnLivenessTransition( + turns: ReadonlyMap, + agentId: string, + transition: TurnLivenessTransition, +): Map { + const current = readTurnLiveness(turns, agentId); + const next = reduceTurnLiveness(current, transition); + if (next === current) return turns as Map; + const updated = new Map(turns); + if (next.phase === "idle") updated.delete(agentId); + else updated.set(agentId, next); + return updated; +} + +export interface TurnPresentation { + isActive: boolean; + startedAt: Date | null; +} + +export function resolveTurnPresentation( + liveness: TurnLiveness, + submissions: readonly { submittedAt: Date }[], +): TurnPresentation { + const latestSubmission = submissions.at(-1); + if (liveness.phase === "open") { + return { + isActive: true, + startedAt: liveness.startedAt ?? latestSubmission?.submittedAt ?? null, + }; + } + return latestSubmission + ? { isActive: true, startedAt: latestSubmission.submittedAt } + : { isActive: false, startedAt: null }; +} + +export function reduceTurnLiveness( + current: TurnLiveness, + transition: TurnLivenessTransition, +): TurnLiveness { + if (transition.type === "directory_running") { + return current.phase === "open" + ? current + : { + phase: "open", + startedAt: transition.startedAt, + evidence: null, + }; + } + + if (transition.type === "stream_open") { + return { + phase: "open", + startedAt: + current.phase === "open" + ? (current.startedAt ?? transition.startedAt) + : transition.startedAt, + evidence: transition.evidence, + }; + } + + if (transition.type === "stream_restart") { + return { + phase: "open", + startedAt: transition.startedAt, + evidence: transition.evidence, + }; + } + + if (transition.type === "stream_close" || transition.type === "destructive_close") { + return TURN_LIVENESS_IDLE; + } + + if ( + current.phase === "open" && + current.evidence?.epoch === transition.coverage.epoch && + current.evidence.seq > transition.coverage.seq + ) { + return current; + } + + if (transition.status === "running") { + return { + phase: "open", + startedAt: + current.phase === "open" + ? (current.startedAt ?? transition.startedAt) + : transition.startedAt, + evidence: transition.coverage, + }; + } + + return TURN_LIVENESS_IDLE; +} diff --git a/packages/app/src/timeline/turn-time.test.ts b/packages/app/src/timeline/turn-time.test.ts index 47387ed42e..b5c13ee00c 100644 --- a/packages/app/src/timeline/turn-time.test.ts +++ b/packages/app/src/timeline/turn-time.test.ts @@ -27,6 +27,7 @@ describe("deriveStreamTurnTiming", () => { const timing = deriveStreamTurnTiming({ isTurnActive: true, + activeTurnStartedAt: submittedAt, tail: [], head: [user("submitted", submittedAt)], }); @@ -40,6 +41,7 @@ describe("deriveStreamTurnTiming", () => { const timing = deriveStreamTurnTiming({ isTurnActive: true, + activeTurnStartedAt: secondUserAt, tail: [ user("u1", firstUserAt), assistant("a1", new Date("2026-05-15T00:00:05.000Z")), @@ -58,6 +60,7 @@ describe("deriveStreamTurnTiming", () => { const timing = deriveStreamTurnTiming({ isTurnActive: false, + activeTurnStartedAt: null, tail: [ user("u1", userAt), assistant("a1", assistantAt), @@ -80,6 +83,7 @@ describe("deriveStreamTurnTiming", () => { const timing = deriveStreamTurnTiming({ isTurnActive: false, + activeTurnStartedAt: null, tail: [ user("u1", userAt), assistant("a1", firstAssistantAt), diff --git a/packages/app/src/timeline/turn-time.ts b/packages/app/src/timeline/turn-time.ts index 0c0af024c2..3cb8d18241 100644 --- a/packages/app/src/timeline/turn-time.ts +++ b/packages/app/src/timeline/turn-time.ts @@ -13,6 +13,7 @@ export interface StreamTurnTiming { export function deriveStreamTurnTiming(params: { isTurnActive: boolean; + activeTurnStartedAt: Date | null; tail: StreamItem[]; head: StreamItem[]; }): StreamTurnTiming { @@ -59,7 +60,7 @@ export function deriveStreamTurnTiming(params: { visitItem(item); } - const runningStartedAt = params.isTurnActive ? currentUserAt : null; + const runningStartedAt = params.isTurnActive ? params.activeTurnStartedAt : null; if (!params.isTurnActive) { flushCompletedTurn(); } diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 5330f8aa2a..243b88a2d9 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -84,11 +84,13 @@ function applyAgentStatus(input: { agentId: string; status: AgentSnapshotPayload["status"]; updatedAt: string; + lastUserMessageAt?: string; }): void { const agent = createAgentPayload({ id: input.agentId, status: input.status, updatedAt: input.updatedAt, + lastUserMessageAt: input.lastUserMessageAt ?? null, }); applyAgentDirectoryDelta({ serverId: input.serverId, @@ -96,6 +98,38 @@ function applyAgentStatus(input: { }); } +describe("turn liveness authority", () => { + it("seeds from a running directory transition without closing from directory idle", () => { + const serverId = "server-turn-liveness"; + const agentId = "agent-1"; + const startedAt = "2026-07-27T10:00:01.000Z"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + applyAgentStatus({ serverId, agentId, status: "idle", updatedAt: startedAt }); + + applyAgentStatus({ + serverId, + agentId, + status: "running", + updatedAt: "2026-07-27T10:00:02.000Z", + lastUserMessageAt: startedAt, + }); + applyAgentStatus({ + serverId, + agentId, + status: "idle", + updatedAt: "2026-07-27T10:00:03.000Z", + }); + + expect(store.getSession(serverId)?.agentTurnLiveness.get(agentId)).toEqual({ + phase: "open", + startedAt: new Date(startedAt), + evidence: null, + }); + store.clearSession(serverId); + }); +}); + describe("message submission authority", () => { it("does not settle a submission from an unrelated running transition", () => { const serverId = "server-running-is-not-submission-ack"; diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index a8acbefd4d..1ee4e4cff5 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -65,17 +65,32 @@ function upsertAgentDirectoryReplica( export function upsertAgentReplica(serverId: string, agent: Agent): Agent { let acceptedAgent = agent; + let previousAgent: Agent | undefined; useSessionStore.getState().setAgents(serverId, (current) => { const currentAgent = current.get(agent.id); + previousAgent = currentAgent; acceptedAgent = acceptAgentDirectoryUpdate(currentAgent, agent); if (acceptedAgent === currentAgent) return current; const next = new Map(current); next.set(agent.id, acceptedAgent); return next; }); + seedAgentTurnLivenessFromDirectory(serverId, previousAgent, acceptedAgent); return acceptedAgent; } +function seedAgentTurnLivenessFromDirectory( + serverId: string, + previousAgent: Agent | undefined, + agent: Agent, +): void { + if (previousAgent?.status === "running" || agent.status !== "running") return; + useSessionStore.getState().applyAgentTurnLiveness(serverId, agent.id, { + type: "directory_running", + startedAt: agent.lastUserMessageAt, + }); +} + export function replaceAgentPendingPermissions(serverId: string, agent: Agent): void { const pendingPermissions = new Map( useSessionStore.getState().sessions[serverId]?.pendingPermissions, @@ -114,7 +129,7 @@ export function removeAgentDirectoryReplica(serverId: string, agentId: string): store.setAgentAuthoritativeHistoryApplied(serverId, agentId, false); store.setAgentStreamTail(serverId, removeKey); store.clearAgentStreamHead(serverId, agentId); - store.setAgentStreamState(serverId, agentId, { turnActive: false }); + store.applyAgentTurnLiveness(serverId, agentId, { type: "destructive_close" }); useSessionStore.setState((state) => { if (!state.agentLastActivity.has(agentId)) return state; const agentLastActivity = new Map(state.agentLastActivity); @@ -173,6 +188,7 @@ export function replaceFetchedAgentDirectory(input: { }): { agents: Map } { const { agents: fetchedAgents, pendingPermissions } = buildAgentDirectoryState(input); const store = useSessionStore.getState(); + const previousAgents = store.sessions[input.serverId]?.agents ?? new Map(); for (const agent of fetchedAgents.values()) { if (agent.archivedAt) { @@ -181,6 +197,9 @@ export function replaceFetchedAgentDirectory(input: { } store.setAgents(input.serverId, fetchedAgents); + for (const agent of fetchedAgents.values()) { + seedAgentTurnLivenessFromDirectory(input.serverId, previousAgents.get(agent.id), agent); + } store.setAgentDetails(input.serverId, (prev) => { let next: Map | null = null; for (const agentId of fetchedAgents.keys()) {