diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index fe339dfa8..bc51d80de 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -6,12 +6,13 @@ import { useClientActivity } from "@/hooks/use-client-activity"; import { usePushTokenRegistration } from "@/hooks/use-push-token-registration"; import { clearArchiveAgentPending } from "@/hooks/use-archive-agent"; import { - applyStreamEvent, generateMessageId, - hydrateStreamState, - reduceStreamUpdate, type StreamItem, } from "@/types/stream"; +import { + processTimelineResponse, + processAgentStreamEvent, +} from "@/contexts/session-stream-reducers"; import type { ActivityLogPayload, AgentStreamEventPayload, @@ -35,7 +36,12 @@ import { import { useDraftStore } from "@/stores/draft-store"; import type { AgentDirectoryEntry } from "@/types/agent-directory"; import { sendOsNotification } from "@/utils/os-notifications"; -import { getInitKey, getInitDeferred, resolveInitDeferred, rejectInitDeferred, createInitDeferred } from "@/utils/agent-initialization"; +import { + getInitKey, + getInitDeferred, + resolveInitDeferred, + rejectInitDeferred, +} from "@/utils/agent-initialization"; import { encodeImages } from "@/utils/encode-images"; import { derivePendingPermissionKey, @@ -114,6 +120,48 @@ type AgentUpdatePayload = Extract< const getAgentIdFromUpdate = (update: AgentUpdatePayload): string => update.kind === "remove" ? update.agentId : update.agent.id; +// --------------------------------------------------------------------------- +// Module-level pending agent updates buffer (scoped by serverId) +// --------------------------------------------------------------------------- +const pendingAgentUpdates = new Map(); + +function pendingKey(serverId: string, agentId: string): string { + return `${serverId}:${agentId}`; +} + +export function bufferPendingAgentUpdate( + serverId: string, + agentId: string, + update: AgentUpdatePayload +): void { + pendingAgentUpdates.set(pendingKey(serverId, agentId), update); +} + +export function flushPendingAgentUpdate( + serverId: string, + agentId: string +): AgentUpdatePayload | undefined { + const key = pendingKey(serverId, agentId); + const update = pendingAgentUpdates.get(key); + pendingAgentUpdates.delete(key); + return update; +} + +export function deletePendingAgentUpdate( + serverId: string, + agentId: string +): void { + pendingAgentUpdates.delete(pendingKey(serverId, agentId)); +} + +export function clearPendingAgentUpdates(serverId: string): void { + for (const key of [...pendingAgentUpdates.keys()]) { + if (key.startsWith(`${serverId}:`)) { + pendingAgentUpdates.delete(key); + } + } +} + const createExplorerState = () => ({ directories: new Map(), files: new Map(), @@ -270,9 +318,6 @@ function SessionProviderInternal({ ); const attentionNotifiedRef = useRef>(new Map()); const appStateRef = useRef(AppState.currentState); - const pendingAgentUpdatesRef = useRef>( - new Map() - ); useEffect(() => { const subscription = AppState.addEventListener("change", (nextState) => { @@ -385,7 +430,7 @@ function SessionProviderInternal({ useEffect(() => { if (!isConnected) { flushAgentLastActivity(); - pendingAgentUpdatesRef.current.clear(); + clearPendingAgentUpdates(serverId); setInitializingAgents(serverId, new Map()); } }, [flushAgentLastActivity, serverId, isConnected, setInitializingAgents]); @@ -395,7 +440,7 @@ function SessionProviderInternal({ if (update.kind === "remove") { const agentId = update.agentId; previousAgentStatusRef.current.delete(agentId); - pendingAgentUpdatesRef.current.delete(agentId); + deletePendingAgentUpdate(serverId, agentId); clearArchiveAgentPending({ queryClient, serverId, agentId }); setAgents(serverId, (prev) => { @@ -635,6 +680,26 @@ function SessionProviderInternal({ }, }); + const requestCanonicalCatchUp = useCallback( + (agentId: string, cursor: { epoch: string; endSeq: number }) => { + void client + .fetchAgentTimeline(agentId, { + direction: "after", + cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, + limit: 0, + projection: "canonical", + }) + .catch((error) => { + console.warn( + "[Session] failed to fetch canonical catch-up timeline", + agentId, + error + ); + }); + }, + [client] + ); + const applyTimelineResponse = useCallback( ( payload: Extract< @@ -645,7 +710,106 @@ function SessionProviderInternal({ const agentId = payload.agentId; const initKey = getInitKey(serverId, agentId); - if (payload.error) { + // Read current store state + const session = useSessionStore.getState().sessions[serverId]; + const isInitializing = session?.initializingAgents.get(agentId) === true; + const activeInitDeferred = getInitDeferred(initKey); + const hasActiveInitDeferred = Boolean(activeInitDeferred); + const currentCursor = session?.agentTimelineCursor.get(agentId); + const currentTail = session?.agentStreamTail.get(agentId) ?? []; + const currentHead = session?.agentStreamHead.get(agentId) ?? []; + + // Call pure reducer + const result = processTimelineResponse({ + payload, + currentTail, + currentHead, + currentCursor, + isInitializing, + hasActiveInitDeferred, + initRequestDirection: activeInitDeferred?.requestDirection ?? "tail", + }); + + // Apply error path + if (result.error) { + if (result.clearInitializing) { + setInitializingAgents(serverId, (prev) => { + if (prev.get(agentId) !== true) { + return prev; + } + const next = new Map(prev); + next.set(agentId, false); + return next; + }); + } + if (result.initResolution === "reject") { + rejectInitDeferred(initKey, new Error(result.error)); + } + return; + } + + // Apply tail patch + if (result.tail !== currentTail) { + setAgentStreamTail(serverId, (prev) => { + const next = new Map(prev); + next.set(agentId, result.tail); + return next; + }); + } + + // Apply head patch + if (result.head !== currentHead) { + if (result.head.length === 0) { + clearAgentStreamHead(serverId, agentId); + } else { + setAgentStreamHead(serverId, (prev) => { + const next = new Map(prev); + next.set(agentId, result.head); + return next; + }); + } + } + + // Apply cursor patch + if (result.cursorChanged) { + setAgentTimelineCursor(serverId, (prev) => { + const current = prev.get(agentId); + if (!result.cursor) { + if (!current) { + return prev; + } + const next = new Map(prev); + next.delete(agentId); + return next; + } + if ( + current && + current.epoch === result.cursor.epoch && + current.startSeq === result.cursor.startSeq && + current.endSeq === result.cursor.endSeq + ) { + return prev; + } + const next = new Map(prev); + next.set(agentId, result.cursor); + return next; + }); + } + + // Execute side effects + for (const effect of result.sideEffects) { + if (effect.type === "catch_up") { + requestCanonicalCatchUp(agentId, effect.cursor); + } else if (effect.type === "flush_pending_updates") { + const deferredUpdate = flushPendingAgentUpdate(serverId, agentId); + if (deferredUpdate) { + applyAgentUpdatePayload(deferredUpdate); + } + } + } + + // Apply init resolution + if (result.clearInitializing) { setInitializingAgents(serverId, (prev) => { if (prev.get(agentId) !== true) { return prev; @@ -654,80 +818,20 @@ function SessionProviderInternal({ 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; - }); + if (result.initResolution === "resolve") { + resolveInitDeferred(initKey); } - - 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); + if (result.clearInitializing) { + markAgentHistorySynchronized(serverId, agentId); } - - 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, + requestCanonicalCatchUp, serverId, setAgentStreamTail, setAgentTimelineCursor, @@ -739,8 +843,8 @@ function SessionProviderInternal({ if (isConnected) { return; } - pendingAgentUpdatesRef.current.clear(); - }, [isConnected]); + clearPendingAgentUpdates(serverId); + }, [isConnected, serverId]); // Daemon message handlers - directly update Zustand store useEffect(() => { @@ -755,11 +859,11 @@ function SessionProviderInternal({ Boolean(getInitDeferred(initKey)); if (isSyncingHistory) { - pendingAgentUpdatesRef.current.set(agentId, update); + bufferPendingAgentUpdate(serverId, agentId, update); return; } - pendingAgentUpdatesRef.current.delete(agentId); + deletePendingAgentUpdate(serverId, agentId); applyAgentUpdatePayload(update); }); @@ -767,7 +871,9 @@ function SessionProviderInternal({ if (message.type !== "agent_stream") return; const { agentId, event, timestamp, seq, epoch } = message.payload; const parsedTimestamp = new Date(timestamp); + const streamEvent = event as AgentStreamEventPayload; + // Attention notification stays in React (not extractable to pure reducer) if (event.type === "attention_required") { if (event.shouldNotify) { notifyAgentAttention({ @@ -779,50 +885,94 @@ function SessionProviderInternal({ } } + // Read current store state const session = useSessionStore.getState().sessions[serverId]; const currentTail = session?.agentStreamTail.get(agentId) ?? []; const currentHead = session?.agentStreamHead.get(agentId) ?? []; - const { tail, head, changedTail, changedHead } = applyStreamEvent({ - tail: currentTail, - head: currentHead, - event: event as AgentStreamEventPayload, + const currentCursor = session?.agentTimelineCursor.get(agentId); + const currentAgentEntry = session?.agents.get(agentId); + const currentAgent = currentAgentEntry + ? { + status: currentAgentEntry.status, + updatedAt: currentAgentEntry.updatedAt, + lastActivityAt: currentAgentEntry.lastActivityAt, + } + : null; + + // Call pure reducer + const result = processAgentStreamEvent({ + event: streamEvent, + seq, + epoch, + currentTail, + currentHead, + currentCursor, + currentAgent, timestamp: parsedTimestamp, }); - if (changedTail || changedHead) { + // Apply tail/head patches + if (result.changedTail || result.changedHead) { setAgentStreamState(serverId, agentId, { - ...(changedTail ? { tail } : {}), - ...(changedHead ? { head } : {}), + ...(result.changedTail ? { tail: result.tail } : {}), + ...(result.changedHead ? { head: result.head } : {}), }); } - if ( - event.type === "timeline" && - typeof seq === "number" && - typeof epoch === "string" - ) { + // Apply cursor patch + if (result.cursorChanged && result.cursor) { setAgentTimelineCursor(serverId, (prev) => { const current = prev.get(agentId); - if (current && current.epoch === epoch) { - // Fast-path: seq only extends the range during streaming. - // Skip the Map copy when nothing actually changes. - if (seq >= current.startSeq && seq <= current.endSeq) { - return prev; - } - const next = new Map(prev); - next.set(agentId, { - epoch, - startSeq: Math.min(current.startSeq, seq), - endSeq: Math.max(current.endSeq, seq), - }); - return next; + if ( + current && + typeof seq === "number" && + typeof epoch === "string" && + current.epoch === epoch && + seq >= current.startSeq && + seq <= current.endSeq + ) { + // Fast-path: seq stays inside the current range during streaming. + return prev; + } + if ( + current && + current.epoch === result.cursor.epoch && + current.startSeq === result.cursor.startSeq && + current.endSeq === result.cursor.endSeq + ) { + return prev; } const next = new Map(prev); - next.set(agentId, { epoch, startSeq: seq, endSeq: seq }); + next.set(agentId, result.cursor); return next; }); } + // Apply agent patch (optimistic lifecycle) + if (result.agentChanged && result.agent) { + setAgents(serverId, (prev) => { + const current = prev.get(agentId); + if (!current) { + return prev; + } + const next = new Map(prev); + next.set(agentId, { + ...current, + status: result.agent.status, + updatedAt: result.agent.updatedAt, + lastActivityAt: result.agent.lastActivityAt, + }); + return next; + }); + } + + // Execute side effects + for (const effect of result.sideEffects) { + if (effect.type === "catch_up") { + requestCanonicalCatchUp(agentId, effect.cursor); + } + } + // 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. @@ -1136,7 +1286,7 @@ function SessionProviderInternal({ } const { agentId } = message.payload; console.log("[Session] Agent deleted:", agentId); - pendingAgentUpdatesRef.current.delete(agentId); + deletePendingAgentUpdate(serverId, agentId); clearArchiveAgentPending({ queryClient, serverId, agentId }); setAgents(serverId, (prev) => { @@ -1272,6 +1422,7 @@ function SessionProviderInternal({ setHasHydratedAgents, clearDraftInput, notifyAgentAttention, + requestCanonicalCatchUp, applyAgentUpdatePayload, applyTimelineResponse, ]); diff --git a/packages/app/src/contexts/session-stream-lifecycle.test.ts b/packages/app/src/contexts/session-stream-lifecycle.test.ts new file mode 100644 index 000000000..f00a2d4b0 --- /dev/null +++ b/packages/app/src/contexts/session-stream-lifecycle.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import type { AgentStreamEventPayload } from "@server/shared/messages"; +import { deriveOptimisticLifecycleStatus } from "./session-stream-lifecycle"; + +const turnCompletedEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", +}; + +const turnFailedEvent: AgentStreamEventPayload = { + type: "turn_failed", + provider: "claude", + error: "failed", +}; + +describe("session stream lifecycle helpers", () => { + it("derives optimistic terminal lifecycle only when current status is running", () => { + expect(deriveOptimisticLifecycleStatus("running", turnCompletedEvent)).toBe( + "idle" + ); + expect(deriveOptimisticLifecycleStatus("running", turnFailedEvent)).toBe( + "error" + ); + expect(deriveOptimisticLifecycleStatus("initializing", turnCompletedEvent)).toBe( + null + ); + expect(deriveOptimisticLifecycleStatus("idle", turnFailedEvent)).toBe(null); + }); +}); diff --git a/packages/app/src/contexts/session-stream-lifecycle.ts b/packages/app/src/contexts/session-stream-lifecycle.ts new file mode 100644 index 000000000..e8c9102bd --- /dev/null +++ b/packages/app/src/contexts/session-stream-lifecycle.ts @@ -0,0 +1,20 @@ +import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; +import type { AgentStreamEventPayload } from "@server/shared/messages"; + +export function deriveOptimisticLifecycleStatus( + currentStatus: AgentLifecycleStatus, + event: AgentStreamEventPayload +): AgentLifecycleStatus | null { + if (currentStatus !== "running") { + return null; + } + switch (event.type) { + case "turn_completed": + case "turn_canceled": + return "idle"; + case "turn_failed": + return "error"; + default: + return null; + } +} diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/contexts/session-stream-reducers.test.ts new file mode 100644 index 000000000..07e23ea31 --- /dev/null +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -0,0 +1,673 @@ +import { describe, expect, it } from "vitest"; +import type { AgentStreamEventPayload } from "@server/shared/messages"; +import type { StreamItem } from "@/types/stream"; +import { + processTimelineResponse, + processAgentStreamEvent, + type ProcessTimelineResponseInput, + type ProcessAgentStreamEventInput, + type TimelineCursor, +} from "./session-stream-reducers"; + +// --------------------------------------------------------------------------- +// Test helpers +// --------------------------------------------------------------------------- + +function makeTimelineEntry( + seq: number, + text: string, + type: string = "assistant_message" +) { + return { + seqStart: seq, + provider: "claude", + item: { type, text }, + timestamp: new Date(1000 + seq).toISOString(), + }; +} + +function makeTimelineEvent( + text: string, + type: string = "assistant_message" +): AgentStreamEventPayload { + return { + type: "timeline", + provider: "claude", + item: { type, text }, + } as AgentStreamEventPayload; +} + +function makeUserTimelineEvent(text: string): AgentStreamEventPayload { + return { + type: "timeline", + provider: "claude", + item: { type: "user_message", text }, + } as AgentStreamEventPayload; +} + +const baseTimelineInput: ProcessTimelineResponseInput = { + payload: { + agentId: "agent-1", + direction: "after", + reset: false, + epoch: "epoch-1", + startCursor: null, + endCursor: null, + entries: [], + error: null, + }, + currentTail: [], + currentHead: [], + currentCursor: undefined, + isInitializing: false, + hasActiveInitDeferred: false, + initRequestDirection: "tail", +}; + +const baseStreamInput: ProcessAgentStreamEventInput = { + event: makeTimelineEvent("hello"), + seq: undefined, + epoch: undefined, + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + timestamp: new Date(2000), +}; + +// --------------------------------------------------------------------------- +// processTimelineResponse +// --------------------------------------------------------------------------- + +describe("processTimelineResponse", () => { + it("returns error path when payload.error is set", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: true, + payload: { + ...baseTimelineInput.payload, + error: "something broke", + }, + }); + + expect(result.error).toBe("something broke"); + expect(result.initResolution).toBe("reject"); + expect(result.clearInitializing).toBe(true); + expect(result.tail).toBe(baseTimelineInput.currentTail); + expect(result.head).toBe(baseTimelineInput.currentHead); + expect(result.cursorChanged).toBe(false); + expect(result.sideEffects).toEqual([]); + }); + + it("returns error with no init resolution when no deferred exists", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: false, + payload: { + ...baseTimelineInput.payload, + error: "timeout", + }, + }); + + expect(result.error).toBe("timeout"); + expect(result.initResolution).toBe(null); + expect(result.clearInitializing).toBe(true); + }); + + it("replaces tail and clears head when reset=true", () => { + const existingTail: StreamItem[] = [ + { + kind: "user_message", + id: "old", + text: "old message", + timestamp: new Date(500), + }, + ]; + const existingHead: StreamItem[] = [ + { + kind: "assistant_message", + id: "head-1", + text: "streaming", + timestamp: new Date(600), + }, + ]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: existingTail, + currentHead: existingHead, + payload: { + ...baseTimelineInput.payload, + reset: true, + startCursor: { seq: 1 }, + endCursor: { seq: 3 }, + entries: [ + makeTimelineEntry(1, "first"), + makeTimelineEntry(2, "second"), + makeTimelineEntry(3, "third"), + ], + }, + }); + + expect(result.tail).not.toBe(existingTail); + expect(result.tail.length).toBeGreaterThan(0); + expect(result.head).toEqual([]); + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }); + expect(result.error).toBe(null); + expect( + result.sideEffects.some((e) => e.type === "flush_pending_updates") + ).toBe(true); + }); + + it("sets cursor to null when reset=true but no cursors in payload", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 5 }, + payload: { + ...baseTimelineInput.payload, + reset: true, + entries: [], + }, + }); + + expect(result.cursor).toBe(null); + expect(result.cursorChanged).toBe(true); + }); + + it("performs bootstrap tail init with catch-up side effect", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: true, + initRequestDirection: "tail", + payload: { + ...baseTimelineInput.payload, + direction: "tail", + epoch: "epoch-1", + startCursor: { seq: 1 }, + endCursor: { seq: 5 }, + entries: [ + makeTimelineEntry(1, "first"), + makeTimelineEntry(5, "last"), + ], + }, + }); + + // Bootstrap tail replaces + expect(result.tail.length).toBeGreaterThan(0); + expect(result.head).toEqual([]); + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }); + + // Should have catch-up side effect + const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); + expect(catchUp).toBeDefined(); + expect(catchUp!.type === "catch_up" && catchUp!.cursor).toEqual({ + epoch: "epoch-1", + endSeq: 5, + }); + }); + + it("appends incrementally for contiguous seqs", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [ + makeTimelineEntry(4, "next-1"), + makeTimelineEntry(5, "next-2"), + ], + }, + }); + + expect(result.tail.length).toBeGreaterThan(0); + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }); + expect(result.error).toBe(null); + }); + + it("detects gap and emits catch-up side effect", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [ + makeTimelineEntry(10, "far ahead"), + ], + }, + }); + + // Gap should trigger catch-up + const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); + expect(catchUp).toBeDefined(); + expect(catchUp!.type === "catch_up" && catchUp!.cursor).toEqual({ + epoch: "epoch-1", + endSeq: 3, + }); + }); + + it("drops stale entries silently", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 8, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [ + makeTimelineEntry(5, "old"), + makeTimelineEntry(7, "also old"), + ], + }, + }); + + // No new items appended (all dropped as stale) + expect(result.tail).toBe(baseTimelineInput.currentTail); + expect(result.cursorChanged).toBe(false); + }); + + it("drops entries with epoch mismatch", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-2", + entries: [ + makeTimelineEntry(6, "different epoch"), + ], + }, + }); + + expect(result.tail).toBe(baseTimelineInput.currentTail); + expect(result.cursorChanged).toBe(false); + }); + + it("resolves init when deferred matches direction", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: true, + initRequestDirection: "after", + payload: { + ...baseTimelineInput.payload, + direction: "after", + entries: [], + }, + }); + + expect(result.initResolution).toBe("resolve"); + expect(result.clearInitializing).toBe(true); + }); + + it("does not resolve init when directions differ (before vs after)", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: true, + initRequestDirection: "after", + payload: { + ...baseTimelineInput.payload, + direction: "before", + entries: [], + }, + }); + + // "before" direction doesn't match "after" initRequestDirection, + // and "before" is not a bootstrap tail path, so init should NOT resolve + expect(result.initResolution).toBe(null); + expect(result.clearInitializing).toBe(false); + }); + + it("clears initializing even without deferred", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: false, + payload: { + ...baseTimelineInput.payload, + direction: "after", + entries: [], + }, + }); + + expect(result.clearInitializing).toBe(true); + expect(result.initResolution).toBe(null); + }); + + it("always includes flush_pending_updates side effect on success", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + payload: { + ...baseTimelineInput.payload, + entries: [], + }, + }); + + expect( + result.sideEffects.some((e) => e.type === "flush_pending_updates") + ).toBe(true); + }); + + it("initializes cursor when no existing cursor on first entries", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: undefined, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [ + makeTimelineEntry(1, "first"), + makeTimelineEntry(2, "second"), + ], + }, + }); + + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 2, + }); + }); +}); + +// --------------------------------------------------------------------------- +// processAgentStreamEvent +// --------------------------------------------------------------------------- + +describe("processAgentStreamEvent", () => { + it("passes through non-timeline events without cursor changes", () => { + const turnEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnEvent, + seq: undefined, + epoch: undefined, + }); + + expect(result.cursorChanged).toBe(false); + expect(result.cursor).toBe(null); + expect(result.sideEffects).toEqual([]); + }); + + it("accepts timeline event with cursor advance", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 4, + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("new chunk"), + seq: 5, + epoch: "epoch-1", + currentCursor: existingCursor, + }); + + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }); + expect(result.sideEffects).toEqual([]); + }); + + it("detects gap and emits catch-up side effect", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 4, + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("far ahead"), + seq: 10, + epoch: "epoch-1", + currentCursor: existingCursor, + }); + + expect(result.cursorChanged).toBe(false); + expect(result.changedTail).toBe(false); + expect(result.changedHead).toBe(false); + + const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); + expect(catchUp).toBeDefined(); + expect(catchUp!.cursor).toEqual({ + epoch: "epoch-1", + endSeq: 4, + }); + }); + + it("drops stale timeline event", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 8, + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("old"), + seq: 5, + epoch: "epoch-1", + currentCursor: existingCursor, + }); + + expect(result.cursorChanged).toBe(false); + expect(result.changedTail).toBe(false); + expect(result.changedHead).toBe(false); + expect(result.sideEffects).toEqual([]); + }); + + it("drops timeline event with epoch mismatch", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("wrong epoch"), + seq: 6, + epoch: "epoch-2", + currentCursor: existingCursor, + }); + + expect(result.cursorChanged).toBe(false); + expect(result.changedTail).toBe(false); + expect(result.changedHead).toBe(false); + expect(result.sideEffects).toEqual([]); + }); + + it("initializes cursor when none exists", () => { + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("first"), + seq: 1, + epoch: "epoch-1", + currentCursor: undefined, + }); + + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 1, + }); + }); + + it("derives optimistic idle status on turn_completed for running agent", () => { + const turnCompletedEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnCompletedEvent, + currentAgent: { + status: "running", + updatedAt: new Date(1000), + lastActivityAt: new Date(1000), + }, + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(true); + expect(result.agent).not.toBe(null); + expect(result.agent!.status).toBe("idle"); + expect(result.agent!.updatedAt.getTime()).toBe(2000); + expect(result.agent!.lastActivityAt.getTime()).toBe(2000); + }); + + it("derives optimistic error status on turn_failed for running agent", () => { + const turnFailedEvent: AgentStreamEventPayload = { + type: "turn_failed", + provider: "claude", + error: "something broke", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnFailedEvent, + currentAgent: { + status: "running", + updatedAt: new Date(1000), + lastActivityAt: new Date(1000), + }, + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(true); + expect(result.agent!.status).toBe("error"); + }); + + it("does not change agent when status is not running", () => { + const turnCompletedEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnCompletedEvent, + currentAgent: { + status: "idle", + updatedAt: new Date(1000), + lastActivityAt: new Date(1000), + }, + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(false); + expect(result.agent).toBe(null); + }); + + it("does not change agent when no agent is provided", () => { + const turnCompletedEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnCompletedEvent, + currentAgent: null, + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(false); + expect(result.agent).toBe(null); + }); + + it("preserves updatedAt when agent timestamp is newer than event", () => { + const turnCompletedEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", + }; + + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: turnCompletedEvent, + currentAgent: { + status: "running", + updatedAt: new Date(5000), + lastActivityAt: new Date(5000), + }, + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(true); + expect(result.agent!.updatedAt.getTime()).toBe(5000); + expect(result.agent!.lastActivityAt.getTime()).toBe(5000); + }); + + it("does not produce agent patch for non-terminal events", () => { + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: makeTimelineEvent("just text"), + currentAgent: { + status: "running", + updatedAt: new Date(1000), + lastActivityAt: new Date(1000), + }, + seq: 1, + epoch: "epoch-1", + timestamp: new Date(2000), + }); + + expect(result.agentChanged).toBe(false); + expect(result.agent).toBe(null); + }); +}); diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts new file mode 100644 index 000000000..fab26012d --- /dev/null +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -0,0 +1,441 @@ +import type { AgentStreamEventPayload } from "@server/shared/messages"; +import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; +import type { StreamItem } from "@/types/stream"; +import { + applyStreamEvent, + hydrateStreamState, + reduceStreamUpdate, +} from "@/types/stream"; +import { + classifySessionTimelineSeq, + type SessionTimelineSeqDecision, +} from "@/contexts/session-timeline-seq-gate"; +import { + deriveBootstrapTailTimelinePolicy, + shouldResolveTimelineInit, +} from "@/contexts/session-timeline-bootstrap-policy"; +import { deriveOptimisticLifecycleStatus } from "@/contexts/session-stream-lifecycle"; + +// --------------------------------------------------------------------------- +// Shared cursor type +// --------------------------------------------------------------------------- + +export type TimelineCursor = { + epoch: string; + startSeq: number; + endSeq: number; +}; + +// --------------------------------------------------------------------------- +// Side-effect discriminated unions +// --------------------------------------------------------------------------- + +export type TimelineReducerSideEffect = + | { type: "catch_up"; cursor: { epoch: string; endSeq: number } } + | { type: "flush_pending_updates" }; + +export type AgentStreamReducerSideEffect = { + type: "catch_up"; + cursor: { epoch: string; endSeq: number }; +}; + +// --------------------------------------------------------------------------- +// processTimelineResponse +// --------------------------------------------------------------------------- + +type TimelineDirection = "tail" | "before" | "after"; +type InitRequestDirection = "tail" | "after"; + +type TimelineResponseEntry = { + seqStart: number; + provider: string; + item: Record; + timestamp: string; +}; + +export interface ProcessTimelineResponseInput { + payload: { + agentId: string; + direction: TimelineDirection; + reset: boolean; + epoch: string; + startCursor: { seq: number } | null; + endCursor: { seq: number } | null; + entries: TimelineResponseEntry[]; + error: string | null; + }; + currentTail: StreamItem[]; + currentHead: StreamItem[]; + currentCursor: TimelineCursor | undefined; + isInitializing: boolean; + hasActiveInitDeferred: boolean; + initRequestDirection: InitRequestDirection; +} + +export interface ProcessTimelineResponseOutput { + tail: StreamItem[]; + head: StreamItem[]; + cursor: TimelineCursor | null | undefined; + cursorChanged: boolean; + initResolution: "resolve" | "reject" | null; + clearInitializing: boolean; + error: string | null; + sideEffects: TimelineReducerSideEffect[]; +} + +export function processTimelineResponse( + input: ProcessTimelineResponseInput +): ProcessTimelineResponseOutput { + const { + payload, + currentTail, + currentHead, + currentCursor, + isInitializing, + hasActiveInitDeferred, + initRequestDirection, + } = input; + + // ------------------------------------------------------------------ + // Error path: reject init and leave stream state unchanged + // ------------------------------------------------------------------ + if (payload.error) { + return { + tail: currentTail, + head: currentHead, + cursor: currentCursor, + cursorChanged: false, + initResolution: hasActiveInitDeferred ? "reject" : null, + clearInitializing: isInitializing, + error: payload.error, + sideEffects: [], + }; + } + + // ------------------------------------------------------------------ + // Convert entries to timeline units + // ------------------------------------------------------------------ + const timelineUnits = payload.entries.map((entry) => ({ + seq: entry.seqStart, + event: { + type: "timeline", + provider: entry.provider, + item: entry.item, + } as AgentStreamEventPayload, + timestamp: new Date(entry.timestamp), + })); + + const toHydratedEvents = ( + units: typeof timelineUnits + ): Array<{ event: AgentStreamEventPayload; timestamp: Date }> => + units.map(({ event, timestamp }) => ({ event, timestamp })); + + // ------------------------------------------------------------------ + // Derive bootstrap policy (replace vs incremental) + // ------------------------------------------------------------------ + const bootstrapPolicy = deriveBootstrapTailTimelinePolicy({ + direction: payload.direction, + reset: payload.reset, + epoch: payload.epoch, + endCursor: payload.endCursor, + isInitializing, + hasActiveInitDeferred, + }); + const replace = bootstrapPolicy.replace; + + let nextTail = currentTail; + let nextHead = currentHead; + let nextCursor: TimelineCursor | null | undefined = currentCursor; + let cursorChanged = false; + const sideEffects: TimelineReducerSideEffect[] = []; + + if (replace) { + // ---------------------------------------------------------------- + // Replace path: full hydration from scratch + // ---------------------------------------------------------------- + nextTail = hydrateStreamState(toHydratedEvents(timelineUnits)); + nextHead = []; + + if (payload.startCursor && payload.endCursor) { + nextCursor = { + epoch: payload.epoch, + startSeq: payload.startCursor.seq, + endSeq: payload.endCursor.seq, + }; + cursorChanged = true; + } else { + nextCursor = null; + cursorChanged = true; + } + + if (bootstrapPolicy.catchUpCursor) { + sideEffects.push({ + type: "catch_up", + cursor: bootstrapPolicy.catchUpCursor, + }); + } + } else if (timelineUnits.length > 0) { + // ---------------------------------------------------------------- + // Incremental append path + // ---------------------------------------------------------------- + const acceptedUnits: typeof timelineUnits = []; + let cursor = currentCursor; + let gapCursor: { epoch: string; endSeq: number } | null = null; + + for (const unit of timelineUnits) { + const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({ + cursor: cursor + ? { epoch: cursor.epoch, endSeq: cursor.endSeq } + : null, + epoch: payload.epoch, + seq: unit.seq, + }); + + if (decision === "gap") { + gapCursor = cursor + ? { epoch: cursor.epoch, endSeq: cursor.endSeq } + : null; + break; + } + if (decision === "drop_stale" || decision === "drop_epoch") { + continue; + } + + acceptedUnits.push(unit); + if (decision === "init") { + cursor = { + epoch: payload.epoch, + startSeq: unit.seq, + endSeq: unit.seq, + }; + continue; + } + if (!cursor) { + continue; + } + cursor = { + ...cursor, + endSeq: unit.seq, + }; + } + + if (acceptedUnits.length > 0) { + nextTail = acceptedUnits.reduce( + (state, { event, timestamp }) => + reduceStreamUpdate(state, event, timestamp), + currentTail + ); + } + + if ( + cursor && + (!currentCursor || + currentCursor.epoch !== cursor.epoch || + currentCursor.startSeq !== cursor.startSeq || + currentCursor.endSeq !== cursor.endSeq) + ) { + nextCursor = cursor; + cursorChanged = true; + } + + if (gapCursor) { + sideEffects.push({ type: "catch_up", cursor: gapCursor }); + } + } + + // ------------------------------------------------------------------ + // Flush pending agent updates side effect + // ------------------------------------------------------------------ + sideEffects.push({ type: "flush_pending_updates" }); + + // ------------------------------------------------------------------ + // Init resolution + // ------------------------------------------------------------------ + const shouldResolveDeferredInit = shouldResolveTimelineInit({ + hasActiveInitDeferred, + isInitializing, + initRequestDirection, + responseDirection: payload.direction, + reset: payload.reset, + }); + const clearInitializing = + shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred); + + const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit + ? "resolve" + : null; + + return { + tail: nextTail, + head: nextHead, + cursor: nextCursor, + cursorChanged, + initResolution, + clearInitializing, + error: null, + sideEffects, + }; +} + +// --------------------------------------------------------------------------- +// processAgentStreamEvent +// --------------------------------------------------------------------------- + +export interface ProcessAgentStreamEventInput { + event: AgentStreamEventPayload; + seq: number | undefined; + epoch: string | undefined; + currentTail: StreamItem[]; + currentHead: StreamItem[]; + currentCursor: TimelineCursor | undefined; + currentAgent: { + status: AgentLifecycleStatus; + updatedAt: Date; + lastActivityAt: Date; + } | null; + timestamp: Date; +} + +export interface AgentPatch { + status: AgentLifecycleStatus; + updatedAt: Date; + lastActivityAt: Date; +} + +export interface ProcessAgentStreamEventOutput { + tail: StreamItem[]; + head: StreamItem[]; + changedTail: boolean; + changedHead: boolean; + cursor: TimelineCursor | null; + cursorChanged: boolean; + agent: AgentPatch | null; + agentChanged: boolean; + sideEffects: AgentStreamReducerSideEffect[]; +} + +export function processAgentStreamEvent( + input: ProcessAgentStreamEventInput +): ProcessAgentStreamEventOutput { + const { + event, + seq, + epoch, + currentTail, + currentHead, + currentCursor, + currentAgent, + timestamp, + } = input; + + let shouldApplyStreamEvent = true; + let nextTimelineCursor: TimelineCursor | null = null; + let cursorChanged = false; + const sideEffects: AgentStreamReducerSideEffect[] = []; + + // ------------------------------------------------------------------ + // Timeline sequencing gate + // ------------------------------------------------------------------ + if ( + event.type === "timeline" && + typeof seq === "number" && + typeof epoch === "string" + ) { + const decision = classifySessionTimelineSeq({ + cursor: currentCursor + ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } + : null, + epoch, + seq, + }); + + if (decision === "init") { + nextTimelineCursor = { epoch, startSeq: seq, endSeq: seq }; + cursorChanged = true; + } else if (decision === "accept") { + nextTimelineCursor = { + ...(currentCursor ?? { epoch, startSeq: seq, endSeq: seq }), + epoch, + endSeq: seq, + }; + cursorChanged = true; + } else if (decision === "gap") { + shouldApplyStreamEvent = false; + if (currentCursor) { + sideEffects.push({ + type: "catch_up", + cursor: { + epoch: currentCursor.epoch, + endSeq: currentCursor.endSeq, + }, + }); + } + } else { + // drop_stale or drop_epoch + shouldApplyStreamEvent = false; + } + } + + // ------------------------------------------------------------------ + // Apply stream event to tail/head + // ------------------------------------------------------------------ + const { tail, head, changedTail, changedHead } = shouldApplyStreamEvent + ? applyStreamEvent({ + tail: currentTail, + head: currentHead, + event, + timestamp, + }) + : { + tail: currentTail, + head: currentHead, + changedTail: false, + changedHead: false, + }; + + // ------------------------------------------------------------------ + // Optimistic lifecycle status + // ------------------------------------------------------------------ + let agentPatch: AgentPatch | null = null; + let agentChanged = false; + + if ( + currentAgent && + (event.type === "turn_completed" || + event.type === "turn_canceled" || + event.type === "turn_failed") + ) { + const optimisticStatus = deriveOptimisticLifecycleStatus( + currentAgent.status, + event + ); + if (optimisticStatus) { + const nextUpdatedAtMs = Math.max( + currentAgent.updatedAt.getTime(), + timestamp.getTime() + ); + const nextLastActivityAtMs = Math.max( + currentAgent.lastActivityAt.getTime(), + timestamp.getTime() + ); + agentPatch = { + status: optimisticStatus, + updatedAt: new Date(nextUpdatedAtMs), + lastActivityAt: new Date(nextLastActivityAtMs), + }; + agentChanged = true; + } + } + + return { + tail, + head, + changedTail, + changedHead, + cursor: nextTimelineCursor, + cursorChanged, + agent: agentPatch, + agentChanged, + sideEffects, + }; +} diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts new file mode 100644 index 000000000..ea2706a07 --- /dev/null +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it } from "vitest"; +import { classifySessionTimelineSeq } from "./session-timeline-seq-gate"; +import { + deriveBootstrapTailTimelinePolicy, + shouldResolveTimelineInit, +} from "./session-timeline-bootstrap-policy"; + +describe("deriveBootstrapTailTimelinePolicy", () => { + it("always replaces on explicit reset without catch-up cursor", () => { + const policy = deriveBootstrapTailTimelinePolicy({ + direction: "after", + reset: true, + epoch: "epoch-1", + endCursor: { seq: 200 }, + isInitializing: false, + hasActiveInitDeferred: false, + }); + + expect(policy.replace).toBe(true); + expect(policy.catchUpCursor).toBeNull(); + }); + + it("forces baseline replace and canonical catch-up for init tail race", () => { + const advancedCursor = { epoch: "epoch-1", endSeq: 205 }; + const tailSeqStart = 101; + const tailSeqEnd = 200; + + let acceptedWithoutBootstrap = 0; + for (let seq = tailSeqStart; seq <= tailSeqEnd; seq += 1) { + const decision = classifySessionTimelineSeq({ + cursor: advancedCursor, + epoch: "epoch-1", + seq, + }); + if (decision === "accept" || decision === "init") { + acceptedWithoutBootstrap += 1; + } + } + expect(acceptedWithoutBootstrap).toBe(0); + + const policy = deriveBootstrapTailTimelinePolicy({ + direction: "tail", + reset: false, + epoch: "epoch-1", + endCursor: { seq: 200 }, + isInitializing: true, + hasActiveInitDeferred: true, + }); + + expect(policy.replace).toBe(true); + expect(policy.catchUpCursor).toEqual({ + epoch: "epoch-1", + endSeq: 200, + }); + }); + + it("does not replace non-bootstrap, non-reset responses", () => { + const policy = deriveBootstrapTailTimelinePolicy({ + direction: "tail", + reset: false, + epoch: "epoch-1", + endCursor: { seq: 200 }, + isInitializing: false, + hasActiveInitDeferred: false, + }); + + expect(policy.replace).toBe(false); + expect(policy.catchUpCursor).toBeNull(); + }); +}); + +describe("shouldResolveTimelineInit", () => { + it("resolves tail init when the tail response arrives", () => { + expect( + shouldResolveTimelineInit({ + hasActiveInitDeferred: true, + isInitializing: true, + initRequestDirection: "tail", + responseDirection: "tail", + reset: false, + }) + ).toBe(true); + }); + + it("does not resolve tail init when an after response arrives first", () => { + expect( + shouldResolveTimelineInit({ + hasActiveInitDeferred: true, + isInitializing: true, + initRequestDirection: "tail", + responseDirection: "after", + reset: false, + }) + ).toBe(false); + }); + + it("resolves after init when an after response arrives", () => { + expect( + shouldResolveTimelineInit({ + hasActiveInitDeferred: true, + isInitializing: true, + initRequestDirection: "after", + responseDirection: "after", + reset: false, + }) + ).toBe(true); + }); +}); diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts new file mode 100644 index 000000000..a4b5b0aca --- /dev/null +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts @@ -0,0 +1,62 @@ +type TimelineDirection = "tail" | "before" | "after"; +type InitRequestDirection = "tail" | "after"; + +type TimelineCursor = { + seq: number; +} | null; + +export function deriveBootstrapTailTimelinePolicy({ + direction, + reset, + epoch, + endCursor, + isInitializing, + hasActiveInitDeferred, +}: { + direction: TimelineDirection; + reset: boolean; + epoch: string; + endCursor: TimelineCursor; + isInitializing: boolean; + hasActiveInitDeferred: boolean; +}): { + replace: boolean; + catchUpCursor: { epoch: string; endSeq: number } | null; +} { + if (reset) { + return { replace: true, catchUpCursor: null }; + } + + const isBootstrapTailInit = + direction === "tail" && isInitializing && hasActiveInitDeferred; + if (!isBootstrapTailInit) { + return { replace: false, catchUpCursor: null }; + } + + return { + replace: true, + catchUpCursor: endCursor ? { epoch, endSeq: endCursor.seq } : null, + }; +} + +export function shouldResolveTimelineInit({ + hasActiveInitDeferred, + isInitializing, + initRequestDirection, + responseDirection, + reset, +}: { + hasActiveInitDeferred: boolean; + isInitializing: boolean; + initRequestDirection: InitRequestDirection; + responseDirection: TimelineDirection; + reset: boolean; +}): boolean { + if (!hasActiveInitDeferred || !isInitializing) { + return false; + } + if (reset) { + return true; + } + return responseDirection === initRequestDirection; +} diff --git a/packages/app/src/contexts/session-timeline-seq-gate.test.ts b/packages/app/src/contexts/session-timeline-seq-gate.test.ts new file mode 100644 index 000000000..c66c79740 --- /dev/null +++ b/packages/app/src/contexts/session-timeline-seq-gate.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "vitest"; +import { classifySessionTimelineSeq } from "./session-timeline-seq-gate"; + +describe("classifySessionTimelineSeq", () => { + it("accepts contiguous forward seq", () => { + expect( + classifySessionTimelineSeq({ + cursor: { epoch: "epoch-1", endSeq: 4 }, + epoch: "epoch-1", + seq: 5, + }) + ).toBe("accept"); + }); + + it("drops stale seq older than the current end", () => { + expect( + classifySessionTimelineSeq({ + cursor: { epoch: "epoch-1", endSeq: 8 }, + epoch: "epoch-1", + seq: 7, + }) + ).toBe("drop_stale"); + }); + + it("drops duplicate replay seq equal to the current end", () => { + expect( + classifySessionTimelineSeq({ + cursor: { epoch: "epoch-1", endSeq: 8 }, + epoch: "epoch-1", + seq: 8, + }) + ).toBe("drop_stale"); + }); + + it("drops epoch mismatch", () => { + expect( + classifySessionTimelineSeq({ + cursor: { epoch: "epoch-1", endSeq: 4 }, + epoch: "epoch-2", + seq: 5, + }) + ).toBe("drop_epoch"); + }); + + it("initializes when cursor is null", () => { + expect( + classifySessionTimelineSeq({ + cursor: null, + epoch: "epoch-1", + seq: 1, + }) + ).toBe("init"); + }); + + it("classifies forward gaps", () => { + expect( + classifySessionTimelineSeq({ + cursor: { epoch: "epoch-1", endSeq: 4 }, + epoch: "epoch-1", + seq: 9, + }) + ).toBe("gap"); + }); +}); diff --git a/packages/app/src/contexts/session-timeline-seq-gate.ts b/packages/app/src/contexts/session-timeline-seq-gate.ts new file mode 100644 index 000000000..a2e56ec39 --- /dev/null +++ b/packages/app/src/contexts/session-timeline-seq-gate.ts @@ -0,0 +1,38 @@ +export type SessionTimelineSeqCursor = + | { + epoch: string; + endSeq: number; + } + | null + | undefined; + +export type SessionTimelineSeqDecision = + | "accept" + | "drop_stale" + | "drop_epoch" + | "gap" + | "init"; + +export function classifySessionTimelineSeq({ + cursor, + epoch, + seq, +}: { + cursor: SessionTimelineSeqCursor; + epoch: string; + seq: number; +}): SessionTimelineSeqDecision { + if (!cursor) { + return "init"; + } + if (cursor.epoch !== epoch) { + return "drop_epoch"; + } + if (seq <= cursor.endSeq) { + return "drop_stale"; + } + if (seq === cursor.endSeq + 1) { + return "accept"; + } + return "gap"; +} diff --git a/packages/app/src/hooks/use-agent-initialization.test.ts b/packages/app/src/hooks/use-agent-initialization.test.ts new file mode 100644 index 000000000..d1280155f --- /dev/null +++ b/packages/app/src/hooks/use-agent-initialization.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import { __private__ } from "./use-agent-initialization"; + +describe("useAgentInitialization timeline request policy", () => { + it("uses canonical tail bootstrap when cursor is missing", () => { + expect(__private__.buildInitialTimelineRequest(undefined)).toEqual({ + direction: "tail", + limit: 200, + projection: "canonical", + }); + }); + + it("uses canonical catch-up after the current cursor when present", () => { + expect( + __private__.buildInitialTimelineRequest({ + epoch: "epoch-1", + endSeq: 42, + }) + ).toEqual({ + direction: "after", + cursor: { epoch: "epoch-1", seq: 42 }, + limit: 0, + projection: "canonical", + }); + }); +}); diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index 61c0e2da1..639d47ce3 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -27,7 +27,7 @@ function buildInitialTimelineRequest( return { direction: "tail", limit: DEFAULT_INITIAL_TIMELINE_LIMIT, - projection: "projected", + projection: "canonical", }; } @@ -36,10 +36,14 @@ function buildInitialTimelineRequest( cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, // Catch up all missing canonical rows by default. limit: 0, - projection: "projected", + projection: "canonical", }; } +export const __private__ = { + buildInitialTimelineRequest, +}; + export function useAgentInitialization({ serverId, client, @@ -70,7 +74,13 @@ export function useAgentInitialization({ return existing.promise; } - const deferred = createInitDeferred(key); + const session = useSessionStore.getState().sessions[serverId]; + const cursor = session?.agentTimelineCursor.get(agentId); + const timelineRequest = buildInitialTimelineRequest(cursor); + const initRequestDirection = + timelineRequest.direction === "after" ? "after" : "tail"; + + const deferred = createInitDeferred(key, initRequestDirection); const timeoutId = setTimeout(() => { setAgentInitializing(agentId, false); rejectInitDeferred( @@ -90,10 +100,6 @@ export function useAgentInitialization({ return deferred.promise; } - const session = useSessionStore.getState().sessions[serverId]; - const cursor = session?.agentTimelineCursor.get(agentId); - const timelineRequest = buildInitialTimelineRequest(cursor); - client .fetchAgentTimeline(agentId, timelineRequest) .then(() => { @@ -125,7 +131,7 @@ export function useAgentInitialization({ await client.fetchAgentTimeline(agentId, { direction: "tail", limit: DEFAULT_INITIAL_TIMELINE_LIMIT, - projection: "projected", + projection: "canonical", }); } catch (error) { setAgentInitializing(agentId, false); diff --git a/packages/app/src/utils/agent-initialization.ts b/packages/app/src/utils/agent-initialization.ts index f9f87c53f..7bc52b518 100644 --- a/packages/app/src/utils/agent-initialization.ts +++ b/packages/app/src/utils/agent-initialization.ts @@ -3,6 +3,7 @@ export interface DeferredInit { resolve: () => void; reject: (error: Error) => void; timeoutId: ReturnType | null; + requestDirection: "tail" | "after"; } const initPromises = new Map(); @@ -15,7 +16,10 @@ export function getInitDeferred(key: string): DeferredInit | undefined { return initPromises.get(key); } -export function createInitDeferred(key: string): DeferredInit { +export function createInitDeferred( + key: string, + requestDirection: "tail" | "after" +): DeferredInit { let resolve!: () => void; let reject!: (error: Error) => void; @@ -24,7 +28,13 @@ export function createInitDeferred(key: string): DeferredInit { reject = rej; }); - const deferred: DeferredInit = { promise, resolve, reject, timeoutId: null }; + const deferred: DeferredInit = { + promise, + resolve, + reject, + timeoutId: null, + requestDirection, + }; initPromises.set(key, deferred); return deferred; }