fix(chat): support nonblocking Codex questions

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -1734,11 +1734,13 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
           threadId: ThreadId.make("thread-stale-user-input"),
           activity: {
             id: EventId.make("activity-stale-user-input-requested"),
+            sequence: 1,
             tone: "info",
             kind: "user-input.requested",
             summary: "User input requested",
             payload: {
               requestId: "user-input-request-stale-1",
+              isBlocking: false,
               questions: [
                 {
                   id: "sandbox_mode",
@@ -1759,6 +1761,16 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
         },
       });
 
+      const pendingRows = yield* sql<{
+        readonly pendingUserInputCount: number;
+        readonly blockingUserInputCount: number;
+      }>`
+        SELECT pending_user_input_count AS "pendingUserInputCount",
+               blocking_user_input_count AS "blockingUserInputCount"
+        FROM projection_threads WHERE thread_id = 'thread-stale-user-input'
+      `;
+      assert.deepEqual(pendingRows, [{ pendingUserInputCount: 1, blockingUserInputCount: 0 }]);
+
       yield* appendAndProject({
         type: "thread.activity-appended",
         eventId: EventId.make("evt-stale-user-input-4"),
@@ -1773,6 +1785,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
           threadId: ThreadId.make("thread-stale-user-input"),
           activity: {
             id: EventId.make("activity-stale-user-input-failed"),
+            sequence: 2,
             tone: "error",
             kind: "provider.user-input.respond.failed",
             summary: "Provider user input response failed",
@@ -1782,19 +1795,21 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
                 "Provider adapter request failed (codex) for item/tool/requestUserInput: Unknown pending Codex user input request: user-input-request-stale-1",
             },
             turnId: null,
-            createdAt: "2026-02-26T12:35:03.000Z",
+            createdAt: "2026-02-26T12:35:01.000Z",
           },
         },
       });
 
       const threadRows = yield* sql<{
         readonly pendingUserInputCount: number;
+        readonly blockingUserInputCount: number;
       }>`
-        SELECT pending_user_input_count AS "pendingUserInputCount"
+        SELECT pending_user_input_count AS "pendingUserInputCount",
+               blocking_user_input_count AS "blockingUserInputCount"
         FROM projection_threads
         WHERE thread_id = 'thread-stale-user-input'
       `;
-      assert.deepEqual(threadRows, [{ pendingUserInputCount: 0 }]);
+      assert.deepEqual(threadRows, [{ pendingUserInputCount: 0, blockingUserInputCount: 0 }]);
     }),
   );
 

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -14,9 +14,8 @@ import * as Path from "effect/Path";
 import * as Stream from "effect/Stream";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 import {
-  collectOpenPendingRequests,
+  countPendingUserInputs,
   isStalePendingRequestFailureDetail,
-  USER_INPUT_ACTIVITY_KINDS,
 } from "@threadlines/shared/pendingRequests";
 
 import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts";
@@ -109,16 +108,6 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
   return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
-function derivePendingUserInputCountFromActivities(
-  activities: ReadonlyArray<ProjectionThreadActivity>,
-): number {
-  const ordered = [...activities].toSorted(
-    (left, right) =>
-      compareTranscriptOrder(left, right) || left.activityId.localeCompare(right.activityId),
-  );
-  return collectOpenPendingRequests(ordered, USER_INPUT_ACTIVITY_KINDS).length;
-}
-
 function activityMayAffectThreadShellSummary(activity: { readonly kind: string }): boolean {
   switch (activity.kind) {
     case "approval.requested":
@@ -558,7 +547,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
       const pendingApprovalCount = pendingApprovals.filter(
         (approval) => approval.status === "pending",
       ).length;
-      const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+      // Provider timestamps can arrive out of order; count in transcript order.
+      const userInputCounts = countPendingUserInputs(
+        [...activities].toSorted(
+          (left, right) =>
+            compareTranscriptOrder(left, right) || left.activityId.localeCompare(right.activityId),
+        ),
+      );
       const hasActionableProposedPlan = deriveHasActionableProposedPlan({
         latestTurnId: existingRow.value.latestTurnId,
         proposedPlans,
@@ -568,7 +563,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
         ...existingRow.value,
         latestUserMessageAt,
         pendingApprovalCount,
-        pendingUserInputCount,
+        ...userInputCounts,
         hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
       });
     });
@@ -603,6 +598,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             latestUserMessageAt: null,
             pendingApprovalCount: 0,
             pendingUserInputCount: 0,
+            blockingUserInputCount: 0,
             hasActionableProposedPlan: 0,
             deletedAt: null,
           });
@@ -1654,7 +1650,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           // rows.  Other activity kinds that happen to carry a requestId
           // (e.g. user-input.requested / user-input.resolved) must not
           // pollute this projection — they have their own accounting via
