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: