fix(cloud): keep resumed messages and progress in sync (#3914)

Author: tatoalo

packages/core/src/cloud-task/cloud-task-engine.ts

@@ -1114,7 +1114,9 @@ export class CloudTaskEngine extends TypedEventEmitter<CloudTaskEvents> {
       watcher.totalEntryCount = watcher.resumeFromEntryCount;
       watcher.hasEmittedSnapshot = true;
       watcher.isBootstrapping = false;
-      void this.connectSse(key, { startLatest: true });
+      // The renderer dedupes this leaf replay against its hydrated resume chain; starting at latest
+      // can lose entries persisted between that hydration request and the stream connection.
+      void this.connectSse(key);
       return;
     }
 

packages/core/src/cloud-task/cloud-task.test.ts

@@ -352,6 +352,62 @@ describe("CloudTaskEngine", () => {
     );
   });
 
+  it("replays a resumed run stream so hydration cannot miss its live tail", async () => {
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+
+    mockNetFetch.mockResolvedValueOnce(
+      createJsonResponse({
+        id: "run-1",
+        status: "in_progress",
+        stage: "build",
+        output: null,
+        error_message: null,
+        branch: "main",
+        updated_at: "2026-01-01T00:00:00Z",
+      }),
+    );
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(
+        'id: 1\ndata: {"type":"notification","timestamp":"2026-01-01T00:00:01Z","notification":{"jsonrpc":"2.0","method":"_posthog/progress","params":{"id":"sandbox","status":"completed","title":"Restored sandbox"}}}\n\n',
+      ),
+    );
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+      resumeFromEntryCount: 3,
+    });
+
+    await waitFor(() =>
+      updates.some((update) => {
+        const payload = update as {
+          kind?: string;
+          totalEntryCount?: number;
+        };
+        return payload.kind === "logs" && payload.totalEntryCount === 4;
+      }),
+    );
+
+    expect(mockStreamFetch).toHaveBeenCalledWith(
+      "https://app.example.com/api/projects/2/tasks/task-1/runs/run-1/stream/",
+      expect.objectContaining({
+        headers: expect.objectContaining({
+          Authorization: "Bearer token",
+          Accept: "text/event-stream",
+        }),
+      }),
+    );
+    expect(
+      mockNetFetch.mock.calls.some(([input]) => {
+        const url = typeof input === "string" ? input : input.toString();
+        return url.includes("/session_logs/");
+      }),
+    ).toBe(false);
+  });
+
   it("drops a re-delivered log entry with a duplicate stream id", async () => {
     const updates: unknown[] = [];
     service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));

packages/core/src/sessions/sessionEvents.test.ts

@@ -8,6 +8,7 @@ import {
   convertStoredEntriesToEvents,
   extractUserPromptsFromEvents,
   hasSessionPromptEvent,
+  hasSessionPromptEventForTaskRun,
   isAbsoluteFolderPath,
   isFatalSessionError,
   promptReferencesAbsoluteFolder,
@@ -196,6 +197,37 @@ describe("hasSessionPromptEvent", () => {
     expect(hasSessionPromptEvent([notification])).toBe(false);
     expect(hasSessionPromptEvent([])).toBe(false);
   });
+
+  it("does not attribute an ancestor prompt to a resumed run", () => {
+    const storedEntry = (event: AcpMessage): StoredLogEntry => ({
+      type: "notification",
+      timestamp: new Date(event.ts).toISOString(),
+      notification: event.message,
+    });
+    const leafPrompt = {
+      ...promptRequest,
+      ts: 3,
+      message: { ...promptRequest.message, id: 2 },
+    };
+    const events = convertStoredEntriesToEvents(
+      [
+        storedEntry(promptRequest),
+        storedEntry(notification),
+        storedEntry(leafPrompt),
+      ],
+      undefined,
+      {
+        taskRunId: "resume-run",
+        startEntryIndex: 0,
+        firstPositionedEntryIndex: 2,
+      },
+    );
+
+    expect(
+      hasSessionPromptEventForTaskRun(events.slice(0, 2), "resume-run"),
+    ).toBe(false);
+    expect(hasSessionPromptEventForTaskRun(events, "resume-run")).toBe(true);
+  });
 });
 
 describe("convertStoredEntriesToEvents — imported user prompts", () => {

packages/core/src/sessions/sessionEvents.ts

@@ -427,6 +427,18 @@ export function hasSessionPromptEvent(events: AcpMessage[]): boolean {
   );
 }
 
+export function hasSessionPromptEventForTaskRun(
+  events: AcpMessage[],
+  taskRunId: string,
+): boolean {
+  return events.some(
+    (event) =>
+      isJsonRpcRequest(event.message) &&
+      event.message.method === "session/prompt" &&
+      getStoredLogEventPosition(event)?.taskRunId === taskRunId,
+  );
+}
+
 /**
  * Whether an event is a turn-complete notification.
  */

packages/core/src/sessions/sessionService.ts

@@ -89,6 +89,7 @@ import {
   getStoredLogEventPosition,
   getUserShellExecutesSinceLastPrompt,
   hasSessionPromptEvent,
+  hasSessionPromptEventForTaskRun,
   isTurnCompleteEvent,
   normalizePromptToBlocks,
   promptReferencesAbsoluteFolder,
@@ -6222,27 +6223,34 @@ export class SessionService {
         );
       }
     }
