Keep older chat history and image previews stable (#2490)
* fix(app): keep timeline history and previews stable Treat persisted timeline replicas as display-only so authoritative pagination always comes from the daemon. Require renewed user intent between older-history pages unless the viewport genuinely remains at the history edge. Model assistant image acquisition as loading, loaded, or failed so recreating a preview URL cannot be rendered as an error. * fix(app): keep file image previews current * fix(app): stabilize history pagination edges * fix(app): recover transient timeline loads * fix(app): retain assistant image previews * Stabilize history pagination lifecycle * Stabilize history settlement and image previews * Retry file images after reconnect * Keep active previews and pagination requests alive * Replace image hook tests with typed ports * Integrate timeline hydration with submission authority * Harden pre-hydration timeline reconciliation * Make rewind E2E use real scroll intent * Preserve live timeline state through hydration * Keep mounted image attachments retained * Protect preview persistence and lifecycle hydration * Preserve idless live assistant continuations * Preserve timeline rows across delayed hydration * test(app): cover real assistant image files * Preserve assistant tool ordering during hydration
This commit is contained in:
@@ -173,6 +173,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
|
||||
hasActiveInitDeferred: false,
|
||||
initRequestDirection: "tail",
|
||||
sendingClientMessageIds: [],
|
||||
hasAuthoritativeBaseline: true,
|
||||
};
|
||||
|
||||
const baseStreamInput: ProcessAgentStreamEventInput = {
|
||||
@@ -362,6 +363,442 @@ describe("processTimelineResponse", () => {
|
||||
expect(result.head).toEqual([liveThought, liveAssistant]);
|
||||
});
|
||||
|
||||
it("keeps a newer live tool completion when bootstrap contains the running call", () => {
|
||||
const callId = "toolu_live_completion";
|
||||
const liveCompletion = hydrateStreamState(
|
||||
[
|
||||
{
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "claude",
|
||||
item: makeToolCallTimelineEntry(2, callId, "completed", {
|
||||
type: "read",
|
||||
filePath: "/tmp/example.ts",
|
||||
}).item,
|
||||
} as AgentStreamEventPayload,
|
||||
timestamp: new Date(2000),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 2 },
|
||||
},
|
||||
],
|
||||
{ source: "canonical" },
|
||||
);
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentHead: liveCompletion,
|
||||
hasAuthoritativeBaseline: false,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
reset: false,
|
||||
epoch: "epoch-1",
|
||||
startCursor: { seq: 1 },
|
||||
endCursor: { seq: 1 },
|
||||
entries: [
|
||||
makeToolCallTimelineEntry(1, callId, "running", {
|
||||
type: "unknown",
|
||||
input: { file_path: "/tmp/example.ts" },
|
||||
output: null,
|
||||
}),
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect(getAgentToolCalls([...result.tail, ...result.head])).toEqual([
|
||||
expect.objectContaining({
|
||||
timelineCursor: { epoch: "epoch-1", seq: 2 },
|
||||
payload: expect.objectContaining({
|
||||
data: expect.objectContaining({ callId, status: "completed" }),
|
||||
}),
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps one newer todo state when bootstrap contains its older state", () => {
|
||||
const liveTodo = hydrateStreamState(
|
||||
[
|
||||
{
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: {
|
||||
type: "todo",
|
||||
items: [{ text: "Verify hydration", completed: true }],
|
||||
},
|
||||
} as AgentStreamEventPayload,
|
||||
timestamp: new Date(2000),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 2 },
|
||||
},
|
||||
],
|
||||
{ source: "canonical" },
|
||||
);
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentHead: liveTodo,
|
||||
hasAuthoritativeBaseline: false,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
reset: false,
|
||||
epoch: "epoch-1",
|
||||
startCursor: { seq: 1 },
|
||||
endCursor: { seq: 1 },
|
||||
entries: [
|
||||
{
|
||||
seqStart: 1,
|
||||
seqEnd: 1,
|
||||
provider: "codex",
|
||||
item: {
|
||||
type: "todo",
|
||||
items: [{ text: "Verify hydration", completed: false }],
|
||||
},
|
||||
timestamp: new Date(1000).toISOString(),
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect([...result.tail, ...result.head].filter((item) => item.kind === "todo_list")).toEqual([
|
||||
expect.objectContaining({
|
||||
timelineCursor: { epoch: "epoch-1", seq: 2 },
|
||||
items: [{ text: "Verify hydration", completed: true }],
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps one completed compaction when bootstrap contains its loading state", () => {
|
||||
const liveCompaction = hydrateStreamState(
|
||||
[
|
||||
{
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: { type: "compaction", status: "completed", trigger: "auto", preTokens: 1200 },
|
||||
} as AgentStreamEventPayload,
|
||||
timestamp: new Date(2000),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 2 },
|
||||
},
|
||||
],
|
||||
{ source: "canonical" },
|
||||
);
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentHead: liveCompaction,
|
||||
hasAuthoritativeBaseline: false,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
reset: false,
|
||||
epoch: "epoch-1",
|
||||
startCursor: { seq: 1 },
|
||||
endCursor: { seq: 1 },
|
||||
entries: [
|
||||
{
|
||||
seqStart: 1,
|
||||
seqEnd: 1,
|
||||
provider: "codex",
|
||||
item: { type: "compaction", status: "loading", trigger: "auto" },
|
||||
timestamp: new Date(1000).toISOString(),
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect([...result.tail, ...result.head].filter((item) => item.kind === "compaction")).toEqual([
|
||||
expect.objectContaining({
|
||||
timelineCursor: { epoch: "epoch-1", seq: 2 },
|
||||
status: "completed",
|
||||
trigger: "auto",
|
||||
preTokens: 1200,
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("replaces a painted replica with an empty authoritative timeline", () => {
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: null,
|
||||
endCursor: null,
|
||||
entries: [],
|
||||
hasOlder: false,
|
||||
},
|
||||
});
|
||||
|
||||
expect(result.tail).toEqual([]);
|
||||
expect(result.cursor).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps a newer live assistant continuation when bootstrap ends on the same message", () => {
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
currentHead: [
|
||||
{
|
||||
kind: "assistant_message",
|
||||
id: "assistant-live",
|
||||
messageId: "assistant-live",
|
||||
text: " continuation",
|
||||
timestamp: new Date(2001),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 101 },
|
||||
},
|
||||
],
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: { seq: 61 },
|
||||
endCursor: { seq: 100 },
|
||||
entries: [
|
||||
{
|
||||
...makeTimelineEntry(100, "canonical prefix"),
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
text: "canonical prefix",
|
||||
messageId: "assistant-live",
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
const assistants = [...result.tail, ...result.head].filter(
|
||||
(item) => item.kind === "assistant_message",
|
||||
);
|
||||
expect(assistants).toHaveLength(1);
|
||||
expect(assistants[0]?.text).toBe("canonical prefix continuation");
|
||||
expect(assistants[0]?.timelineCursor).toEqual({ epoch: "epoch-1", seq: 101 });
|
||||
});
|
||||
|
||||
it("keeps assistant segments separate when a live tool sits between them", () => {
|
||||
const messageId = "assistant-with-tool";
|
||||
const liveTool = hydrateStreamState(
|
||||
[
|
||||
{
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "claude",
|
||||
item: makeToolCallTimelineEntry(101, "tool-between-segments", "completed", {
|
||||
type: "read",
|
||||
filePath: "/tmp/example.ts",
|
||||
}).item,
|
||||
} as AgentStreamEventPayload,
|
||||
timestamp: new Date(2101),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 101 },
|
||||
},
|
||||
{
|
||||
event: makeAssistantTimelineEvent("after tool", messageId),
|
||||
timestamp: new Date(2102),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 102 },
|
||||
},
|
||||
],
|
||||
{ source: "canonical" },
|
||||
);
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
currentHead: liveTool,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: { seq: 61 },
|
||||
endCursor: { seq: 100 },
|
||||
entries: [
|
||||
{
|
||||
...makeTimelineEntry(100, "before tool"),
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
text: "before tool",
|
||||
messageId,
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([
|
||||
"assistant_message",
|
||||
"tool_call",
|
||||
"assistant_message",
|
||||
]);
|
||||
expect(getAssistantTexts([...result.tail, ...result.head])).toEqual([
|
||||
"before tool",
|
||||
"after tool",
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps a newer live assistant continuation without a provider message ID", () => {
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
currentHead: [
|
||||
{
|
||||
kind: "assistant_message",
|
||||
id: "assistant-live",
|
||||
text: "canonical prefix continuation",
|
||||
timestamp: new Date(2001),
|
||||
timelineCursor: { epoch: "epoch-1", seq: 101 },
|
||||
},
|
||||
],
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: { seq: 61 },
|
||||
endCursor: { seq: 100 },
|
||||
entries: [makeTimelineEntry(100, "canonical prefix")],
|
||||
},
|
||||
});
|
||||
|
||||
const assistants = [...result.tail, ...result.head].filter(
|
||||
(item) => item.kind === "assistant_message",
|
||||
);
|
||||
expect(assistants).toHaveLength(1);
|
||||
expect(assistants[0]?.text).toBe("canonical prefix continuation");
|
||||
expect(assistants[0]?.timelineCursor).toEqual({ epoch: "epoch-1", seq: 101 });
|
||||
});
|
||||
|
||||
it("keeps a canonical live user row newer than the bootstrap page", () => {
|
||||
const live = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "claude",
|
||||
item: {
|
||||
type: "user_message",
|
||||
text: "remote live prompt",
|
||||
clientMessageId: "remote-client-message",
|
||||
},
|
||||
},
|
||||
seq: 101,
|
||||
epoch: "epoch-1",
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
hasAuthoritativeBaseline: false,
|
||||
});
|
||||
expect(live.head[0]).toMatchObject({
|
||||
kind: "user_message",
|
||||
timelineCursor: { epoch: "epoch-1", seq: 101 },
|
||||
});
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: live.tail,
|
||||
currentHead: live.head,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: { seq: 61 },
|
||||
endCursor: { seq: 100 },
|
||||
entries: [makeTimelineEntry(100, "canonical answer")],
|
||||
},
|
||||
});
|
||||
|
||||
expect(getUserTexts(result.head)).toEqual(["remote live prompt"]);
|
||||
});
|
||||
|
||||
it("keeps a provider-acknowledged painted-tail prompt newer than the bootstrap page", () => {
|
||||
const submitted = makeSubmittedUserMessage("submitted before bootstrap", "client-message-1");
|
||||
const live = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "claude",
|
||||
item: {
|
||||
type: "user_message",
|
||||
text: "canonical presentation",
|
||||
clientMessageId: "client-message-1",
|
||||
messageId: "provider-message-1",
|
||||
},
|
||||
},
|
||||
seq: 51,
|
||||
epoch: "epoch-1",
|
||||
currentTail: [makeAssistantItem("painted replica"), submitted],
|
||||
currentCursor: undefined,
|
||||
hasAuthoritativeBaseline: false,
|
||||
});
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: live.tail,
|
||||
currentHead: live.head,
|
||||
sendingClientMessageIds: [],
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: { seq: 11 },
|
||||
endCursor: { seq: 50 },
|
||||
entries: [makeTimelineEntry(50, "canonical answer")],
|
||||
},
|
||||
});
|
||||
|
||||
expect(getUserTexts([...result.tail, ...result.head])).toEqual(["submitted before bootstrap"]);
|
||||
expect(result.head[0]).toMatchObject({
|
||||
kind: "user_message",
|
||||
clientMessageId: "client-message-1",
|
||||
messageId: "provider-message-1",
|
||||
timelineCursor: { epoch: "epoch-1", seq: 51 },
|
||||
});
|
||||
});
|
||||
|
||||
it("does not duplicate a live row already covered by the bootstrap page", () => {
|
||||
const live = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: makeTimelineEvent("thinking", "reasoning"),
|
||||
seq: 100,
|
||||
epoch: "epoch-1",
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
hasAuthoritativeBaseline: false,
|
||||
});
|
||||
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
currentTail: live.tail,
|
||||
currentHead: live.head,
|
||||
isInitializing: true,
|
||||
hasActiveInitDeferred: true,
|
||||
hasAuthoritativeBaseline: false,
|
||||
payload: {
|
||||
...baseTimelineInput.payload,
|
||||
direction: "tail",
|
||||
startCursor: { seq: 61 },
|
||||
endCursor: { seq: 100 },
|
||||
entries: [makeTimelineEntry(100, "thinking", "reasoning")],
|
||||
},
|
||||
});
|
||||
|
||||
expect([...result.tail, ...result.head].filter((item) => item.kind === "thought")).toHaveLength(
|
||||
1,
|
||||
);
|
||||
});
|
||||
|
||||
it("uses the timeline entry timestamp as canonical", () => {
|
||||
const result = processTimelineResponse({
|
||||
...baseTimelineInput,
|
||||
@@ -1873,7 +2310,10 @@ describe("processTimelineResponse", () => {
|
||||
|
||||
expect(getAssistantTexts(result.tail)).toEqual(["older chunk newer chunk"]);
|
||||
expect(result.tail[0]).toEqual(
|
||||
expect.objectContaining({ timelineCursor: { epoch: "epoch-1", seq: 3 } }),
|
||||
expect.objectContaining({
|
||||
id: "assistant-newer",
|
||||
timelineCursor: { epoch: "epoch-1", seq: 3 },
|
||||
}),
|
||||
);
|
||||
expect(result.cursor).toEqual({
|
||||
epoch: "epoch-1",
|
||||
@@ -2417,6 +2857,61 @@ describe("processAgentStreamEvent", () => {
|
||||
endSeq: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it("renders a live row over painted items without creating an authoritative cursor", () => {
|
||||
const result = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: makeTimelineEvent("live before bootstrap", "user_message"),
|
||||
seq: 51,
|
||||
epoch: "epoch-1",
|
||||
currentTail: [makeAssistantItem("painted replica")],
|
||||
currentCursor: undefined,
|
||||
hasAuthoritativeBaseline: false,
|
||||
});
|
||||
|
||||
expect(getAssistantTexts(result.tail)).toEqual(["painted replica"]);
|
||||
expect(getUserTexts(result.head)).toEqual(["live before bootstrap"]);
|
||||
expect(result.head[0]).toMatchObject({
|
||||
kind: "user_message",
|
||||
timelineCursor: { epoch: "epoch-1", seq: 51 },
|
||||
});
|
||||
expect(result.cursorChanged).toBe(false);
|
||||
expect(result.cursor).toBeNull();
|
||||
expect(result.sideEffects).toEqual([]);
|
||||
});
|
||||
|
||||
it("reconciles a pre-bootstrap echo with its submitted row", () => {
|
||||
const submitted = makeSubmittedUserMessage("submitted before bootstrap", "client-message-1");
|
||||
const result = processAgentStreamEvent({
|
||||
...baseStreamInput,
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "claude",
|
||||
item: {
|
||||
type: "user_message",
|
||||
text: "canonical presentation",
|
||||
clientMessageId: "client-message-1",
|
||||
messageId: "provider-message-1",
|
||||
},
|
||||
},
|
||||
seq: 51,
|
||||
epoch: "epoch-1",
|
||||
currentTail: [makeAssistantItem("painted replica"), submitted],
|
||||
currentCursor: undefined,
|
||||
hasAuthoritativeBaseline: false,
|
||||
});
|
||||
|
||||
const users = [...result.tail, ...result.head].filter((item) => item.kind === "user_message");
|
||||
expect(users).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "client-message-1",
|
||||
clientMessageId: "client-message-1",
|
||||
messageId: "provider-message-1",
|
||||
text: "submitted before bootstrap",
|
||||
timelineCursor: { epoch: "epoch-1", seq: 51 },
|
||||
}),
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("processAgentStreamEvents", () => {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
|
||||
import { useSessionStore } from "@/stores/session-store";
|
||||
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
|
||||
import type { AssistantMessageItem, StreamItem } from "@/types/stream";
|
||||
import {
|
||||
applyStreamEvent,
|
||||
@@ -89,6 +89,7 @@ export interface ProcessTimelineResponseInput {
|
||||
hasActiveInitDeferred: boolean;
|
||||
initRequestDirection: InitRequestDirection;
|
||||
sendingClientMessageIds: readonly string[];
|
||||
hasAuthoritativeBaseline: boolean;
|
||||
}
|
||||
|
||||
export interface ProcessTimelineResponseOutput {
|
||||
@@ -233,6 +234,10 @@ function applyTimelineReplacePath(args: {
|
||||
previousHead: currentHead,
|
||||
sendingClientMessageIds,
|
||||
preserveLiveHead,
|
||||
canonicalCoverage: {
|
||||
epoch: payload.epoch,
|
||||
endSeq: payload.endCursor?.seq ?? null,
|
||||
},
|
||||
});
|
||||
const cursor: TimelineCursor | null =
|
||||
payload.startCursor && payload.endCursor
|
||||
@@ -255,7 +260,6 @@ function applyTimelineReplacePath(args: {
|
||||
acknowledgedClientMessageIds,
|
||||
};
|
||||
}
|
||||
|
||||
interface IncrementalAcceptResult {
|
||||
acceptedUnits: TimelineUnit[];
|
||||
cursor: TimelineCursor | undefined;
|
||||
@@ -406,16 +410,15 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
|
||||
return [...olderTail, ...currentTail];
|
||||
}
|
||||
|
||||
return [
|
||||
...olderTail.slice(0, -1),
|
||||
{
|
||||
...olderLast,
|
||||
text: `${olderLast.text}${currentFirst.text}`,
|
||||
timestamp: currentFirst.timestamp,
|
||||
...(currentFirst.timelineCursor ? { timelineCursor: currentFirst.timelineCursor } : {}),
|
||||
},
|
||||
...currentTail.slice(1),
|
||||
];
|
||||
const mergedAssistant: AssistantMessageItem = {
|
||||
...currentFirst,
|
||||
text: `${olderLast.text}${currentFirst.text}`,
|
||||
};
|
||||
if (mergedAssistant.messageId === undefined && olderLast.messageId !== undefined) {
|
||||
mergedAssistant.messageId = olderLast.messageId;
|
||||
}
|
||||
|
||||
return [...olderTail.slice(0, -1), mergedAssistant, ...currentTail.slice(1)];
|
||||
}
|
||||
|
||||
function replaceLiveAssistantWithProjectedText(params: {
|
||||
@@ -820,6 +823,7 @@ export function processTimelineResponse(
|
||||
hasActiveInitDeferred,
|
||||
initRequestDirection,
|
||||
sendingClientMessageIds,
|
||||
hasAuthoritativeBaseline,
|
||||
} = input;
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
@@ -891,7 +895,8 @@ export function processTimelineResponse(
|
||||
currentTail,
|
||||
currentHead,
|
||||
sendingClientMessageIds,
|
||||
preserveLiveHead: currentCursor?.epoch === payload.epoch,
|
||||
preserveLiveHead:
|
||||
currentCursor?.epoch === payload.epoch || (!payload.reset && !hasAuthoritativeBaseline),
|
||||
toHydratedEvents,
|
||||
})
|
||||
: applyTimelineIncrementalPath({
|
||||
@@ -955,6 +960,7 @@ export interface ProcessAgentStreamEventInput {
|
||||
currentTail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
currentCursor: TimelineCursor | undefined;
|
||||
hasAuthoritativeBaseline?: boolean;
|
||||
timestamp: Date;
|
||||
}
|
||||
|
||||
@@ -989,6 +995,7 @@ export interface ProcessAgentStreamEventsInput {
|
||||
currentTail: StreamItem[];
|
||||
currentHead: StreamItem[];
|
||||
currentCursor: TimelineCursor | undefined;
|
||||
hasAuthoritativeBaseline?: boolean;
|
||||
}
|
||||
|
||||
export type AgentStreamReducerSnapshot = Omit<ProcessAgentStreamEventsInput, "events">;
|
||||
@@ -1017,8 +1024,9 @@ function processTimelineSequencingGate(input: {
|
||||
seq: number | undefined;
|
||||
epoch: string | undefined;
|
||||
currentCursor: TimelineCursor | undefined;
|
||||
hasAuthoritativeBaseline: boolean;
|
||||
}): TimelineSequencingGateResult {
|
||||
const { event, seq, epoch, currentCursor } = input;
|
||||
const { event, seq, epoch, currentCursor, hasAuthoritativeBaseline } = input;
|
||||
const base: TimelineSequencingGateResult = {
|
||||
shouldApplyStreamEvent: true,
|
||||
nextTimelineCursor: null,
|
||||
@@ -1029,6 +1037,9 @@ function processTimelineSequencingGate(input: {
|
||||
if (event.type !== "timeline" || typeof seq !== "number" || typeof epoch !== "string") {
|
||||
return base;
|
||||
}
|
||||
if (!hasAuthoritativeBaseline) {
|
||||
return base;
|
||||
}
|
||||
|
||||
const decision = classifySessionTimelineSeq({
|
||||
cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null,
|
||||
@@ -1085,9 +1096,24 @@ function processTimelineSequencingGate(input: {
|
||||
export function processAgentStreamEvent(
|
||||
input: ProcessAgentStreamEventInput,
|
||||
): ProcessAgentStreamEventOutput {
|
||||
const { event, seq, epoch, currentTail, currentHead, currentCursor, timestamp } = input;
|
||||
const {
|
||||
event,
|
||||
seq,
|
||||
epoch,
|
||||
currentTail,
|
||||
currentHead,
|
||||
currentCursor,
|
||||
timestamp,
|
||||
hasAuthoritativeBaseline = true,
|
||||
} = input;
|
||||
|
||||
const sequencing = processTimelineSequencingGate({ event, seq, epoch, currentCursor });
|
||||
const sequencing = processTimelineSequencingGate({
|
||||
event,
|
||||
seq,
|
||||
epoch,
|
||||
currentCursor,
|
||||
hasAuthoritativeBaseline,
|
||||
});
|
||||
const timelineCursor =
|
||||
event.type === "timeline" && seq !== undefined && epoch !== undefined
|
||||
? { epoch, seq }
|
||||
@@ -1096,30 +1122,61 @@ export function processAgentStreamEvent(
|
||||
// ------------------------------------------------------------------
|
||||
// Apply stream event to tail/head
|
||||
// ------------------------------------------------------------------
|
||||
const applied = sequencing.shouldApplyStreamEvent
|
||||
? applyStreamEvent({
|
||||
tail: sequencing.resetLiveTimeline ? [] : currentTail,
|
||||
head: sequencing.resetLiveTimeline ? [] : currentHead,
|
||||
let streamResult: ReturnType<typeof applyStreamEvent>;
|
||||
if (!sequencing.shouldApplyStreamEvent) {
|
||||
streamResult = {
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
changedTail: false,
|
||||
changedHead: false,
|
||||
};
|
||||
} else if (!hasAuthoritativeBaseline) {
|
||||
if (event.type === "timeline" && event.item.type === "user_message") {
|
||||
streamResult = applyStreamEvent({
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
})
|
||||
: {
|
||||
unmatchedUserMessageInsert: "head",
|
||||
});
|
||||
} else {
|
||||
const overlay = applyStreamEvent({
|
||||
tail: currentHead,
|
||||
head: [],
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
});
|
||||
streamResult = {
|
||||
tail: currentTail,
|
||||
head: currentHead,
|
||||
head: [...overlay.tail, ...overlay.head],
|
||||
changedTail: false,
|
||||
changedHead: false,
|
||||
changedHead: overlay.changedTail || overlay.changedHead,
|
||||
};
|
||||
}
|
||||
} else {
|
||||
streamResult = applyStreamEvent({
|
||||
tail: sequencing.resetLiveTimeline ? [] : currentTail,
|
||||
head: sequencing.resetLiveTimeline ? [] : currentHead,
|
||||
event,
|
||||
timestamp,
|
||||
source: "live",
|
||||
timelineCursor,
|
||||
});
|
||||
}
|
||||
const { tail, head, changedTail, changedHead } = streamResult;
|
||||
|
||||
return {
|
||||
tail: applied.tail,
|
||||
head: applied.head,
|
||||
changedTail: applied.changedTail,
|
||||
changedHead: applied.changedHead,
|
||||
tail,
|
||||
head,
|
||||
changedTail,
|
||||
changedHead,
|
||||
cursor: sequencing.nextTimelineCursor,
|
||||
cursorChanged: sequencing.cursorChanged,
|
||||
acknowledgedClientMessageIds: applied.acknowledgedClientMessageIds ?? [],
|
||||
acknowledgedClientMessageIds: streamResult.acknowledgedClientMessageIds ?? [],
|
||||
sideEffects: sequencing.sideEffects,
|
||||
};
|
||||
}
|
||||
@@ -1144,6 +1201,7 @@ export function processAgentStreamEvents(
|
||||
currentTail: tail,
|
||||
currentHead: head,
|
||||
currentCursor: cursor,
|
||||
hasAuthoritativeBaseline: input.hasAuthoritativeBaseline,
|
||||
timestamp: reducerEvent.timestamp,
|
||||
});
|
||||
|
||||
@@ -1281,10 +1339,12 @@ export function createSessionAgentStreamReducerQueue(
|
||||
return createAgentStreamReducerQueue({
|
||||
getSnapshot: (agentId) => {
|
||||
const session = useSessionStore.getState().sessions[serverId];
|
||||
const timeline = selectAgentTimelineState(session, agentId);
|
||||
return {
|
||||
currentTail: session?.agentStreamTail.get(agentId) ?? [],
|
||||
currentTail: timeline.status === "cold" ? [] : timeline.items,
|
||||
currentHead: session?.agentStreamHead.get(agentId) ?? [],
|
||||
currentCursor: session?.agentTimelineCursor.get(agentId),
|
||||
currentCursor: timeline.status === "synced" ? (timeline.range ?? undefined) : undefined,
|
||||
hasAuthoritativeBaseline: timeline.status === "synced",
|
||||
};
|
||||
},
|
||||
commit: (agentId, result, events) => {
|
||||
|
||||
Reference in New Issue
Block a user