Allow failed agent turns to be forked (#2063)

* fix(chat): allow failed turns to be forked Failed turns can end in Paseo-generated timeline items without provider message IDs. Anchor forks to canonical timeline positions while retaining legacy boundaries for compatibility. Place the fork summary before the new prompt and delimit it as structured chat history. * fix(chat): tighten fork boundary handling
2026-07-15 16:30:42 +02:00
parent f06792ae89
commit 04e893417e
23 changed files with 549 additions and 108 deletions
@@ -187,6 +187,25 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
 // ---------------------------------------------------------------------------

 describe("processTimelineResponse", () => {
+  it("preserves the canonical end cursor on a projected assistant message", () => {
+    const result = processTimelineResponse({
+      ...baseTimelineInput,
+      payload: {
+        ...baseTimelineInput.payload,
+        epoch: "timeline-1",
+        entries: [makeTimelineEntry(40, "[System Error] failed", "assistant_message", 42)],
+      },
+    });
+
+    expect(result.tail).toEqual([
+      expect.objectContaining({
+        kind: "assistant_message",
+        text: "[System Error] failed",
+        timelineCursor: { epoch: "timeline-1", seq: 42 },
+      }),
+    ]);
+  });
+
  it("returns error path when payload.error is set", () => {
    const result = processTimelineResponse({
      ...baseTimelineInput,
@@ -892,7 +911,12 @@ describe("processTimelineResponse", () => {
  });

  it("merges assistant chunks across the older-page prepend boundary", () => {
-    const currentTail = [makeAssistantItem("newer chunk", "assistant-newer")];
+    const currentTail = [
+      {
+        ...makeAssistantItem("newer chunk", "assistant-newer"),
+        timelineCursor: { epoch: "epoch-1", seq: 3 },
+      },
+    ];
    const existingCursor: TimelineCursor = {
      epoch: "epoch-1",
      startSeq: 3,
@@ -914,6 +938,9 @@ 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(result.cursor).toEqual({
      epoch: "epoch-1",
      startSeq: 1,
@@ -1222,6 +1249,23 @@ describe("processTimelineResponse", () => {
 // ---------------------------------------------------------------------------

 describe("processAgentStreamEvent", () => {
+  it("preserves the live timeline cursor on an assistant error", () => {
+    const result = processAgentStreamEvent({
+      ...baseStreamInput,
+      event: makeAssistantTimelineEvent("[System Error] failed"),
+      epoch: "timeline-1",
+      seq: 42,
+    });
+
+    expect(result.head).toEqual([
+      expect.objectContaining({
+        kind: "assistant_message",
+        text: "[System Error] failed",
+        timelineCursor: { epoch: "timeline-1", seq: 42 },
+      }),
+    ]);
+  });
+
  it("passes through non-timeline events without cursor changes", () => {
    const turnEvent: AgentStreamEventPayload = {
      type: "turn_completed",
@@ -510,6 +510,7 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
      ...olderLast,
      text: `${olderLast.text}${currentFirst.text}`,
      timestamp: currentFirst.timestamp,
+      ...(currentFirst.timelineCursor ? { timelineCursor: currentFirst.timelineCursor } : {}),
    },
    ...currentTail.slice(1),
  ];
@@ -519,8 +520,9 @@ function replaceLiveAssistantWithProjectedText(params: {
  head: StreamItem[];
  event: AgentStreamEventPayload;
  timestamp: Date;
+  timelineCursor: { epoch: string; seq: number };
 }): StreamItem[] | null {
-  const { head, event, timestamp } = params;
+  const { head, event, timestamp, timelineCursor } = params;
  if (event.type !== "timeline" || event.item.type !== "assistant_message") {
    return null;
  }
@@ -537,6 +539,7 @@ function replaceLiveAssistantWithProjectedText(params: {
    ...current,
    text: event.item.text,
    timestamp,
+    timelineCursor,
  };
  return next;
 }
@@ -575,19 +578,22 @@ function applyTimelineIncrementalPath(args: {
  if (acceptedUnits.length > 0) {
    if (payload.direction === "before") {
      const olderTail = hydrateStreamState(
-        acceptedUnits.map(({ event, timestamp }) => ({
+        acceptedUnits.map(({ event, timestamp, seqEnd }) => ({
          event,
          timestamp,
+          timelineCursor: { epoch: payload.epoch, seq: seqEnd },
        })),
        { source: "canonical" },
      );
      nextTail = mergePrependedCanonicalTail(olderTail, currentTail);
    } else if (currentHead.length > 0) {
-      for (const { event, timestamp } of acceptedUnits) {
+      for (const { event, timestamp, seqEnd } of acceptedUnits) {
+        const timelineCursor = { epoch: payload.epoch, seq: seqEnd };
        const replacedHead = replaceLiveAssistantWithProjectedText({
          head: nextHead,
          event,
          timestamp,
+          timelineCursor,
        });
        if (replacedHead) {
          nextHead = replacedHead;
@@ -599,15 +605,17 @@ function applyTimelineIncrementalPath(args: {
          event,
          timestamp,
          source: "canonical",
+          timelineCursor,
        });
        nextTail = applied.tail;
        nextHead = applied.head;
      }
    } else {
      nextTail = acceptedUnits.reduce<StreamItem[]>(
-        (state, { event, timestamp }) =>
+        (state, { event, timestamp, seqEnd }) =>
          reduceStreamUpdate(state, event, timestamp, {
            source: "canonical",
+            timelineCursor: { epoch: payload.epoch, seq: seqEnd },
          }),
        currentTail,
      );
@@ -681,8 +689,16 @@ export function processTimelineResponse(

  const toHydratedEvents = (
    units: TimelineUnit[],
-  ): Array<{ event: AgentStreamEventPayload; timestamp: Date }> =>
-    units.map(({ event, timestamp }) => ({ event, timestamp }));
+  ): Array<{
+    event: AgentStreamEventPayload;
+    timestamp: Date;
+    timelineCursor: { epoch: string; seq: number };
+  }> =>
+    units.map(({ event, timestamp, seqEnd }) => ({
+      event,
+      timestamp,
+      timelineCursor: { epoch: payload.epoch, seq: seqEnd },
+    }));

  // ------------------------------------------------------------------
  // Derive bootstrap policy (replace vs incremental)
@@ -934,6 +950,10 @@ export function processAgentStreamEvent(
    input;

  const sequencing = processTimelineSequencingGate({ event, seq, epoch, currentCursor });
+  const timelineCursor =
+    event.type === "timeline" && seq !== undefined && epoch !== undefined
+      ? { epoch, seq }
+      : undefined;

  // ------------------------------------------------------------------
  // Apply stream event to tail/head
@@ -945,6 +965,7 @@ export function processAgentStreamEvent(
        event,
        timestamp,
        source: "live",
+        timelineCursor,
      })
    : {
        tail: currentTail,