Merge branch 'storage-terminal-ui-dev' into dev
# Conflicts: # nix/package.nix # packages/app/src/app/_layout.tsx # packages/app/src/components/sidebar-workspace-list.tsx # packages/app/src/hooks/use-command-center.ts # packages/app/src/screens/agent/draft-agent-screen.tsx # packages/app/src/screens/workspace/workspace-desktop-tabs-row.tsx # packages/app/src/screens/workspace/workspace-screen.tsx # packages/server/src/server/session.ts # packages/server/src/server/session.workspaces.test.ts # packages/server/src/terminal/terminal.test.ts # packages/server/src/terminal/terminal.ts
This commit is contained in:
Mohamed Boudra
@@ -521,9 +521,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
void client
|
||||
.fetchAgentTimeline(agentId, {
|
||||
direction: "after",
|
||||
cursor: { epoch: cursor.epoch, seq: cursor.endSeq },
|
||||
cursor: { seq: cursor.endSeq },
|
||||
limit: 0,
|
||||
projection: "canonical",
|
||||
})
|
||||
.catch((error) => {
|
||||
console.warn("[Session] failed to fetch catch-up timeline on resume", agentId, error);
|
||||
@@ -749,13 +748,12 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
);
|
||||
|
||||
const requestCanonicalCatchUp = useCallback(
|
||||
(agentId: string, cursor: { epoch: string; endSeq: number }) => {
|
||||
(agentId: string, cursor: { endSeq: number }) => {
|
||||
void client
|
||||
.fetchAgentTimeline(agentId, {
|
||||
direction: "after",
|
||||
cursor: { epoch: cursor.epoch, seq: cursor.endSeq },
|
||||
cursor: { seq: cursor.endSeq },
|
||||
limit: 0,
|
||||
projection: "canonical",
|
||||
})
|
||||
.catch((error) => {
|
||||
console.warn("[Session] failed to fetch canonical catch-up timeline", agentId, error);
|
||||
@@ -858,7 +856,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
}
|
||||
if (
|
||||
current &&
|
||||
current.epoch === result.cursor.epoch &&
|
||||
current.startSeq === result.cursor.startSeq &&
|
||||
current.endSeq === result.cursor.endSeq
|
||||
) {
|
||||
@@ -963,7 +960,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, epoch } = message.payload;
|
||||
const { agentId, event, timestamp, seq } = message.payload;
|
||||
const parsedTimestamp = new Date(timestamp);
|
||||
const streamEvent = event as AgentStreamEventPayload;
|
||||
if (
|
||||
@@ -1005,7 +1002,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
const result = processAgentStreamEvent({
|
||||
event: streamEvent,
|
||||
seq,
|
||||
epoch,
|
||||
currentTail,
|
||||
currentHead,
|
||||
currentCursor,
|
||||
@@ -1029,8 +1025,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
if (
|
||||
current &&
|
||||
typeof seq === "number" &&
|
||||
typeof epoch === "string" &&
|
||||
current.epoch === epoch &&
|
||||
seq >= current.startSeq &&
|
||||
seq <= current.endSeq
|
||||
) {
|
||||
@@ -1039,7 +1033,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
}
|
||||
if (
|
||||
current &&
|
||||
current.epoch === nextCursor.epoch &&
|
||||
current.startSeq === nextCursor.startSeq &&
|
||||
current.endSeq === nextCursor.endSeq
|
||||
) {
|
||||
@@ -1090,7 +1083,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
||||
const unsubWorkspaceUpdate = client.on("workspace_update", (message) => {
|
||||
if (message.type !== "workspace_update") return;
|
||||
if (message.payload.kind === "remove") {
|
||||
removeWorkspace(serverId, message.payload.id);
|
||||
removeWorkspace(serverId, String(message.payload.id));
|
||||
return;
|
||||
}
|
||||
mergeWorkspaces(serverId, [normalizeWorkspaceDescriptor(message.payload.workspace)]);
|
||||
|
||||
@@ -7,6 +7,7 @@ function createAgent(status: Agent["status"]): Agent {
|
||||
serverId: "server-1",
|
||||
id: "agent-1",
|
||||
provider: "codex",
|
||||
terminal: false,
|
||||
status,
|
||||
createdAt: new Date(0),
|
||||
updatedAt: new Date(0),
|
||||
@@ -19,6 +20,7 @@ function createAgent(status: Agent["status"]): Agent {
|
||||
supportsMcpServers: true,
|
||||
supportsReasoningStream: true,
|
||||
supportsToolInvocations: true,
|
||||
supportsTerminalMode: false,
|
||||
},
|
||||
currentModeId: null,
|
||||
availableModes: [],
|
||||
|
||||
@@ -9,13 +9,9 @@ import {
|
||||
type TimelineCursor,
|
||||
} from "./session-stream-reducers";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Test helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") {
|
||||
return {
|
||||
seqStart: seq,
|
||||
seq,
|
||||
provider: "claude",
|
||||
item: { type, text },
|
||||
timestamp: new Date(1000 + seq).toISOString(),
|
||||
@@ -33,22 +29,30 @@ function makeTimelineEvent(
|
||||
} as AgentStreamEventPayload;
|
||||
}
|
||||
|
||||
function makeUserTimelineEvent(text: string): AgentStreamEventPayload {
|
||||
function makeToolCallEvent(status: "running" | "completed"): AgentStreamEventPayload {
|
||||
return {
|
||||
type: "timeline",
|
||||
provider: "claude",
|
||||
item: { type: "user_message", text },
|
||||
} as AgentStreamEventPayload;
|
||||
item: {
|
||||
type: "tool_call",
|
||||
callId: "call-1",
|
||||
name: "shell",
|
||||
status,
|
||||
detail: {
|
||||
type: "shell",
|
||||
command: "pwd",
|
||||
},
|
||||
error: null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const baseTimelineInput: ProcessTimelineResponseInput = {
|
||||
payload: {
|
||||
agentId: "agent-1",
|
||||
direction: "after",
|
||||
reset: false,
|
||||
epoch: "epoch-1",
|
||||
startCursor: null,
|
||||
endCursor: null,
|
||||
startSeq: null,
|
||||
endSeq: null,
|
||||
entries: [],
|
||||
error: null,
|
||||
},
|
||||
@@ -63,7 +67,6 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
|
||||
const baseStreamInput: ProcessAgentStreamEventInput = {
|
||||
event: makeTimelineEvent("hello"),
|
||||
seq: undefined,
|
||||
epoch: undefined,
|
||||
currentTail: [],
|
||||
currentHead: [],
|
||||
currentCursor: undefined,
|
||||
@@ -71,10 +74,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
|
||||
timestamp: new Date(2000),
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// processTimelineResponse
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describe("processTimelineResponse", () => {
|
||||
it("returns error path when payload.error is set", () => {
|
||||
const result = processTimelineResponse({
|
||||
@@ -93,35 +92,10 @@ describe("processTimelineResponse", () => {
|
||||
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[] = [
|
||||
it("replaces tail during bootstrap tail init and schedules committed catch-up", () => {
|
||||
const provisionalHead: StreamItem[] = [
|
||||
{
|
||||
kind: "assistant_message",
|
||||
id: "head-1",
|
||||
@@ -132,518 +106,280 @@ describe("processTimelineResponse", () => {
|
||||
|
||||
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,
|
||||
currentHead: provisionalHead,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
initRequestDirection: "tail",
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
epoch: "epoch-1",
|
||||
startCursor: { seq: 1 },
|
||||
endCursor: { seq: 5 },
|
||||
startSeq: 1,
|
||||
endSeq: 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,
|
||||
const catchUp = result.sideEffects.find((effect) => effect.type === "catch_up");
|
||||
expect(catchUp).toEqual({
|
||||
type: "catch_up",
|
||||
cursor: { 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")],
|
||||
it("prepends older committed history for before pagination", () => {
|
||||
const currentTail: StreamItem[] = [
|
||||
{
|
||||
kind: "assistant_message",
|
||||
id: "tail-3",
|
||||
text: "newer",
|
||||
timestamp: new Date(3000),
|
||||
},
|
||||
});
|
||||
|
||||
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 currentCursor: TimelineCursor = { startSeq: 3, endSeq: 4 };
|
||||
|
||||
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",
|
||||
currentTail,
|
||||
currentCursor,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "before",
|
||||
entries: [],
|
||||
startSeq: 1,
|
||||
endSeq: 2,
|
||||
entries: [
|
||||
makeTimelineEntry(1, "hello", "user_message"),
|
||||
makeTimelineEntry(2, "older"),
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
// "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);
|
||||
expect(result.cursorChanged).toBe(true);
|
||||
expect(result.cursor).toEqual({
|
||||
startSeq: 1,
|
||||
endSeq: 4,
|
||||
});
|
||||
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]);
|
||||
});
|
||||
|
||||
it("clears initializing even without deferred", () => {
|
||||
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 };
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: false,
|
||||
currentHead,
|
||||
currentCursor,
|
||||
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")],
|
||||
startSeq: 121,
|
||||
endSeq: 121,
|
||||
entries: [makeTimelineEntry(121, "finalized reply")],
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.head).toEqual([]);
|
||||
expect(result.cursorChanged).toBe(true);
|
||||
expect(result.cursor).toEqual({
|
||||
epoch: "epoch-1",
|
||||
startSeq: 1,
|
||||
endSeq: 2,
|
||||
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);
|
||||
});
|
||||
|
||||
it("requests catch-up when committed rows arrive with a forward gap", () => {
|
||||
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentCursor,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "after",
|
||||
startSeq: 125,
|
||||
endSeq: 125,
|
||||
entries: [makeTimelineEntry(125, "far ahead")],
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.cursorChanged).toBe(false);
|
||||
expect(result.tail).toBe(baseTimelineInput.currentTail);
|
||||
expect(result.sideEffects).toContainEqual({
|
||||
type: "catch_up",
|
||||
cursor: { endSeq: 120 },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// processAgentStreamEvent
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describe("processAgentStreamEvent", () => {
|
||||
it("passes through non-timeline events without cursor changes", () => {
|
||||
const turnEvent: AgentStreamEventPayload = {
|
||||
type: "turn_completed",
|
||||
provider: "claude",
|
||||
};
|
||||
|
||||
it("treats seq-less timeline events as provisional head updates", () => {
|
||||
const result = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: turnEvent,
|
||||
event: makeTimelineEvent("partial"),
|
||||
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("accepts timeline event with cursor advance", () => {
|
||||
const existingCursor: TimelineCursor = {
|
||||
epoch: "epoch-1",
|
||||
startSeq: 1,
|
||||
endSeq: 4,
|
||||
};
|
||||
it("appends committed live rows to tail and clears superseded provisional assistant state", () => {
|
||||
const currentHead: StreamItem[] = [
|
||||
{
|
||||
kind: "assistant_message",
|
||||
id: "head-assistant",
|
||||
text: "partial",
|
||||
timestamp: new Date(1000),
|
||||
},
|
||||
];
|
||||
const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 };
|
||||
|
||||
const result = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: makeTimelineEvent("new chunk"),
|
||||
seq: 5,
|
||||
epoch: "epoch-1",
|
||||
currentCursor: existingCursor,
|
||||
event: makeTimelineEvent("finalized reply"),
|
||||
seq: 121,
|
||||
currentHead,
|
||||
currentCursor,
|
||||
});
|
||||
|
||||
expect(result.changedTail).toBe(true);
|
||||
expect(result.changedHead).toBe(true);
|
||||
expect(result.head).toEqual([]);
|
||||
expect(result.cursorChanged).toBe(true);
|
||||
expect(result.cursor).toEqual({
|
||||
epoch: "epoch-1",
|
||||
startSeq: 1,
|
||||
endSeq: 5,
|
||||
endSeq: 121,
|
||||
});
|
||||
expect(result.tail[result.tail.length - 1]).toMatchObject({
|
||||
kind: "assistant_message",
|
||||
text: "finalized reply",
|
||||
});
|
||||
expect(result.sideEffects).toEqual([]);
|
||||
});
|
||||
|
||||
it("detects gap and emits catch-up side effect", () => {
|
||||
const existingCursor: TimelineCursor = {
|
||||
epoch: "epoch-1",
|
||||
startSeq: 1,
|
||||
endSeq: 4,
|
||||
};
|
||||
it("replaces provisional tool progress when the committed tool row arrives", () => {
|
||||
const provisional = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: makeToolCallEvent("running"),
|
||||
seq: undefined,
|
||||
});
|
||||
|
||||
const committed = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: makeToolCallEvent("completed"),
|
||||
seq: 8,
|
||||
currentHead: provisional.head,
|
||||
currentTail: provisional.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",
|
||||
},
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("requests catch-up when a committed live row skips ahead", () => {
|
||||
const result = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: makeTimelineEvent("far ahead"),
|
||||
seq: 10,
|
||||
epoch: "epoch-1",
|
||||
currentCursor: existingCursor,
|
||||
seq: 125,
|
||||
currentCursor: { startSeq: 1, endSeq: 120 },
|
||||
});
|
||||
|
||||
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.sideEffects).toContainEqual({
|
||||
type: "catch_up",
|
||||
cursor: { endSeq: 120 },
|
||||
});
|
||||
});
|
||||
|
||||
it("clears provisional head on terminal turn events without committing it to tail", () => {
|
||||
const result = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: {
|
||||
type: "turn_completed",
|
||||
provider: "claude",
|
||||
},
|
||||
currentHead: [
|
||||
{
|
||||
kind: "thought",
|
||||
id: "reasoning-1",
|
||||
text: "thinking",
|
||||
timestamp: new Date(1000),
|
||||
status: "loading",
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
expect(result.changedHead).toBe(true);
|
||||
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);
|
||||
expect(result.head).toEqual([]);
|
||||
expect(result.tail).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
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 { hydrateStreamState, reduceStreamUpdate } from "@/types/stream";
|
||||
import {
|
||||
classifySessionTimelineSeq,
|
||||
type SessionTimelineSeqDecision,
|
||||
@@ -12,38 +12,25 @@ 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: { epoch: string; endSeq: number } }
|
||||
| { type: "catch_up"; cursor: { endSeq: number } }
|
||||
| { type: "flush_pending_updates" };
|
||||
|
||||
export type AgentStreamReducerSideEffect = {
|
||||
type: "catch_up";
|
||||
cursor: { epoch: string; endSeq: number };
|
||||
cursor: { endSeq: number };
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// processTimelineResponse
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type TimelineDirection = "tail" | "before" | "after";
|
||||
type InitRequestDirection = "tail" | "after";
|
||||
|
||||
type TimelineResponseEntry = {
|
||||
seqStart: number;
|
||||
seq: number;
|
||||
provider: string;
|
||||
item: Record<string, unknown>;
|
||||
timestamp: string;
|
||||
@@ -53,10 +40,8 @@ export interface ProcessTimelineResponseInput {
|
||||
payload: {
|
||||
agentId: string;
|
||||
direction: TimelineDirection;
|
||||
reset: boolean;
|
||||
epoch: string;
|
||||
startCursor: { seq: number } | null;
|
||||
endCursor: { seq: number } | null;
|
||||
startSeq: number | null;
|
||||
endSeq: number | null;
|
||||
entries: TimelineResponseEntry[];
|
||||
error: string | null;
|
||||
};
|
||||
@@ -79,205 +64,9 @@ 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) {
|
||||
nextTail = acceptedUnits.reduce<StreamItem[]>(
|
||||
(state, { event, timestamp }) =>
|
||||
reduceStreamUpdate(state, event, timestamp, {
|
||||
source: "canonical",
|
||||
}),
|
||||
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;
|
||||
@@ -307,75 +96,250 @@ 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, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } =
|
||||
input;
|
||||
const { event, seq, currentTail, currentHead, currentCursor, currentAgent, timestamp } = input;
|
||||
|
||||
let shouldApplyStreamEvent = true;
|
||||
let nextTail = currentTail;
|
||||
let nextHead = currentHead;
|
||||
let changedTail = false;
|
||||
let changedHead = false;
|
||||
let nextTimelineCursor: TimelineCursor | null = null;
|
||||
let cursorChanged = false;
|
||||
const sideEffects: AgentStreamReducerSideEffect[] = [];
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Timeline sequencing gate
|
||||
// ------------------------------------------------------------------
|
||||
if (event.type === "timeline" && typeof seq === "number" && typeof epoch === "string") {
|
||||
if (event.type === "timeline" && typeof seq === "number") {
|
||||
const decision = classifySessionTimelineSeq({
|
||||
cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null,
|
||||
epoch,
|
||||
cursor: currentCursor ? { endSeq: currentCursor.endSeq } : null,
|
||||
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 (decision === "gap") {
|
||||
if (currentCursor) {
|
||||
sideEffects.push({
|
||||
type: "catch_up",
|
||||
cursor: {
|
||||
epoch: currentCursor.epoch,
|
||||
endSeq: currentCursor.endSeq,
|
||||
},
|
||||
cursor: { endSeq: currentCursor.endSeq },
|
||||
});
|
||||
}
|
||||
} else {
|
||||
// drop_stale or drop_epoch
|
||||
shouldApplyStreamEvent = false;
|
||||
} else if (decision !== "drop_stale") {
|
||||
nextTail = reduceStreamUpdate(currentTail, event, timestamp, {
|
||||
source: "canonical",
|
||||
});
|
||||
changedTail = nextTail !== currentTail;
|
||||
|
||||
nextHead = removeSupersededProvisionalItems(currentHead, event);
|
||||
changedHead = nextHead !== currentHead;
|
||||
|
||||
nextTimelineCursor =
|
||||
decision === "init"
|
||||
? { startSeq: seq, endSeq: seq }
|
||||
: { ...(currentCursor ?? { startSeq: seq, endSeq: seq }), endSeq: seq };
|
||||
cursorChanged = !cursorsEqual(currentCursor, nextTimelineCursor);
|
||||
}
|
||||
} else if (event.type === "timeline") {
|
||||
nextHead = reduceStreamUpdate(currentHead, event, timestamp, {
|
||||
source: "live",
|
||||
});
|
||||
changedHead = nextHead !== currentHead;
|
||||
} else if (
|
||||
(event.type === "turn_completed" ||
|
||||
event.type === "turn_canceled" ||
|
||||
event.type === "turn_failed") &&
|
||||
currentHead.length > 0
|
||||
) {
|
||||
nextHead = [];
|
||||
changedHead = true;
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Apply stream event to tail/head
|
||||
// ------------------------------------------------------------------
|
||||
const { tail, head, changedTail, changedHead } = shouldApplyStreamEvent
|
||||
? applyStreamEvent({
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
})
|
||||
: {
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
changedTail: false,
|
||||
changedHead: false,
|
||||
};
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Optimistic lifecycle status
|
||||
// ------------------------------------------------------------------
|
||||
let agentPatch: AgentPatch | null = null;
|
||||
let agentChanged = false;
|
||||
|
||||
@@ -402,8 +366,8 @@ export function processAgentStreamEvent(
|
||||
}
|
||||
|
||||
return {
|
||||
tail,
|
||||
head,
|
||||
tail: nextTail,
|
||||
head: nextHead,
|
||||
changedTail,
|
||||
changedHead,
|
||||
cursor: nextTimelineCursor,
|
||||
|
||||
@@ -2,26 +2,42 @@ import { describe, expect, it } from "vitest";
|
||||
import { classifySessionTimelineSeq } from "./session-timeline-seq-gate";
|
||||
import {
|
||||
deriveBootstrapTailTimelinePolicy,
|
||||
deriveInitialTimelineRequest,
|
||||
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,
|
||||
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,
|
||||
});
|
||||
|
||||
expect(policy.replace).toBe(true);
|
||||
expect(policy.catchUpCursor).toBeNull();
|
||||
});
|
||||
|
||||
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("forces baseline replace and canonical catch-up for init tail race", () => {
|
||||
const advancedCursor = { epoch: "epoch-1", endSeq: 205 };
|
||||
const advancedCursor = { endSeq: 205 };
|
||||
const tailSeqStart = 101;
|
||||
const tailSeqEnd = 200;
|
||||
|
||||
@@ -29,7 +45,6 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
|
||||
for (let seq = tailSeqStart; seq <= tailSeqEnd; seq += 1) {
|
||||
const decision = classifySessionTimelineSeq({
|
||||
cursor: advancedCursor,
|
||||
epoch: "epoch-1",
|
||||
seq,
|
||||
});
|
||||
if (decision === "accept" || decision === "init") {
|
||||
@@ -40,16 +55,13 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
|
||||
|
||||
const policy = deriveBootstrapTailTimelinePolicy({
|
||||
direction: "tail",
|
||||
reset: false,
|
||||
epoch: "epoch-1",
|
||||
endCursor: { seq: 200 },
|
||||
endSeq: 200,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
});
|
||||
|
||||
expect(policy.replace).toBe(true);
|
||||
expect(policy.catchUpCursor).toEqual({
|
||||
epoch: "epoch-1",
|
||||
endSeq: 200,
|
||||
});
|
||||
});
|
||||
@@ -57,9 +69,7 @@ describe("deriveBootstrapTailTimelinePolicy", () => {
|
||||
it("does not replace non-bootstrap, non-reset responses", () => {
|
||||
const policy = deriveBootstrapTailTimelinePolicy({
|
||||
direction: "tail",
|
||||
reset: false,
|
||||
epoch: "epoch-1",
|
||||
endCursor: { seq: 200 },
|
||||
endSeq: 200,
|
||||
isInitializing: false,
|
||||
hasActiveInitDeferred: false,
|
||||
});
|
||||
@@ -77,7 +87,6 @@ describe("shouldResolveTimelineInit", () => {
|
||||
isInitializing: true,
|
||||
initRequestDirection: "tail",
|
||||
responseDirection: "tail",
|
||||
reset: false,
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
@@ -89,7 +98,6 @@ describe("shouldResolveTimelineInit", () => {
|
||||
isInitializing: true,
|
||||
initRequestDirection: "tail",
|
||||
responseDirection: "after",
|
||||
reset: false,
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
@@ -101,7 +109,6 @@ describe("shouldResolveTimelineInit", () => {
|
||||
isInitializing: true,
|
||||
initRequestDirection: "after",
|
||||
responseDirection: "after",
|
||||
reset: false,
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
@@ -6,7 +6,6 @@ type BootstrapTailCursor = {
|
||||
} | null;
|
||||
|
||||
type InitialTimelineCursor = {
|
||||
epoch: string;
|
||||
seq: number;
|
||||
} | null;
|
||||
|
||||
@@ -20,48 +19,37 @@ export function deriveInitialTimelineRequest({
|
||||
initialTimelineLimit: number;
|
||||
}): {
|
||||
direction: "tail" | "after";
|
||||
cursor?: { epoch: string; seq: number };
|
||||
cursor?: { seq: number };
|
||||
limit: number;
|
||||
projection: "canonical";
|
||||
} {
|
||||
if (!hasAuthoritativeHistory || !cursor) {
|
||||
return {
|
||||
direction: "tail",
|
||||
limit: initialTimelineLimit,
|
||||
projection: "canonical",
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
direction: "after",
|
||||
cursor: { epoch: cursor.epoch, seq: cursor.seq },
|
||||
cursor: { seq: cursor.seq },
|
||||
limit: 0,
|
||||
projection: "canonical",
|
||||
};
|
||||
}
|
||||
|
||||
export function deriveBootstrapTailTimelinePolicy({
|
||||
direction,
|
||||
reset,
|
||||
epoch,
|
||||
endCursor,
|
||||
endSeq,
|
||||
isInitializing,
|
||||
hasActiveInitDeferred,
|
||||
}: {
|
||||
direction: TimelineDirection;
|
||||
reset: boolean;
|
||||
epoch: string;
|
||||
endCursor: BootstrapTailCursor;
|
||||
endSeq: number | null;
|
||||
isInitializing: boolean;
|
||||
hasActiveInitDeferred: boolean;
|
||||
}): {
|
||||
replace: boolean;
|
||||
catchUpCursor: { epoch: string; endSeq: number } | null;
|
||||
catchUpCursor: { endSeq: number } | null;
|
||||
} {
|
||||
if (reset) {
|
||||
return { replace: true, catchUpCursor: null };
|
||||
}
|
||||
|
||||
const isBootstrapTailInit = direction === "tail" && isInitializing && hasActiveInitDeferred;
|
||||
if (!isBootstrapTailInit) {
|
||||
return { replace: false, catchUpCursor: null };
|
||||
@@ -69,7 +57,7 @@ export function deriveBootstrapTailTimelinePolicy({
|
||||
|
||||
return {
|
||||
replace: true,
|
||||
catchUpCursor: endCursor ? { epoch, endSeq: endCursor.seq } : null,
|
||||
catchUpCursor: typeof endSeq === "number" ? { endSeq } : null,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -78,19 +66,14 @@ 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,8 +5,7 @@ describe("classifySessionTimelineSeq", () => {
|
||||
it("accepts contiguous forward seq", () => {
|
||||
expect(
|
||||
classifySessionTimelineSeq({
|
||||
cursor: { epoch: "epoch-1", endSeq: 4 },
|
||||
epoch: "epoch-1",
|
||||
cursor: { endSeq: 4 },
|
||||
seq: 5,
|
||||
}),
|
||||
).toBe("accept");
|
||||
@@ -15,8 +14,7 @@ describe("classifySessionTimelineSeq", () => {
|
||||
it("drops stale seq older than the current end", () => {
|
||||
expect(
|
||||
classifySessionTimelineSeq({
|
||||
cursor: { epoch: "epoch-1", endSeq: 8 },
|
||||
epoch: "epoch-1",
|
||||
cursor: { endSeq: 8 },
|
||||
seq: 7,
|
||||
}),
|
||||
).toBe("drop_stale");
|
||||
@@ -25,28 +23,16 @@ describe("classifySessionTimelineSeq", () => {
|
||||
it("drops duplicate replay seq equal to the current end", () => {
|
||||
expect(
|
||||
classifySessionTimelineSeq({
|
||||
cursor: { epoch: "epoch-1", endSeq: 8 },
|
||||
epoch: "epoch-1",
|
||||
cursor: { endSeq: 8 },
|
||||
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");
|
||||
@@ -55,8 +41,7 @@ describe("classifySessionTimelineSeq", () => {
|
||||
it("classifies forward gaps", () => {
|
||||
expect(
|
||||
classifySessionTimelineSeq({
|
||||
cursor: { epoch: "epoch-1", endSeq: 4 },
|
||||
epoch: "epoch-1",
|
||||
cursor: { endSeq: 4 },
|
||||
seq: 9,
|
||||
}),
|
||||
).toBe("gap");
|
||||
|
||||
@@ -1,28 +1,22 @@
|
||||
export type SessionTimelineSeqCursor =
|
||||
| {
|
||||
epoch: string;
|
||||
endSeq: number;
|
||||
}
|
||||
| null
|
||||
| undefined;
|
||||
|
||||
export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "drop_epoch" | "gap" | "init";
|
||||
export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "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";
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user