From 279211e110ff2673fbf0ddd40ac42efa38e57f08 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 9 Nov 2026 08:28:52 +0100 Subject: [PATCH] fix(app): keep agent details and stay quiet after a current timeline answer The timeline fetch that a current answer skips also delivered the full agent row, and the replica cache stores only part of it. After a cold reload with an empty incremental directory page, usage, modes, permissions and runtime info stayed empty. The subscribe answer now carries the row for each current chat, gated with the rest of the freshness check. The panel also reported ready before the timeline owner was published, so a connected render with cached content showed Updating messages. No owner now reads as a check that has not begun. --- docs/timeline-sync.md | 7 +- packages/app/src/panels/agent-panel.tsx | 12 ++-- .../app/src/runtime/directory-sync/index.ts | 21 ++++++ packages/app/src/runtime/host-runtime.ts | 15 ++++- .../app/src/timeline/viewed-timeline-sync.ts | 7 +- packages/protocol/src/messages.ts | 3 + .../selective-timeline-delivery.e2e.test.ts | 10 ++- packages/server/src/server/session.ts | 66 +++++++++++++------ 8 files changed, 106 insertions(+), 35 deletions(-) diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 111f370d45..82d45f0242 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -110,9 +110,10 @@ A chat with a cached cursor asks the daemon first. The `agent.timeline.set_subsc that attaches the chat carries its `{ epoch, seq }`, and the response answers `current` or `behind` per chat. The daemon attaches before it compares, so an entry committed in between arrives live or counts as behind. `current` means the agent is loaded, the epoch matches, and the cursor is at the -head; it completes catch-up with no fetch. `behind`, or no entry, runs the bounded catch-up. The -subscribe handler never loads an agent, so an agent the daemon has not loaded answers `behind` and -resumes in that fetch. Hosts without `features.timelineFreshnessCheck` get no cursors and always fetch. +head; it completes catch-up with no fetch. The response carries the agent row for each `current` +chat, because the timeline fetch is also what refreshes that row and the replica cache stores only +part of it. `behind`, or no entry, runs the bounded catch-up. The subscribe handler never loads an +agent, so an agent the daemon has not loaded answers `behind` and resumes in that fetch. Hosts without `features.timelineFreshnessCheck` get no cursors and always fetch. Cached history remains readable during recovery. The chat shows Reconnecting to host while the host is offline and until a reconnected chat's check answers, so an unchanged chat never shows a second diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index a91f335048..ba13bb8a9f 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -726,13 +726,11 @@ function ChatAgentContent({ (listener: () => void) => viewedTimelineSync?.subscribe(listener) ?? (() => {}), [viewedTimelineSync], ); - const readTimelineStatus = useCallback( - () => - !agentId || !viewedTimelineSync - ? ("ready" as const) - : viewedTimelineSync.getAgentTimelineStatus(agentId), - [agentId, viewedTimelineSync], - ); + const readTimelineStatus = useCallback((): ViewedTimelineStatus => { + if (!agentId) return "ready"; + // Until the owner is published nothing has confirmed this chat, so its check has not begun. + return viewedTimelineSync?.getAgentTimelineStatus(agentId) ?? "checking"; + }, [agentId, viewedTimelineSync]); const timelineStatus = useSyncExternalStore( subscribeToVisibilityCatchUp, readTimelineStatus, diff --git a/packages/app/src/runtime/directory-sync/index.ts b/packages/app/src/runtime/directory-sync/index.ts index 1ffb4fae5f..0f539a2b1d 100644 --- a/packages/app/src/runtime/directory-sync/index.ts +++ b/packages/app/src/runtime/directory-sync/index.ts @@ -4,6 +4,7 @@ import type { FetchAgentsEntry, FetchAgentsOptions, } from "@getpaseo/client/internal/daemon-client"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once"; import { normalizeProjectDescriptor, @@ -482,6 +483,26 @@ export class DirectorySync { return page; } + /** + * Captures these chats' lifecycle before a subscribe. The returned function keeps the agent rows + * its answer carries, which is what a timeline fetch would have delivered for each of them. + */ + captureSubscribedAgents( + agentIds: string[], + ): (agents: Record) => void { + const tokens = new Map(agentIds.map((id) => [id, this.agents.captureTimeline(id)])); + return (agents) => { + let accepted = false; + for (const [agentId, agent] of Object.entries(agents)) { + const token = tokens.get(agentId); + if (token && this.agents.submitTimelineAgent(token, agent)) accepted = true; + } + if (!accepted) return; + this.revision += 1; + this.persistCheckpoint(); + }; + } + async refreshAgents( input: RefreshAgentDirectoryInput = {}, ): Promise { diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index d6fe75ab12..15a86b7bb4 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -2880,7 +2880,20 @@ export class HostRuntimeStore { drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(serverId, agentId), applyTurnLiveness: (agentId, transitions) => directory.applyAgentTurnLiveness(agentId, transitions), - ports, + ports: { + ...ports, + observe: (agentIds, cursors) => { + const keepAgents = directory.captureSubscribedAgents(agentIds); + const observation = ports.observe(agentIds, cursors); + return { + ready: observation.ready.then((answer) => { + if (answer?.agents) keepAgents(answer.agents); + return answer; + }), + release: () => observation.release(), + }; + }, + }, }); } diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 7c2d16e50f..eb79d87a69 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -12,7 +12,7 @@ import { type AgentTimelineState, } from "@/stores/session-store"; import { useCreateFlowStore } from "@/stores/create-flow-store"; -import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; +import type { AgentSnapshotPayload, SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { getSendingClientMessageIds } from "@/composer/submission/model"; import { getInitDeferred, @@ -381,7 +381,10 @@ export interface ViewedTimelineSyncPorts { agentIds: string[], cursors?: Record, ): { - ready: Promise<{ timelines?: Record } | void>; + ready: Promise<{ + timelines?: Record; + agents?: Record; + } | void>; release(): Promise; }; /** The daemon confirmed the cached timeline is at its head, as a zero-row after page would. */ diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 4cff2be935..72a3249fa7 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -5259,6 +5259,9 @@ export const SetAgentTimelineSubscriptionResponseMessageSchema = z.object({ requestId: z.string(), // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. timelines: z.record(z.string(), TimelineFreshnessSchema).optional(), + // The agent row for each chat answered "current", which gets no timeline fetch to carry it. + // COMPAT(timelineFreshnessCheck): added after v0.15.3, remove after 2027-05-09. + agents: z.record(z.string(), AgentSnapshotPayloadSchema).optional(), }), }); diff --git a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts index 2f602407ec..7b096256cc 100644 --- a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts +++ b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts @@ -262,11 +262,15 @@ test("subscribing with a cursor answers whether each chat is at its head", async }, ); - expect((await observation.ready).timelines).toEqual({ + const answer = await observation.ready; + expect(answer.timelines).toEqual({ [atHead.agentId]: "current", [staleSeq.agentId]: "behind", [otherEpoch.agentId]: "behind", }); + // Only a current chat skips the fetch that would have delivered its agent row. + expect(Object.keys(answer.agents ?? {})).toEqual([atHead.agentId]); + expect(answer.agents?.[atHead.agentId]).toMatchObject({ id: atHead.agentId, title: "At head" }); connected.clear(); await daemon.daemon.agentManager.appendTimelineItem(atHead.agentId, { @@ -292,7 +296,9 @@ test("subscribing with a cursor answers behind for a stored agent and leaves it cursors: { [stored.agentId]: stored.head }, }); - expect((await observation.ready).timelines).toEqual({ [stored.agentId]: "behind" }); + const answer = await observation.ready; + expect(answer.timelines).toEqual({ [stored.agentId]: "behind" }); + expect(answer.agents).toEqual({}); expect(daemon.daemon.agentManager.getAgent(stored.agentId)).toBeNull(); await observation.release(); }); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index e96b5e6320..13f8db7397 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1423,6 +1423,50 @@ export class Session { return timelines; } + private async answerTimelineFreshness( + agentIds: string[], + cursors: Record, + ): Promise<{ + timelines: Record; + agents: Record; + }> { + const snapshots = await Promise.all( + agentIds.flatMap((agentId) => { + const agent = cursors[agentId] ? this.agentManager.getAgent(agentId) : null; + return agent ? [this.buildAgentPayload(agent)] : []; + }), + ); + // Compared after the await, so a row appended while reading the snapshots answers behind. + const timelines = this.compareTimelineCursors(agentIds, cursors); + const agents: Record = {}; + for (const snapshot of snapshots) { + if (timelines[snapshot.id] === "current") agents[snapshot.id] = snapshot; + } + return { timelines, agents }; + } + + private async handleAgentTimelineSubscription( + msg: Extract, + source?: object, + ): Promise { + const agentIds = [...new Set(msg.agentIds)].sort(); + // Attach before comparing: a row appended after the comparison reaches this owner live. + const owner = this.subscribeAgentTimelines(agentIds); + const freshness = msg.cursors ? await this.answerTimelineFreshness(agentIds, msg.cursors) : {}; + this.emitForSource( + { + type: "agent.timeline.set_subscription.response", + payload: { + agentIds, + requestId: msg.requestId, + subscriptionId: owner.responseId, + ...freshness, + }, + }, + source, + ); + } + private subscribeAgentTimelines(agentIds: string[]): OwnedSubscription { const owner = this.delivery.begin("timelines", undefined, (id) => { this.timelineSubscriptions.delete(id); @@ -3058,26 +3102,8 @@ export class Session { throw error; }); } - case "agent.timeline.set_subscription.request": { - const agentIds = [...new Set(msg.agentIds)].sort(); - // Attach before comparing: a row appended after the comparison reaches this owner live. - const owner = this.subscribeAgentTimelines(agentIds); - this.emitForSource( - { - type: "agent.timeline.set_subscription.response", - payload: { - agentIds, - requestId: msg.requestId, - subscriptionId: owner.responseId, - ...(msg.cursors - ? { timelines: this.compareTimelineCursors(agentIds, msg.cursors) } - : {}), - }, - }, - source, - ); - return undefined; - } + case "agent.timeline.set_subscription.request": + return this.handleAgentTimelineSubscription(msg, source); case "agent.fork_context.request": return this.handleAgentForkContextRequest(msg); default: