fix(cloud-task): make question answered state durable across chat switches and restarts (#3241)

Author: tatoalo

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

@@ -135,6 +135,62 @@ describe("CloudTaskService", () => {
     vi.unstubAllGlobals();
   });
 
+  it("emits a replayed permission_request frame only once", 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",
+        }),
+      )
+      .mockResolvedValue(
+        createJsonResponse([], 200, { "X-Has-More": "false" }),
+      );
+
+    const frame =
+      'id: 5\ndata: {"type":"permission_request","requestId":"req-1","toolCall":{"toolCallId":"tool-1"},"options":[]}\n\n';
+    const trailingEntry =
+      'id: 6\ndata: {"type":"notification","timestamp":"2026-01-01T00:00:02Z","notification":{"jsonrpc":"2.0","method":"_posthog/console","params":{"sessionId":"run-1","level":"info","message":"after replay"}}}\n\n';
+    // The durable stream re-sends the tail on reconnect/replay — the same
+    // frame arriving twice must not re-surface the question a second time.
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(frame + frame + trailingEntry),
+    );
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() =>
+      updates.some((u) => (u as { kind?: string }).kind === "logs"),
+    );
+
+    const permissionUpdates = updates.filter(
+      (u) => (u as { kind?: string }).kind === "permission_request",
+    );
+    expect(permissionUpdates).toEqual([
+      {
+        taskId: "task-1",
+        runId: "run-1",
+        kind: "permission_request",
+        requestId: "req-1",
+        toolCall: { toolCallId: "tool-1" },
+        options: [],
+      },
+    ]);
+  });
+
   it("bootstraps paged backlog for active runs and drains deduped live SSE entries", async () => {
     const updates: unknown[] = [];
     service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));

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

@@ -1138,6 +1138,22 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       return null;
     }
 
