puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-06 08:24:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Integrate timeline hydration with submission authority

Browse Source
This commit is contained in:
Mohamed Boudra
2026-07-28 16:45:40 +02:00
parent 4735bfb8c8
commit 05c2869e2c
7 changed files with 213 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/agent-stream/strategy-web.test.tsx
+9
View File
@@ -5,11 +5,16 @@ import React from "react";
import { act } from "react";
import { createRoot, type Root } from "react-dom/client";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import "@/styles/unistyles";
import type { StreamItem } from "@/types/stream";
import type { StreamRenderInput, StreamSegmentRenderers, StreamViewportHandle } from "./strategy";
import { createWebStreamStrategy } from "./strategy-web";
vi.hoisted(() => {
Object.defineProperty(globalThis, "__DEV__", {
configurable: true,
value: false,
});
Object.defineProperty(window, "matchMedia", {
configurable: true,
value: () => ({
@@ -23,6 +28,10 @@ vi.hoisted(() => {
removeListener: () => {},
}),
});
Object.defineProperty(window.screen, "orientation", {
configurable: true,
value: { type: "landscape-primary" },
});
});
function userMessage(index: number): StreamItem {
packages/app/src/contexts/session-context.tsx
+5 -2
View File
@@ -575,8 +575,10 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const isInitializing = session?.initializingAgents.get(agentId) === true;
const activeInitDeferred = getInitDeferred(initKey);
const hasActiveInitDeferred = Boolean(activeInitDeferred);
const currentCursor = session?.agentTimelineCursor.get(agentId);
const currentTail = session?.agentStreamTail.get(agentId) ?? [];
const timeline = selectAgentTimelineState(session, agentId);
const currentCursor =
timeline.status === "synced" ? (timeline.range ?? undefined) : undefined;
const currentTail = timeline.status === "cold" ? [] : timeline.items;
const currentHead = session?.agentStreamHead.get(agentId) ?? [];
const sendingClientMessageIds = getSendingClientMessageIds(
session?.messageSubmissions.get(agentId),
@@ -592,6 +594,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
hasActiveInitDeferred,
initRequestDirection: activeInitDeferred?.requestDirection ?? "tail",
sendingClientMessageIds,
hasAuthoritativeBaseline: timeline.status === "synced",
});
if (result.error) {
packages/app/src/stores/session-store.test.ts
+2
View File
@@ -73,6 +73,7 @@ describe("agent timeline state", () => {
range: { epoch: "epoch-1", startSeq: 51, endSeq: 100 },
older: "available",
synchronized: true,
acknowledgedClientMessageIds: [],
});
expect(
@@ -93,6 +94,7 @@ describe("agent timeline state", () => {
range: null,
older: "none",
synchronized: true,
acknowledgedClientMessageIds: [],
});
expect(
packages/app/src/timeline/session-stream-reducers.test.ts
+128
View File
@@ -173,6 +173,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
hasActiveInitDeferred: false,
initRequestDirection: "tail",
sendingClientMessageIds: [],
hasAuthoritativeBaseline: true,
};
const baseStreamInput: ProcessAgentStreamEventInput = {
@@ -362,6 +363,111 @@ describe("processTimelineResponse", () => {
expect(result.head).toEqual([liveThought, liveAssistant]);
});
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 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("uses the timeline entry timestamp as canonical", () => {
const result = processTimelineResponse({
...baseTimelineInput,
@@ -2417,6 +2523,28 @@ 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([]);
});
});
describe("processAgentStreamEvents", () => {
packages/app/src/timeline/session-stream-reducers.ts
+15 -48
View File
@@ -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
@@ -574,43 +579,6 @@ function reconcileOverlappingProjectedReasoning(params: {
: { tail: params.tail, head: params.head, reconciled: false };
}
function reconcileProjectedUserOverlay(params: {
tail: StreamItem[];
head: StreamItem[];
unit: TimelineUnit;
}): { tail: StreamItem[]; head: StreamItem[]; reconciled: boolean } {
const { event } = params.unit;
if (event.type !== "timeline" || event.item.type !== "user_message") {
return { tail: params.tail, head: params.head, reconciled: false };
}
const userItem = event.item;
const overlayIndex = params.head.findIndex(
(item) =>
item.kind === "user_message" &&
matchesLocalUserMessageIdentity(
{
messageId: userItem.messageId,
clientMessageId: userItem.clientMessageId,
text: userItem.text,
},
item,
),
);
const overlay = params.head[overlayIndex];
if (overlayIndex < 0 || !overlay || overlay.kind !== "user_message") {
return { tail: params.tail, head: params.head, reconciled: false };
}
const canonical = reduceStreamUpdate([], event, params.unit.timestamp, {
source: "canonical",
})[0];
if (!canonical || canonical.kind !== "user_message") {
return { tail: params.tail, head: params.head, reconciled: false };
}
const head = [...params.head];
head[overlayIndex] = mergeCanonicalUserWithLocalPresentation(canonical, overlay);
return { tail: params.tail, head, reconciled: true };
}
function reconcileOverlappingProjectedStreamItems(params: {
tail: StreamItem[];
head: StreamItem[];
@@ -624,16 +592,13 @@ function reconcileOverlappingProjectedStreamItems(params: {
if (params.currentEndSeq === undefined) return { tail, head, reconciledUnits };
for (const unit of params.units) {
let reconciled = reconcileProjectedUserOverlay({ tail, head, unit });
if (!reconciled.reconciled) {
reconciled = reconcileOverlappingProjectedAssistant({
tail,
head,
unit,
epoch: params.epoch,
currentEndSeq: params.currentEndSeq,
});
}
let reconciled = reconcileOverlappingProjectedAssistant({
tail,
head,
unit,
epoch: params.epoch,
currentEndSeq: params.currentEndSeq,
});
if (!reconciled.reconciled) {
reconciled = reconcileOverlappingProjectedReasoning({
tail,
@@ -859,6 +824,7 @@ export function processTimelineResponse(
hasActiveInitDeferred,
initRequestDirection,
sendingClientMessageIds,
hasAuthoritativeBaseline,
} = input;
// ------------------------------------------------------------------
@@ -930,7 +896,8 @@ export function processTimelineResponse(
currentTail,
currentHead,
sendingClientMessageIds,
preserveLiveHead: currentCursor?.epoch === payload.epoch,
preserveLiveHead:
currentCursor?.epoch === payload.epoch || (!payload.reset && !hasAuthoritativeBaseline),
toHydratedEvents,
})
: applyTimelineIncrementalPath({
packages/app/src/types/stream.test.ts
+5 -1
View File
@@ -1350,7 +1350,10 @@ describe("turn lifecycle events", () => {
},
},
new Date("2025-01-01T15:03:11Z"),
{ source: "canonical" },
{
source: "canonical",
timelineCursor: { epoch: "epoch-1", seq: 42 },
},
);
const userMessages = state.filter((item) => item.kind === "user_message");
@@ -1361,6 +1364,7 @@ describe("turn lifecycle events", () => {
assert.strictEqual(userMessage.messageId, "provider-owned-canonical");
assert.strictEqual(userMessage.text, "Analyze this");
assert.strictEqual(userMessage.timestamp.getTime(), submittedTimestamp.getTime());
assert.deepStrictEqual(userMessage.timelineCursor, { epoch: "epoch-1", seq: 42 });
assert.deepStrictEqual(userMessage.images, [image]);
assert.deepStrictEqual(userMessage.attachments, [attachment]);
});
packages/app/src/types/stream.ts
+49 -5
View File
@@ -88,6 +88,7 @@ export interface UserMessageItem {
id: string;
clientMessageId?: string;
messageId?: string;
timelineCursor?: TimelinePosition;
text: string;
timestamp: Date;
images?: UserMessageImageAttachment[];
@@ -98,6 +99,7 @@ export interface UserMessageInput {
id?: string;
clientMessageId?: string;
messageId?: string;
timelineCursor?: TimelinePosition;
text: string;
timestamp: Date;
images?: UserMessageImageAttachment[];
@@ -114,6 +116,7 @@ export function createUserMessage(input: UserMessageInput): UserMessageItem {
id,
...(input.clientMessageId ? { clientMessageId: input.clientMessageId } : {}),
...(input.messageId ? { messageId: input.messageId } : {}),
...(input.timelineCursor ? { timelineCursor: input.timelineCursor } : {}),
text: input.text,
timestamp: input.timestamp,
...(input.images && input.images.length > 0 ? { images: input.images } : {}),
@@ -236,11 +239,13 @@ function produceUserMessage(
...presentation,
clientMessageId: incoming.clientMessageId ?? existing.clientMessageId,
messageId: incoming.messageId ?? existing.messageId,
timelineCursor: incoming.timelineCursor ?? existing.timelineCursor,
});
if (
existing.id === merged.id &&
existing.clientMessageId === merged.clientMessageId &&
existing.messageId === merged.messageId &&
existing.timelineCursor === merged.timelineCursor &&
existing.text === merged.text &&
existing.timestamp === merged.timestamp &&
existing.images === merged.images &&
@@ -392,6 +397,7 @@ export interface CanonicalStreamReplacementInput {
previousHead: StreamItem[];
sendingClientMessageIds: readonly string[];
preserveLiveHead: boolean;
canonicalCoverage: { epoch: string; endSeq: number | null };
}
export interface CanonicalStreamReplacementResult {
@@ -434,12 +440,33 @@ function preserveReplacementHead(
if (
liveAssistant.kind !== "assistant_message" ||
!tailAssistant ||
tailAssistant.kind !== "assistant_message" ||
!liveAssistant.text.startsWith(tailAssistant.text)
tailAssistant.kind !== "assistant_message"
) {
return { tail, head: unreconciledHead, acknowledgedClientMessageIds: [] };
}
const isNewerContinuation =
liveAssistant.messageId !== undefined &&
liveAssistant.messageId === tailAssistant.messageId &&
liveAssistant.timelineCursor !== undefined &&
tailAssistant.timelineCursor !== undefined &&
liveAssistant.timelineCursor.epoch === tailAssistant.timelineCursor.epoch &&
liveAssistant.timelineCursor.seq > tailAssistant.timelineCursor.seq;
if (isNewerContinuation) {
const text = liveAssistant.text.startsWith(tailAssistant.text)
? liveAssistant.text
: `${tailAssistant.text}${liveAssistant.text}`;
const head = [
...unreconciledHead.slice(0, liveAssistantIndex),
{ ...liveAssistant, text },
...unreconciledHead.slice(liveAssistantIndex + 1),
];
return { tail: tail.slice(0, -1), head, acknowledgedClientMessageIds: [] };
}
if (!liveAssistant.text.startsWith(tailAssistant.text)) {
return { tail, head: unreconciledHead, acknowledgedClientMessageIds: [] };
}
const head = [
...unreconciledHead.slice(0, liveAssistantIndex),
{ ...liveAssistant, text: tailAssistant.text },
@@ -507,7 +534,13 @@ export function replaceWithCanonicalStream(
nextHead = nextHead.filter((item) => {
if (item.kind !== "user_message" || !item.clientMessageId) return true;
return sendingClientMessageIds.has(item.clientMessageId);
if (sendingClientMessageIds.has(item.clientMessageId)) return true;
if (!input.preserveLiveHead || !item.timelineCursor) return false;
return (
item.timelineCursor.epoch === input.canonicalCoverage.epoch &&
(input.canonicalCoverage.endSeq === null ||
item.timelineCursor.seq > input.canonicalCoverage.endSeq)
);
});
const replacement = preserveReplacementHead(
@@ -675,6 +708,7 @@ function appendUserMessage(
_source: StreamUpdateSource,
messageId?: string,
clientMessageId?: string,
timelineCursor?: TimelinePosition,
): StreamItem[] {
const { chunk, hasContent } = normalizeChunk(text);
if (!hasContent) {
@@ -686,6 +720,7 @@ function appendUserMessage(
id: messageId ?? createUniqueTimelineId(state, "user", chunkSeed, timestamp),
clientMessageId,
messageId,
timelineCursor,
text: chunk,
timestamp,
});
@@ -1148,6 +1183,7 @@ function reduceTimelineEvent(
source,
item.messageId,
item.clientMessageId,
timelineCursor,
),
);
case "assistant_message":
@@ -1510,8 +1546,9 @@ function applyCanonicalUserMessageEvent(params: {
head: StreamItem[];
event: AgentStreamEventPayload;
timestamp: Date;
timelineCursor?: TimelinePosition;
}): ApplyStreamEventResult | null {
const { tail, head, event, timestamp } = params;
const { tail, head, event, timestamp, timelineCursor } = params;
if (event.type !== "timeline" || event.item.type !== "user_message") return null;
const normalized = normalizeChunk(event.item.text);
@@ -1523,6 +1560,7 @@ function applyCanonicalUserMessageEvent(params: {
createUniqueTimelineId([...tail, ...head], "user", normalized.chunk.trim(), timestamp),
messageId: event.item.messageId,
clientMessageId: event.item.clientMessageId,
timelineCursor,
text: normalized.chunk,
timestamp,
});
@@ -1559,7 +1597,13 @@ export function applyStreamEvent(params: {
timelineCursor?: TimelinePosition;
}): ApplyStreamEventResult {
const { tail, head, event, timestamp } = params;
const canonicalUserResult = applyCanonicalUserMessageEvent({ tail, head, event, timestamp });
const canonicalUserResult = applyCanonicalUserMessageEvent({
tail,
head,
event,
timestamp,
timelineCursor: params.timelineCursor,
});
if (canonicalUserResult) return canonicalUserResult;
const source = params.source ?? "live";
let nextTail = tail;