From 8169134ac3902c8cb504e7502f710364881f4adb Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 14 Feb 2026 01:03:36 +0700 Subject: [PATCH] refactor timeline loading to cursor-based fetch API --- packages/app/src/contexts/session-context.tsx | 267 ++++++++------- .../app/src/hooks/use-agent-initialization.ts | 98 +++--- packages/app/src/stores/session-store.ts | 38 +++ packages/cli/src/commands/agent/attach.ts | 40 +-- packages/cli/src/commands/agent/logs.ts | 70 +--- packages/cli/src/utils/timeline.ts | 33 ++ .../server/src/client/daemon-client.test.ts | 133 +++++--- packages/server/src/client/daemon-client.ts | 46 ++- .../src/server/agent/agent-manager.test.ts | 114 +++++++ .../server/src/server/agent/agent-manager.ts | 317 +++++++++++++++++- .../providers/codex-app-server-agent.test.ts | 120 +++++++ .../agent/providers/codex-app-server-agent.ts | 31 ++ .../server/agent/timeline-projection.test.ts | 118 +++++++ .../src/server/agent/timeline-projection.ts | 209 ++++++++++++ .../src/server/daemon-client.e2e.test.ts | 28 +- .../daemon-e2e/agent-operations.e2e.test.ts | 24 +- .../server/daemon-e2e/persistence.e2e.test.ts | 25 +- packages/server/src/server/exports.ts | 1 - packages/server/src/server/session.ts | 181 +++++----- ...gacy-agent-stream-snapshot-inProgress.json | 24 -- .../shared/messages.stream-parsing.test.ts | 58 +++- packages/server/src/shared/messages.ts | 96 ++++-- 22 files changed, 1567 insertions(+), 504 deletions(-) create mode 100644 packages/cli/src/utils/timeline.ts create mode 100644 packages/server/src/server/agent/timeline-projection.test.ts create mode 100644 packages/server/src/server/agent/timeline-projection.ts delete mode 100644 packages/server/src/shared/__fixtures__/legacy-agent-stream-snapshot-inProgress.json diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 9b39cb8a0..b85dfcc85 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -9,6 +9,7 @@ import { applyStreamEvent, generateMessageId, hydrateStreamState, + reduceStreamUpdate, type StreamItem, } from "@/types/stream"; import type { @@ -303,6 +304,9 @@ export function SessionProvider({ const clearAgentStreamHead = useSessionStore( (state) => state.clearAgentStreamHead ); + const setAgentTimelineCursor = useSessionStore( + (state) => state.setAgentTimelineCursor + ); const setInitializingAgents = useSessionStore( (state) => state.setInitializingAgents ); @@ -371,7 +375,7 @@ export function SessionProvider({ ) => Promise) | null >(null); - const hasRequestedInitialSnapshotRef = useRef(false); + const hasBootstrappedAgentUpdatesRef = useRef(false); const agentUpdatesSubscriptionIdRef = useRef(null); const sessionStateTimeoutRef = useRef | null>( null @@ -580,6 +584,15 @@ export function SessionProvider({ return next; }); + setAgentTimelineCursor(serverId, (prev) => { + if (!prev.has(agentId)) { + return prev; + } + const next = new Map(prev); + next.delete(agentId); + return next; + }); + return; } @@ -644,6 +657,7 @@ export function SessionProvider({ setAgentLastActivity, setPendingPermissions, setQueuedMessages, + setAgentTimelineCursor, ] ); @@ -665,18 +679,6 @@ export function SessionProvider({ [serverId, setFileExplorer] ); - const refreshAgentMutation = useMutation({ - mutationFn: async ({ agentId }: { agentId: string }) => { - if (!agentId) { - throw new Error("Agent id is required"); - } - if (!client) { - throw new Error("Daemon client unavailable"); - } - return await client.refreshAgent(agentId); - }, - }); - const directoryListingMutation = useMutation({ mutationFn: async ({ agentId, path }: { agentId: string; path: string }) => { if (!agentId) { @@ -740,9 +742,109 @@ export function SessionProvider({ }, }); + const applyTimelineResponse = useCallback( + ( + payload: Extract< + SessionOutboundMessage, + { type: "fetch_agent_timeline_response" } + >["payload"] + ) => { + const agentId = payload.agentId; + const initKey = getInitKey(serverId, agentId); + + if (payload.error) { + setInitializingAgents(serverId, (prev) => { + if (prev.get(agentId) !== true) { + return prev; + } + const next = new Map(prev); + next.set(agentId, false); + return next; + }); + rejectInitDeferred(initKey, new Error(payload.error)); + return; + } + + const hydratedEvents: Array<{ + event: AgentStreamEventPayload; + timestamp: Date; + }> = payload.entries.map((entry) => ({ + event: { + type: "timeline", + provider: entry.provider, + item: entry.item, + }, + timestamp: new Date(entry.timestamp), + })); + + const replace = payload.reset || payload.direction !== "after"; + if (replace) { + const hydrated = hydrateStreamState(hydratedEvents); + setAgentStreamTail(serverId, (prev) => { + const next = new Map(prev); + next.set(agentId, hydrated); + return next; + }); + clearAgentStreamHead(serverId, agentId); + } else if (hydratedEvents.length > 0) { + setAgentStreamTail(serverId, (prev) => { + const next = new Map(prev); + const current = next.get(agentId) ?? []; + const updated = hydratedEvents.reduce( + (state, { event, timestamp }) => reduceStreamUpdate(state, event, timestamp), + current + ); + next.set(agentId, updated); + return next; + }); + } + + setAgentTimelineCursor(serverId, (prev) => { + const next = new Map(prev); + if (payload.startCursor && payload.endCursor) { + next.set(agentId, { + epoch: payload.epoch, + startSeq: payload.startCursor.seq, + endSeq: payload.endCursor.seq, + }); + } else if (payload.reset) { + next.delete(agentId); + } + return next; + }); + + const deferredUpdate = pendingAgentUpdatesRef.current.get(agentId); + pendingAgentUpdatesRef.current.delete(agentId); + if (deferredUpdate) { + applyAgentUpdatePayload(deferredUpdate); + } + + setInitializingAgents(serverId, (prev) => { + if (prev.get(agentId) !== true) { + return prev; + } + const next = new Map(prev); + next.set(agentId, false); + return next; + }); + + resolveInitDeferred(initKey); + markAgentHistorySynchronized(serverId, agentId); + }, + [ + applyAgentUpdatePayload, + clearAgentStreamHead, + markAgentHistorySynchronized, + serverId, + setAgentStreamTail, + setAgentTimelineCursor, + setInitializingAgents, + ] + ); + useEffect(() => { if (!connectionSnapshot.isConnected) { - hasRequestedInitialSnapshotRef.current = false; + hasBootstrappedAgentUpdatesRef.current = false; const subscriptionId = agentUpdatesSubscriptionIdRef.current; if (subscriptionId && client) { try { @@ -754,10 +856,10 @@ export function SessionProvider({ agentUpdatesSubscriptionIdRef.current = null; return; } - if (hasRequestedInitialSnapshotRef.current) { + if (hasBootstrappedAgentUpdatesRef.current) { return; } - hasRequestedInitialSnapshotRef.current = true; + hasBootstrappedAgentUpdatesRef.current = true; try { if (!agentUpdatesSubscriptionIdRef.current) { @@ -803,7 +905,7 @@ export function SessionProvider({ const unsubAgentStream = client.on("agent_stream", (message) => { if (message.type !== "agent_stream") return; - const { agentId, event, timestamp } = message.payload; + const { agentId, event, timestamp, seq, epoch } = message.payload; const parsedTimestamp = new Date(timestamp); if (event.type === "attention_required") { @@ -846,51 +948,37 @@ export function SessionProvider({ }); } + if ( + event.type === "timeline" && + typeof seq === "number" && + typeof epoch === "string" + ) { + setAgentTimelineCursor(serverId, (prev) => { + const current = prev.get(agentId); + const next = new Map(prev); + if (!current || current.epoch !== epoch) { + next.set(agentId, { epoch, startSeq: seq, endSeq: seq }); + return next; + } + next.set(agentId, { + epoch, + startSeq: Math.min(current.startSeq, seq), + endSeq: Math.max(current.endSeq, seq), + }); + return next; + }); + } + // NOTE: We don't update lastActivityAt on every stream event to prevent // cascading rerenders. The agent_update handler updates agent.lastActivityAt // on status changes, which is sufficient for sorting and display purposes. }); - const unsubAgentStreamSnapshot = client.on( - "agent_stream_snapshot", + const unsubAgentTimeline = client.on( + "fetch_agent_timeline_response", (message) => { - if (message.type !== "agent_stream_snapshot") return; - const { agentId, events } = message.payload; - - const hydrated = hydrateStreamState( - events.map(({ event, timestamp }) => ({ - event: event as AgentStreamEventPayload, - timestamp: new Date(timestamp), - })) - ); - const initKey = getInitKey(serverId, agentId); - const hasInFlightHistorySync = Boolean(getInitDeferred(initKey)); - - setAgentStreamTail(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, hydrated); - return next; - }); - clearAgentStreamHead(serverId, agentId); - - const deferredUpdate = pendingAgentUpdatesRef.current.get(agentId); - pendingAgentUpdatesRef.current.delete(agentId); - if (hasInFlightHistorySync && deferredUpdate) { - applyAgentUpdatePayload(deferredUpdate); - } - - setInitializingAgents(serverId, (prev) => { - if (prev.get(agentId) !== true) { - return prev; - } - const next = new Map(prev); - next.set(agentId, false); - return next; - }); - - // Resolve the initialization promise (even for empty history) - resolveInitDeferred(initKey); - markAgentHistorySynchronized(serverId, agentId); + if (message.type !== "fetch_agent_timeline_response") return; + applyTimelineResponse(message.payload); } ); @@ -1233,6 +1321,14 @@ export function SessionProvider({ return next; }); clearAgentStreamHead(serverId, agentId); + setAgentTimelineCursor(serverId, (prev) => { + if (!prev.has(agentId)) { + return prev; + } + const next = new Map(prev); + next.delete(agentId); + return next; + }); // Remove draft input clearDraftInput(agentId); @@ -1292,7 +1388,7 @@ export function SessionProvider({ return () => { unsubAgentUpdate(); unsubAgentStream(); - unsubAgentStreamSnapshot(); + unsubAgentTimeline(); unsubStatus(); unsubPermissionRequest(); unsubPermissionResolved(); @@ -1313,6 +1409,7 @@ export function SessionProvider({ setAgentStreamTail, setAgentStreamHead, clearAgentStreamHead, + setAgentTimelineCursor, setInitializingAgents, setAgents, setAgentLastActivity, @@ -1323,63 +1420,9 @@ export function SessionProvider({ clearDraftInput, notifyAgentAttention, applyAgentUpdatePayload, - markAgentHistorySynchronized, + applyTimelineResponse, ]); - const initializeAgent = useCallback( - ({ agentId, requestId }: { agentId: string; requestId?: string }) => { - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, true); - return next; - }); - - if (!client) { - console.warn("[Session] initializeAgent skipped: daemon unavailable"); - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; - }); - return; - } - - client - .initializeAgent(agentId, requestId) - .catch((error) => { - console.warn("[Session] initializeAgent failed", { agentId, error }); - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; - }); - }); - }, - [serverId, client, setInitializingAgents] - ); - - const refreshAgent = useCallback( - ({ agentId }: { agentId: string }) => { - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, true); - return next; - }); - - refreshAgentMutation - .mutateAsync({ agentId }) - .catch((error) => { - console.warn("[Session] refreshAgent failed", { agentId, error }); - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; - }); - }); - }, - [serverId, refreshAgentMutation, setInitializingAgents] - ); - const sendAgentMessage = useCallback( async ( agentId: string, diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index 337db27a2..669b86f04 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -1,5 +1,6 @@ import { useCallback } from "react"; import { useSessionStore } from "@/stores/session-store"; +import type { FetchAgentTimelineOptions } from "@server/client/daemon-client"; import { attachInitTimeout, createInitDeferred, @@ -9,10 +10,49 @@ import { } from "@/utils/agent-initialization"; const INIT_TIMEOUT_MS = 5 * 60_000; +const DEFAULT_INITIAL_TIMELINE_LIMIT = 200; + +type TimelineCursorState = { + epoch: string; + endSeq: number; +}; + +function buildInitialTimelineRequest( + cursor: TimelineCursorState | undefined +): FetchAgentTimelineOptions { + if (!cursor) { + return { + direction: "tail", + limit: DEFAULT_INITIAL_TIMELINE_LIMIT, + projection: "projected", + }; + } + + return { + direction: "after", + cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, + // Catch up all missing canonical rows by default. + limit: 0, + projection: "projected", + }; +} export function useAgentInitialization(serverId: string) { const client = useSessionStore((state) => state.sessions[serverId]?.client ?? null); const setInitializingAgents = useSessionStore((state) => state.setInitializingAgents); + const setAgentInitializing = useCallback( + (agentId: string, initializing: boolean) => { + setInitializingAgents(serverId, (prev) => { + if (prev.get(agentId) === initializing) { + return prev; + } + const next = new Map(prev); + next.set(agentId, initializing); + return next; + }); + }, + [serverId, setInitializingAgents] + ); const ensureAgentIsInitialized = useCallback( (agentId: string): Promise => { @@ -24,14 +64,7 @@ export function useAgentInitialization(serverId: string) { const deferred = createInitDeferred(key); const timeoutId = setTimeout(() => { - setInitializingAgents(serverId, (prev) => { - if (prev.get(agentId) !== true) { - return prev; - } - const next = new Map(prev); - next.set(agentId, false); - return next; - }); + setAgentInitializing(agentId, false); rejectInitDeferred( key, new Error( @@ -41,34 +74,26 @@ export function useAgentInitialization(serverId: string) { }, INIT_TIMEOUT_MS); attachInitTimeout(key, timeoutId); - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, true); - return next; - }); + setAgentInitializing(agentId, true); if (!client) { - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; - }); + setAgentInitializing(agentId, false); rejectInitDeferred(key, new Error("Host is not connected")); return deferred.promise; } + const session = useSessionStore.getState().sessions[serverId]; + const cursor = session?.agentTimelineCursor.get(agentId); + const timelineRequest = buildInitialTimelineRequest(cursor); + client - .initializeAgent(agentId) + .fetchAgentTimeline(agentId, timelineRequest) .then(() => { - // No-op: the actual "timeline hydrated" signal is the `agent_stream_snapshot` - // message, handled in SessionContext. + // No-op: hydration completion is handled by SessionContext + // when it processes fetch_agent_timeline_response. }) .catch((error) => { - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; - }); + setAgentInitializing(agentId, false); rejectInitDeferred( key, error instanceof Error ? error : new Error(String(error)) @@ -77,7 +102,7 @@ export function useAgentInitialization(serverId: string) { return deferred.promise; }, - [client, serverId, setInitializingAgents] + [client, serverId, setAgentInitializing] ); const refreshAgent = useCallback( @@ -85,24 +110,21 @@ export function useAgentInitialization(serverId: string) { if (!client) { throw new Error("Host is not connected"); } - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, true); - return next; - }); + setAgentInitializing(agentId, true); try { await client.refreshAgent(agentId); - } catch (error) { - setInitializingAgents(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, false); - return next; + await client.fetchAgentTimeline(agentId, { + direction: "tail", + limit: DEFAULT_INITIAL_TIMELINE_LIMIT, + projection: "projected", }); + } catch (error) { + setAgentInitializing(agentId, false); throw error; } }, - [client, serverId, setInitializingAgents] + [client, setAgentInitializing] ); return { ensureAgentIsInitialized, refreshAgent }; diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 0956c22ef..214354ca3 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -157,6 +157,12 @@ export type DaemonServerInfo = { hostname: string | null; }; +export interface AgentTimelineCursorState { + epoch: string; + startSeq: number; + endSeq: number; +} + // Per-session state export interface SessionState { serverId: string; @@ -189,6 +195,7 @@ export interface SessionState { // Stream state (head/tail model) agentStreamTail: Map; agentStreamHead: Map; + agentTimelineCursor: Map; historySyncGeneration: number; agentHistorySyncGeneration: Map; @@ -240,6 +247,12 @@ interface SessionStoreActions { setAgentStreamTail: (serverId: string, state: Map | ((prev: Map) => Map)) => void; setAgentStreamHead: (serverId: string, state: Map | ((prev: Map) => Map)) => void; clearAgentStreamHead: (serverId: string, agentId: string) => void; + setAgentTimelineCursor: ( + serverId: string, + state: + | Map + | ((prev: Map) => Map) + ) => void; bumpHistorySyncGeneration: (serverId: string) => void; markAgentHistorySynchronized: (serverId: string, agentId: string) => void; @@ -321,6 +334,7 @@ function createInitialSessionState(serverId: string, client: DaemonClient, audio currentAssistantMessage: "", agentStreamTail: new Map(), agentStreamHead: new Map(), + agentTimelineCursor: new Map(), historySyncGeneration: 0, agentHistorySyncGeneration: new Map(), initializingAgents: new Map(), @@ -603,6 +617,30 @@ export const useSessionStore = create()( }); }, + setAgentTimelineCursor: (serverId, state) => { + set((prev) => { + const session = prev.sessions[serverId]; + if (!session) { + return prev; + } + const nextState = + typeof state === "function" ? state(session.agentTimelineCursor) : state; + if (session.agentTimelineCursor === nextState) { + return prev; + } + logSessionStoreUpdate("setAgentTimelineCursor", serverId, { + agentCount: nextState.size, + }); + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { ...session, agentTimelineCursor: nextState }, + }, + }; + }); + }, + bumpHistorySyncGeneration: (serverId) => { set((prev) => { const session = prev.sessions[serverId]; diff --git a/packages/cli/src/commands/agent/attach.ts b/packages/cli/src/commands/agent/attach.ts index abf5d5a38..02a8c465a 100644 --- a/packages/cli/src/commands/agent/attach.ts +++ b/packages/cli/src/commands/agent/attach.ts @@ -1,9 +1,9 @@ import type { Command } from 'commander' import { connectToDaemon, getDaemonHost } from '../../utils/client.js' +import { fetchProjectedTimelineItems } from '../../utils/timeline.js' import type { DaemonClient, AgentStreamMessage, - AgentStreamSnapshotMessage, AgentStreamEventPayload, AgentTimelineItem, } from '@getpaseo/server' @@ -131,37 +131,19 @@ export async function runAttachCommand( console.log(`Attaching to agent ${resolvedId.substring(0, 7)}...`) console.log(`(Press Ctrl+C to detach)\n`) - // Get existing output from snapshot - const snapshotPromise = new Promise((resolve) => { - const timeout = setTimeout(() => resolve(), 5000) - - const unsubscribe = client.on('agent_stream_snapshot', (msg: unknown) => { - const message = msg as AgentStreamSnapshotMessage - if (message.type !== 'agent_stream_snapshot') return - if (message.payload.agentId !== resolvedId) return - - clearTimeout(timeout) - unsubscribe() - - // Print recent events from snapshot - for (const e of message.payload.events) { - printStreamEvent(e.event) - } - - resolve() - }) - }) - - // Initialize agent to trigger snapshot + // Print existing output from timeline fetch. try { - await client.initializeAgent(resolvedId) - } catch { - // Agent might already be initialized, continue + const timelineItems = await fetchProjectedTimelineItems({ + client, + agentId: resolvedId, + }) + for (const item of timelineItems) { + printTimelineItem(item) + } + } catch (error) { + console.warn('Warning: failed to fetch existing timeline', error) } - // Wait for snapshot - await snapshotPromise - // Subscribe to new events const unsubscribe = client.on('agent_stream', (msg: unknown) => { const message = msg as AgentStreamMessage diff --git a/packages/cli/src/commands/agent/logs.ts b/packages/cli/src/commands/agent/logs.ts index 7f28f6252..8e50c748a 100644 --- a/packages/cli/src/commands/agent/logs.ts +++ b/packages/cli/src/commands/agent/logs.ts @@ -1,10 +1,10 @@ import type { Command } from 'commander' import { connectToDaemon, getDaemonHost } from '../../utils/client.js' import type { CommandOptions } from '../../output/index.js' +import { fetchProjectedTimelineItems } from '../../utils/timeline.js' import type { DaemonClient, AgentStreamMessage, - AgentStreamSnapshotMessage, AgentTimelineItem, } from '@getpaseo/server' import { curateAgentActivity } from '@getpaseo/server' @@ -54,19 +54,6 @@ function matchesFilter(item: AgentTimelineItem, filter?: string): boolean { } } -/** - * Extract timeline items from an agent_stream_snapshot message - */ -function extractTimelineFromSnapshot(message: AgentStreamSnapshotMessage): AgentTimelineItem[] { - const items: AgentTimelineItem[] = [] - for (const e of message.payload.events) { - if (e.event.type === 'timeline') { - items.push(e.event.item) - } - } - return items -} - export async function runLogsCommand( id: string, options: AgentLogsOptions, @@ -112,32 +99,12 @@ export async function runLogsCommand( return } - // For non-follow mode, initialize the agent to get timeline snapshot - // Set up handler for timeline events before initializing - const snapshotPromise = new Promise((resolve) => { - const timeout = setTimeout(() => resolve([]), 10000) - - const unsubscribe = client.on('agent_stream_snapshot', (msg: unknown) => { - const message = msg as AgentStreamSnapshotMessage - if (message.type !== 'agent_stream_snapshot') return - if (message.payload.agentId !== resolvedId) return - - clearTimeout(timeout) - unsubscribe() - resolve(extractTimelineFromSnapshot(message)) - }) + // Fetch timeline directly via cursor RPC. + let timelineItems = await fetchProjectedTimelineItems({ + client, + agentId: resolvedId, }) - // Initialize agent to trigger timeline snapshot - try { - await client.initializeAgent(resolvedId) - } catch { - // Agent might already be initialized, continue to collect from queue - } - - // Get timeline from snapshot - let timelineItems = await snapshotPromise - // Apply filter if (options.filter) { timelineItems = timelineItems.filter((item) => matchesFilter(item, options.filter)) @@ -179,31 +146,12 @@ async function runFollowMode( const DEFAULT_FOLLOW_TAIL = 10 const tailCount = parseTailCount(options.tail) ?? DEFAULT_FOLLOW_TAIL - // First, get existing timeline - const snapshotPromise = new Promise((resolve) => { - const timeout = setTimeout(() => resolve([]), 10000) - - const unsubscribe = client.on('agent_stream_snapshot', (msg: unknown) => { - const message = msg as AgentStreamSnapshotMessage - if (message.type !== 'agent_stream_snapshot') return - if (message.payload.agentId !== agentId) return - - clearTimeout(timeout) - unsubscribe() - resolve(extractTimelineFromSnapshot(message)) - }) + // First, get existing timeline. + let existingItems = await fetchProjectedTimelineItems({ + client, + agentId, }) - // Initialize agent to trigger timeline snapshot - try { - await client.initializeAgent(agentId) - } catch { - // Agent might already be initialized - } - - // Get existing timeline - let existingItems = await snapshotPromise - // Apply filter to existing items if (options.filter) { existingItems = existingItems.filter((item) => matchesFilter(item, options.filter)) diff --git a/packages/cli/src/utils/timeline.ts b/packages/cli/src/utils/timeline.ts new file mode 100644 index 000000000..dde26dda1 --- /dev/null +++ b/packages/cli/src/utils/timeline.ts @@ -0,0 +1,33 @@ +import type { AgentTimelineItem, DaemonClient } from '@getpaseo/server' + +type TimelineFetchResult = { + entries: Array<{ item: AgentTimelineItem }> +} + +type TimelineFetchClient = { + fetchAgentTimeline: ( + agentId: string, + options: { + direction: 'tail' + limit: 0 + projection: 'projected' + } + ) => Promise +} + +type FetchProjectedTimelineItemsInput = { + client: DaemonClient + agentId: string +} + +export async function fetchProjectedTimelineItems( + input: FetchProjectedTimelineItemsInput +): Promise { + const timelineClient = input.client as unknown as TimelineFetchClient + const timeline = await timelineClient.fetchAgentTimeline(input.agentId, { + direction: 'tail', + limit: 0, + projection: 'projected', + }) + return timeline.entries.map((entry) => entry.item) +} diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index f7cb5aa44..b3b5b13e7 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -1,4 +1,3 @@ -import { readFileSync } from "node:fs"; import { afterEach, describe, expect, expectTypeOf, test, vi } from "vitest"; import { DaemonClient, type DaemonTransport } from "./daemon-client"; import { @@ -60,11 +59,6 @@ function createMockTransport() { }; } -function loadLegacySnapshotFixture(): unknown { - const url = new URL("../shared/__fixtures__/legacy-agent-stream-snapshot-inProgress.json", import.meta.url); - return JSON.parse(readFileSync(url, "utf8")); -} - function wrapSessionMessage(message: unknown): string { return JSON.stringify({ type: "session", @@ -930,7 +924,7 @@ describe("DaemonClient", () => { expect(logger.warn).toHaveBeenCalled(); }); - test("parses canonical agent_stream_snapshot tool_call payloads without crashing", async () => { + test("parses canonical fetch_agent_timeline_response payloads without crashing", async () => { const logger = createMockLogger(); const mock = createMockTransport(); @@ -947,35 +941,49 @@ describe("DaemonClient", () => { await connectPromise; const received: unknown[] = []; - const unsubscribe = client.on("agent_stream_snapshot", (msg) => { + const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => { received.push(msg); }); mock.triggerMessage( wrapSessionMessage({ - type: "agent_stream_snapshot", + type: "fetch_agent_timeline_response", payload: { + requestId: "req-1", agentId: "agent_cli", - events: [ + direction: "tail", + projection: "projected", + epoch: "epoch-1", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 1, maxSeq: 1, nextSeq: 2 }, + startCursor: { epoch: "epoch-1", seq: 1 }, + endCursor: { epoch: "epoch-1", seq: 1 }, + hasOlder: false, + hasNewer: false, + entries: [ { timestamp: "2026-02-08T20:20:00.000Z", - event: { - type: "timeline", - provider: "codex", - item: { - type: "tool_call", - callId: "call_cli_snapshot", - name: "shell", - status: "running", - detail: { - type: "shell", - command: "pwd", - }, - error: null, + provider: "codex", + seqStart: 1, + seqEnd: 1, + sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], + collapsed: [], + item: { + type: "tool_call", + callId: "call_cli_snapshot", + name: "shell", + status: "running", + detail: { + type: "shell", + command: "pwd", }, + error: null, }, }, ], + error: null, }, }) ); @@ -983,35 +991,32 @@ describe("DaemonClient", () => { unsubscribe(); expect(received).toHaveLength(1); - const snapshotMsg = received[0] as { + const timelineMsg = received[0] as { payload: { - events: Array<{ - event: { - type: "timeline"; - item: { - type: "tool_call"; - status: string; - error: unknown; - detail: { - type: string; - }; + entries: Array<{ + item: { + type: "tool_call"; + status: string; + error: unknown; + detail: { + type: string; }; }; }>; }; }; - const firstTimeline = snapshotMsg.payload.events[0]?.event; - expect(firstTimeline?.type).toBe("timeline"); - if (firstTimeline?.type === "timeline" && firstTimeline.item.type === "tool_call") { - expect(firstTimeline.item.status).toBe("running"); - expect(firstTimeline.item.error).toBeNull(); - expect(firstTimeline.item.detail.type).toBe("shell"); + const firstEntry = timelineMsg.payload.entries[0]; + expect(firstEntry?.item.type).toBe("tool_call"); + if (firstEntry?.item.type === "tool_call") { + expect(firstEntry.item.status).toBe("running"); + expect(firstEntry.item.error).toBeNull(); + expect(firstEntry.item.detail.type).toBe("shell"); } expect(logger.warn).not.toHaveBeenCalled(); }); - test("drops legacy agent_stream_snapshot tool_call payloads and logs validation warning", async () => { + test("drops invalid fetch_agent_timeline_response tool_call payloads and logs validation warning", async () => { const logger = createMockLogger(); const mock = createMockTransport(); @@ -1028,12 +1033,52 @@ describe("DaemonClient", () => { await connectPromise; const received: unknown[] = []; - const unsubscribe = client.on("agent_stream_snapshot", (msg) => { + const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => { received.push(msg); }); - const snapshot = loadLegacySnapshotFixture(); - mock.triggerMessage(wrapSessionMessage(snapshot)); + mock.triggerMessage( + wrapSessionMessage({ + type: "fetch_agent_timeline_response", + payload: { + requestId: "req-invalid", + agentId: "agent_cli", + direction: "tail", + projection: "projected", + epoch: "epoch-1", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 1, maxSeq: 1, nextSeq: 2 }, + startCursor: { epoch: "epoch-1", seq: 1 }, + endCursor: { epoch: "epoch-1", seq: 1 }, + hasOlder: false, + hasNewer: false, + entries: [ + { + timestamp: "2026-02-08T20:20:00.000Z", + provider: "codex", + seqStart: 1, + seqEnd: 1, + sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], + collapsed: [], + item: { + type: "tool_call", + callId: "call_cli_invalid", + name: "shell", + status: "inProgress", + detail: { + type: "unknown", + input: { command: "pwd" }, + output: null, + }, + }, + }, + ], + error: null, + }, + }) + ); unsubscribe(); diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index 638131cda..080213f61 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -15,6 +15,7 @@ import type { CreateAgentRequestMessage, FileDownloadTokenResponse, FileExplorerResponse, + FetchAgentTimelineResponseMessage, GitSetupOptions, CheckoutStatusResponse, CheckoutCommitResponse, @@ -139,6 +140,8 @@ export type DaemonEvent = agentId: string; event: AgentStreamEventPayload; timestamp: string; + seq?: number; + epoch?: string; } | { type: "status"; payload: { status: string } & Record } | { type: "agent_deleted"; agentId: string } @@ -235,6 +238,20 @@ type TerminalOutputPayload = TerminalOutput["payload"]; type KillTerminalPayload = KillTerminalResponse["payload"]; type AttachTerminalStreamPayload = AttachTerminalStreamResponse["payload"]; type DetachTerminalStreamPayload = DetachTerminalStreamResponse["payload"]; +export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"]; + +export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"]; +export type FetchAgentTimelineProjection = FetchAgentTimelinePayload["projection"]; +export type FetchAgentTimelineCursor = NonNullable< + FetchAgentTimelinePayload["startCursor"] +>; +export type FetchAgentTimelineOptions = { + direction?: FetchAgentTimelineDirection; + cursor?: FetchAgentTimelineCursor; + limit?: number; + projection?: FetchAgentTimelineProjection; + requestId?: string; +}; export type TerminalStreamChunk = { streamId: number; @@ -1244,23 +1261,28 @@ export class DaemonClient { }); } - async initializeAgent( + async fetchAgentTimeline( agentId: string, - requestId?: string - ): Promise { - const resolvedRequestId = this.createRequestId(requestId); + options: FetchAgentTimelineOptions = {} + ): Promise { + const resolvedRequestId = this.createRequestId(options.requestId); const message = SessionInboundMessageSchema.parse({ - type: "initialize_agent_request", + type: "fetch_agent_timeline_request", agentId, requestId: resolvedRequestId, + ...(options.direction ? { direction: options.direction } : {}), + ...(options.cursor ? { cursor: options.cursor } : {}), + ...(typeof options.limit === "number" ? { limit: options.limit } : {}), + ...(options.projection ? { projection: options.projection } : {}), }); + const payload = await this.sendRequest({ requestId: resolvedRequestId, message, - timeout: 10000, + timeout: 15000, options: { skipQueue: true }, select: (msg) => { - if (msg.type !== "initialize_agent_request") { + if (msg.type !== "fetch_agent_timeline_response") { return null; } if (msg.payload.requestId !== resolvedRequestId) { @@ -1269,14 +1291,12 @@ export class DaemonClient { return msg.payload; }, }); + if (payload.error) { throw new Error(payload.error); } - const agent = await this.fetchAgent(agentId); - if (!agent) { - throw new Error(`Agent not found after initialize: ${agentId}`); - } - return agent; + + return payload; } // ============================================================================ @@ -3079,6 +3099,8 @@ export class DaemonClient { agentId: msg.payload.agentId, event: msg.payload.event, timestamp: msg.payload.timestamp, + ...(typeof msg.payload.seq === "number" ? { seq: msg.payload.seq } : {}), + ...(typeof msg.payload.epoch === "string" ? { epoch: msg.payload.epoch } : {}), }; case "status": return { type: "status", payload: msg.payload }; diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index e724fb6a4..86a74d620 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -351,6 +351,120 @@ describe("AgentManager", () => { expect(afterHydrate).toEqual(beforeReload); }); + test("fetchTimeline returns full timeline with reset when cursor epoch is stale", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-stale-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000118", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "one", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "two", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "three", + }); + + const baseline = manager.fetchTimeline(snapshot.id, { + direction: "tail", + limit: 2, + }); + expect(baseline.rows).toHaveLength(2); + + const result = manager.fetchTimeline(snapshot.id, { + direction: "after", + cursor: { + epoch: "stale-epoch", + seq: baseline.rows[baseline.rows.length - 1]!.seq, + }, + limit: 1, + }); + + expect(result.reset).toBe(true); + expect(result.staleCursor).toBe(true); + expect(result.gap).toBe(false); + expect(result.rows).toHaveLength(3); + expect(result.rows[0]?.seq).toBe(1); + expect(result.rows[result.rows.length - 1]?.seq).toBe(3); + }); + + test("fetchTimeline returns full timeline with reset when cursor seq falls behind retention window", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-gap-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + maxTimelineItems: 2, + idFactory: () => "00000000-0000-4000-8000-000000000119", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "first", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "second", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "third", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "fourth", + }); + + const fresh = manager.fetchTimeline(snapshot.id, { + direction: "tail", + limit: 0, + }); + expect(fresh.window.minSeq).toBe(3); + expect(fresh.window.maxSeq).toBe(4); + + const result = manager.fetchTimeline(snapshot.id, { + direction: "after", + cursor: { + epoch: fresh.epoch, + seq: 1, + }, + limit: 10, + }); + + expect(result.reset).toBe(true); + expect(result.staleCursor).toBe(false); + expect(result.gap).toBe(true); + expect(result.rows).toHaveLength(2); + expect(result.rows[0]?.seq).toBe(3); + expect(result.rows[1]?.seq).toBe(4); + }); + test("createAgent fails when generated agent ID is not a UUID", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index c37e71fc5..89d963352 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -35,7 +35,13 @@ export { AGENT_LIFECYCLE_STATUSES, type AgentLifecycleStatus }; export type AgentManagerEvent = | { type: "agent_state"; agent: ManagedAgent } - | { type: "agent_stream"; agentId: string; event: AgentStreamEvent }; + | { + type: "agent_stream"; + agentId: string; + event: AgentStreamEvent; + seq?: number; + epoch?: string; + }; export type AgentSubscriber = (event: AgentManagerEvent) => void; @@ -84,6 +90,48 @@ export type WaitForAgentStartOptions = { signal?: AbortSignal; }; +export type AgentTimelineRow = { + seq: number; + timestamp: string; + item: AgentTimelineItem; +}; + +export type AgentTimelineCursor = { + epoch: string; + seq: number; +}; + +export type AgentTimelineFetchDirection = "tail" | "before" | "after"; + +export type AgentTimelineFetchOptions = { + direction?: AgentTimelineFetchDirection; + cursor?: AgentTimelineCursor; + /** + * Number of canonical rows to return. + * - undefined: manager default + * - 0: all rows in the selected window + */ + limit?: number; +}; + +export type AgentTimelineWindow = { + minSeq: number; + maxSeq: number; + nextSeq: number; +}; + +export type AgentTimelineFetchResult = { + epoch: string; + direction: AgentTimelineFetchDirection; + reset: boolean; + staleCursor: boolean; + gap: boolean; + window: AgentTimelineWindow; + hasOlder: boolean; + hasNewer: boolean; + rows: AgentTimelineRow[]; +}; + type AttentionState = | { requiresAttention: false } | { @@ -105,6 +153,9 @@ type ManagedAgentBase = { currentModeId: string | null; pendingPermissions: Map; timeline: AgentTimelineItem[]; + timelineRows: AgentTimelineRow[]; + timelineEpoch: string; + timelineNextSeq: number; persistence: AgentPersistenceHandle | null; historyPrimed: boolean; lastUserMessageAt: Date | null; @@ -187,6 +238,7 @@ type SubscriptionRecord = { }; const DEFAULT_MAX_TIMELINE_ITEMS = 2000; +const DEFAULT_TIMELINE_FETCH_LIMIT = 200; const BUSY_STATUSES: AgentLifecycleStatus[] = [ "initializing", "running", @@ -386,6 +438,153 @@ export class AgentManager { return [...agent.timeline]; } + getTimelineRows(id: string): AgentTimelineRow[] { + const agent = this.requireAgent(id); + const { rows } = this.ensureTimelineState(agent); + return rows.map((row) => ({ ...row })); + } + + fetchTimeline( + id: string, + options?: AgentTimelineFetchOptions + ): AgentTimelineFetchResult { + const agent = this.requireAgent(id); + const { + rows, + epoch, + nextSeq, + minSeq, + maxSeq, + } = this.ensureTimelineState(agent); + const direction = options?.direction ?? "tail"; + const requestedLimit = options?.limit; + const limit = + requestedLimit === undefined + ? DEFAULT_TIMELINE_FETCH_LIMIT + : Math.max(0, Math.floor(requestedLimit)); + const cursor = options?.cursor; + + const window: AgentTimelineWindow = { minSeq, maxSeq, nextSeq }; + + if (cursor && cursor.epoch !== epoch) { + return { + epoch, + direction, + reset: true, + staleCursor: true, + gap: false, + window, + hasOlder: false, + hasNewer: false, + rows: rows.map((row) => ({ ...row })), + }; + } + + const selectAll = limit === 0; + const cloneRows = (items: AgentTimelineRow[]) => items.map((row) => ({ ...row })); + + if (direction === "after" && cursor && rows.length > 0 && cursor.seq < minSeq - 1) { + return { + epoch, + direction, + reset: true, + staleCursor: false, + gap: true, + window, + hasOlder: false, + hasNewer: false, + rows: cloneRows(rows), + }; + } + + if (rows.length === 0) { + return { + epoch, + direction, + reset: false, + staleCursor: false, + gap: false, + window, + hasOlder: false, + hasNewer: false, + rows: [], + }; + } + + if (direction === "tail") { + const selected = + selectAll || limit >= rows.length ? rows : rows.slice(rows.length - limit); + const hasOlder = selected.length > 0 && selected[0]!.seq > minSeq; + return { + epoch, + direction, + reset: false, + staleCursor: false, + gap: false, + window, + hasOlder, + hasNewer: false, + rows: cloneRows(selected), + }; + } + + if (direction === "after") { + const baseSeq = cursor?.seq ?? 0; + const startIdx = rows.findIndex((row) => row.seq > baseSeq); + if (startIdx < 0) { + return { + epoch, + direction, + reset: false, + staleCursor: false, + gap: false, + window, + hasOlder: baseSeq >= minSeq, + hasNewer: false, + rows: [], + }; + } + + const selected = + selectAll ? rows.slice(startIdx) : rows.slice(startIdx, startIdx + limit); + const lastSelected = selected[selected.length - 1]; + return { + epoch, + direction, + reset: false, + staleCursor: false, + gap: false, + window, + hasOlder: selected[0]!.seq > minSeq, + hasNewer: Boolean(lastSelected && lastSelected.seq < maxSeq), + rows: cloneRows(selected), + }; + } + + // direction === "before" + const beforeSeq = cursor?.seq ?? nextSeq; + const endExclusive = rows.findIndex((row) => row.seq >= beforeSeq); + const boundedRows = + endExclusive < 0 ? rows : rows.slice(0, endExclusive); + const selected = + selectAll || limit >= boundedRows.length + ? boundedRows + : boundedRows.slice(boundedRows.length - limit); + const hasOlder = selected.length > 0 && selected[0]!.seq > minSeq; + const hasNewer = endExclusive >= 0; + return { + epoch, + direction, + reset: false, + staleCursor: false, + gap: false, + window, + hasOlder, + hasNewer, + rows: cloneRows(selected), + }; + } + async createAgent( config: AgentSessionConfig, agentId?: string, @@ -459,7 +658,11 @@ export class AgentManager { overrides?: Partial ): Promise { const existing = this.requireAgent(agentId); + const timelineState = this.ensureTimelineState(existing); const preservedTimeline = [...existing.timeline]; + const preservedTimelineRows = timelineState.rows.map((row) => ({ ...row })); + const preservedTimelineEpoch = timelineState.epoch; + const preservedTimelineNextSeq = timelineState.nextSeq; const preservedHistoryPrimed = existing.historyPrimed; const preservedLastUsage = existing.lastUsage; const preservedLastError = existing.lastError; @@ -496,6 +699,9 @@ export class AgentManager { updatedAt: existing.updatedAt, lastUserMessageAt: existing.lastUserMessageAt, timeline: preservedTimeline, + timelineRows: preservedTimelineRows, + timelineEpoch: preservedTimelineEpoch, + timelineNextSeq: preservedTimelineNextSeq, historyPrimed: preservedHistoryPrimed, lastUsage: preservedLastUsage, lastError: preservedLastError, @@ -651,11 +857,14 @@ export class AgentManager { }; agent.updatedAt = new Date(); agent.lastUserMessageAt = agent.updatedAt; - this.recordTimeline(agent, item); + const row = this.recordTimeline(agent, item); this.dispatchStream(agentId, { type: "timeline", item, provider: agent.provider, + }, { + seq: row.seq, + epoch: this.ensureTimelineState(agent).epoch, }); this.emitState(agent); } @@ -663,11 +872,14 @@ export class AgentManager { async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise { const agent = this.requireAgent(agentId); agent.updatedAt = new Date(); - this.recordTimeline(agent, item); + const row = this.recordTimeline(agent, item); this.dispatchStream(agentId, { type: "timeline", item, provider: agent.provider, + }, { + seq: row.seq, + epoch: this.ensureTimelineState(agent).epoch, }); await this.persistSnapshot(agent); } @@ -1106,6 +1318,9 @@ export class AgentManager { lastUserMessageAt?: Date | null; labels?: Record; timeline?: AgentTimelineItem[]; + timelineRows?: AgentTimelineRow[]; + timelineEpoch?: string; + timelineNextSeq?: number; historyPrimed?: boolean; lastUsage?: AgentUsage; lastError?: string; @@ -1118,6 +1333,21 @@ export class AgentManager { } const now = new Date(); + const initialTimeline = options?.timeline ? [...options.timeline] : []; + const initialTimelineRows = + options?.timelineRows?.length + ? options.timelineRows.map((row) => ({ ...row })) + : this.buildTimelineRowsFromItems( + initialTimeline, + options?.timelineNextSeq ?? 1, + (options?.updatedAt ?? options?.createdAt ?? now).toISOString() + ); + const derivedNextSeq = + options?.timelineNextSeq ?? + (initialTimelineRows.length + ? initialTimelineRows[initialTimelineRows.length - 1]!.seq + 1 + : 1); + const managed = { id: resolvedAgentId, provider: config.provider, @@ -1133,7 +1363,10 @@ export class AgentManager { currentModeId: null, pendingPermissions: new Map(), pendingRun: null, - timeline: options?.timeline ? [...options.timeline] : [], + timeline: initialTimeline, + timelineRows: initialTimelineRows, + timelineEpoch: options?.timelineEpoch ?? randomUUID(), + timelineNextSeq: derivedNextSeq, persistence: attachPersistenceCwd(session.describePersistence(), config.cwd), historyPrimed: options?.historyPrimed ?? false, lastUserMessageAt: options?.lastUserMessageAt ?? null, @@ -1169,6 +1402,44 @@ export class AgentManager { return { ...managed }; } + private buildTimelineRowsFromItems( + items: readonly AgentTimelineItem[], + startSeq: number, + timestamp: string + ): AgentTimelineRow[] { + let nextSeq = startSeq; + return items.map((item) => { + const row: AgentTimelineRow = { + seq: nextSeq, + timestamp, + item, + }; + nextSeq += 1; + return row; + }); + } + + private ensureTimelineState(agent: ManagedAgent): { + rows: AgentTimelineRow[]; + epoch: string; + nextSeq: number; + minSeq: number; + maxSeq: number; + } { + const minSeq = agent.timelineRows.length ? agent.timelineRows[0]!.seq : 0; + const maxSeq = agent.timelineRows.length + ? agent.timelineRows[agent.timelineRows.length - 1]!.seq + : 0; + + return { + rows: agent.timelineRows, + epoch: agent.timelineEpoch, + nextSeq: agent.timelineNextSeq, + minSeq, + maxSeq, + }; + } + private async persistSnapshot( agent: ManagedAgent, options?: { title?: string | null; internal?: boolean } @@ -1256,6 +1527,8 @@ export class AgentManager { agent.updatedAt = new Date(); } + let timelineRow: AgentTimelineRow | null = null; + switch (event.type) { case "thread_started": // Update persistence with the new session ID from the provider. @@ -1268,7 +1541,7 @@ export class AgentManager { } break; case "timeline": - this.recordTimeline(agent, event.item); + timelineRow = this.recordTimeline(agent, event.item); if ( !options?.fromHistory && event.item.type === "user_message" @@ -1325,17 +1598,35 @@ export class AgentManager { } // Skip dispatching individual stream events during history replay. - // The caller will send a batched agent_stream_snapshot after priming. if (!options?.fromHistory) { - this.dispatchStream(agent.id, event); + this.dispatchStream(agent.id, event, timelineRow + ? { + seq: timelineRow.seq, + epoch: this.ensureTimelineState(agent).epoch, + } + : undefined); } } - private recordTimeline(agent: ManagedAgent, item: AgentTimelineItem): void { + private recordTimeline( + agent: ManagedAgent, + item: AgentTimelineItem + ): AgentTimelineRow { + const timelineState = this.ensureTimelineState(agent); + const row: AgentTimelineRow = { + seq: timelineState.nextSeq, + timestamp: new Date().toISOString(), + item, + }; + agent.timelineNextSeq = timelineState.nextSeq + 1; agent.timeline.push(item); + timelineState.rows.push(row); if (agent.timeline.length > this.maxTimelineItems) { - agent.timeline.splice(0, agent.timeline.length - this.maxTimelineItems); + const removeCount = agent.timeline.length - this.maxTimelineItems; + agent.timeline.splice(0, removeCount); + timelineState.rows.splice(0, removeCount); } + return row; } private emitState(agent: ManagedAgent): void { @@ -1439,8 +1730,12 @@ export class AgentManager { }); } - private dispatchStream(agentId: string, event: AgentStreamEvent): void { - this.dispatch({ type: "agent_stream", agentId, event }); + private dispatchStream( + agentId: string, + event: AgentStreamEvent, + metadata?: { seq?: number; epoch?: string } + ): void { + this.dispatch({ type: "agent_stream", agentId, event, ...metadata }); } private dispatch(event: AgentManagerEvent): void { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index bed0acead..d0becd9e7 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -1061,6 +1061,126 @@ describe("Codex app-server provider (integration)", () => { 120000 ); + test.runIf(isCodexInstalled())( + "prefers exec_command notifications over mirrored item/completed for shell tool calls", + async () => { + const cleanup = useTempCodexSessionDir(); + const cwd = tmpCwd("codex-command-lifecycle-dedupe-"); + + try { + const client = new CodexAppServerAgentClient(logger); + const session = await client.createSession({ + provider: "codex", + cwd, + modeId: "full-access", + approvalPolicy: "never", + model: CODEX_TEST_MODEL, + thinkingOptionId: CODEX_TEST_THINKING_OPTION_ID, + }); + + const commandLifecycleByCallId = new Map(); + const rawClient = (session as any).client as + | { + notificationHandler?: (method: string, params: unknown) => void; + setNotificationHandler?: (handler: (method: string, params: unknown) => void) => void; + } + | null; + const originalHandler = rawClient?.notificationHandler; + rawClient?.setNotificationHandler?.((method: string, params: unknown) => { + if (method === "codex/event/exec_command_end") { + const callId = + params && + typeof params === "object" && + "msg" in (params as Record) && + typeof (params as { msg?: { call_id?: unknown } }).msg?.call_id === "string" + ? ((params as { msg?: { call_id?: string } }).msg?.call_id ?? null) + : null; + if (callId) { + const existing = commandLifecycleByCallId.get(callId) ?? { + execEnd: false, + itemCompleted: false, + }; + existing.execEnd = true; + commandLifecycleByCallId.set(callId, existing); + } + } + if (method === "item/completed") { + const item = + params && typeof params === "object" + ? ((params as { item?: { id?: unknown; type?: unknown } }).item ?? null) + : null; + const itemId = typeof item?.id === "string" ? item.id : null; + const normalizedType = + typeof item?.type === "string" + ? item.type.replace(/[._-]/g, "").toLowerCase() + : ""; + if (itemId && normalizedType === "commandexecution") { + const existing = commandLifecycleByCallId.get(itemId) ?? { + execEnd: false, + itemCompleted: false, + }; + existing.itemCompleted = true; + commandLifecycleByCallId.set(itemId, existing); + } + } + originalHandler?.(method, params); + }); + + const marker = "PASEO_COMMAND_DEDUPE_CHECK_4D0E96C8"; + const shellCalls: Array<{ + callId: string; + status: string; + output: string; + }> = []; + let failure: string | null = null; + for await (const event of session.stream( + `Run exactly this shell command and then reply exactly DONE: printf '${marker}\\n'` + )) { + if (event.type === "timeline" && event.item.type === "tool_call" && event.item.name === "shell") { + const output = event.item.detail.type === "shell" ? event.item.detail.output ?? "" : ""; + shellCalls.push({ + callId: event.item.callId, + status: event.item.status, + output, + }); + } + if (event.type === "turn_failed") { + failure = event.error; + break; + } + if (event.type === "turn_completed") { + break; + } + } + + await session.close(); + if (failure) { + throw new Error(failure); + } + + const targetCall = shellCalls.find( + (entry) => entry.status === "completed" && entry.output.includes(marker) + ); + expect(targetCall).toBeDefined(); + if (!targetCall) { + return; + } + const lifecycle = commandLifecycleByCallId.get(targetCall.callId); + expect(lifecycle?.execEnd).toBe(true); + expect(lifecycle?.itemCompleted).toBe(true); + + const statusesForCall = shellCalls + .filter((entry) => entry.callId === targetCall.callId) + .map((entry) => entry.status); + expect(statusesForCall.filter((status) => status === "completed").length).toBe(1); + } finally { + cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } + }, + 120000 + ); + test.runIf(isCodexInstalled())( "interrupts long-running commands and emits a canceled turn", async () => { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 58b4663dd..bf2cfa827 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -1751,6 +1751,8 @@ class CodexAppServerAgentSession implements AgentSession { private pendingReasoning = new Map(); private pendingCommandOutputDeltas = new Map(); private pendingFileChangeOutputDeltas = new Map(); + private emittedExecCommandStartedCallIds = new Set(); + private emittedExecCommandCompletedCallIds = new Set(); private emittedItemStartedIds = new Set(); private emittedItemCompletedIds = new Set(); private warnedUnknownNotificationMethods = new Set(); @@ -2440,6 +2442,8 @@ class CodexAppServerAgentSession implements AgentSession { this.currentTurnId = parsed.turnId; this.emittedItemStartedIds.clear(); this.emittedItemCompletedIds.clear(); + this.emittedExecCommandStartedCallIds.clear(); + this.emittedExecCommandCompletedCallIds.clear(); this.pendingCommandOutputDeltas.clear(); this.pendingFileChangeOutputDeltas.clear(); this.warnedIncompleteEditToolCallIds.clear(); @@ -2461,6 +2465,8 @@ class CodexAppServerAgentSession implements AgentSession { } this.emittedItemStartedIds.clear(); this.emittedItemCompletedIds.clear(); + this.emittedExecCommandStartedCallIds.clear(); + this.emittedExecCommandCompletedCallIds.clear(); this.pendingCommandOutputDeltas.clear(); this.pendingFileChangeOutputDeltas.clear(); this.warnedIncompleteEditToolCallIds.clear(); @@ -2526,6 +2532,7 @@ class CodexAppServerAgentSession implements AgentSession { if (parsed.kind === "exec_command_started") { if (parsed.callId) { + this.emittedExecCommandStartedCallIds.add(parsed.callId); this.pendingCommandOutputDeltas.delete(parsed.callId); } const timelineItem = mapCodexExecNotificationToToolCall({ @@ -2556,6 +2563,7 @@ class CodexAppServerAgentSession implements AgentSession { running: false, }); if (timelineItem) { + this.emittedExecCommandCompletedCallIds.add(timelineItem.callId); this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); } return; @@ -2618,7 +2626,21 @@ class CodexAppServerAgentSession implements AgentSession { cwd: this.config.cwd ?? null, }); if (timelineItem) { + const normalizedItemType = normalizeCodexThreadItemType( + typeof parsed.item.type === "string" ? parsed.item.type : undefined + ); const itemId = parsed.item.id; + // For commandExecution items, codex/event/exec_command_* is authoritative. + // Keep item/completed as fallback only when no exec_command completion was seen. + if ( + timelineItem.type === "tool_call" && + normalizedItemType === "commandExecution" + ) { + const callId = timelineItem.callId || itemId; + if (callId && this.emittedExecCommandCompletedCallIds.has(callId)) { + return; + } + } if (itemId && this.emittedItemCompletedIds.has(itemId)) { return; } @@ -2657,7 +2679,16 @@ class CodexAppServerAgentSession implements AgentSession { cwd: this.config.cwd ?? null, }); if (timelineItem && timelineItem.type === "tool_call") { + const normalizedItemType = normalizeCodexThreadItemType( + typeof parsed.item.type === "string" ? parsed.item.type : undefined + ); const itemId = parsed.item.id; + if (normalizedItemType === "commandExecution") { + const callId = timelineItem.callId || itemId; + if (callId && this.emittedExecCommandStartedCallIds.has(callId)) { + return; + } + } if (itemId && this.emittedItemStartedIds.has(itemId)) { return; } diff --git a/packages/server/src/server/agent/timeline-projection.test.ts b/packages/server/src/server/agent/timeline-projection.test.ts new file mode 100644 index 000000000..81330be4c --- /dev/null +++ b/packages/server/src/server/agent/timeline-projection.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, test } from "vitest"; + +import type { AgentTimelineRow } from "./agent-manager.js"; +import { projectTimelineRows } from "./timeline-projection.js"; + +describe("projectTimelineRows", () => { + test("merges adjacent assistant chunks in projected mode", () => { + const rows: AgentTimelineRow[] = [ + { + seq: 1, + timestamp: "2026-02-13T00:00:00.000Z", + item: { type: "assistant_message", text: "Hel" }, + }, + { + seq: 2, + timestamp: "2026-02-13T00:00:00.100Z", + item: { type: "assistant_message", text: "lo" }, + }, + { + seq: 3, + timestamp: "2026-02-13T00:00:00.200Z", + item: { type: "user_message", text: "next" }, + }, + ]; + + const projected = projectTimelineRows(rows, "codex", "projected"); + + expect(projected).toHaveLength(2); + expect(projected[0]?.item).toEqual({ + type: "assistant_message", + text: "Hello", + }); + expect(projected[0]?.seqStart).toBe(1); + expect(projected[0]?.seqEnd).toBe(2); + expect(projected[0]?.sourceSeqRanges).toEqual([{ startSeq: 1, endSeq: 2 }]); + expect(projected[0]?.collapsed).toContain("assistant_merge"); + }); + + test("collapses tool lifecycle by callId and reports exact source seq ranges", () => { + const rows: AgentTimelineRow[] = [ + { + seq: 1, + timestamp: "2026-02-13T00:00:00.000Z", + item: { + type: "tool_call", + callId: "call_1", + name: "shell", + status: "running", + error: null, + detail: { + type: "unknown", + input: { cmd: "pwd" }, + output: null, + }, + }, + }, + { + seq: 2, + timestamp: "2026-02-13T00:00:00.100Z", + item: { type: "assistant_message", text: "working" }, + }, + { + seq: 3, + timestamp: "2026-02-13T00:00:00.200Z", + item: { + type: "tool_call", + callId: "call_1", + name: "shell", + status: "completed", + error: null, + detail: { + type: "unknown", + input: { cmd: "pwd" }, + output: { stdout: "/tmp" }, + }, + }, + }, + ]; + + const projected = projectTimelineRows(rows, "codex", "projected"); + + expect(projected).toHaveLength(2); + const tool = projected[0]; + expect(tool?.item.type).toBe("tool_call"); + if (tool?.item.type === "tool_call") { + expect(tool.item.status).toBe("completed"); + expect(tool.item.callId).toBe("call_1"); + } + expect(tool?.sourceSeqRanges).toEqual([ + { startSeq: 1, endSeq: 1 }, + { startSeq: 3, endSeq: 3 }, + ]); + expect(tool?.collapsed).toContain("tool_lifecycle"); + }); + + test("returns canonical rows unchanged in canonical mode", () => { + const rows: AgentTimelineRow[] = [ + { + seq: 10, + timestamp: "2026-02-13T00:00:00.000Z", + item: { type: "assistant_message", text: "A" }, + }, + { + seq: 11, + timestamp: "2026-02-13T00:00:00.100Z", + item: { type: "assistant_message", text: "B" }, + }, + ]; + + const projected = projectTimelineRows(rows, "codex", "canonical"); + + expect(projected).toHaveLength(2); + expect(projected[0]?.item).toEqual(rows[0]?.item); + expect(projected[1]?.item).toEqual(rows[1]?.item); + expect(projected[0]?.collapsed).toEqual([]); + expect(projected[1]?.collapsed).toEqual([]); + }); +}); diff --git a/packages/server/src/server/agent/timeline-projection.ts b/packages/server/src/server/agent/timeline-projection.ts new file mode 100644 index 000000000..966af0196 --- /dev/null +++ b/packages/server/src/server/agent/timeline-projection.ts @@ -0,0 +1,209 @@ +import type { AgentProvider, AgentTimelineItem, ToolCallDetail } from "./agent-sdk-types.js"; +import type { AgentTimelineRow } from "./agent-manager.js"; + +export type TimelineProjectionMode = "canonical" | "projected"; + +export type TimelineSeqRange = { + startSeq: number; + endSeq: number; +}; + +export type TimelineProjectionKind = "assistant_merge" | "tool_lifecycle"; + +export type TimelineProjectionEntry = { + provider: AgentProvider; + item: AgentTimelineItem; + timestamp: string; + seqStart: number; + seqEnd: number; + sourceSeqRanges: TimelineSeqRange[]; + collapsed: TimelineProjectionKind[]; +}; + +type WorkingEntry = TimelineProjectionEntry; + +function appendSeqToRanges(ranges: TimelineSeqRange[], seq: number): TimelineSeqRange[] { + if (ranges.length === 0) { + return [{ startSeq: seq, endSeq: seq }]; + } + + const next = [...ranges]; + const last = next[next.length - 1]; + if (!last) { + return [{ startSeq: seq, endSeq: seq }]; + } + + if (seq <= last.endSeq + 1) { + last.endSeq = Math.max(last.endSeq, seq); + return next; + } + + next.push({ startSeq: seq, endSeq: seq }); + return next; +} + +function mergeSeqRanges( + existing: TimelineSeqRange[], + incoming: TimelineSeqRange[] +): TimelineSeqRange[] { + let merged = [...existing]; + for (const range of incoming) { + for (let seq = range.startSeq; seq <= range.endSeq; seq += 1) { + merged = appendSeqToRanges(merged, seq); + } + } + return merged; +} + +function mergeToolCallDetail(existing: ToolCallDetail, incoming: ToolCallDetail): ToolCallDetail { + if (existing.type === "unknown" && incoming.type !== "unknown") { + return incoming; + } + if (incoming.type === "unknown" && existing.type !== "unknown") { + return existing; + } + return incoming; +} + +function mergeToolCallItems( + existing: Extract, + incoming: Extract +): Extract { + const mergedDetail = mergeToolCallDetail(existing.detail, incoming.detail); + const mergedMetadata = + existing.metadata || incoming.metadata + ? { ...(existing.metadata ?? {}), ...(incoming.metadata ?? {}) } + : undefined; + + const merged: Extract = { + ...existing, + ...incoming, + detail: mergedDetail, + metadata: mergedMetadata, + }; + + if (incoming.status === "failed") { + merged.error = incoming.error; + } else if (incoming.status === "completed" || incoming.status === "canceled") { + merged.error = null; + } else if (incoming.error !== undefined) { + merged.error = incoming.error; + } + + return merged; +} + +function makeCanonicalEntries( + rows: readonly AgentTimelineRow[], + provider: AgentProvider +): WorkingEntry[] { + return rows.map((row) => ({ + provider, + item: row.item, + timestamp: row.timestamp, + seqStart: row.seq, + seqEnd: row.seq, + sourceSeqRanges: [{ startSeq: row.seq, endSeq: row.seq }], + collapsed: [], + })); +} + +function collapseToolLifecycle(entries: readonly WorkingEntry[]): WorkingEntry[] { + const output: WorkingEntry[] = []; + const toolIndexByCallId = new Map(); + + for (const entry of entries) { + if (entry.item.type !== "tool_call") { + output.push(entry); + continue; + } + + const existingIndex = toolIndexByCallId.get(entry.item.callId); + if (existingIndex === undefined) { + toolIndexByCallId.set(entry.item.callId, output.length); + output.push(entry); + continue; + } + + const existing = output[existingIndex]; + if (!existing || existing.item.type !== "tool_call") { + output.push(entry); + continue; + } + + const mergedItem = mergeToolCallItems(existing.item, entry.item); + const mergedRanges = mergeSeqRanges(existing.sourceSeqRanges, entry.sourceSeqRanges); + const collapsed = existing.collapsed.includes("tool_lifecycle") + ? existing.collapsed + : ([...existing.collapsed, "tool_lifecycle"] as TimelineProjectionKind[]); + + output[existingIndex] = { + ...existing, + item: mergedItem, + seqEnd: Math.max(existing.seqEnd, entry.seqEnd), + sourceSeqRanges: mergedRanges, + collapsed, + }; + } + + return output; +} + +function mergeAssistantChunks(entries: readonly WorkingEntry[]): WorkingEntry[] { + const output: WorkingEntry[] = []; + + for (const entry of entries) { + const previous = output[output.length - 1]; + const shouldMerge = + previous && + previous.item.type === "assistant_message" && + entry.item.type === "assistant_message" && + previous.seqEnd + 1 === entry.seqStart; + + if (!shouldMerge || !previous) { + output.push(entry); + continue; + } + const previousAssistant = previous.item as Extract< + AgentTimelineItem, + { type: "assistant_message" } + >; + const entryAssistant = entry.item as Extract< + AgentTimelineItem, + { type: "assistant_message" } + >; + + const collapsedKinds = new Set([ + ...previous.collapsed, + ...entry.collapsed, + "assistant_merge", + ]); + + output[output.length - 1] = { + ...previous, + item: { + type: "assistant_message", + text: `${previousAssistant.text}${entryAssistant.text}`, + }, + seqEnd: entry.seqEnd, + sourceSeqRanges: mergeSeqRanges(previous.sourceSeqRanges, entry.sourceSeqRanges), + collapsed: Array.from(collapsedKinds), + }; + } + + return output; +} + +export function projectTimelineRows( + rows: readonly AgentTimelineRow[], + provider: AgentProvider, + mode: TimelineProjectionMode +): TimelineProjectionEntry[] { + const canonical = makeCanonicalEntries(rows, provider); + if (mode === "canonical") { + return canonical; + } + + const toolCollapsed = collapseToolLifecycle(canonical); + return mergeAssistantChunks(toolCollapsed); +} diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index b1025770d..25c239fd1 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -335,8 +335,12 @@ describe("daemon client E2E", () => { agent.id ); - const initResult = await ctx.client.initializeAgent(agent.id); - expect(initResult.id).toBe(agent.id); + const timelineResult = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 1, + projection: "projected", + }); + expect(timelineResult.agentId).toBe(agent.id); const nextMode = agent.availableModes.find( (mode) => mode.id !== agent.currentModeId @@ -577,22 +581,12 @@ describe("daemon client E2E", () => { await ctx.client.sendMessage(agent.id, "Say 'hello' and nothing else"); await ctx.client.waitForFinish(agent.id, 120000); - const snapshotPromise = waitForSignal(15000, (resolve) => { - const unsubscribe = ctx.client.on("agent_stream_snapshot", (message) => { - if (message.type !== "agent_stream_snapshot") { - return; - } - if (message.payload.agentId !== agent.id) { - return; - } - resolve(message); - }); - return unsubscribe; + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "projected", }); - - await ctx.client.initializeAgent(agent.id); - const snapshot = await snapshotPromise; - expect(snapshot.payload.events.length).toBeGreaterThan(0); + expect(timeline.entries.length).toBeGreaterThan(0); await ctx.client.deleteAgent(agent.id); rmSync(cwd, { recursive: true, force: true }); diff --git a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts index c65089998..47f3091a8 100644 --- a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts @@ -61,25 +61,35 @@ describe("daemon E2E", () => { // Clear captured messages before the "click" action messages.length = 0; - // Simulate clicking on the agent (initialize_agent_request) - // This is what happens when the user opens an agent in the UI - const refreshedState = await ctx.client.initializeAgent(agent.id); + // Simulate opening the agent in the UI: + // fetch timeline window + fetch current agent snapshot. + await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 200, + projection: "projected", + }); + const refreshedState = await ctx.client.fetchAgent(agent.id); // Verify agent is still idle - expect(refreshedState.status).toBe("idle"); + expect(refreshedState?.status).toBe("idle"); // CRITICAL: The timestamp should NOT have changed // Just opening/clicking an agent should not update its updatedAt - expect(refreshedState.updatedAt).toBe(initialUpdatedAt); + expect(refreshedState?.updatedAt).toBe(initialUpdatedAt); // Also clear attention (what happens when opening an agent with notification) await ctx.client.clearAgentAttention(agent.id); // Get the state again after clearing attention - const stateAfterClear = await ctx.client.initializeAgent(agent.id); + await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 200, + projection: "projected", + }); + const stateAfterClear = await ctx.client.fetchAgent(agent.id); // Timestamp should STILL not have changed - expect(stateAfterClear.updatedAt).toBe(initialUpdatedAt); + expect(stateAfterClear?.updatedAt).toBe(initialUpdatedAt); // Cleanup rmSync(cwd, { recursive: true, force: true }); diff --git a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts index 61c694ec1..e02894de6 100644 --- a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts @@ -4,7 +4,6 @@ import { tmpdir } from "node:os"; import path from "node:path"; import { createDaemonTestContext, type DaemonTestContext } from "../test-utils/index.js"; -import type { AgentTimelineItem } from "../agent/agent-sdk-types.js"; import type { SessionOutboundMessage } from "../messages.js"; function tmpCwd(): string { @@ -116,9 +115,12 @@ describe("daemon E2E - persistence", () => { messages.push(message); }); - await ctx.client.initializeAgent(agentId); - - const timelineItems = extractTimelineSnapshotItems(messages, agentId); + const timeline = await ctx.client.fetchAgentTimeline(agentId, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + const timelineItems = timeline.entries.map((entry) => entry.item); expect(timelineItems.length).toBeGreaterThan(0); expect(timelineItems.some((item) => item.type === "assistant_message")).toBe(true); } finally { @@ -145,18 +147,3 @@ function extractAssistantText(queue: SessionOutboundMessage[], agentId: string): } return parts.join(""); } - -function extractTimelineSnapshotItems(queue: SessionOutboundMessage[], agentId: string): AgentTimelineItem[] { - const items: AgentTimelineItem[] = []; - for (const m of queue) { - if (m.type !== "agent_stream_snapshot") continue; - if ((m.payload as { agentId: string }).agentId !== agentId) continue; - const events = (m.payload as { events: Array<{ event: { type: string; item?: AgentTimelineItem } }> }).events; - for (const e of events) { - if (e.event.type === "timeline" && e.event.item) { - items.push(e.event.item); - } - } - } - return items; -} diff --git a/packages/server/src/server/exports.ts b/packages/server/src/server/exports.ts index 75dd9fd87..f4c8c5406 100644 --- a/packages/server/src/server/exports.ts +++ b/packages/server/src/server/exports.ts @@ -31,5 +31,4 @@ export type { AgentSnapshotPayload, AgentStreamEventPayload, AgentStreamMessage, - AgentStreamSnapshotMessage, } from "../shared/messages.js"; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 539708c6c..3737ff8f6 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -60,9 +60,14 @@ export type AgentMcpTransportFactory = () => Promise; import { buildProviderRegistry } from "./agent/provider-registry.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; import { AgentManager } from "./agent/agent-manager.js"; -import type { ManagedAgent } from "./agent/agent-manager.js"; +import type { + AgentTimelineCursor, + AgentTimelineFetchDirection, + ManagedAgent, +} from "./agent/agent-manager.js"; import { scheduleAgentMetadataGeneration } from "./agent/agent-metadata-generator.js"; import { toAgentPayload } from "./agent/agent-projections.js"; +import { projectTimelineRows, type TimelineProjectionMode } from "./agent/timeline-projection.js"; import { StructuredAgentResponseError, generateStructuredAgentResponse, @@ -839,9 +844,7 @@ export class Session { // Reduce bandwidth/CPU on mobile: only forward high-frequency agent stream events // for the focused agent, with a short grace window while backgrounded. - // - // History catch-up is handled via explicit `initialize_agent_request` which emits a - // batched `agent_stream_snapshot`. + // History catch-up is handled via pull-based `fetch_agent_timeline_request`. const activity = this.clientActivity; if (activity?.deviceType === "mobile") { if (!activity.focusedAgentId) { @@ -867,6 +870,8 @@ export class Session { agentId: event.agentId, event: serializedEvent, timestamp: new Date().toISOString(), + ...(typeof event.seq === "number" ? { seq: event.seq } : {}), + ...(typeof event.epoch === "string" ? { epoch: event.epoch } : {}), } as const; this.emit({ @@ -1224,17 +1229,17 @@ export class Session { await this.handleRefreshAgentRequest(msg); break; - case "cancel_agent_request": - await this.handleCancelAgentRequest(msg.agentId); - break; + case "cancel_agent_request": + await this.handleCancelAgentRequest(msg.agentId); + break; case "restart_server_request": await this.handleRestartServerRequest(msg.requestId, msg.reason); break; - case "initialize_agent_request": - await this.handleInitializeAgentRequest(msg.agentId, msg.requestId); - break; + case "fetch_agent_timeline_request": + await this.handleFetchAgentTimelineRequest(msg); + break; case "set_agent_mode_request": await this.handleSetAgentModeRequest(msg.agentId, msg.modeId, msg.requestId); @@ -2200,45 +2205,6 @@ export class Session { this.startAgentStream(agentId, prompt, runOptions); } - /** - * Handle on-demand agent initialization request from client - */ - private async handleInitializeAgentRequest( - agentId: string, - requestId: string - ): Promise { - try { - const snapshot = await this.ensureAgentLoaded(agentId); - await this.forwardAgentUpdate(snapshot); - - // Send timeline snapshot after hydration (if any) - const timelineSize = this.emitAgentTimelineSnapshot(snapshot); - - this.emit({ - type: "initialize_agent_request", - payload: { - agentId, - agentStatus: snapshot.lifecycle, - timelineSize, - requestId, - }, - }); - } catch (error: any) { - this.sessionLogger.error( - { err: error, agentId }, - `Failed to initialize agent ${agentId}` - ); - this.emit({ - type: "initialize_agent_request", - payload: { - agentId, - requestId, - error: error?.message ?? "Failed to initialize agent", - }, - }); - } - } - /** * Handle create agent request */ @@ -2383,7 +2349,7 @@ export class Session { ); await this.agentManager.hydrateTimelineFromProvider(snapshot.id); await this.forwardAgentUpdate(snapshot); - const timelineSize = this.emitAgentTimelineSnapshot(snapshot); + const timelineSize = this.agentManager.getTimeline(snapshot.id).length; if (requestId) { const agentPayload = await this.getAgentPayloadById(snapshot.id); if (!agentPayload) { @@ -2458,7 +2424,7 @@ export class Session { } await this.agentManager.hydrateTimelineFromProvider(agentId); await this.forwardAgentUpdate(snapshot); - const timelineSize = this.emitAgentTimelineSnapshot(snapshot); + const timelineSize = this.agentManager.getTimeline(agentId).length; if (requestId) { this.emit({ type: "status", @@ -4925,6 +4891,92 @@ export class Session { }); } + private async handleFetchAgentTimelineRequest( + msg: Extract + ): Promise { + const direction: AgentTimelineFetchDirection = + msg.direction ?? (msg.cursor ? "after" : "tail"); + const projection: TimelineProjectionMode = msg.projection ?? "projected"; + const limit = msg.limit ?? (direction === "after" ? 0 : undefined); + const cursor: AgentTimelineCursor | undefined = msg.cursor + ? { + epoch: msg.cursor.epoch, + seq: msg.cursor.seq, + } + : undefined; + + try { + const snapshot = await this.ensureAgentLoaded(msg.agentId); + + const timeline = this.agentManager.fetchTimeline(msg.agentId, { + direction, + cursor, + limit, + }); + + const projected = projectTimelineRows( + timeline.rows, + snapshot.provider, + projection + ); + + const firstRow = timeline.rows[0]; + const lastRow = timeline.rows[timeline.rows.length - 1]; + const startCursor = firstRow + ? { epoch: timeline.epoch, seq: firstRow.seq } + : null; + const endCursor = lastRow + ? { epoch: timeline.epoch, seq: lastRow.seq } + : null; + + this.emit({ + type: "fetch_agent_timeline_response", + payload: { + requestId: msg.requestId, + agentId: msg.agentId, + direction, + projection, + epoch: timeline.epoch, + reset: timeline.reset, + staleCursor: timeline.staleCursor, + gap: timeline.gap, + window: timeline.window, + startCursor, + endCursor, + hasOlder: timeline.hasOlder, + hasNewer: timeline.hasNewer, + entries: projected, + error: null, + }, + }); + } catch (error) { + this.sessionLogger.error( + { err: error, agentId: msg.agentId }, + "Failed to handle fetch_agent_timeline_request" + ); + this.emit({ + type: "fetch_agent_timeline_response", + payload: { + requestId: msg.requestId, + agentId: msg.agentId, + direction, + projection, + epoch: "", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, + startCursor: null, + endCursor: null, + hasOlder: false, + hasNewer: false, + entries: [], + error: error instanceof Error ? error.message : String(error), + }, + }); + } + } + private async handleSendAgentMessageRequest( msg: Extract ): Promise { @@ -5125,33 +5177,6 @@ export class Session { } } - private emitAgentTimelineSnapshot(agent: ManagedAgent): number { - const timeline = this.agentManager.getTimeline(agent.id); - const events = timeline.flatMap((item) => { - const serializedEvent = serializeAgentStreamEvent({ - type: "timeline", - provider: agent.provider, - item, - }); - if (!serializedEvent) { - return []; - } - return [ - { - event: serializedEvent, - timestamp: new Date().toISOString(), - }, - ]; - }); - - this.emit({ - type: "agent_stream_snapshot", - payload: { agentId: agent.id, events }, - }); - - return timeline.length; - } - /** * Handle audio chunk for buffering and transcription */ diff --git a/packages/server/src/shared/__fixtures__/legacy-agent-stream-snapshot-inProgress.json b/packages/server/src/shared/__fixtures__/legacy-agent-stream-snapshot-inProgress.json deleted file mode 100644 index 7ed165b59..000000000 --- a/packages/server/src/shared/__fixtures__/legacy-agent-stream-snapshot-inProgress.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "type": "agent_stream_snapshot", - "payload": { - "agentId": "agent_fixture_legacy", - "events": [ - { - "timestamp": "2026-02-08T20:00:00.000Z", - "event": { - "type": "timeline", - "provider": "codex", - "item": { - "type": "tool_call", - "callId": "call_fixture_legacy", - "name": "shell", - "status": "inProgress", - "input": { - "command": "pwd" - } - } - } - } - ] - } -} diff --git a/packages/server/src/shared/messages.stream-parsing.test.ts b/packages/server/src/shared/messages.stream-parsing.test.ts index 6c5c5af85..246f5f185 100644 --- a/packages/server/src/shared/messages.stream-parsing.test.ts +++ b/packages/server/src/shared/messages.stream-parsing.test.ts @@ -1,24 +1,48 @@ -import { readFileSync } from "node:fs"; import { describe, expect, it } from "vitest"; import { AgentStreamMessageSchema, - AgentStreamSnapshotMessageSchema, + FetchAgentTimelineResponseMessageSchema, SessionInboundMessageSchema, SessionOutboundMessageSchema, WSOutboundMessageSchema, } from "./messages.js"; -function loadFixture(name: string): unknown { - const url = new URL(`./__fixtures__/${name}`, import.meta.url); - return JSON.parse(readFileSync(url, "utf8")); -} - describe("shared messages stream parsing", () => { - it("rejects legacy inProgress tool_call snapshots", () => { - const fixture = loadFixture("legacy-agent-stream-snapshot-inProgress.json"); - const parsed = AgentStreamSnapshotMessageSchema.safeParse(fixture); - expect(parsed.success).toBe(false); + it("parses representative fetch_agent_timeline_response payload", () => { + const parsed = FetchAgentTimelineResponseMessageSchema.parse({ + type: "fetch_agent_timeline_response", + payload: { + requestId: "req-1", + agentId: "agent_live", + direction: "tail", + projection: "projected", + epoch: "epoch-1", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 1, maxSeq: 2, nextSeq: 3 }, + startCursor: { epoch: "epoch-1", seq: 1 }, + endCursor: { epoch: "epoch-1", seq: 2 }, + hasOlder: false, + hasNewer: false, + entries: [ + { + provider: "codex", + item: { type: "assistant_message", text: "hello" }, + timestamp: "2026-02-08T20:10:00.000Z", + seqStart: 1, + seqEnd: 2, + sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }], + collapsed: ["assistant_merge"], + }, + ], + error: null, + }, + }); + + expect(parsed.payload.entries).toHaveLength(1); + expect(parsed.payload.entries[0]?.item.type).toBe("assistant_message"); }); it("parses representative agent_stream tool_call event", () => { @@ -54,12 +78,14 @@ describe("shared messages stream parsing", () => { } }); - it("rejects websocket envelope for agent_stream_snapshot with legacy status", () => { - const fixture = loadFixture("legacy-agent-stream-snapshot-inProgress.json") as { - type: "agent_stream_snapshot"; - payload: unknown; + it("rejects websocket envelope for removed agent_stream_snapshot message type", () => { + const fixture = { + type: "agent_stream_snapshot", + payload: { + agentId: "agent-legacy", + events: [], + }, }; - const wrapped = WSOutboundMessageSchema.safeParse({ type: "session", message: fixture, diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index cc4ed0d52..d241d3357 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -605,22 +605,21 @@ export const RestartServerRequestMessageSchema = z.object({ requestId: z.string(), }); -export const InitializeAgentRequestMessageSchema = z.object({ - type: z.literal("initialize_agent_request"), - agentId: z.string(), - requestId: z.string(), +export const AgentTimelineCursorSchema = z.object({ + epoch: z.string(), + seq: z.number().int().nonnegative(), }); -export const InitializeAgentResponseMessageSchema = z.object({ - // Response shares the same type literal as the request for simple requestId matching - type: z.literal("initialize_agent_request"), - payload: z.object({ - agentId: z.string(), - agentStatus: z.string().optional(), - timelineSize: z.number().optional(), - requestId: z.string(), - error: z.string().optional(), - }), +export const FetchAgentTimelineRequestMessageSchema = z.object({ + type: z.literal("fetch_agent_timeline_request"), + agentId: z.string(), + requestId: z.string(), + direction: z.enum(["tail", "before", "after"]).optional(), + cursor: AgentTimelineCursorSchema.optional(), + // 0 means "all matching rows for this query window". + limit: z.number().int().nonnegative().optional(), + // Default should be projected for app timeline loading. + projection: z.enum(["projected", "canonical"]).optional(), }); export const SetAgentModeRequestMessageSchema = z.object({ @@ -1023,7 +1022,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ RefreshAgentRequestMessageSchema, CancelAgentRequestMessageSchema, RestartServerRequestMessageSchema, - InitializeAgentRequestMessageSchema, + FetchAgentTimelineRequestMessageSchema, SetAgentModeRequestMessageSchema, SetAgentModelRequestMessageSchema, SetAgentThinkingRequestMessageSchema, @@ -1316,19 +1315,9 @@ export const AgentStreamMessageSchema = z.object({ agentId: z.string(), event: AgentStreamEventPayloadSchema, timestamp: z.string(), - }), -}); - -export const AgentStreamSnapshotMessageSchema = z.object({ - type: z.literal("agent_stream_snapshot"), - payload: z.object({ - agentId: z.string(), - events: z.array( - z.object({ - event: AgentStreamEventPayloadSchema, - timestamp: z.string(), - }) - ), + // Present for timeline events. Maps 1:1 to canonical in-memory timeline rows. + seq: z.number().int().nonnegative().optional(), + epoch: z.string().optional(), }), }); @@ -1384,6 +1373,46 @@ export const FetchAgentResponseMessageSchema = z.object({ }), }); +const AgentTimelineSeqRangeSchema = z.object({ + startSeq: z.number().int().nonnegative(), + endSeq: z.number().int().nonnegative(), +}); + +export const AgentTimelineEntryPayloadSchema = z.object({ + provider: AgentProviderSchema, + item: AgentTimelineItemPayloadSchema, + timestamp: z.string(), + seqStart: z.number().int().nonnegative(), + seqEnd: z.number().int().nonnegative(), + sourceSeqRanges: z.array(AgentTimelineSeqRangeSchema), + collapsed: z.array(z.enum(["assistant_merge", "tool_lifecycle"])), +}); + +export const FetchAgentTimelineResponseMessageSchema = z.object({ + type: z.literal("fetch_agent_timeline_response"), + payload: z.object({ + requestId: z.string(), + agentId: z.string(), + direction: z.enum(["tail", "before", "after"]), + projection: z.enum(["projected", "canonical"]), + epoch: z.string(), + reset: z.boolean(), + staleCursor: z.boolean(), + gap: z.boolean(), + window: z.object({ + minSeq: z.number().int().nonnegative(), + maxSeq: z.number().int().nonnegative(), + nextSeq: z.number().int().nonnegative(), + }), + startCursor: AgentTimelineCursorSchema.nullable(), + endCursor: AgentTimelineCursorSchema.nullable(), + hasOlder: z.boolean(), + hasNewer: z.boolean(), + entries: z.array(AgentTimelineEntryPayloadSchema), + error: z.string().nullable(), + }), +}); + export const SendAgentMessageResponseMessageSchema = z.object({ type: z.literal("send_agent_message_response"), payload: z.object({ @@ -1880,15 +1909,14 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ StatusMessageSchema, PongMessageSchema, RpcErrorMessageSchema, - InitializeAgentResponseMessageSchema, ArtifactMessageSchema, AgentUpdateMessageSchema, AgentStreamMessageSchema, - AgentStreamSnapshotMessageSchema, AgentStatusMessageSchema, FetchAgentsResponseMessageSchema, FetchAgentsGroupedByProjectResponseMessageSchema, FetchAgentResponseMessageSchema, + FetchAgentTimelineResponseMessageSchema, SendAgentMessageResponseMessageSchema, SetVoiceModeResponseMessageSchema, SetAgentModeResponseMessageSchema, @@ -1946,9 +1974,6 @@ export type RpcErrorMessage = z.infer; export type ArtifactMessage = z.infer; export type AgentUpdateMessage = z.infer; export type AgentStreamMessage = z.infer; -export type AgentStreamSnapshotMessage = z.infer< - typeof AgentStreamSnapshotMessageSchema ->; export type AgentStatusMessage = z.infer; export type ProjectCheckoutLitePayload = z.infer; export type ProjectPlacementPayload = z.infer; @@ -1961,6 +1986,9 @@ export type FetchAgentsGroupedByProjectResponseMessage = z.infer< export type FetchAgentResponseMessage = z.infer< typeof FetchAgentResponseMessageSchema >; +export type FetchAgentTimelineResponseMessage = z.infer< + typeof FetchAgentTimelineResponseMessageSchema +>; export type SendAgentMessageResponseMessage = z.infer< typeof SendAgentMessageResponseMessageSchema >; @@ -1982,7 +2010,6 @@ export type ListAvailableProvidersResponse = z.infer< >; export type SpeechModelsListResponse = z.infer; export type SpeechModelsDownloadResponse = z.infer; -export type InitializeAgentResponseMessage = z.infer; // Type exports for payload types export type ActivityLogPayload = z.infer; @@ -2014,7 +2041,6 @@ export type SpeechModelsDownloadRequestMessage = z.infer< export type ResumeAgentRequestMessage = z.infer; export type DeleteAgentRequestMessage = z.infer; export type UpdateAgentRequestMessage = z.infer; -export type InitializeAgentRequestMessage = z.infer; export type SetAgentModeRequestMessage = z.infer; export type SetAgentModelRequestMessage = z.infer; export type SetAgentThinkingRequestMessage = z.infer;