-          // derivePendingUserInputCountFromActivities.
+          // countPendingUserInputs.
           if (event.payload.activity.kind !== "approval.requested") {
             return;
           }

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -519,6 +519,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
           latestUserMessageAt: "2026-02-24T00:00:04.000Z",
           hasPendingApprovals: true,
           hasPendingUserInput: false,
+          hasBlockingUserInput: false,
           hasActionableProposedPlan: false,
           cumulativeDiffStat: { additions: 2, deletions: 1 },
           diffStatBaselineTurnCount: 0,
@@ -1264,6 +1265,51 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
       assert.equal(snapshot.threads[0]?.activities.length, MAX_THREAD_ACTIVITIES);
       assert.equal(snapshot.threads[0]?.subagents?.[0]?.id, "agent-durable");
 
+      // Old open questions and approvals must survive a busy turn. Closed
+      // prompt history still leaves the recent activity window as usual.
+      const oldPrompts = [
+        {
+          id: "open-question",
+          kind: "user-input.requested",
+          payload: { requestId: "question", isBlocking: false },
+        },
+        { id: "open-approval", kind: "approval.requested", payload: { requestId: "approval" } },
+        { id: "closed-question", kind: "user-input.requested", payload: { requestId: "closed" } },
+        {
+          id: "closed-question-resolved",
+          kind: "user-input.resolved",
+          payload: { requestId: "closed" },
+        },
+      ];
+      for (const [index, prompt] of oldPrompts.entries()) {
+        yield* sql`
+          INSERT INTO projection_thread_activities (
+            activity_id, thread_id, tone, kind, summary, payload_json, sequence, created_at
+          ) VALUES (
+            ${prompt.id}, 'thread-activity-cap', 'info', ${prompt.kind}, 'Prompt',
+            ${JSON.stringify(prompt.payload)}, ${index}, '2026-03-01T00:00:00.000Z'
+          )
+        `;
+      }
+      const retainedSnapshot = yield* snapshotQuery.getSnapshot();
+      const retainedDetail = yield* snapshotQuery.getThreadDetailById(
+        ThreadId.make("thread-activity-cap"),
+      );
+      assert.equal(retainedDetail._tag, "Some");
+      if (retainedDetail._tag === "Some") {
+        const activities = retainedDetail.value.activities;
+        assert.equal(activities.length, MAX_THREAD_ACTIVITIES + 2);
+        assert.deepEqual(
+          activities.slice(0, 2).map((activity) => activity.id),
+          ["open-question", "open-approval"],
+        );
+        assert.equal(
+          activities.some((activity) => activity.id === "closed-question"),
+          false,
+        );
+        assert.deepEqual(retainedSnapshot.threads[0]?.activities, activities);
+      }
+
       yield* sql`DELETE FROM projection_thread_activities`;
       yield* sql`
         INSERT INTO projection_thread_activities (

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -39,6 +39,7 @@ import * as Struct from "effect/Struct";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 import * as SqlSchema from "effect/unstable/sql/SqlSchema";
 import { MAX_THREAD_ACTIVITIES, MAX_THREAD_MESSAGES } from "@threadlines/shared/threadLimits";
+import { retainRecentActivitiesAndOpenRequests } from "@threadlines/shared/pendingRequests";
 
 import {
   isPersistenceError,
@@ -531,6 +532,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           latest_user_message_at AS "latestUserMessageAt",
           pending_approval_count AS "pendingApprovalCount",
           pending_user_input_count AS "pendingUserInputCount",
+          blocking_user_input_count AS "blockingUserInputCount",
           has_actionable_proposed_plan AS "hasActionableProposedPlan",
           deleted_at AS "deletedAt"
         FROM projection_threads
@@ -568,6 +570,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           latest_user_message_at AS "latestUserMessageAt",
           pending_approval_count AS "pendingApprovalCount",
           pending_user_input_count AS "pendingUserInputCount",
+          blocking_user_input_count AS "blockingUserInputCount",
           has_actionable_proposed_plan AS "hasActionableProposedPlan",
           deleted_at AS "deletedAt"
         FROM projection_threads
@@ -607,6 +610,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           latest_user_message_at AS "latestUserMessageAt",
           pending_approval_count AS "pendingApprovalCount",
           pending_user_input_count AS "pendingUserInputCount",
+          blocking_user_input_count AS "blockingUserInputCount",
           has_actionable_proposed_plan AS "hasActionableProposedPlan",
           deleted_at AS "deletedAt"
         FROM projection_threads
@@ -729,6 +733,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           created_at AS "createdAt"
         FROM ranked_activities
         WHERE activity_rank <= ${MAX_THREAD_ACTIVITIES}
+          OR kind IN (
+            'approval.requested', 'approval.resolved', 'provider.approval.respond.failed',
+            'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed'
+          )
         ORDER BY
           thread_id ASC,
           event_sequence ASC,
@@ -1190,6 +1198,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           latest_user_message_at AS "latestUserMessageAt",
           pending_approval_count AS "pendingApprovalCount",
           pending_user_input_count AS "pendingUserInputCount",
+          blocking_user_input_count AS "blockingUserInputCount",
           has_actionable_proposed_plan AS "hasActionableProposedPlan",
           deleted_at AS "deletedAt"
         FROM projection_threads
@@ -1260,6 +1269,15 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
             created_at DESC,
             activity_id DESC
           LIMIT ${MAX_THREAD_ACTIVITIES}
+        ), retained_candidates AS (
+          SELECT * FROM limited_activities
+          UNION
+          SELECT * FROM projection_thread_activities
+          WHERE thread_id = ${threadId}
+            AND kind IN (
+              'approval.requested', 'approval.resolved', 'provider.approval.respond.failed',
+              'user-input.requested', 'user-input.resolved', 'provider.user-input.respond.failed'
+            )
         )
         SELECT
           activity_id AS "activityId",
@@ -1272,7 +1290,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           payload_json AS "payload",
           sequence,
           created_at AS "createdAt"
-        FROM limited_activities
+        FROM retained_candidates
         ORDER BY
           event_sequence ASC,
           sequence ASC,
@@ -1739,7 +1757,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
                 messages: messagesByThread.get(row.threadId) ?? [],
                 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
                 activities: dropStaleContextWindowActivities(
-                  activitiesByThread.get(row.threadId) ?? [],
+                  retainRecentActivitiesAndOpenRequests(
+                    activitiesByThread.get(row.threadId) ?? [],
+                    MAX_THREAD_ACTIVITIES,
+                  ),
                 ),
                 subagents: subagentsByThread.get(row.threadId) ?? [],
                 checkpoints: checkpointsByThread.get(row.threadId) ?? [],
@@ -2130,6 +2151,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
                     latestUserMessageAt: row.latestUserMessageAt,
                     hasPendingApprovals: row.pendingApprovalCount > 0,
                     hasPendingUserInput: row.pendingUserInputCount > 0,
+                    hasBlockingUserInput: row.blockingUserInputCount > 0,
                     hasActionableProposedPlan: row.hasActionableProposedPlan > 0,
                     cumulativeDiffStat: mapThreadDiffStat(diffStatByThread.get(row.threadId)),
                     diffStatBaselineTurnCount: row.diffStatBaselineTurnCount ?? 0,
@@ -2280,6 +2302,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
                   latestUserMessageAt: row.latestUserMessageAt,
                   hasPendingApprovals: row.pendingApprovalCount > 0,
                   hasPendingUserInput: row.pendingUserInputCount > 0,
+                  hasBlockingUserInput: row.blockingUserInputCount > 0,
                   hasActionableProposedPlan: row.hasActionableProposedPlan > 0,
                   cumulativeDiffStat: mapThreadDiffStat(diffStatByThread.get(row.threadId)),
                   diffStatBaselineTurnCount: row.diffStatBaselineTurnCount ?? 0,
@@ -2556,6 +2579,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
         latestUserMessageAt: threadRow.value.latestUserMessageAt,
         hasPendingApprovals: threadRow.value.pendingApprovalCount > 0,
         hasPendingUserInput: threadRow.value.pendingUserInputCount > 0,
+        hasBlockingUserInput: threadRow.value.blockingUserInputCount > 0,
         hasActionableProposedPlan: threadRow.value.hasActionableProposedPlan > 0,
         cumulativeDiffStat: mapThreadDiffStat(Option.getOrUndefined(diffStatRow)),
         diffStatBaselineTurnCount: threadRow.value.diffStatBaselineTurnCount ?? 0,
@@ -2667,7 +2691,12 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
         deletedAt: null,
         messages: messageRows.map(mapThreadMessageRow),
         proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
-        activities: dropStaleContextWindowActivities(activityRows.map(mapThreadActivityRow)),
+        activities: dropStaleContextWindowActivities(
+          retainRecentActivitiesAndOpenRequests(
+            activityRows.map(mapThreadActivityRow),
+            MAX_THREAD_ACTIVITIES,
+          ),
+        ),
         subagents: subagentRows.map(mapThreadSubagentRow),
         checkpoints: checkpointRows.map((row) => ({
           turnId: row.turnId,

apps/server/src/orchestration/Layers/ProviderActivityProjection.ts

@@ -717,6 +717,7 @@ export function projectRuntimeEventToActivities(
             ...(event.payload.isBlocking !== undefined
               ? { isBlocking: event.payload.isBlocking }
               : {}),
+            ...(event.payload.responseMode ? { responseMode: event.payload.responseMode } : {}),
           },
         }),
       ];
@@ -727,10 +728,11 @@ export function projectRuntimeEventToActivities(
           id: event.eventId,
           tone: "info",
           kind: "user-input.resolved",
-          summary: "User input submitted",
+          summary: event.payload.reason ? "Question closed" : "User input submitted",
           payload: {
             ...(event.requestId ? { requestId: event.requestId } : {}),
             answers: event.payload.answers,
+            ...(event.payload.reason ? { reason: event.payload.reason } : {}),
           },
         }),
       ];