From 1b5d082d1a7d61d64d2a9223b278e3020e674e43 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 2 May 2026 15:19:09 +0800 Subject: [PATCH] refactor(app): fold timeline sequencing helpers into stream reducer (#664) Three sibling modules (session-stream-lifecycle, session-timeline-seq-gate, and the bootstrap-tail policy from session-timeline-bootstrap-policy) only served the stream reducer. Inline them as private helpers in session-stream-reducers.ts; keep deriveInitialTimelineRequest exported for the hook caller. Removes shallow seams without changing behavior. --- .../contexts/session-stream-lifecycle.test.ts | 30 ---- .../src/contexts/session-stream-lifecycle.ts | 23 --- .../src/contexts/session-stream-reducers.ts | 119 +++++++++++++-- .../session-timeline-bootstrap-policy.test.ts | 143 +++++------------- .../session-timeline-bootstrap-policy.ts | 62 -------- .../session-timeline-seq-gate.test.ts | 64 -------- .../src/contexts/session-timeline-seq-gate.ts | 33 ---- 7 files changed, 151 insertions(+), 323 deletions(-) delete mode 100644 packages/app/src/contexts/session-stream-lifecycle.test.ts delete mode 100644 packages/app/src/contexts/session-stream-lifecycle.ts delete mode 100644 packages/app/src/contexts/session-timeline-seq-gate.test.ts delete mode 100644 packages/app/src/contexts/session-timeline-seq-gate.ts diff --git a/packages/app/src/contexts/session-stream-lifecycle.test.ts b/packages/app/src/contexts/session-stream-lifecycle.test.ts deleted file mode 100644 index 3a004cc07..000000000 --- a/packages/app/src/contexts/session-stream-lifecycle.test.ts +++ /dev/null @@ -1,30 +0,0 @@ -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", -}; - -const turnCanceledEvent: AgentStreamEventPayload = { - type: "turn_canceled", - provider: "codex", - reason: "interrupted", -}; - -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("running", turnCanceledEvent)).toBe(null); - 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 deleted file mode 100644 index f31e65e32..000000000 --- a/packages/app/src/contexts/session-stream-lifecycle.ts +++ /dev/null @@ -1,23 +0,0 @@ -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": - return "idle"; - case "turn_failed": - return "error"; - case "turn_canceled": - // A canceled turn can be either a final user cancel or an interrupt before - // a replacement turn starts. The daemon snapshot is authoritative here. - return null; - default: - return null; - } -} diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts index b8aa311ad..7459192e2 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -4,15 +4,6 @@ import type { Agent } from "@/stores/session-store"; import { useSessionStore } from "@/stores/session-store"; 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"; const AGENT_STREAM_REDUCER_FLUSH_DELAY_MS = 16 * 3; @@ -46,6 +37,16 @@ export interface AgentStreamReducerSideEffect { type TimelineDirection = "tail" | "before" | "after"; type InitRequestDirection = "tail" | "after"; +type SessionTimelineSeqCursor = + | { + epoch: string; + endSeq: number; + } + | null + | undefined; + +type SessionTimelineSeqDecision = "accept" | "drop_stale" | "drop_epoch" | "gap" | "init"; + interface TimelineResponseEntry { seqStart: number; seqEnd: number; @@ -99,6 +100,106 @@ interface TimelinePathResult { sideEffects: TimelineReducerSideEffect[]; } +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"; +} + +function deriveBootstrapTailTimelinePolicy({ + direction, + reset, + epoch, + endCursor, + isInitializing, + hasActiveInitDeferred, +}: { + direction: TimelineDirection; + reset: boolean; + epoch: string; + endCursor: { seq: number } | null; + 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, + }; +} + +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; +} + +function deriveOptimisticLifecycleStatus( + currentStatus: AgentLifecycleStatus, + event: AgentStreamEventPayload, +): AgentLifecycleStatus | null { + if (currentStatus !== "running") { + return null; + } + switch (event.type) { + case "turn_completed": + return "idle"; + case "turn_failed": + return "error"; + case "turn_canceled": + // A canceled turn can be either a final user cancel or an interrupt before + // a replacement turn starts. The daemon snapshot is authoritative here. + return null; + default: + return null; + } +} + function preserveReplacePathAssistantHead(params: { tail: StreamItem[]; currentHead: StreamItem[]; diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts index e233b8742..b25cb485d 100644 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts @@ -1,108 +1,47 @@ import { describe, expect, it } from "vitest"; -import { classifySessionTimelineSeq } from "./session-timeline-seq-gate"; -import { - deriveBootstrapTailTimelinePolicy, - shouldResolveTimelineInit, -} from "./session-timeline-bootstrap-policy"; +import { deriveInitialTimelineRequest } from "./session-timeline-bootstrap-policy"; -describe("deriveBootstrapTailTimelinePolicy", () => { - it("always replaces on explicit reset without catch-up cursor", () => { - const policy = deriveBootstrapTailTimelinePolicy({ +describe("deriveInitialTimelineRequest", () => { + it("requests tail history when authoritative history has not been applied", () => { + expect( + deriveInitialTimelineRequest({ + cursor: { epoch: "epoch-1", seq: 42 }, + hasAuthoritativeHistory: false, + initialTimelineLimit: 100, + }), + ).toEqual({ + direction: "tail", + limit: 100, + projection: "canonical", + }); + }); + + it("requests tail history when no cursor is available", () => { + expect( + deriveInitialTimelineRequest({ + cursor: null, + hasAuthoritativeHistory: true, + initialTimelineLimit: 100, + }), + ).toEqual({ + direction: "tail", + limit: 100, + projection: "canonical", + }); + }); + + it("requests only entries after the known cursor when authoritative history exists", () => { + expect( + deriveInitialTimelineRequest({ + cursor: { epoch: "epoch-1", seq: 42 }, + hasAuthoritativeHistory: true, + initialTimelineLimit: 100, + }), + ).toEqual({ direction: "after", - reset: true, - epoch: "epoch-1", - endCursor: { seq: 200 }, - isInitializing: false, - hasActiveInitDeferred: false, + cursor: { epoch: "epoch-1", seq: 42 }, + limit: 0, + projection: "canonical", }); - - 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 index 706ab5eee..ac26e9a37 100644 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts @@ -1,10 +1,3 @@ -type TimelineDirection = "tail" | "before" | "after"; -type InitRequestDirection = "tail" | "after"; - -type BootstrapTailCursor = { - seq: number; -} | null; - type InitialTimelineCursor = { epoch: string; seq: number; @@ -39,58 +32,3 @@ export function deriveInitialTimelineRequest({ projection: "canonical", }; } - -export function deriveBootstrapTailTimelinePolicy({ - direction, - reset, - epoch, - endCursor, - isInitializing, - hasActiveInitDeferred, -}: { - direction: TimelineDirection; - reset: boolean; - epoch: string; - endCursor: BootstrapTailCursor; - 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 deleted file mode 100644 index d18af910a..000000000 --- a/packages/app/src/contexts/session-timeline-seq-gate.test.ts +++ /dev/null @@ -1,64 +0,0 @@ -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 deleted file mode 100644 index e7e35b567..000000000 --- a/packages/app/src/contexts/session-timeline-seq-gate.ts +++ /dev/null @@ -1,33 +0,0 @@ -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"; -}