fix(aio): harden cloud run terminal state

Author: tatoalo

packages/agent/tsup.config.ts

@@ -88,6 +88,7 @@ const sharedOptions = {
     "@posthog/git",
     "@posthog/enricher",
     "@posthog/harness",
+    /^@opentelemetry\//,
     "fflate",
   ],
   external: [

packages/core/src/sessions/sessionService.ts

@@ -5508,6 +5508,27 @@ export class SessionService {
       if (onStatusChange) {
         existingWatcher.onStatusChange = onStatusChange;
       }
+      // The run finished while a live watcher was still attached: apply the
+      // final transcript, mark the terminal status, and tear the watcher down
+      // instead of leaving a live stream on a dead run.
+      if (isTerminalStatus(runStatus)) {
+        const terminalSession = this.d.store.getSessionByTaskId(taskId);
+        if (terminalSession?.taskRunId === taskRunId) {
+          void this.hydrateCloudTaskSessionFromLogs(
+            taskId,
+            taskRunId,
+            logUrl,
+            taskDescription,
+            runStatus,
+            runState,
+          );
+        }
+        this.finalizeTerminalCloudTask(taskId, taskRunId, {
+          status: runStatus,
+        });
+        this.stopCloudTaskWatch(taskId);
+        return () => {};
+      }
       // Ensure configOptions is populated on revisit
       const existing = this.d.store.getSessionByTaskId(taskId);
       if (existing) {
@@ -5687,6 +5708,26 @@ export class SessionService {
       initialReasoningEffort,
     );
 
+    // A run that is already terminal has no live stream to subscribe to:
+    // hydrate the final transcript, settle the terminal status, and return
+    // without registering a watcher. Plain hydration already fetches the
+    // full resume chain for terminal runs, and the resume wrapper only
+    // exists to buffer a live stream that does not exist here.
+    if (isTerminalStatus(runStatus)) {
+      void this.hydrateCloudTaskSessionFromLogs(
+        taskId,
+        taskRunId,
+        logUrl,
+        taskDescription,
+        runStatus,
+        runState,
+      );
+      this.finalizeTerminalCloudTask(taskId, taskRunId, {
+        status: runStatus,
+      });
+      return () => {};
+    }
+
     const processCloudUpdate = (update: CloudTaskUpdatePayload): void => {
       if (update.kind === "logs" || update.kind === "snapshot") {
         this.d.store.updateSession(taskRunId, {
@@ -5707,11 +5748,18 @@ export class SessionService {
               ),
             }
           : update;
+      // Evaluate staleness before handleCloudTaskUpdate mutates cloudStatus:
+      // a late non-terminal update must not re-notify after the run settled.
+      const isStaleNonTerminalStatus = this.isStaleNonTerminalCloudUpdate(
+        taskRunId,
+        normalizedUpdate,
+      );
       this.handleCloudTaskUpdate(taskRunId, normalizedUpdate);
       if (
         (update.kind === "status" ||
           update.kind === "snapshot" ||
           update.kind === "error") &&
+        !isStaleNonTerminalStatus &&
         watcher?.onStatusChange
       ) {
         watcher.onStatusChange();
@@ -5928,7 +5976,8 @@ export class SessionService {
         ? runState.resume_from_run_id
         : undefined;
     const isResumeRun = Boolean(resumeFromRunId);
-    if (isTerminalStatus(runStatus) || isResumeRun) {
+    const isTerminalRun = isTerminalStatus(runStatus);
+    if (isTerminalRun || isResumeRun) {
       // Resume chains need the full history even while the leaf run is still
       // active; otherwise a renderer restart hydrates only the final run.
       // Non-resume in-progress runs keep using the single-run log so hydrate
@@ -6002,6 +6051,16 @@ export class SessionService {
         }
         rawEntries = result.entries;
         liveStreamLineCount = rawEntries.length;
+        // A terminal run whose persisted chain comes back empty can still
+        // have a complete S3 session log (persistence raced teardown); fall
+        // back to it rather than hydrating an empty final transcript.
+        if (rawEntries.length === 0 && logUrl) {
+          const parsed = await this.fetchSessionLogs(logUrl, taskRunId);
+          if (parsed.rawEntries.length > 0) {
+            rawEntries = parsed.rawEntries;
+            liveStreamLineCount = parsed.totalLineCount;
+          }
+        }
       }
     } else {
       const parsed = await this.fetchSessionLogs(logUrl, taskRunId);
@@ -6049,7 +6108,7 @@ export class SessionService {
     // its chip renders right away) over the bare task description.
     const seedContent =
       this.initialCloudOptimisticPrompt.get(taskId) ?? taskDescription;
-    if (!hasUserPrompt && seedContent?.trim()) {
+    if (!isTerminalRun && !hasUserPrompt && seedContent?.trim()) {
       this.d.store.appendOptimisticItem(taskRunId, {
         type: "user_message",
         content: seedContent,
@@ -6061,6 +6120,12 @@ export class SessionService {
       this.initialCloudOptimisticPrompt.delete(taskId);
       this.d.store.clearTailOptimisticItems(taskRunId);
     }
+    if (isTerminalRun) {
+      // A finished run gets no further echoes: any optimistic leftovers would
+      // otherwise linger as phantom tail items on the final transcript.
+      this.initialCloudOptimisticPrompt.delete(taskId);
+      this.d.store.clearTailOptimisticItems(taskRunId);
+    }
 
     if (rawEntries.length === 0) {
       this.pendingPermissionHydratedRuns.add(taskRunId);
@@ -6072,7 +6137,19 @@ export class SessionService {
 
     // If live updates already populated a processed count, don't overwrite
     // that newer state with the persisted baseline fetched during startup.
-    if (
+    // Terminal hydration is different: it is the final transcript, so apply
+    // it whenever the persisted chain has more lines than the local stream.
+    const effectiveLineCount = Math.max(liveStreamLineCount, rawEntries.length);
+    if (isTerminalRun) {
+      if ((session.processedLineCount ?? 0) >= effectiveLineCount) {
+        this.surfacePersistedPendingPermissions(taskRunId, rawEntries);
+        this.pendingPermissionHydratedRuns.add(taskRunId);
+        return {
+          historyEntryCount: rawEntries.length,
+          liveStreamLineCount: session.processedLineCount ?? liveStreamLineCount,
+        };
+      }
+    } else if (
       session.processedLineCount !== undefined &&
       session.processedLineCount > 0 &&
       !isResumeRun
@@ -6090,20 +6167,83 @@ export class SessionService {
       isCloud: true,
       logUrl: logUrl ?? session.logUrl,
       cloudTranscriptEntryCount: rawEntries.length,
-      processedLineCount: liveStreamLineCount,
+      // Terminal hydration records the whole chain as processed so nothing
+      // re-applies it; live resume runs keep the leaf-stream cursor.
+      processedLineCount: isTerminalRun ? effectiveLineCount : liveStreamLineCount,
     });
     this.surfacePersistedPendingPermissions(taskRunId, rawEntries);
     this.pendingPermissionHydratedRuns.add(taskRunId);
     // Without this the "Galumphing…" indicator stays hidden when the hydrated
     // baseline already contains an in-flight session/prompt — the live delta
     // path otherwise sees delta <= 0 and never re-evaluates the tail.
     this.updatePromptStateFromEvents(taskRunId, events);
+    if (isTerminalRun) {
+      this.clearTerminalCloudPromptState(taskId, taskRunId);
+    }
     return {
       historyEntryCount: rawEntries.length,
       liveStreamLineCount,
     };
   }
 
+  private finalizeTerminalCloudTask(
+    taskId: string,
+    taskRunId: string,
+    fields: {
+      status?: TaskRunStatus;
+      stage?: string | null;
+      output?: Record<string, unknown> | null;
+      errorMessage?: string | null;
+      branch?: string | null;
+    },
+  ): void {
+    this.d.store.updateCloudStatus(taskRunId, fields);
+    this.clearTerminalCloudPromptState(taskId, taskRunId);
+  }
+
+  /**
+   * A terminal run can never flush its queue or answer a pending prompt;
+   * leaving those set keeps the composer in a busy state forever.
+   */
+  private clearTerminalCloudPromptState(
+    taskId: string,
+    taskRunId: string,
+  ): void {
+    const session = this.d.store.getSessions()[taskRunId];
+    if (
+      !session ||
+      (!session.isPromptPending && session.messageQueue.length === 0)
+    ) {
+      return;
+    }
+
+    this.d.store.clearMessageQueue(taskId);
+    this.d.store.updateSession(taskRunId, {
+      isPromptPending: false,
+    });
+  }
+
+  /**
+   * SSE replays and out-of-order bursts can deliver a non-terminal status
+   * after the run already settled; applying it would revive a finished run.
+   */
+  private isStaleNonTerminalCloudUpdate(
+    taskRunId: string,
+    update: CloudTaskUpdatePayload,
+  ): boolean {
+    if (update.kind !== "status" && update.kind !== "snapshot") {
+      return false;
+    }
+    if (update.status === undefined) {
+      return false;
+    }
+    const currentCloudStatus =
+      this.d.store.getSessions()[taskRunId]?.cloudStatus;
+    return (
+      isTerminalStatus(currentCloudStatus) && !isTerminalStatus(update.status)
+    );
+  }
+
   private isCurrentCloudTaskWatcher(
     taskId: string,
     runId: string,
@@ -7028,7 +7168,9 @@ export class SessionService {
     }
 
     if (update.kind === "snapshot" && !isTerminalStatus(update.status)) {
-      this.surfacePersistedPendingPermissions(taskRunId, update.newEntries);
+      if (!this.isStaleNonTerminalCloudUpdate(taskRunId, update)) {
+        this.surfacePersistedPendingPermissions(taskRunId, update.newEntries);
+      }
     }
 
     // NOTE: Don't auto-flush on `!isPromptPending && queue.length > 0` here.
@@ -7041,18 +7183,20 @@ export class SessionService {
 
     // Update cloud status fields if present
     if (update.kind === "status" || update.kind === "snapshot") {
-      this.d.store.updateCloudStatus(taskRunId, {
-        status: update.status,
-        stage: update.stage,
-        output: update.output,
-        errorMessage: update.errorMessage,
-        branch: update.branch,
-      });
-
-      if (update.status === "in_progress") {
-        this.tryRecoverIdleCloudQueue(taskRunId, {
-          serverSandboxAlive: update.sandboxAlive,
+      if (!this.isStaleNonTerminalCloudUpdate(taskRunId, update)) {
+        this.d.store.updateCloudStatus(taskRunId, {
+          status: update.status,
+          stage: update.stage,
+          output: update.output,
+          errorMessage: update.errorMessage,
+          branch: update.branch,
         });
+
+        if (update.status === "in_progress") {
+          this.tryRecoverIdleCloudQueue(taskRunId, {
+            serverSandboxAlive: update.sandboxAlive,
+          });
+        }
       }
 
       if (isTerminalStatus(update.status)) {

packages/core/src/sessions/sessionStore.ts

@@ -8,6 +8,7 @@ import {
   sendableQueuePrefixLength,
   type TaskRunStatus,
 } from "@posthog/shared";
+import { isTerminalStatus } from "@posthog/shared/domain-types";
 import { setAutoFreeze } from "immer";
 import { immer } from "zustand/middleware/immer";
 import { createStore } from "zustand/vanilla";
@@ -180,7 +181,16 @@ export const sessionStoreSetters = {
     sessionStore.setState((state) => {
       const session = state.sessions[taskRunId];
       if (!session) return;
-      if (fields.status !== undefined) session.cloudStatus = fields.status;
+      if (fields.status !== undefined) {
+        const currentStatus = session.cloudStatus;
+        if (
+          isTerminalStatus(currentStatus) &&
+          !isTerminalStatus(fields.status)
+        ) {
+          return;
+        }
+        session.cloudStatus = fields.status;
+      }
       if (fields.stage !== undefined) session.cloudStage = fields.stage;
       if (fields.output !== undefined) session.cloudOutput = fields.output;
       if (fields.errorMessage !== undefined)

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

@@ -0,0 +1,87 @@
+import type { AgentSession } from "@posthog/shared";
+import type { Task, TaskRunStatus } from "@posthog/shared/domain-types";
+import { describe, expect, it } from "vitest";
+import { deriveSessionViewState } from "./sessionViewState";
+
+function makeTask(runStatus: TaskRunStatus, runId = "run-1"): Task {
+  return {
+    id: "task-1",
+    task_number: 1,
+    slug: "task-1",
+    title: "Task",
+    description: "",
+    created_at: "2026-01-01T00:00:00.000Z",
+    updated_at: "2026-01-01T00:00:00.000Z",
+    origin_product: "user_created",
+    latest_run: {
+      id: runId,
+      status: runStatus,
+      environment: "cloud",
+    } as Task["latest_run"],
+  };
+}
+
+function makeSession(
+  cloudStatus: TaskRunStatus,
+  taskRunId = "run-1",
+): AgentSession {
+  return {
+    taskId: "task-1",
+    taskRunId,
+    taskTitle: "Task",
+    channel: `agent-event:${taskRunId}`,
+    status: "connected",
+    events: [],
+    startedAt: 0,
+    isCloud: true,
+    cloudStatus,
+    isPromptPending: false,
+    isCompacting: false,
+    promptStartedAt: null,
+    pendingPermissions: new Map(),
+    pausedDurationMs: 0,
+    messageQueue: [],
+    optimisticItems: [],
+  };
+}
+
+describe("deriveSessionViewState", () => {
+  it("uses terminal task status over stale same-run session status", () => {
+    const state = deriveSessionViewState(
+      makeSession("in_progress"),
+      makeTask("completed"),
+      null,
+      true,
+    );
+
+    expect(state.cloudStatus).toBe("completed");
+    expect(state.isCloudRunTerminal).toBe(true);
+    expect(state.isInitializing).toBe(false);
+  });
+
+  it("uses the task status when the session belongs to an older run", () => {
+    const state = deriveSessionViewState(
+      makeSession("completed", "old-run"),
+      makeTask("in_progress", "new-run"),
+      null,
+      true,
+    );
+
+    expect(state.cloudStatus).toBe("in_progress");
+    expect(state.isCloudRunNotTerminal).toBe(true);
+  });
+
+  it("treats not_started as a non-terminal cloud state", () => {
+    const state = deriveSessionViewState(
+      undefined,
+      makeTask("not_started"),
+      null,
+      true,
+    );
+
+    expect(state.cloudStatus).toBe("not_started");
+    expect(state.isCloudRunNotTerminal).toBe(true);
+    expect(state.isCloudRunTerminal).toBe(false);
+    expect(state.isInitializing).toBe(true);
+  });
+});