+    // Drop a re-delivered event by its stream id. The durable stream re-sends
+    // the tail on reconnect/replay: each re-sent log entry would otherwise be
+    // counted as a new entry (advancing totalEntryCount past the renderer's
+    // processedLineCount guard) and emitted again — the root cause of duplicate
+    // transcript entries and back-to-back completion notifications — and a
+    // re-sent permission_request frame would re-surface an already-answered
+    // question as a fresh pending card. Events without an id (legacy servers)
+    // fall through and are handled downstream.
+    const eventId = event.id;
+    if (eventId !== undefined) {
+      if (watcher.seenEventIds.has(eventId)) {
+        return null;
+      }
+      watcher.seenEventIds.add(eventId);
+    }
+
     if (isPermissionRequestEvent(event.data)) {
       this.emit(CloudTaskEvent.Update, {
         taskId: watcher.taskId,
@@ -1150,20 +1166,6 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       return null;
     }
 
-    // Drop a re-delivered log entry by its stream id. The durable stream
-    // re-sends the tail on reconnect/replay, and each resend would otherwise be
-    // counted as a new entry (advancing totalEntryCount past the renderer's
-    // processedLineCount guard) and emitted again — the root cause of duplicate
-    // transcript entries and back-to-back completion notifications. Entries
-    // without an id (legacy servers) fall through and are handled downstream.
-    const eventId = event.id;
-    if (eventId !== undefined) {
-      if (watcher.seenEventIds.has(eventId)) {
-        return null;
-      }
-      watcher.seenEventIds.add(eventId);
-    }
-
     watcher.pendingLogEntries.push(event.data as StoredLogEntry);
     if (watcher.pendingLogEntries.length >= EVENT_BATCH_MAX_SIZE) {
       this.flushLogBatch(key);

packages/core/src/sessions/sessionService.ts

@@ -98,6 +98,7 @@ const AUTO_RETRY_MAX_ATTEMPTS = 2;
 const AUTO_RETRY_DELAY_MS = 10_000;
 const AUTH_RESTORE_MAX_RETRY_WAITS = 6;
 const MAX_SUPERSEDED_RUN_IDS = 100;
+const MAX_RESPONDED_PERMISSION_REQUEST_IDS = 500;
 /**
  * Streamed events are buffered and flushed on this cadence so a burst of tokens
  * coalesces into one processing pass (and roughly one render) instead of one
@@ -607,6 +608,13 @@ export class SessionService {
   private cloudLogGapReconciler: CloudLogGapReconciler;
   /** Maps toolCallId → cloud requestId for routing permission responses */
   private cloudPermissionRequestIds = new Map<string, string>();
+  /**
+   * Cloud permission requestIds the user has already responded to this app
+   * session. A stale snapshot (a resolved marker not yet flushed to storage)
+   * or a replayed stream frame can re-deliver an answered request; without
+   * this guard it would re-surface as a fresh pending card.
+   */
+  private respondedCloudPermissionRequestIds = new Set<string>();
   private liveTurnContent = new Map<
     string,
     { startedAtTs: number; agentTextChunks: number; agentOutputEvents: number }
@@ -2167,6 +2175,15 @@ export class SessionService {
       return;
     }
 
+    if (this.respondedCloudPermissionRequestIds.has(update.requestId)) {
+      this.d.log.debug("Skipping already-answered cloud permission request", {
+        taskRunId,
+        requestId: update.requestId,
+        toolCallId: update.toolCall.toolCallId,
+      });
+      return;
+    }
+
     if (
       isPermissionRequestAlreadySurfaced(
         session.pendingPermissions,
@@ -3180,6 +3197,7 @@ export class SessionService {
     session: AgentSession,
     permission: PermissionRequest | undefined,
     toolCallId: string,
+    requestId: string | undefined,
     optionId: string,
     customInput?: string,
     answers?: Record<string, string>,
@@ -3192,23 +3210,88 @@ export class SessionService {
       customInput,
       answers,
     );
-    if (!answerPrompt) {
+    if (answerPrompt) {
+      await this.sendCloudPrompt(
+        { ...session, cloudStatus: cloudStatus ?? session.cloudStatus },
+        answerPrompt,
+      );
+      this.d.log.info("Permission answer resumed terminal cloud run", {
+        taskId: session.taskId,
+        toolCallId,
+        optionId,
+      });
+    } else {
       this.d.log.info("Dropped permission response for terminal cloud run", {
         taskId: session.taskId,
         toolCallId,
         optionId,
       });
-      return;
     }
-    await this.sendCloudPrompt(
-      { ...session, cloudStatus: cloudStatus ?? session.cloudStatus },
-      answerPrompt,
-    );
-    this.d.log.info("Permission answer resumed terminal cloud run", {
-      taskId: session.taskId,
+    await this.persistCloudPermissionResolution(
+      session.taskId,
+      permission?.taskRunId ?? session.taskRunId,
       toolCallId,
+      requestId,
       optionId,
-    });
+    );
+  }
+
+  /**
+   * Record a response to a permission request whose sandbox is gone. A live
+   * sandbox writes `_posthog/permission_resolved` to the run log when it
+   * resolves a request, but a request answered after its run terminalized has
+   * no sandbox left to do that — without this record the request stays
+   * pending in the persisted log forever, and every future derivation (app
+   * restart, another device, a session rebuild) re-surfaces the
+   * already-answered question as a fresh card.
+   */
+  private async persistCloudPermissionResolution(
+    taskId: string,
+    taskRunId: string,
+    toolCallId: string,
+    requestId: string | undefined,
+    optionId: string,
+  ): Promise<void> {
+    if (!requestId) return;
+    this.markCloudPermissionResponded(requestId);
+
+    const client = await this.d.getAuthenticatedClient();
+    if (!client) return;
+    try {
+      await client.appendTaskRunLog(taskId, taskRunId, [
+        {
+          type: "notification",
+          timestamp: new Date().toISOString(),
+          notification: {
+            method: POSTHOG_NOTIFICATIONS.PERMISSION_RESOLVED,
+            params: { requestId, toolCallId, optionId },
+          },
+        },
+      ]);
+    } catch (error) {
+      this.d.log.warn("Failed to persist permission resolution to run log", {
+        taskId,
+        taskRunId,
+        toolCallId,
+        error,
+      });
+    }
+  }
+
+  private markCloudPermissionResponded(requestId: string): void {
+    this.respondedCloudPermissionRequestIds.add(requestId);
+    // add() grows the set by at most one, so one eviction restores the cap.
+    if (
+      this.respondedCloudPermissionRequestIds.size >
+      MAX_RESPONDED_PERMISSION_REQUEST_IDS
+    ) {
+      const oldest = this.respondedCloudPermissionRequestIds
+        .values()
+        .next().value;
+      if (oldest !== undefined) {
+        this.respondedCloudPermissionRequestIds.delete(oldest);
+      }
+    }
   }
 
   // --- Permissions ---
@@ -3270,6 +3353,7 @@ export class SessionService {
           session,
           permission,
           toolCallId,
+          cloudRequestId,
           optionId,
           customInput,
           answers,
@@ -3293,6 +3377,7 @@ export class SessionService {
               session,
               permission,
               toolCallId,
+              cloudRequestId,
               optionId,
               customInput,
               answers,
@@ -3302,6 +3387,10 @@ export class SessionService {
           }
           throw error;
         }
+        // The live sandbox persists its own resolved marker; remember the
+        // response locally so a snapshot fetched before that marker flushes
+        // to storage cannot re-surface the question.
+        this.markCloudPermissionResponded(cloudRequestId);
       } else {
         await this.d.trpc.agent.respondToPermission.mutate({
           taskRunId: session.taskRunId,
@@ -3353,8 +3442,16 @@ export class SessionService {
     try {
       if (session.isCloud && isTerminalStatus(session.cloudStatus)) {
         // The run is over — the card was resolved locally above and there is no
-        // live permission promise left to reject.
+        // live permission promise left to reject. Persist the dismissal so the
+        // request is not re-derived as pending from the run log later.
         this.cloudPermissionRequestIds.delete(toolCallId);
+        await this.persistCloudPermissionResolution(
+          session.taskId,
+          permission?.taskRunId ?? session.taskRunId,
+          toolCallId,
+          cloudRequestId,
+          "cancelled",
+        );
         return;
       }
       if (session.isCloud && cloudRequestId) {
@@ -3364,6 +3461,7 @@ export class SessionService {
           optionId: "reject_with_feedback",
           customInput: "User cancelled the permission request.",
         });
+        this.markCloudPermissionResponded(cloudRequestId);
       } else {
         await this.d.trpc.agent.cancelPermission.mutate({
           taskRunId: session.taskRunId,

packages/ui/src/features/permissions/QuestionPermission.test.tsx

@@ -0,0 +1,95 @@
+import { Theme } from "@radix-ui/themes";
+import { render, screen } from "@testing-library/react";
+import userEvent from "@testing-library/user-event";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { QuestionPermission } from "./QuestionPermission";
+import { useQuestionDraftStore } from "./questionDraftStore";
+import type { PermissionToolCall } from "./types";
+
+const toolCall = {
+  toolCallId: "question-1",
+  title: "Questions",
+  _meta: {
+    codeToolKind: "question",
+    questions: [
+      {
+        question: "Which framework?",
+        header: "Framework",
+        options: [{ label: "React" }, { label: "Vue" }],
+      },
+      {
+        question: "Which color?",
+        header: "Color",
+        options: [{ label: "Red" }, { label: "Blue" }],
+      },
+    ],
+  },
+} as unknown as PermissionToolCall;
+
+function renderQuestion() {
+  const onSelect = vi.fn();
+  const onCancel = vi.fn();
+  const view = render(
+    <Theme>
+      <QuestionPermission
+        toolCall={toolCall}
+        options={[]}
+        onSelect={onSelect}
+        onCancel={onCancel}
+      />
+    </Theme>,
+  );
+  return { onSelect, onCancel, view };
+}
+
+describe("QuestionPermission", () => {
+  beforeEach(() => {
+    useQuestionDraftStore.setState({ drafts: new Map() });
+  });
+
+  it("restores in-progress answers when the card remounts", async () => {
+    const user = userEvent.setup();
+    const first = renderQuestion();
+
+    await user.click(screen.getByText("React"));
+    await user.click(screen.getByText("Next"));
+    expect(screen.getByText("Which color?")).toBeDefined();
+
+    // Switching chats unmounts the card; the store keeps the draft.
+    first.view.unmount();
+
+    const second = renderQuestion();
+    expect(screen.getByText("Which color?")).toBeDefined();
+
+    await user.click(screen.getByText("Blue"));
+    await user.click(screen.getByText("Next"));
+
+    // The review step summarizes the answer given before the remount.
+    expect(screen.getByText("Ready to submit your answers?")).toBeDefined();
+    expect(screen.getByText("React")).toBeDefined();
+
+    const submitOptions = screen.getAllByText("Submit");
+    await user.click(submitOptions[submitOptions.length - 1] as HTMLElement);
+
+    expect(second.onSelect).toHaveBeenCalledWith("_submit", undefined, {
+      "Which framework?": "React",
+      "Which color?": "Blue",
+    });
+    expect(useQuestionDraftStore.getState().drafts.size).toBe(0);
+  });
+
+  it("clears the draft when the card is cancelled", async () => {
+    const user = userEvent.setup();
+    const { onCancel } = renderQuestion();
+
+    await user.click(screen.getByText("React"));
+    await user.click(screen.getByText("Next"));
+    await user.click(screen.getByText("Red"));
+    await user.click(screen.getByText("Next"));
+
+    await user.click(screen.getByText("Cancel"));
+
+    expect(onCancel).toHaveBeenCalledTimes(1);
+    expect(useQuestionDraftStore.getState().drafts.size).toBe(0);
+  });
+});