refactor(app): remove prompt revision reconciliation

Submitted prompt rows are immutable on the wire, so resume and live reducers no longer need to apply already-covered same-sequence updates.
This commit is contained in:
Mohamed Boudra
2026-08-01 11:48:07 +02:00
parent 20a542126e
commit 09a411a54f
2 changed files with 23 additions and 241 deletions
@@ -258,63 +258,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
// ---------------------------------------------------------------------------
describe("processTimelineResponse", () => {
it("applies a missed same-sequence provider identity revision from an unchanged resume tail", () => {
const canonical = createUserMessage({
id: "canonical-prompt",
clientMessageId: "client-message",
text: "local prompt",
timestamp: new Date(1000),
timelineCursor: { epoch: "epoch-1", seq: 5 },
});
const unrelatedTail = makeAssistantItem("unrelated tail", "unrelated-tail");
const currentTail = [canonical, unrelatedTail];
const currentHead = [makeAssistantItem("live head", "live-head")];
const currentCursor: TimelineCursor = { epoch: "epoch-1", startSeq: 1, endSeq: 8 };
const result = processTimelineResponse({
...baseTimelineInput,
currentTail,
currentHead,
currentCursor,
payload: {
...baseTimelineInput.payload,
direction: "tail",
window: { minSeq: 1, maxSeq: 8, nextSeq: 9 },
startCursor: { seq: 5 },
endCursor: { seq: 8 },
entries: [
{
...makeTimelineEntry(5, "provider prompt", "user_message"),
item: {
type: "user_message",
text: "provider prompt",
messageId: "provider-message",
clientMessageId: "client-message",
},
},
],
},
});
expect(result.commit).toBe("apply");
expect(result.tail).not.toBe(currentTail);
expect(result.tail[0]).toEqual(
expect.objectContaining({
id: "canonical-prompt",
clientMessageId: "client-message",
messageId: "provider-message",
text: "local prompt",
timestamp: canonical.timestamp,
timelineCursor: { epoch: "epoch-1", seq: 5 },
}),
);
expect(result.tail[1]).toBe(unrelatedTail);
expect(result.head).toBe(currentHead);
expect(result.cursor).toBe(currentCursor);
expect(result.cursorChanged).toBe(false);
expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]);
});
it("discards an unchanged resume tail without replacing timeline state", () => {
const canonical = createUserMessage({
id: "canonical-prompt",
@@ -357,7 +300,7 @@ describe("processTimelineResponse", () => {
expect(result.tail).toBe(currentTail);
expect(result.head).toBe(currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]);
expect(result.acknowledgedClientMessageIds).toEqual([]);
expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]);
});
@@ -365,6 +308,7 @@ describe("processTimelineResponse", () => {
const loaded = createUserMessage({
id: "local-prompt-30",
clientMessageId: "client-message-30",
messageId: "client-message-30",
text: "local presentation",
timestamp: new Date(1030),
timelineCursor: { epoch: "epoch-1", seq: 30 },
@@ -407,7 +351,7 @@ describe("processTimelineResponse", () => {
});
expect(result.commit).toBe("apply");
expect(result.tail[0]).toEqual({ ...loaded, messageId: "provider-message-30" });
expect(result.tail[0]).toBe(loaded);
expect(result.tail[1]).toBe(unrelated);
expect(
[...result.tail, ...result.head].filter(
@@ -419,7 +363,7 @@ describe("processTimelineResponse", () => {
]);
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.acknowledgedClientMessageIds).toEqual([]);
expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]);
});
@@ -3242,48 +3186,6 @@ describe("processAgentStreamEvent", () => {
expect(result.sideEffects).toEqual([]);
});
it("applies a same-sequence provider identity revision to its canonical prompt", () => {
const canonical = createUserMessage({
id: "canonical-prompt",
clientMessageId: "client-message-1",
text: "Revise this prompt",
timestamp: new Date(1000),
timelineCursor: { epoch: "epoch-1", seq: 5 },
});
const currentHead = [makeAssistantItem("Still streaming")];
const result = processAgentStreamEvent({
...baseStreamInput,
currentTail: [canonical],
currentHead,
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 8 },
event: {
type: "timeline",
provider: "claude",
item: {
type: "user_message",
text: "Revise this prompt",
clientMessageId: "client-message-1",
messageId: "provider-message-1",
},
},
seq: 5,
epoch: "epoch-1",
});
expect(result.tail).toEqual([
expect.objectContaining({
id: "canonical-prompt",
clientMessageId: "client-message-1",
messageId: "provider-message-1",
timelineCursor: { epoch: "epoch-1", seq: 5 },
}),
]);
expect(result.head).toBe(currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.sideEffects).toEqual([]);
});
it("drops timeline event with epoch mismatch", () => {
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
@@ -4,7 +4,6 @@ import type { AssistantMessageItem, StreamItem } from "@/types/stream";
import type { TurnLivenessTransition } from "@/timeline/turn-liveness";
import {
applyStreamEvent,
createUserMessage,
flushHeadToTail,
hydrateStreamState,
isAgentToolCallItem,
@@ -185,8 +184,8 @@ function deriveBootstrapTailTimelinePolicy({
type ResumeTailPolicy =
| { kind: "not_resume" }
| { kind: "discard"; coveredEndSeq: number }
| { kind: "append"; coveredEndSeq: number }
| { kind: "discard" }
| { kind: "append" }
| { kind: "replace"; preserveContinuity: boolean };
function deriveResumeTailPolicy(input: {
@@ -205,13 +204,13 @@ function deriveResumeTailPolicy(input: {
return { kind: "replace", preserveContinuity: false };
}
if (input.windowMaxSeq === input.currentCursor.endSeq) {
return { kind: "discard", coveredEndSeq: input.currentCursor.endSeq };
return { kind: "discard" };
}
if (input.windowMaxSeq < input.currentCursor.endSeq) {
return { kind: "replace", preserveContinuity: false };
}
if (input.pageStartSeq !== null && input.pageStartSeq <= input.currentCursor.endSeq + 1) {
return { kind: "append", coveredEndSeq: input.currentCursor.endSeq };
return { kind: "append" };
}
return { kind: "replace", preserveContinuity: true };
}
@@ -746,19 +745,24 @@ function applyAcceptedForwardTimelineUnits(params: {
});
let tail = reconciled.tail;
let head = reconciled.head;
const acknowledgedClientMessageIds = new Set<string>();
for (const unit of params.units) {
if (reconciled.reconciledUnits.has(unit)) continue;
const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch });
tail = applied.tail;
head = applied.head;
for (const clientMessageId of applied.acknowledgedClientMessageIds) {
acknowledgedClientMessageIds.add(clientMessageId);
}
}
return { tail, head, acknowledgedClientMessageIds: [...acknowledgedClientMessageIds] };
return {
tail,
head,
acknowledgedClientMessageIds: deriveCanonicalAcknowledgements({
units: params.units,
epoch: params.epoch,
currentTail: params.currentTail,
currentHead: params.currentHead,
}),
};
}
function deriveCanonicalAcknowledgements(params: {
@@ -782,20 +786,6 @@ function deriveCanonicalAcknowledgements(params: {
return [...acknowledged];
}
function shouldDiscardUnchangedTimelineResult(params: {
discardPolicy: boolean;
currentTail: StreamItem[];
currentHead: StreamItem[];
nextTail: StreamItem[];
nextHead: StreamItem[];
}): boolean {
return (
params.discardPolicy &&
params.nextTail === params.currentTail &&
params.nextHead === params.currentHead
);
}
function applyTimelineIncrementalPath(args: {
timelineUnits: TimelineUnit[];
payload: ProcessTimelineResponseInput["payload"];
@@ -976,31 +966,11 @@ 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) {
timelineResult = {
tail: revision.tail,
head: revision.head,
tail: currentTail,
head: currentHead,
cursor: currentCursor,
cursorChanged: false,
older: "unchanged",
@@ -1031,20 +1001,11 @@ export function processTimelineResponse(
timelineResult = applyTimelineIncrementalPath({
timelineUnits: incrementalUnits,
payload,
currentTail: revision.tail,
currentHead: revision.head,
currentTail,
currentHead,
currentCursor,
});
}
timelineResult = {
...timelineResult,
acknowledgedClientMessageIds: [
...new Set([
...coveredRevisionAcknowledgements,
...timelineResult.acknowledgedClientMessageIds,
]),
],
};
const nextTail = timelineResult.tail;
const nextHead = timelineResult.head;
@@ -1074,15 +1035,7 @@ export function processTimelineResponse(
timelineResponseComplete;
const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null;
const commit = shouldDiscardUnchangedTimelineResult({
discardPolicy: discard,
currentTail,
currentHead,
nextTail,
nextHead,
})
? "discard"
: "apply";
const commit = discard ? "discard" : "apply";
return {
commit,
@@ -1140,55 +1093,6 @@ interface TimelineSequencingGateResult {
sideEffects: AgentStreamReducerSideEffect[];
}
function applyCanonicalUserMessageRevision(input: {
event: AgentStreamEventPayload;
seq: number | undefined;
epoch: string | undefined;
timestamp: Date;
currentTail: StreamItem[];
currentHead: StreamItem[];
currentCursor: TimelineCursor | undefined;
}): ReturnType<typeof upsertUserMessageAcrossStream> | null {
const { event, seq, epoch, currentCursor } = input;
if (
event.type !== "timeline" ||
event.item.type !== "user_message" ||
!event.item.clientMessageId ||
!event.item.messageId ||
typeof seq !== "number" ||
typeof epoch !== "string" ||
currentCursor?.epoch !== epoch ||
seq > currentCursor.endSeq
) {
return null;
}
const clientMessageId = event.item.clientMessageId;
const existing = [...input.currentTail, ...input.currentHead].find(
(item) =>
item.kind === "user_message" &&
item.clientMessageId === clientMessageId &&
item.timelineCursor?.epoch === epoch &&
item.timelineCursor.seq === seq,
);
if (!existing || existing.kind !== "user_message") {
return null;
}
return upsertUserMessageAcrossStream({
tail: input.currentTail,
head: input.currentHead,
message: createUserMessage({
id: existing.id,
clientMessageId,
messageId: event.item.messageId,
text: event.item.text,
timestamp: input.timestamp,
timelineCursor: existing.timelineCursor,
}),
insert: "none",
presentation: "existing",
});
}
export interface ProcessAgentStreamEventsInput {
events: AgentStreamReducerEvent[];
currentTail: StreamItem[];
@@ -1306,30 +1210,6 @@ export function processAgentStreamEvent(
hasAuthoritativeBaseline = true,
} = input;
const revision = applyCanonicalUserMessageRevision({
event,
seq,
epoch,
timestamp,
currentTail,
currentHead,
currentCursor,
});
if (revision) {
return {
tail: revision.tail,
head: revision.head,
changedTail: revision.changedTail,
changedHead: revision.changedHead,
cursor: null,
cursorChanged: false,
acknowledgedClientMessageIds: revision.location?.message.clientMessageId
? [revision.location.message.clientMessageId]
: [],
sideEffects: [],
};
}
const sequencing = processTimelineSequencingGate({
event,
seq,