refactor timeline loading to cursor-based fetch API

This commit is contained in:
Mohamed Boudra
2026-02-14 01:03:36 +07:00
parent c41815dcb7
commit 8169134ac3
22 changed files with 1567 additions and 504 deletions
+155 -112
View File
@@ -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<void>)
| null
>(null);
const hasRequestedInitialSnapshotRef = useRef(false);
const hasBootstrappedAgentUpdatesRef = useRef(false);
const agentUpdatesSubscriptionIdRef = useRef<string | null>(null);
const sessionStateTimeoutRef = useRef<ReturnType<typeof setTimeout> | 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<StreamItem[]>(
(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,
@@ -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<void> => {
@@ -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 };
+38
View File
@@ -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<string, StreamItem[]>;
agentStreamHead: Map<string, StreamItem[]>;
agentTimelineCursor: Map<string, AgentTimelineCursorState>;
historySyncGeneration: number;
agentHistorySyncGeneration: Map<string, number>;
@@ -240,6 +247,12 @@ interface SessionStoreActions {
setAgentStreamTail: (serverId: string, state: Map<string, StreamItem[]> | ((prev: Map<string, StreamItem[]>) => Map<string, StreamItem[]>)) => void;
setAgentStreamHead: (serverId: string, state: Map<string, StreamItem[]> | ((prev: Map<string, StreamItem[]>) => Map<string, StreamItem[]>)) => void;
clearAgentStreamHead: (serverId: string, agentId: string) => void;
setAgentTimelineCursor: (
serverId: string,
state:
| Map<string, AgentTimelineCursorState>
| ((prev: Map<string, AgentTimelineCursorState>) => Map<string, AgentTimelineCursorState>)
) => 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<SessionStore>()(
});
},
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];
+11 -29
View File
@@ -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<void>((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
+9 -61
View File
@@ -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<AgentTimelineItem[]>((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<AgentTimelineItem[]>((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))
+33
View File
@@ -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<TimelineFetchResult>
}
type FetchProjectedTimelineItemsInput = {
client: DaemonClient
agentId: string
}
export async function fetchProjectedTimelineItems(
input: FetchProjectedTimelineItemsInput
): Promise<AgentTimelineItem[]> {
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)
}
@@ -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();
+34 -12
View File
@@ -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<string, unknown> }
| { 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<AgentSnapshotPayload> {
const resolvedRequestId = this.createRequestId(requestId);
options: FetchAgentTimelineOptions = {}
): Promise<FetchAgentTimelinePayload> {
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 };
@@ -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");
+306 -11
View File
@@ -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<string, AgentPermissionRequest>;
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<AgentSessionConfig>
): Promise<ManagedAgent> {
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<void> {
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<string, string>;
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 {
@@ -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<string, { execEnd: boolean; itemCompleted: boolean }>();
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<string, unknown>) &&
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 () => {
@@ -1751,6 +1751,8 @@ class CodexAppServerAgentSession implements AgentSession {
private pendingReasoning = new Map<string, string[]>();
private pendingCommandOutputDeltas = new Map<string, string[]>();
private pendingFileChangeOutputDeltas = new Map<string, string[]>();
private emittedExecCommandStartedCallIds = new Set<string>();
private emittedExecCommandCompletedCallIds = new Set<string>();
private emittedItemStartedIds = new Set<string>();
private emittedItemCompletedIds = new Set<string>();
private warnedUnknownNotificationMethods = new Set<string>();
@@ -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;
}
@@ -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([]);
});
});
@@ -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<AgentTimelineItem, { type: "tool_call" }>,
incoming: Extract<AgentTimelineItem, { type: "tool_call" }>
): Extract<AgentTimelineItem, { type: "tool_call" }> {
const mergedDetail = mergeToolCallDetail(existing.detail, incoming.detail);
const mergedMetadata =
existing.metadata || incoming.metadata
? { ...(existing.metadata ?? {}), ...(incoming.metadata ?? {}) }
: undefined;
const merged: Extract<AgentTimelineItem, { type: "tool_call" }> = {
...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<string, number>();
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<TimelineProjectionKind>([
...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);
}
@@ -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 });
@@ -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 });
@@ -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;
}
-1
View File
@@ -31,5 +31,4 @@ export type {
AgentSnapshotPayload,
AgentStreamEventPayload,
AgentStreamMessage,
AgentStreamSnapshotMessage,
} from "../shared/messages.js";
+103 -78
View File
@@ -60,9 +60,14 @@ export type AgentMcpTransportFactory = () => Promise<Transport>;
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<void> {
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<SessionInboundMessage, { type: "fetch_agent_timeline_request" }>
): Promise<void> {
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<SessionInboundMessage, { type: "send_agent_message_request" }>
): Promise<void> {
@@ -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
*/
@@ -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"
}
}
}
}
]
}
}
@@ -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,
+61 -35
View File
@@ -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<typeof RpcErrorMessageSchema>;
export type ArtifactMessage = z.infer<typeof ArtifactMessageSchema>;
export type AgentUpdateMessage = z.infer<typeof AgentUpdateMessageSchema>;
export type AgentStreamMessage = z.infer<typeof AgentStreamMessageSchema>;
export type AgentStreamSnapshotMessage = z.infer<
typeof AgentStreamSnapshotMessageSchema
>;
export type AgentStatusMessage = z.infer<typeof AgentStatusMessageSchema>;
export type ProjectCheckoutLitePayload = z.infer<typeof ProjectCheckoutLitePayloadSchema>;
export type ProjectPlacementPayload = z.infer<typeof ProjectPlacementPayloadSchema>;
@@ -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<typeof SpeechModelsListResponseSchema>;
export type SpeechModelsDownloadResponse = z.infer<typeof SpeechModelsDownloadResponseSchema>;
export type InitializeAgentResponseMessage = z.infer<typeof InitializeAgentResponseMessageSchema>;
// Type exports for payload types
export type ActivityLogPayload = z.infer<typeof ActivityLogPayloadSchema>;
@@ -2014,7 +2041,6 @@ export type SpeechModelsDownloadRequestMessage = z.infer<
export type ResumeAgentRequestMessage = z.infer<typeof ResumeAgentRequestMessageSchema>;
export type DeleteAgentRequestMessage = z.infer<typeof DeleteAgentRequestMessageSchema>;
export type UpdateAgentRequestMessage = z.infer<typeof UpdateAgentRequestMessageSchema>;
export type InitializeAgentRequestMessage = z.infer<typeof InitializeAgentRequestMessageSchema>;
export type SetAgentModeRequestMessage = z.infer<typeof SetAgentModeRequestMessageSchema>;
export type SetAgentModelRequestMessage = z.infer<typeof SetAgentModelRequestMessageSchema>;
export type SetAgentThinkingRequestMessage = z.infer<typeof SetAgentThinkingRequestMessageSchema>;