puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-26 03:03:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(timeline): apply covered revisions during resume append

Browse Source 
Run the exact canonical revision pass before overlapping resume pages append newer rows, while leaving cursor advancement to the incremental path.
This commit is contained in:
Mohamed Boudra
2026-08-01 05:38:55 +02:00
parent 74883869a1
commit d3bffa91b1
2 changed files with 76 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/timeline/session-stream-reducers.test.ts
+40 -12
View File
@@ -362,18 +362,28 @@ describe("processTimelineResponse", () => {
});
it("appends only rows newer than the cursor from an overlapping resume tail", () => {
const loaded: StreamItem = {
kind: "user_message",
id: "loaded-30",
text: "already loaded",
const loaded = createUserMessage({
id: "local-prompt-30",
clientMessageId: "client-message-30",
text: "local presentation",
timestamp: new Date(1030),
timelineCursor: { epoch: "epoch-1", seq: 30 },
};
attachments: [{ type: "text", mimeType: "text/plain", text: "local attachment" }],
});
const unrelated = createUserMessage({
id: "unrelated-row",
messageId: "unrelated-provider-message",
text: "unrelated",
timestamp: new Date(1035),
timelineCursor: { epoch: "epoch-1", seq: 35 },
});
const currentTail = [loaded, unrelated];
const currentCursor: TimelineCursor = { epoch: "epoch-1", startSeq: 1, endSeq: 40 };
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [loaded],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
currentTail,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "tail",
@@ -381,17 +391,35 @@ describe("processTimelineResponse", () => {
startCursor: { seq: 30 },
endCursor: { seq: 42 },
entries: [
makeTimelineEntry(30, "already loaded", "user_message"),
makeTimelineEntry(41, "new 41", "user_message"),
makeTimelineEntry(42, "new 42", "user_message"),
{
...makeTimelineEntry(30, "provider presentation", "user_message"),
item: {
type: "user_message",
text: "provider presentation",
clientMessageId: "client-message-30",
messageId: "provider-message-30",
},
},
makeTimelineEntry(41, "new output 41"),
makeTimelineEntry(42, "new output 42"),
],
},
});
expect(result.commit).toBe("apply");
expect(result.tail[0]).toBe(loaded);
expect(getUserTexts(result.tail)).toEqual(["already loaded", "new 41", "new 42"]);
expect(result.tail[0]).toEqual({ ...loaded, messageId: "provider-message-30" });
expect(result.tail[1]).toBe(unrelated);
expect(
[...result.tail, ...result.head].filter(
(item) => item.kind === "user_message" && item.clientMessageId === "client-message-30",
),
).toHaveLength(1);
expect(getAssistantTexts([...result.tail, ...result.head])).toEqual([
"new output 41new output 42",
]);
expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 1, endSeq: 42 });
expect(result.cursorChanged).toBe(true);
expect(result.acknowledgedClientMessageIds).toEqual(["client-message-30"]);
expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]);
});
packages/app/src/timeline/session-stream-reducers.ts
+36 -23
View File
@@ -185,8 +185,8 @@ function deriveBootstrapTailTimelinePolicy({
type ResumeTailPolicy =
| { kind: "not_resume" }
| { kind: "discard" }
| { kind: "append" }
| { kind: "discard"; coveredEndSeq: number }
| { kind: "append"; coveredEndSeq: number }
| { kind: "replace"; preserveContinuity: boolean };
function deriveResumeTailPolicy(input: {
@@ -205,13 +205,13 @@ function deriveResumeTailPolicy(input: {
return { kind: "replace", preserveContinuity: false };
}
if (input.windowMaxSeq === input.currentCursor.endSeq) {
return { kind: "discard" };
return { kind: "discard", coveredEndSeq: input.currentCursor.endSeq };
}
if (input.windowMaxSeq < input.currentCursor.endSeq) {
return { kind: "replace", preserveContinuity: false };
}
if (input.pageStartSeq !== null && input.pageStartSeq <= input.currentCursor.endSeq + 1) {
return { kind: "append" };
return { kind: "append", coveredEndSeq: input.currentCursor.endSeq };
}
return { kind: "replace", preserveContinuity: true };
}
@@ -976,19 +976,28 @@ export function processTimelineResponse(
bootstrapReplace: replace,
});
const discard = resumeTailPolicy.kind === "discard";
const coveredUnits = timelineUnits.filter(
(unit) => "coveredEndSeq" in resumeTailPolicy && unit.seqEnd <= resumeTailPolicy.coveredEndSeq,
);
const revision = processAgentStreamEvents({
events: coveredUnits.map((unit) => ({
event: unit.event,
seq: unit.seqEnd,
epoch: payload.epoch,
timestamp: unit.timestamp,
})),
currentTail,
currentHead,
currentCursor,
});
const coveredRevisionAcknowledgements = deriveCanonicalAcknowledgements({
units: coveredUnits,
epoch: payload.epoch,
currentTail,
currentHead,
});
let timelineResult: TimelinePathResult;
if (discard) {
const revision = processAgentStreamEvents({
events: timelineUnits.map((unit) => ({
event: unit.event,
seq: unit.seqEnd,
epoch: payload.epoch,
timestamp: unit.timestamp,
})),
currentTail,
currentHead,
currentCursor,
});
timelineResult = {
tail: revision.tail,
head: revision.head,
@@ -996,12 +1005,7 @@ export function processTimelineResponse(
cursorChanged: false,
older: "unchanged",
sideEffects: [],
acknowledgedClientMessageIds: deriveCanonicalAcknowledgements({
units: timelineUnits,
epoch: payload.epoch,
currentTail,
currentHead,
}),
acknowledgedClientMessageIds: [],
};
} else if (replace || resumeTailPolicy.kind === "replace") {
const isResumeReplacement = resumeTailPolicy.kind === "replace";
@@ -1027,11 +1031,20 @@ export function processTimelineResponse(
timelineResult = applyTimelineIncrementalPath({
timelineUnits: incrementalUnits,
payload,
currentTail,
currentHead,
currentTail: revision.tail,
currentHead: revision.head,
currentCursor,
});
}
timelineResult = {
...timelineResult,
acknowledgedClientMessageIds: [
...new Set([
...coveredRevisionAcknowledgements,
...timelineResult.acknowledgedClientMessageIds,
]),
],
};
const nextTail = timelineResult.tail;
const nextHead = timelineResult.head;