fix(aio): never lower a settled terminal cursor from a late hydration

Review finding: cloudHydrationPromises memoizes terminal-chain and
resume-chain hydrations under different mode keys, so both can be in
flight for the same run. If the run settles while a resume-chain
hydration's fetches are pending, the terminal-chain hydration records
the full chain as processed - and the late resume-chain write then
clobbered processedLineCount with its smaller leaf-only cursor
(updateSession is a blind assign with no monotonic guard).

Impact was bounded (the transcript content is identical and the next
terminal hydration self-heals), but it violated the invariant the
mode-keying fix set out to guarantee: the terminal transcript is
authoritative. Re-read the session at write time and floor a
non-terminal cursor at the settled value once cloudStatus is terminal.

The regression test resolves the resume-chain fetches after the run
settles and asserts the settled cursor survives; verified to fail
without the fix.

Generated-By: PostHog Code
Task-Id: 0c511836-2180-455a-9b58-45df2a0661ec

Author: tatoalo

packages/core/src/sessions/sessionService.ts

@@ -6220,6 +6220,15 @@ export class SessionService {
       };
     }
 
+    // A concurrent terminal-chain hydration (they memoize under different
+    // mode keys) may have already recorded the full chain as processed; a
+    // leaf-stream cursor from this older hydration must never lower it once
+    // the run has settled.
+    const settled = this.d.store.getSessions()[taskRunId];
+    const settledCursor = isTerminalStatus(settled?.cloudStatus)
+      ? (settled?.processedLineCount ?? 0)
+      : 0;
+
     this.d.store.updateSession(taskRunId, {
       events,
       isCloud: true,
@@ -6229,7 +6238,7 @@ export class SessionService {
       // re-applies it; live resume runs keep the leaf-stream cursor.
       processedLineCount: isTerminalRun
         ? effectiveLineCount
-        : liveStreamLineCount,
+        : Math.max(liveStreamLineCount, settledCursor),
     });
     this.surfacePersistedPendingPermissions(taskRunId, rawEntries);
     this.pendingPermissionHydratedRuns.add(taskRunId);

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

@@ -1929,6 +1929,93 @@ describe("SessionService", () => {
       resolveFirstHydration({ entries: [], complete: true });
     });
 
+    it("keeps the settled terminal cursor when a resume-chain hydration resolves late", async () => {
+      const service = getSessionService();
+      const session = createMockSession({
+        taskId: "task-123",
+        taskRunId: "run-123",
+        isCloud: true,
+        events: [],
+      });
+      mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session);
+      mockSessionStoreSetters.getSessions.mockReturnValue({
+        "run-123": session,
+      });
+      mockTrpcLogs.readLocalLogs.query.mockResolvedValue("");
+      mockTrpcLogs.fetchS3Logs.query.mockResolvedValue("");
+      const entry = (text: string) => ({
+        timestamp: "2026-07-15T10:00:00+00:00",
+        notification: {
+          method: "session/update",
+          params: {
+            update: {
+              sessionUpdate: "agent_message",
+              content: { type: "text", text },
+            },
+          },
+        },
+      });
+      let resolveAncestor!: (r: {
+        entries: object[];
+        complete: boolean;
+      }) => void;
+      let resolveCurrent!: (r: {
+        entries: object[];
+        complete: boolean;
+      }) => void;
+      mockAuthenticatedClient.getTaskRunSessionLogsResult
+        .mockReturnValueOnce(
+          new Promise((r) => {
+            resolveAncestor = r;
+          }),
+        )
+        .mockReturnValueOnce(
+          new Promise((r) => {
+            resolveCurrent = r;
+          }),
+        );
+
+      service.watchCloudTask(
+        "task-123",
+        "run-123",
+        "https://api.anthropic.com",
+        123,
+        undefined,
+        "https://example.com/logs/run-123",
+        undefined,
+        "claude",
+        undefined,
+        undefined,
+        undefined,
+        "in_progress",
+        undefined,
+        { resume_from_run_id: "previous-run" },
+      );
+
+      await vi.waitFor(() => {
+        expect(
+          mockAuthenticatedClient.getTaskRunSessionLogsResult,
+        ).toHaveBeenCalledTimes(2);
+      });
+
+      // The run settles while the resume-chain fetches are still in flight,
+      // exactly as a concurrent terminal-chain hydration records it.
+      session.cloudStatus = "completed";
+      session.processedLineCount = 5;
+
+      resolveAncestor({ entries: [entry("a"), entry("b")], complete: true });
+      resolveCurrent({ entries: [entry("c")], complete: true });
+
+      // The late resume-chain write must not lower the settled cursor to its
+      // leaf-only count.
+      await vi.waitFor(() => {
+        expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith(
+          "run-123",
+          expect.objectContaining({ processedLineCount: 5 }),
+        );
+      });
+    });
+
     it("does not re-subscribe across repeated calls for a hydrated terminal run", () => {
       const service = getSessionService();
       mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(