fix: sync dev branch with main timeline/schema changes and remove stale providers

Forward-port main's timeline epoch tracking, projected window selection,
and schema updates into dev. Remove aider/amp/gemini providers and
provider-history-compatibility layers that are no longer needed.
This commit is contained in:
Mohamed Boudra
2026-04-15 10:52:28 +07:00
parent 326dad93e4
commit 9d30b7e85a
42 changed files with 1251 additions and 2051 deletions
@@ -1,14 +0,0 @@
import Svg, { Path } from "react-native-svg";
interface AiderIconProps {
size?: number;
color?: string;
}
export function AiderIcon({ size = 16, color = "currentColor" }: AiderIconProps) {
return (
<Svg width={size} height={size} viewBox="0 0 24 24" fill={color}>
<Path d="M9 3.75h5.25V4.5H9V3.75Zm-.75.75h6v2.25h-6V4.5Zm6 3h2.25v3h-2.25v-3Zm-6 3h8.25v3h-8.25v-3Zm-1.5 3h1.5v3h-1.5v-3Zm7.5 0h2.25v3h-2.25v-3Zm-6 3h6v3h-6v-3Zm8.25 0H18v3h-1.5v-3Z" />
</Svg>
);
}
@@ -1,29 +0,0 @@
import Svg, { Path } from "react-native-svg";
interface AmpIconProps {
size?: number;
color?: string;
}
export function AmpIcon({ size = 16, color = "currentColor" }: AmpIconProps) {
return (
<Svg width={size} height={size} viewBox="0 0 24 24" fill="none">
<Path
fill={color}
d="m9.686 6.949 2.907.775-1.98-7.404-3.353.903 1.194 4.49a1.94 1.94 0 0 0 1.232 1.236Z"
/>
<Path
fill={color}
d="m4.771 22 6.34-6.327 2.307 8.62 3.352-.903L13.432 10.87.912 7.533 0 10.906l8.61 2.3-6.31 6.317L4.77 22Z"
/>
<Path
fill={color}
d="m13.254 11.707.778 2.917a1.937 1.937 0 0 0 1.23 1.234l4.511 1.199.89-3.37-7.409-1.98Z"
/>
<Path
fill={color}
d="m15.916 2.484-2.883 2.88a2.063 2.063 0 0 0-.512 1.193l-.046 1.825 1.69.06-.022-.001c.463 0 .898-.181 1.225-.507L18.35 4.95l-2.434-2.467Z"
/>
</Svg>
);
}
@@ -1,17 +0,0 @@
import Svg, { Path } from "react-native-svg";
interface GeminiIconProps {
size?: number;
color?: string;
}
export function GeminiIcon({ size = 16, color = "currentColor" }: GeminiIconProps) {
return (
<Svg width={size} height={size} viewBox="0 0 24 24" fill="none">
<Path
fill={color}
d="M20.616 10.835a14.147 14.147 0 0 1-4.45-3.001 14.111 14.111 0 0 1-3.678-6.452.503.503 0 0 0-.975 0 14.134 14.134 0 0 1-3.679 6.452 14.155 14.155 0 0 1-4.45 3.001c-.65.28-1.318.505-2.002.678a.502.502 0 0 0 0 .975c.684.172 1.35.397 2.002.677a14.147 14.147 0 0 1 4.45 3.001 14.112 14.112 0 0 1 3.679 6.453.502.502 0 0 0 .975 0c.172-.685.397-1.351.677-2.003a14.145 14.145 0 0 1 3.001-4.45 14.113 14.113 0 0 1 6.453-3.678.503.503 0 0 0 0-.975 13.245 13.245 0 0 1-2.003-.678Z"
/>
</Svg>
);
}
@@ -1,9 +1,6 @@
import { Bot } from "lucide-react-native";
import { AiderIcon } from "@/components/icons/aider-icon";
import { AmpIcon } from "@/components/icons/amp-icon";
import { ClaudeIcon } from "@/components/icons/claude-icon";
import { CodexIcon } from "@/components/icons/codex-icon";
import { GeminiIcon } from "@/components/icons/gemini-icon";
import { CopilotIcon } from "@/components/icons/copilot-icon";
import { OpenCodeIcon } from "@/components/icons/opencode-icon";
import { PiIcon } from "@/components/icons/pi-icon";
@@ -11,9 +8,6 @@ import { PiIcon } from "@/components/icons/pi-icon";
const PROVIDER_ICONS: Record<string, typeof Bot> = {
claude: ClaudeIcon as unknown as typeof Bot,
codex: CodexIcon as unknown as typeof Bot,
gemini: GeminiIcon as unknown as typeof Bot,
amp: AmpIcon as unknown as typeof Bot,
aider: AiderIcon as unknown as typeof Bot,
copilot: CopilotIcon as unknown as typeof Bot,
opencode: OpenCodeIcon as unknown as typeof Bot,
pi: PiIcon as unknown as typeof Bot,
+11 -4
View File
@@ -591,8 +591,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
void client
.fetchAgentTimeline(agentId, {
direction: "after",
cursor: { seq: cursor.endSeq },
cursor: { epoch: cursor.epoch, seq: cursor.endSeq },
limit: 0,
projection: "canonical",
})
.catch((error) => {
console.warn("[Session] failed to fetch catch-up timeline on resume", agentId, error);
@@ -853,12 +854,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
);
const requestCanonicalCatchUp = useCallback(
(agentId: string, cursor: { endSeq: number }) => {
(agentId: string, cursor: { epoch: string; endSeq: number }) => {
void client
.fetchAgentTimeline(agentId, {
direction: "after",
cursor: { seq: cursor.endSeq },
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);
@@ -961,6 +963,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
}
if (
current &&
current.epoch === result.cursor.epoch &&
current.startSeq === result.cursor.startSeq &&
current.endSeq === result.cursor.endSeq
) {
@@ -1065,7 +1068,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const unsubAgentStream = client.on("agent_stream", (message) => {
if (message.type !== "agent_stream") return;
const { agentId, event, timestamp, seq } = message.payload;
const { agentId, event, timestamp, seq, epoch } = message.payload;
const parsedTimestamp = new Date(timestamp);
const streamEvent = event as AgentStreamEventPayload;
if (
@@ -1107,6 +1110,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const result = processAgentStreamEvent({
event: streamEvent,
seq,
epoch,
currentTail,
currentHead,
currentCursor,
@@ -1130,6 +1134,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
if (
current &&
typeof seq === "number" &&
typeof epoch === "string" &&
current.epoch === epoch &&
seq >= current.startSeq &&
seq <= current.endSeq
) {
@@ -1138,6 +1144,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
}
if (
current &&
current.epoch === nextCursor.epoch &&
current.startSeq === nextCursor.startSeq &&
current.endSeq === nextCursor.endSeq
) {
@@ -9,9 +9,13 @@ import {
type TimelineCursor,
} from "./session-stream-reducers";
// ---------------------------------------------------------------------------
// Test helpers
// ---------------------------------------------------------------------------
function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") {
return {
seq,
seqStart: seq,
provider: "claude",
item: { type, text },
timestamp: new Date(1000 + seq).toISOString(),
@@ -29,30 +33,22 @@ function makeTimelineEvent(
} as AgentStreamEventPayload;
}
function makeToolCallEvent(status: "running" | "completed"): AgentStreamEventPayload {
function makeUserTimelineEvent(text: string): AgentStreamEventPayload {
return {
type: "timeline",
provider: "claude",
item: {
type: "tool_call",
callId: "call-1",
name: "shell",
status,
detail: {
type: "shell",
command: "pwd",
},
error: null,
},
};
item: { type: "user_message", text },
} as AgentStreamEventPayload;
}
const baseTimelineInput: ProcessTimelineResponseInput = {
payload: {
agentId: "agent-1",
direction: "after",
startSeq: null,
endSeq: null,
reset: false,
epoch: "epoch-1",
startCursor: null,
endCursor: null,
entries: [],
error: null,
},
@@ -67,6 +63,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
const baseStreamInput: ProcessAgentStreamEventInput = {
event: makeTimelineEvent("hello"),
seq: undefined,
epoch: undefined,
currentTail: [],
currentHead: [],
currentCursor: undefined,
@@ -74,6 +71,10 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
timestamp: new Date(2000),
};
// ---------------------------------------------------------------------------
// processTimelineResponse
// ---------------------------------------------------------------------------
describe("processTimelineResponse", () => {
it("returns error path when payload.error is set", () => {
const result = processTimelineResponse({
@@ -92,10 +93,35 @@ describe("processTimelineResponse", () => {
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.head).toBe(baseTimelineInput.currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.sideEffects).toEqual([]);
});
it("replaces tail during bootstrap tail init and schedules committed catch-up", () => {
const provisionalHead: StreamItem[] = [
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",
@@ -106,325 +132,518 @@ describe("processTimelineResponse", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentHead: provisionalHead,
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",
startSeq: 1,
endSeq: 5,
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,
});
const catchUp = result.sideEffects.find((effect) => effect.type === "catch_up");
expect(catchUp).toEqual({
type: "catch_up",
cursor: { 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("prepends older committed history for before pagination", () => {
const currentTail: StreamItem[] = [
{
kind: "assistant_message",
id: "tail-3",
text: "newer",
timestamp: new Date(3000),
},
];
const currentCursor: TimelineCursor = { startSeq: 3, endSeq: 4 };
it("appends incrementally for contiguous seqs", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 3,
};
const result = processTimelineResponse({
...baseTimelineInput,
currentTail,
currentCursor,
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
direction: "before",
startSeq: 1,
endSeq: 2,
entries: [makeTimelineEntry(1, "hello", "user_message"), makeTimelineEntry(2, "older")],
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: 4,
endSeq: 5,
});
expect(result.tail).toHaveLength(3);
expect(result.tail[0]?.kind).toBe("user_message");
expect(result.tail[1]?.kind).toBe("assistant_message");
expect(result.tail[2]).toBe(currentTail[0]);
expect(result.error).toBe(null);
});
it("replaces stale provisional assistant UI when fetch-after returns committed row 121", () => {
const currentHead: StreamItem[] = [
{
kind: "assistant_message",
id: "head-assistant",
text: "partial",
timestamp: new Date(120000),
},
];
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
it("detects gap and emits catch-up side effect", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 3,
};
const result = processTimelineResponse({
...baseTimelineInput,
currentHead,
currentCursor,
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",
startSeq: 121,
endSeq: 121,
entries: [makeTimelineEntry(121, "finalized reply")],
},
});
expect(result.head).toEqual([]);
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
startSeq: 1,
endSeq: 121,
});
expect(result.tail[result.tail.length - 1]).toMatchObject({
kind: "assistant_message",
text: "finalized reply",
});
});
it("keeps provisional head when reconnect catch-up has no new committed rows yet", () => {
const currentHead: StreamItem[] = [
{
kind: "assistant_message",
id: "head-assistant",
text: "still streaming",
timestamp: new Date(120000),
},
];
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
const result = processTimelineResponse({
...baseTimelineInput,
currentHead,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "after",
startSeq: null,
endSeq: null,
entries: [],
},
});
expect(result.head).toBe(currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.initResolution).toBe("resolve");
expect(result.clearInitializing).toBe(true);
});
it("requests catch-up when committed rows arrive with a forward gap", () => {
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
it("does not resolve init when directions differ (before vs after)", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentCursor,
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",
startSeq: 125,
endSeq: 125,
entries: [makeTimelineEntry(125, "far ahead")],
entries: [],
},
});
expect(result.cursorChanged).toBe(false);
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.sideEffects).toContainEqual({
type: "catch_up",
cursor: { endSeq: 120 },
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("treats seq-less timeline events as provisional head updates", () => {
it("passes through non-timeline events without cursor changes", () => {
const turnEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("partial"),
event: turnEvent,
seq: undefined,
epoch: undefined,
});
expect(result.changedHead).toBe(true);
expect(result.changedTail).toBe(false);
expect(result.head).toHaveLength(1);
expect(result.head[0]).toMatchObject({
kind: "assistant_message",
text: "partial",
});
expect(result.cursorChanged).toBe(false);
expect(result.cursor).toBe(null);
expect(result.sideEffects).toEqual([]);
});
it("keeps tool call rows anchored in tail across live and committed updates", () => {
const running = processAgentStreamEvent({
it("accepts timeline event with cursor advance", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 4,
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("running"),
seq: undefined,
});
expect(running.head).toEqual([]);
expect(running.tail).toHaveLength(1);
expect(running.tail[0]).toMatchObject({
kind: "tool_call",
payload: {
source: "agent",
data: {
callId: "call-1",
status: "running",
},
},
event: makeTimelineEvent("new chunk"),
seq: 5,
epoch: "epoch-1",
currentCursor: existingCursor,
});
const completedLive = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("completed"),
seq: undefined,
currentHead: running.head,
currentTail: running.tail,
currentCursor: { startSeq: 1, endSeq: 7 },
});
expect(completedLive.head).toEqual([]);
expect(completedLive.tail).toHaveLength(1);
expect(completedLive.tail[0]).toMatchObject({
kind: "tool_call",
payload: {
source: "agent",
data: {
callId: "call-1",
status: "completed",
},
},
});
const committed = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("completed"),
seq: 8,
currentHead: completedLive.head,
currentTail: completedLive.tail,
currentCursor: { startSeq: 1, endSeq: 7 },
});
expect(committed.head).toEqual([]);
expect(committed.tail).toHaveLength(1);
expect(committed.tail[0]).toMatchObject({
kind: "tool_call",
payload: {
source: "agent",
data: {
callId: "call-1",
status: "completed",
},
},
expect(result.cursorChanged).toBe(true);
expect(result.cursor).toEqual({
epoch: "epoch-1",
startSeq: 1,
endSeq: 5,
});
expect(result.sideEffects).toEqual([]);
});
it("preserves assistant/tool interleaving while a turn is streaming", () => {
const assistantBeforeTool = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("before"),
seq: undefined,
});
it("detects gap and emits catch-up side effect", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 4,
};
const runningTool = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("running"),
seq: undefined,
currentHead: assistantBeforeTool.head,
currentTail: assistantBeforeTool.tail,
});
const assistantAfterTool = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("after"),
seq: undefined,
currentHead: runningTool.head,
currentTail: runningTool.tail,
});
const completedTool = processAgentStreamEvent({
...baseStreamInput,
event: makeToolCallEvent("completed"),
seq: undefined,
currentHead: assistantAfterTool.head,
currentTail: assistantAfterTool.tail,
});
expect(completedTool.head).toEqual([]);
expect(completedTool.tail.map((item) => item.kind)).toEqual([
"assistant_message",
"tool_call",
"assistant_message",
]);
expect(
completedTool.tail[0]?.kind === "assistant_message" ? completedTool.tail[0].text : null,
).toBe("before");
expect(
completedTool.tail[2]?.kind === "assistant_message" ? completedTool.tail[2].text : null,
).toBe("after");
});
it("requests catch-up when a committed live row skips ahead", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("far ahead"),
seq: 125,
currentCursor: { startSeq: 1, endSeq: 120 },
seq: 10,
epoch: "epoch-1",
currentCursor: existingCursor,
});
expect(result.cursorChanged).toBe(false);
expect(result.changedTail).toBe(false);
expect(result.changedHead).toBe(false);
expect(result.cursorChanged).toBe(false);
expect(result.sideEffects).toContainEqual({
type: "catch_up",
cursor: { endSeq: 120 },
const catchUp = result.sideEffects.find((e) => e.type === "catch_up");
expect(catchUp).toBeDefined();
expect(catchUp!.cursor).toEqual({
epoch: "epoch-1",
endSeq: 4,
});
});
it("flushes provisional head into tail on terminal turn events", () => {
const withHead = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("streaming"),
seq: undefined,
});
it("drops stale timeline event", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 1,
endSeq: 8,
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: {
type: "turn_completed",
provider: "claude",
},
currentHead: withHead.head,
currentTail: withHead.tail,
event: makeTimelineEvent("old"),
seq: 5,
epoch: "epoch-1",
currentCursor: existingCursor,
});
expect(result.changedHead).toBe(true);
expect(result.changedTail).toBe(true);
expect(result.head).toEqual([]);
expect(result.tail).toHaveLength(1);
expect(result.tail[0]).toMatchObject({
kind: "assistant_message",
text: "streaming",
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);
});
});
@@ -1,7 +1,12 @@
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 {
applyStreamEvent,
flushHeadToTail,
hydrateStreamState,
reduceStreamUpdate,
} from "@/types/stream";
import {
classifySessionTimelineSeq,
type SessionTimelineSeqDecision,
@@ -12,25 +17,38 @@ import {
} 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: { endSeq: number } }
| { type: "catch_up"; cursor: { epoch: string; endSeq: number } }
| { type: "flush_pending_updates" };
export type AgentStreamReducerSideEffect = {
type: "catch_up";
cursor: { endSeq: number };
cursor: { epoch: string; endSeq: number };
};
// ---------------------------------------------------------------------------
// processTimelineResponse
// ---------------------------------------------------------------------------
type TimelineDirection = "tail" | "before" | "after";
type InitRequestDirection = "tail" | "after";
type TimelineResponseEntry = {
seq: number;
seqStart: number;
provider: string;
item: Record<string, unknown>;
timestamp: string;
@@ -40,8 +58,10 @@ export interface ProcessTimelineResponseInput {
payload: {
agentId: string;
direction: TimelineDirection;
startSeq: number | null;
endSeq: number | null;
reset: boolean;
epoch: string;
startCursor: { seq: number } | null;
endCursor: { seq: number } | null;
entries: TimelineResponseEntry[];
error: string | null;
};
@@ -64,9 +84,217 @@ export interface ProcessTimelineResponseOutput {
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), {
source: "canonical",
});
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) {
// Flush head to tail before appending canonical entries so that
// chronological ordering is preserved. This matters on mobile where
// the server drops live events for backgrounded/unfocused agents:
// when the client catches up, the head may still hold stale live
// items from before the gap that must be committed ahead of the
// canonical gap-fill entries.
const baseTail =
currentHead.length > 0 ? flushHeadToTail(currentTail, currentHead) : currentTail;
if (currentHead.length > 0) {
nextHead = [];
}
nextTail = acceptedUnits.reduce<StreamItem[]>(
(state, { event, timestamp }) =>
reduceStreamUpdate(state, event, timestamp, {
source: "canonical",
}),
baseTail,
);
}
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;
@@ -96,229 +324,57 @@ export interface ProcessAgentStreamEventOutput {
sideEffects: AgentStreamReducerSideEffect[];
}
function cursorsEqual(
left: TimelineCursor | null | undefined,
right: TimelineCursor | null | undefined,
): boolean {
if (!left || !right) {
return left === right;
}
return left.startSeq === right.startSeq && left.endSeq === right.endSeq;
}
function removeSupersededProvisionalItems(
head: StreamItem[],
event: AgentStreamEventPayload,
): StreamItem[] {
if (head.length === 0 || event.type !== "timeline") {
return head;
}
let nextHead = head;
if (event.item.type === "assistant_message") {
nextHead = head.filter((item) => item.kind !== "assistant_message");
} else if (event.item.type === "tool_call") {
const committedToolCall = event.item;
nextHead = head.filter(
(item) =>
item.kind !== "tool_call" ||
item.payload.source !== "agent" ||
item.payload.data.callId !== committedToolCall.callId,
);
}
return nextHead.length === head.length ? head : nextHead;
}
export function processTimelineResponse(
input: ProcessTimelineResponseInput,
): ProcessTimelineResponseOutput {
const {
payload,
currentTail,
currentHead,
currentCursor,
isInitializing,
hasActiveInitDeferred,
initRequestDirection,
} = input;
if (payload.error) {
return {
tail: currentTail,
head: currentHead,
cursor: currentCursor,
cursorChanged: false,
initResolution: hasActiveInitDeferred ? "reject" : null,
clearInitializing: isInitializing,
error: payload.error,
sideEffects: [],
};
}
const timelineUnits = payload.entries.map((entry) => ({
seq: entry.seq,
event: {
type: "timeline",
provider: entry.provider,
item: entry.item,
} as AgentStreamEventPayload,
timestamp: new Date(entry.timestamp),
}));
const bootstrapPolicy = deriveBootstrapTailTimelinePolicy({
direction: payload.direction,
endSeq: payload.endSeq,
isInitializing,
hasActiveInitDeferred,
});
let nextTail = currentTail;
let nextHead = currentHead;
let nextCursor: TimelineCursor | null | undefined = currentCursor;
let cursorChanged = false;
const sideEffects: TimelineReducerSideEffect[] = [];
if (bootstrapPolicy.replace) {
nextTail = hydrateStreamState(
timelineUnits.map(({ event, timestamp }) => ({ event, timestamp })),
{ source: "canonical" },
);
nextHead = [];
nextCursor =
typeof payload.startSeq === "number" && typeof payload.endSeq === "number"
? {
startSeq: payload.startSeq,
endSeq: payload.endSeq,
}
: null;
cursorChanged = !cursorsEqual(currentCursor, nextCursor);
if (bootstrapPolicy.catchUpCursor) {
sideEffects.push({
type: "catch_up",
cursor: bootstrapPolicy.catchUpCursor,
});
}
} else if (payload.direction === "before") {
const prepended = hydrateStreamState(
timelineUnits.map(({ event, timestamp }) => ({ event, timestamp })),
{ source: "canonical" },
);
nextTail = prepended.length > 0 ? [...prepended, ...currentTail] : currentTail;
const derivedCursor =
typeof payload.startSeq === "number"
? {
startSeq: payload.startSeq,
endSeq: currentCursor?.endSeq ?? payload.endSeq ?? payload.startSeq,
}
: currentCursor;
nextCursor = derivedCursor;
cursorChanged = !cursorsEqual(currentCursor, derivedCursor);
} else if (timelineUnits.length > 0) {
const acceptedUnits: typeof timelineUnits = [];
let cursor = currentCursor;
let gapCursor: { endSeq: number } | null = null;
for (const unit of timelineUnits) {
const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({
cursor: cursor ? { endSeq: cursor.endSeq } : null,
seq: unit.seq,
});
if (decision === "gap") {
gapCursor = cursor ? { endSeq: cursor.endSeq } : null;
break;
}
if (decision === "drop_stale") {
continue;
}
acceptedUnits.push(unit);
cursor =
decision === "init"
? { startSeq: unit.seq, endSeq: unit.seq }
: { ...(cursor ?? { startSeq: unit.seq, endSeq: unit.seq }), endSeq: unit.seq };
nextHead = removeSupersededProvisionalItems(nextHead, unit.event);
}
if (acceptedUnits.length > 0) {
nextTail = acceptedUnits.reduce<StreamItem[]>(
(state, { event, timestamp }) =>
reduceStreamUpdate(state, event, timestamp, {
source: "canonical",
}),
currentTail,
);
}
if (cursor && !cursorsEqual(currentCursor, cursor)) {
nextCursor = cursor;
cursorChanged = true;
}
if (gapCursor) {
sideEffects.push({ type: "catch_up", cursor: gapCursor });
}
}
sideEffects.push({ type: "flush_pending_updates" });
const shouldResolveDeferredInit = shouldResolveTimelineInit({
hasActiveInitDeferred,
isInitializing,
initRequestDirection,
responseDirection: payload.direction,
});
const clearInitializing = shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred);
return {
tail: nextTail,
head: nextHead,
cursor: nextCursor,
cursorChanged,
initResolution: shouldResolveDeferredInit ? "resolve" : null,
clearInitializing,
error: null,
sideEffects,
};
}
export function processAgentStreamEvent(
input: ProcessAgentStreamEventInput,
): ProcessAgentStreamEventOutput {
const { event, seq, currentTail, currentHead, currentCursor, currentAgent, timestamp } = input;
const { event, seq, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } =
input;
let shouldApplyStreamEvent = true;
let nextTimelineCursor: TimelineCursor | null = null;
let cursorChanged = false;
const sideEffects: AgentStreamReducerSideEffect[] = [];
if (event.type === "timeline" && typeof seq === "number") {
// ------------------------------------------------------------------
// Timeline sequencing gate
// ------------------------------------------------------------------
if (event.type === "timeline" && typeof seq === "number" && typeof epoch === "string") {
const decision = classifySessionTimelineSeq({
cursor: currentCursor ? { endSeq: currentCursor.endSeq } : null,
cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null,
epoch,
seq,
});
if (decision === "gap") {
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: { endSeq: currentCursor.endSeq },
cursor: {
epoch: currentCursor.epoch,
endSeq: currentCursor.endSeq,
},
});
}
} else if (decision === "drop_stale") {
shouldApplyStreamEvent = false;
} else {
nextTimelineCursor =
decision === "init"
? { startSeq: seq, endSeq: seq }
: { ...(currentCursor ?? { startSeq: seq, endSeq: seq }), endSeq: seq };
cursorChanged = !cursorsEqual(currentCursor, nextTimelineCursor);
// drop_stale or drop_epoch
shouldApplyStreamEvent = false;
}
}
// ------------------------------------------------------------------
// Apply stream event to tail/head
// ------------------------------------------------------------------
const { tail, head, changedTail, changedHead } = shouldApplyStreamEvent
? applyStreamEvent({
tail: currentTail,
@@ -334,6 +390,9 @@ export function processAgentStreamEvent(
changedHead: false,
};
// ------------------------------------------------------------------
// Optimistic lifecycle status
// ------------------------------------------------------------------
let agentPatch: AgentPatch | null = null;
let agentChanged = false;
@@ -2,42 +2,26 @@ import { describe, expect, it } from "vitest";
import { classifySessionTimelineSeq } from "./session-timeline-seq-gate";
import {
deriveBootstrapTailTimelinePolicy,
deriveInitialTimelineRequest,
shouldResolveTimelineInit,
} from "./session-timeline-bootstrap-policy";
describe("deriveInitialTimelineRequest", () => {
it("uses tail bootstrap when history has not synced yet", () => {
expect(
deriveInitialTimelineRequest({
cursor: { seq: 42 },
hasAuthoritativeHistory: false,
initialTimelineLimit: 200,
}),
).toEqual({
direction: "tail",
limit: 200,
});
});
it("uses catch-up after the committed cursor once history is synced", () => {
expect(
deriveInitialTimelineRequest({
cursor: { seq: 42 },
hasAuthoritativeHistory: true,
initialTimelineLimit: 200,
}),
).toEqual({
direction: "after",
cursor: { seq: 42 },
limit: 0,
});
});
});
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 = { endSeq: 205 };
const advancedCursor = { epoch: "epoch-1", endSeq: 205 };
const tailSeqStart = 101;
const tailSeqEnd = 200;
@@ -45,6 +29,7 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
for (let seq = tailSeqStart; seq <= tailSeqEnd; seq += 1) {
const decision = classifySessionTimelineSeq({
cursor: advancedCursor,
epoch: "epoch-1",
seq,
});
if (decision === "accept" || decision === "init") {
@@ -55,13 +40,16 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
const policy = deriveBootstrapTailTimelinePolicy({
direction: "tail",
endSeq: 200,
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,
});
});
@@ -69,7 +57,9 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
it("does not replace non-bootstrap, non-reset responses", () => {
const policy = deriveBootstrapTailTimelinePolicy({
direction: "tail",
endSeq: 200,
reset: false,
epoch: "epoch-1",
endCursor: { seq: 200 },
isInitializing: false,
hasActiveInitDeferred: false,
});
@@ -87,6 +77,7 @@ describe("shouldResolveTimelineInit", () => {
isInitializing: true,
initRequestDirection: "tail",
responseDirection: "tail",
reset: false,
}),
).toBe(true);
});
@@ -98,6 +89,7 @@ describe("shouldResolveTimelineInit", () => {
isInitializing: true,
initRequestDirection: "tail",
responseDirection: "after",
reset: false,
}),
).toBe(false);
});
@@ -109,6 +101,7 @@ describe("shouldResolveTimelineInit", () => {
isInitializing: true,
initRequestDirection: "after",
responseDirection: "after",
reset: false,
}),
).toBe(true);
});
@@ -6,6 +6,7 @@ type BootstrapTailCursor = {
} | null;
type InitialTimelineCursor = {
epoch: string;
seq: number;
} | null;
@@ -19,37 +20,48 @@ export function deriveInitialTimelineRequest({
initialTimelineLimit: number;
}): {
direction: "tail" | "after";
cursor?: { seq: number };
cursor?: { epoch: string; seq: number };
limit: number;
projection: "canonical";
} {
if (!hasAuthoritativeHistory || !cursor) {
return {
direction: "tail",
limit: initialTimelineLimit,
projection: "canonical",
};
}
return {
direction: "after",
cursor: { seq: cursor.seq },
cursor: { epoch: cursor.epoch, seq: cursor.seq },
limit: 0,
projection: "canonical",
};
}
export function deriveBootstrapTailTimelinePolicy({
direction,
endSeq,
reset,
epoch,
endCursor,
isInitializing,
hasActiveInitDeferred,
}: {
direction: TimelineDirection;
endSeq: number | null;
reset: boolean;
epoch: string;
endCursor: BootstrapTailCursor;
isInitializing: boolean;
hasActiveInitDeferred: boolean;
}): {
replace: boolean;
catchUpCursor: { endSeq: number } | null;
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 };
@@ -57,7 +69,7 @@ export function deriveBootstrapTailTimelinePolicy({
return {
replace: true,
catchUpCursor: typeof endSeq === "number" ? { endSeq } : null,
catchUpCursor: endCursor ? { epoch, endSeq: endCursor.seq } : null,
};
}
@@ -66,14 +78,19 @@ export function shouldResolveTimelineInit({
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;
}
@@ -5,7 +5,8 @@ describe("classifySessionTimelineSeq", () => {
it("accepts contiguous forward seq", () => {
expect(
classifySessionTimelineSeq({
cursor: { endSeq: 4 },
cursor: { epoch: "epoch-1", endSeq: 4 },
epoch: "epoch-1",
seq: 5,
}),
).toBe("accept");
@@ -14,7 +15,8 @@ describe("classifySessionTimelineSeq", () => {
it("drops stale seq older than the current end", () => {
expect(
classifySessionTimelineSeq({
cursor: { endSeq: 8 },
cursor: { epoch: "epoch-1", endSeq: 8 },
epoch: "epoch-1",
seq: 7,
}),
).toBe("drop_stale");
@@ -23,16 +25,28 @@ describe("classifySessionTimelineSeq", () => {
it("drops duplicate replay seq equal to the current end", () => {
expect(
classifySessionTimelineSeq({
cursor: { endSeq: 8 },
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");
@@ -41,7 +55,8 @@ describe("classifySessionTimelineSeq", () => {
it("classifies forward gaps", () => {
expect(
classifySessionTimelineSeq({
cursor: { endSeq: 4 },
cursor: { epoch: "epoch-1", endSeq: 4 },
epoch: "epoch-1",
seq: 9,
}),
).toBe("gap");
@@ -1,22 +1,28 @@
export type SessionTimelineSeqCursor =
| {
epoch: string;
endSeq: number;
}
| null
| undefined;
export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "gap" | "init";
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";
}
@@ -2,10 +2,11 @@ import { describe, expect, it } from "vitest";
import { __private__ } from "./use-agent-initialization";
describe("useAgentInitialization timeline request policy", () => {
it("uses committed tail bootstrap when history has not synced yet", () => {
it("uses canonical tail bootstrap when history has not synced yet", () => {
expect(
__private__.deriveInitialTimelineRequest({
cursor: {
epoch: "epoch-1",
seq: 42,
},
hasAuthoritativeHistory: false,
@@ -14,10 +15,11 @@ describe("useAgentInitialization timeline request policy", () => {
).toEqual({
direction: "tail",
limit: 200,
projection: "canonical",
});
});
it("uses committed tail bootstrap when cursor is missing", () => {
it("uses canonical tail bootstrap when cursor is missing", () => {
expect(
__private__.deriveInitialTimelineRequest({
cursor: null,
@@ -27,13 +29,15 @@ describe("useAgentInitialization timeline request policy", () => {
).toEqual({
direction: "tail",
limit: 200,
projection: "canonical",
});
});
it("uses committed catch-up after the current cursor once history is synced", () => {
it("uses canonical catch-up after the current cursor once history is synced", () => {
expect(
__private__.deriveInitialTimelineRequest({
cursor: {
epoch: "epoch-1",
seq: 42,
},
hasAuthoritativeHistory: true,
@@ -41,8 +45,9 @@ describe("useAgentInitialization timeline request policy", () => {
}),
).toEqual({
direction: "after",
cursor: { seq: 42 },
cursor: { epoch: "epoch-1", seq: 42 },
limit: 0,
projection: "canonical",
});
});
@@ -56,6 +61,7 @@ describe("useAgentInitialization timeline request policy", () => {
).toEqual({
direction: "tail",
limit: 0,
projection: "canonical",
});
});
@@ -60,7 +60,7 @@ export function useAgentInitialization({
const hasAuthoritativeHistory =
session?.agentAuthoritativeHistoryApplied.get(agentId) === true;
const timelineRequest = deriveInitialTimelineRequest({
cursor: cursor ? { seq: cursor.endSeq } : null,
cursor: cursor ? { epoch: cursor.epoch, seq: cursor.endSeq } : null,
hasAuthoritativeHistory,
initialTimelineLimit,
});
@@ -107,6 +107,7 @@ export function useAgentInitialization({
await client.fetchAgentTimeline(agentId, {
direction: "tail",
limit: initialTimelineLimit,
projection: "canonical",
});
} catch (error) {
setAgentInitializing(agentId, false);
@@ -1,61 +0,0 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
vi.mock("@react-native-async-storage/async-storage", () => {
const storage = new Map<string, string>();
return {
default: {
getItem: vi.fn(async (key: string) => storage.get(key) ?? null),
setItem: vi.fn(async (key: string, value: string) => {
storage.set(key, value);
}),
removeItem: vi.fn(async (key: string) => {
storage.delete(key);
}),
},
};
});
import { AGENT_PROVIDER_DEFINITIONS } from "@server/server/agent/provider-manifest";
import {
__providerRecencyStoreTestUtils,
sortProvidersByRecency,
useProviderRecencyStore,
} from "./provider-recency-store";
describe("provider-recency-store", () => {
beforeEach(() => {
useProviderRecencyStore.setState({
recentProviderIds: [],
recordUsage: useProviderRecencyStore.getState().recordUsage,
});
});
it("sorts used providers first and keeps unused providers in default order", () => {
const sorted = sortProvidersByRecency(AGENT_PROVIDER_DEFINITIONS, ["codex"]);
expect(sorted.map((provider) => provider.id)).toEqual([
"codex",
...AGENT_PROVIDER_DEFINITIONS.filter((provider) => provider.id !== "codex").map(
(provider) => provider.id,
),
]);
});
it("moves the latest provider to the front without duplicating prior entries", () => {
useProviderRecencyStore.getState().recordUsage("codex");
useProviderRecencyStore.getState().recordUsage("opencode");
useProviderRecencyStore.getState().recordUsage("codex");
expect(useProviderRecencyStore.getState().recentProviderIds).toEqual(["codex", "opencode"]);
});
it("filters invalid and duplicate providers during migration", () => {
expect(
__providerRecencyStoreTestUtils.migratePersistedState({
recentProviderIds: ["codex", "invalid", "codex", "claude"],
}),
).toEqual({
recentProviderIds: ["codex", "claude"],
});
});
});
@@ -1,131 +0,0 @@
import { useMemo } from "react";
import AsyncStorage from "@react-native-async-storage/async-storage";
import type { AgentProvider } from "@server/server/agent/agent-sdk-types";
import {
AGENT_PROVIDER_DEFINITIONS,
isValidAgentProvider,
type AgentProviderDefinition,
} from "@server/server/agent/provider-manifest";
import { create } from "zustand";
import { createJSONStorage, persist } from "zustand/middleware";
const PROVIDER_RECENCY_STORE_VERSION = 1;
interface ProviderRecencyStoreState {
recentProviderIds: AgentProvider[];
recordUsage: (providerId: AgentProvider) => void;
}
function sanitizeRecentProviderIds(providerIds: readonly string[] | undefined): AgentProvider[] {
if (!providerIds || providerIds.length === 0) {
return [];
}
const seen = new Set<AgentProvider>();
const sanitized: AgentProvider[] = [];
for (const providerId of providerIds) {
if (!isValidAgentProvider(providerId)) {
continue;
}
if (seen.has(providerId)) {
continue;
}
seen.add(providerId);
sanitized.push(providerId);
}
return sanitized;
}
export function sortProvidersByRecency<T extends { id: string }>(
providers: readonly T[],
recentProviderIds: readonly string[],
): T[] {
if (providers.length <= 1) {
return [...providers];
}
const recentRank = new Map<string, number>();
for (const providerId of recentProviderIds) {
if (recentRank.has(providerId)) {
continue;
}
recentRank.set(providerId, recentRank.size);
}
return providers
.map((provider, defaultIndex) => ({
provider,
defaultIndex,
recentIndex: recentRank.get(provider.id) ?? Number.POSITIVE_INFINITY,
}))
.sort((left, right) => {
if (left.recentIndex !== right.recentIndex) {
return left.recentIndex - right.recentIndex;
}
return left.defaultIndex - right.defaultIndex;
})
.map((entry) => entry.provider);
}
function migratePersistedState(
state: unknown,
): Pick<ProviderRecencyStoreState, "recentProviderIds"> {
const record = state as { recentProviderIds?: string[] } | null | undefined;
return {
recentProviderIds: sanitizeRecentProviderIds(record?.recentProviderIds),
};
}
export const useProviderRecencyStore = create<ProviderRecencyStoreState>()(
persist(
(set) => ({
recentProviderIds: [],
recordUsage: (providerId) => {
if (!isValidAgentProvider(providerId)) {
return;
}
set((state) => ({
recentProviderIds: [
providerId,
...state.recentProviderIds.filter((id) => id !== providerId),
],
}));
},
}),
{
name: "agent-provider-recency",
version: PROVIDER_RECENCY_STORE_VERSION,
storage: createJSONStorage(() => AsyncStorage),
partialize: (state) => ({
recentProviderIds: state.recentProviderIds,
}),
migrate: (persistedState) => migratePersistedState(persistedState),
},
),
);
export function useProviderRecency(
availableProviders: readonly AgentProviderDefinition[] = AGENT_PROVIDER_DEFINITIONS,
): {
providers: AgentProviderDefinition[];
recordUsage: (providerId: AgentProvider) => void;
} {
const recentProviderIds = useProviderRecencyStore((state) => state.recentProviderIds);
const recordUsage = useProviderRecencyStore((state) => state.recordUsage);
const providers = useMemo(
() => sortProvidersByRecency(availableProviders, recentProviderIds),
[availableProviders, recentProviderIds],
);
return {
providers,
recordUsage,
};
}
export const __providerRecencyStoreTestUtils = {
migratePersistedState,
sanitizeRecentProviderIds,
};
+1
View File
@@ -212,6 +212,7 @@ export type DaemonServerInfo = {
};
export interface AgentTimelineCursorState {
epoch: string;
startSeq: number;
endSeq: number;
}
+1 -1
View File
@@ -729,7 +729,7 @@ function finalizeHeadItems(head: StreamItem[]): StreamItem[] {
/**
* Flush head items to tail, avoiding duplicates.
*/
function flushHeadToTail(tail: StreamItem[], head: StreamItem[]): StreamItem[] {
export function flushHeadToTail(tail: StreamItem[], head: StreamItem[]): StreamItem[] {
if (head.length === 0) {
return tail;
}