-    const hasUserPrompt = events.some(
-      (e: AcpMessage) =>
-        isJsonRpcRequest(e.message) && e.message.method === "session/prompt",
+    const hasCurrentRunUserPrompt = isResumeRun
+      ? hasSessionPromptEventForTaskRun(events, taskRunId)
+      : hasSessionPromptEvent(events);
+
+    // A reload loses the in-memory placeholder; restore it from the resume
+    // state or initial task description until the active run records its prompt.
+    const seedContent = isResumeRun
+      ? typeof runState?.pending_user_message === "string"
+        ? runState.pending_user_message
+        : undefined
+      : (this.initialCloudOptimisticPrompt.get(taskId) ?? taskDescription);
+    const hasOptimisticUserPrompt = session.optimisticItems.some(
+      (item) => item.type === "user_message",
     );
-
-    // Seed the optimistic user-message bubble whenever the agent has
-    // not yet recorded an initial `session/prompt` request — covers the
-    // brand-new task case as well as "agent has emitted lifecycle
-    // notifications but hasn't received its first prompt yet". Prefer the
-    // stashed initial prompt (which carries the channel CONTEXT.md block, so
-    // its chip renders right away) over the bare task description.
-    const seedContent =
-      this.initialCloudOptimisticPrompt.get(taskId) ?? taskDescription;
-    if (!isTerminalRun && !hasUserPrompt && seedContent?.trim()) {
+    if (
+      !isTerminalRun &&
+      !hasCurrentRunUserPrompt &&
+      !hasOptimisticUserPrompt &&
+      seedContent?.trim()
+    ) {
       this.d.store.appendOptimisticItem(taskRunId, {
         type: "user_message",
         content: seedContent,
         timestamp: Date.now(),
+        ...(isResumeRun ? { pinToTop: false } : {}),
       });
     }
-    if (hasUserPrompt || isTerminalRun) {
+    if (hasCurrentRunUserPrompt || isTerminalRun) {
       // The stash is no longer needed once the real prompt lands - and a
       // finished run gets no further echoes, so leftover optimistic items
       // would otherwise linger as phantom tail items on the final transcript.

packages/ui/src/features/sessions/components/buildConversationItems.ts

@@ -87,7 +87,7 @@ export interface BuildResult {
 interface ProgressCardState {
   /** Step key → full step entry. Key order reflects arrival order. */
   steps: Map<string, Step>;
-  /** Reference to the pushed render item; mutated in place as events arrive. */
+  /** Replaced when steps change so memoized rows observe live progress. */
   renderItem: {
     sessionUpdate: "progress_group";
     steps: Step[];
@@ -818,8 +818,20 @@ function syncProgressCard(
       ? { ...step, status: "in_progress" as StepStatus }
       : step,
   );
-  card.renderItem.steps = ordered;
-  card.renderItem.isActive = ordered.some((s) => s.status === "in_progress");
+  const renderItem = {
+    sessionUpdate: "progress_group" as const,
+    steps: ordered,
+    isActive: ordered.some((step) => step.status === "in_progress"),
+  };
+  card.renderItem = renderItem;
+
+  const item = b.items[card.itemIndex];
+  if (
+    item?.type === "session_update" &&
+    item.update.sessionUpdate === "progress_group"
+  ) {
+    b.items[card.itemIndex] = { ...item, update: renderItem };
+  }
 }
 
 function handleProgress(

packages/ui/src/features/sessions/components/incrementalConversationItems.test.ts

@@ -364,6 +364,52 @@ describe("createIncrementalConversationBuilder", () => {
     );
   });
 
+  it("replaces a pre-prompt progress row when it completes after the prompt starts", () => {
+    const inc = createIncrementalConversationBuilder();
+    const events = [
+      progressMsg(
+        1,
+        "sandbox",
+        "in_progress",
+        "Restoring sandbox",
+        "setup:run-1",
+      ),
+      userPromptMsg(2, 1, "continue"),
+      progressMsg(3, "sandbox", "completed", "Restored sandbox", "setup:run-1"),
+    ];
+
+    const before = inc.update(events.slice(0, 2), true);
+    const beforeProgress = before.items.find(
+      (item) =>
+        item.type === "session_update" &&
+        item.update.sessionUpdate === "progress_group",
+    );
+
+    const after = inc.update(events, true);
+    const afterProgress = after.items.find(
+      (item) =>
+        item.type === "session_update" &&
+        item.update.sessionUpdate === "progress_group",
+    );
+
+    expect(afterProgress).not.toBe(beforeProgress);
+    expect(
+      afterProgress?.type === "session_update" &&
+        afterProgress.update.sessionUpdate === "progress_group"
+        ? afterProgress.update
+        : null,
+    ).toMatchObject({
+      isActive: false,
+      steps: [
+        {
+          key: "sandbox",
+          label: "Restored sandbox",
+          status: "completed",
+        },
+      ],
+    });
+  });
+
   it("keeps completed-turn item references stable while the active turn streams", () => {
     const inc = createIncrementalConversationBuilder();
     const base = [

packages/ui/src/features/sessions/components/mergeConversationItems.test.ts

@@ -2,6 +2,24 @@ import { describe, expect, it } from "vitest";
 import type { ConversationItem } from "./buildConversationItems";
 import { mergeConversationItems } from "./mergeConversationItems";
 
+function progressGroup(id: string): ConversationItem {
+  return {
+    type: "session_update",
+    id,
+    update: {
+      sessionUpdate: "progress_group",
+      steps: [],
+      isActive: true,
+    },
+    turnContext: {
+      toolCalls: new Map(),
+      childItems: new Map(),
+      turnCancelled: false,
+      turnComplete: false,
+    },
+  };
+}
+
 function userMessage(
   id: string,
   content: string,
@@ -160,6 +178,19 @@ describe("mergeConversationItems", () => {
     expect(result.map((i) => i.id)).toEqual(["setup", "opt"]);
   });
 
+  it("cloud: keeps a resumed prompt before trailing setup progress", () => {
+    const result = mergeConversationItems({
+      conversationItems: [
+        userMessage("old", "previous prompt"),
+        progressGroup("restore"),
+      ],
+      optimisticItems: [userMessage("opt", "follow up", false)],
+      isCloud: true,
+    });
+
+    expect(result.map((item) => item.id)).toEqual(["old", "opt", "restore"]);
+  });
+
   it("cloud: does not dedupe historical messages against tail follow-up optimistics", () => {
     const result = mergeConversationItems({
       conversationItems: [

packages/ui/src/features/sessions/components/mergeConversationItems.ts

@@ -28,8 +28,9 @@ function strippedUserContent(content: string): string {
 }
 
 // Cloud's initial optimistic is pinned to the top so the user's prompt stays
-// visible above setup progress. Follow-up optimistics render at the tail until
-// the streamed `session/prompt` arrives and replaces them.
+// visible above setup progress. Follow-up optimistics render at the tail, but
+// before trailing progress cards, to match where the streamed `session/prompt`
+// will appear.
 //
 // Local sessions keep optimistic at the chronological end — they rely on
 // `replaceOptimisticWithEvent` to swap optimistic↔real in place.
@@ -103,9 +104,23 @@ export function mergeConversationItems({
           };
         });
 
+  let tailInsertionIndex = dedupedConversation.length;
+  while (tailInsertionIndex > 0) {
+    const item = dedupedConversation[tailInsertionIndex - 1];
+    if (
+      item.type === "session_update" &&
+      item.update.sessionUpdate === "progress_group"
+    ) {
+      tailInsertionIndex--;
+    } else {
+      break;
+    }
+  }
+
   return [
     ...resolvedPinnedItems,
-    ...dedupedConversation,
+    ...dedupedConversation.slice(0, tailInsertionIndex),
     ...tailOptimisticItems,
+    ...dedupedConversation.slice(tailInsertionIndex),
   ];
 }

packages/ui/src/features/sessions/sessionServiceHost.test.ts

@@ -356,6 +356,9 @@ const mockConvertStoredEntriesToEvents = vi.hoisted(() =>
     ) => unknown[]
   >(() => []),
 );
+const mockHasSessionPromptEventForTaskRun = vi.hoisted(() =>
+  vi.fn(() => false),
+);
 
 vi.mock("@posthog/core/sessions/sessionEvents", async () => {
   const actual = await vi.importActual<
@@ -387,6 +390,7 @@ vi.mock("@posthog/core/sessions/sessionEvents", async () => {
     getStoredLogEventPosition: actual.getStoredLogEventPosition,
     getUserShellExecutesSinceLastPrompt: vi.fn(() => []),
     hasSessionPromptEvent: actual.hasSessionPromptEvent,
+    hasSessionPromptEventForTaskRun: mockHasSessionPromptEventForTaskRun,
     isAbsoluteFolderPath: actual.isAbsoluteFolderPath,
     isFatalSessionError: actual.isFatalSessionError,
     isRateLimitError: actual.isRateLimitError,
@@ -446,6 +450,7 @@ describe("SessionService", () => {
   beforeEach(() => {
     vi.clearAllMocks();
     mockConvertStoredEntriesToEvents.mockImplementation(() => []);
+    mockHasSessionPromptEventForTaskRun.mockReturnValue(false);
     resetSessionService();
     mockSettingsState.customInstructions = "";
     mockSettingsState.spokenNotifications = false;
@@ -4374,6 +4379,7 @@ describe("SessionService", () => {
           resumePrompt,
           resumeCompletion,
         ]);
+        mockHasSessionPromptEventForTaskRun.mockReturnValueOnce(true);
 
         service.watchCloudTask(
           "task-123",