perf(server): request_secret stops re-reading the thread twice a second while it waits

apps/server/src/mcp/OrchestratorMcpService.activity.test.ts

@@ -17,6 +17,7 @@ import * as Effect from "effect/Effect";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Metric from "effect/Metric";
+import * as Stream from "effect/Stream";
 import * as TestClock from "effect/testing/TestClock";
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import { expect, it } from "vite-plus/test";
@@ -805,6 +806,8 @@ it("requestSecret returns the unsaved note for a declined, cancelled or timed-ou
             getThreadRecords: () => Effect.sync(projection),
             getThreadShell: () =>
               Effect.sync(() => projection().thread as unknown as OrchestrationV2ThreadShell),
+            getThreadEventSequence: () => Effect.succeed(0),
+            streamStoredEventsFrom: () => Stream.never,
             dispatch: (command) => Effect.sync(() => void dispatched.push(command) as never),
           } satisfies Partial<ThreadManagementService.ThreadManagementService["Service"]>),
           ServerSettings.layerTest(),

apps/server/src/mcp/OrchestratorMcpService.test.ts

@@ -21,13 +21,16 @@ import {
   ProviderThreadId,
   ProviderTurnId,
   RunId,
+  SECRET_REQUEST_UNSAVED_NOTE,
+  SecretRef,
   ThreadId,
   type OrchestrationV2ServerCommand,
   type OrchestrationV2StoredEvent,
   type OrchestrationV2ThreadProjection,
   type ServerProvider,
 } from "@t3tools/contracts";
 import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
+import * as Cause from "effect/Cause";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
@@ -37,6 +40,7 @@ import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
 
 import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
 import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
@@ -1387,6 +1391,122 @@ describe("OrchestratorMcpService provider resolution", () => {
     }),
   );
 
+  it.effect.each([
+    "saved",
+    "declined",
+    "the run ending",
+    "an answer before the wait",
+    "an answer during the first read",
+    "no answer",
+  ] as const)("a secret request returns on %s", (outcome) =>
+    Effect.gen(function* () {
+      const events = yield* Queue.unbounded<OrchestrationV2StoredEvent, Cause.Done>();
+      const subscribed = yield* Deferred.make<void>();
+      const secretRef = SecretRef.make("secret-ref:mcp-secret");
+      const eventType = outcome === "the run ending" ? "run.updated" : "turn-item.updated";
+      let runStatus = "running";
+      let card: { readonly id: string; secretStatus: string } | undefined;
+      let sequence = 0;
+      const commit = (type: string, payload: unknown) =>
+        Queue.offer(events, {
+          sequence: ++sequence,
+          event: { type, threadId: parentThreadId, payload },
+        } as unknown as OrchestrationV2StoredEvent);
+      const dependencies = Layer.mergeAll(
+        NodeServices.layer,
+        NodeCrypto.layer,
+        Layer.mock(ThreadManagementService.ThreadManagementService)({
+          getThreadRecords: () =>
+            Effect.gen(function* () {
+              const projection = parentProjection([]);
+              const records = {
+                ...projection,
+                runs: projection.runs.map((run) => ({ ...run, status: runStatus })),
+                turnItems: card === undefined ? [] : [{ ...card, type: "secret_request" }],
+              } as unknown as OrchestrationV2ThreadProjection;
+              // The answer commits after this read's snapshot, before the wait subscribes.
+              if (
+                outcome === "an answer during the first read" &&
+                card?.secretStatus === "pending"
+              ) {
+                card = { id: card.id, secretStatus: "saved" };
+                yield* commit("turn-item.updated", card);
+              }
+              return records;
+            }),
+          getThreadEventSequence: () => Effect.sync(() => sequence),
+          streamStoredEventsFrom: (input) =>
+            input?.eventType === eventType
+              ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe(
+                  Stream.drain,
+                  Stream.concat(Stream.fromQueue(events)),
+                  Stream.filter((stored) => stored.sequence > (input.afterSequence ?? 0)),
+                )
+              : Stream.empty,
+          dispatch: (command) =>
+            Effect.sync(() => {
+              if (command.type === "secret_request.record") {
+                card = {
+                  id: command.turnItemId,
+                  secretStatus:
+                    outcome === "an answer before the wait" ? "saved" : command.secretStatus,
+                };
+              }
+              return { sequence: ++sequence, storedEvents: [] } as never;
+            }),
+        }),
+        ServerSettings.layerTest(),
+        providerRegistryLayer([]),
+        adapterRegistryLayer([codexInstanceId]),
+        Layer.mock(ProjectService.ProjectService)({}),
+        Layer.mock(SecretRequests.SecretRequests)({
+          savedRef: () => Effect.succeedSome(secretRef),
+        }),
+        Layer.mock(RemoteDelegation.RemoteDelegation)({}),
+        Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
+      );
+      yield* Effect.gen(function* () {
+        const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+        const waiting = yield* service
+          .requestSecret(scope, { label: "API key", reason: "Deploy." })
+          .pipe(Effect.forkChild);
+        if (outcome !== "an answer before the wait") yield* Deferred.await(subscribed);
+        if (outcome === "saved" || outcome === "declined") {
+          card = { id: card!.id, secretStatus: outcome };
+          yield* commit("turn-item.updated", card);
+        } else if (outcome === "the run ending") {
+          runStatus = "interrupted";
+          yield* commit("run.updated", { id: parentRunId });
+        } else if (outcome === "an answer during the first read") {
+          // Nothing else arrives, so a wait that skipped the answer ends without one.
+          yield* Queue.end(events);
+        } else if (outcome === "no answer") {
+          yield* TestClock.adjust("10 minutes");
+        }
+        const result = yield* Fiber.join(waiting);
+        const status = (
+          {
+            saved: "saved",
+            declined: "declined",
+            "the run ending": "cancelled",
+            "an answer before the wait": "saved",
+            "an answer during the first read": "saved",
+            "no answer": "timed_out",
+          } as const
+        )[outcome];
+        assert.deepEqual(
+          result,
+          status === "saved"
+            ? { status, secretRef }
+            : { status, note: SECRET_REQUEST_UNSAVED_NOTE[status] },
+        );
+        if (outcome === "the run ending" || outcome === "no answer") {
+          assert.equal(card?.secretStatus, "cancelled");
+        }
+      }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(dependencies))));
+    }),
+  );
+
   it.effect(
     "advertises orchestration capability from registered adapters rather than a driver allowlist",
     () =>

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -82,6 +82,7 @@ import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
+import * as Predicate from "effect/Predicate";
 import * as Ref from "effect/Ref";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
@@ -138,7 +139,6 @@ const TASK_WAKE_EVENTS = [
   { thread: "child", eventType: "provider-thread.updated" },
   { thread: "child", eventType: "runtime-request.updated" },
 ] as const;
-const SECRET_REQUEST_POLL_INTERVAL_MS = 500;
 const DEFAULT_THREAD_LIST_LIMIT = 50;
 const DEFAULT_THREAD_READ_LIMIT = 50;
 const DEFAULT_THREAD_RUN_LIMIT = 10;
@@ -2233,41 +2233,68 @@ const make = Effect.gen(function* () {
         ),
       );
 
+      const readError = (error: unknown) =>
+        failure("orchestration_error", `Unable to read the secret request: ${errorMessage(error)}`);
+      const watchError = (error: unknown) =>
+        failure(
+          "orchestration_error",
+          `Unable to watch the secret request: ${errorMessage(error)}`,
+        );
+      const readAnswer = Effect.gen(function* () {
+        const projection = yield* threadManagement
+          .getThreadRecords(threadId, ["runs", "turnItems"], {
+            turnItemTypes: ["secret_request"],
+            messageRoles: [],
+          })
+          .pipe(Effect.mapError(readError));
+        const item = projection.turnItems.find((candidate) => candidate.id === turnItemId);
+        if (item?.type === "secret_request" && item.secretStatus !== "pending") {
+          return item.secretStatus;
+        }
+        const current = projection.runs.find((candidate) => candidate.id === runId);
+        if (current === undefined || ThreadManagementService.isTerminalRunStatus(current.status)) {
+          yield* record("cancelled");
+          return "cancelled" as const;
+        }
+        return undefined;
+      });
+
       const answered = yield* Effect.gen(function* () {
-        while (true) {
-          const projection = yield* threadManagement
-            .getThreadRecords(threadId, ["runs", "turnItems"], {
-              turnItemTypes: ["secret_request"],
-              messageRoles: [],
-            })
+        // Taken after the pending record: earlier answers show in the first read, later ones replay.
+        const afterSequence = yield* threadManagement
+          .getThreadEventSequence(threadId)
+          .pipe(Effect.mapError(watchError));
+        const initial = yield* readAnswer;
+        if (initial !== undefined) return Option.some(initial);
+        return yield* Stream.merge(
+          threadManagement
+            .streamStoredEventsFrom({ threadId, afterSequence, eventType: "turn-item.updated" })
             .pipe(
-              Effect.mapError((error) =>
-                failure(
-                  "orchestration_error",
-                  `Unable to read the secret request: ${errorMessage(error)}`,
-                ),
+              Stream.filter(
+                ({ event }) =>
+                  event.type === "turn-item.updated" && event.payload.id === turnItemId,
               ),
-            );
-          const item = projection.turnItems.find((candidate) => candidate.id === turnItemId);
-          if (item?.type === "secret_request" && item.secretStatus !== "pending") {
-            return item.secretStatus;
-          }
-          const current = projection.runs.find((candidate) => candidate.id === runId);
-          if (
-            current === undefined ||
-            ThreadManagementService.isTerminalRunStatus(current.status)
-          ) {
-            yield* record("cancelled");
-            return "cancelled" as const;
-          }
-          yield* Effect.sleep(Duration.millis(SECRET_REQUEST_POLL_INTERVAL_MS));
-        }
+            ),
+          threadManagement
+            .streamStoredEventsFrom({ threadId, afterSequence, eventType: "run.updated" })
+            .pipe(
+              Stream.filter(
+                ({ event }) => event.type === "run.updated" && event.payload.id === runId,
+              ),
+            ),
+        ).pipe(
+          Stream.mapError(watchError),
+          Stream.mapEffect(() => readAnswer),
+          Stream.filter(Predicate.isNotUndefined),
+          Stream.runHead,
+        );
       }).pipe(
         Effect.timeoutOption(
           Duration.millis(
             Math.min(input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS, MAX_WAIT_TIMEOUT_MS),
           ),
         ),
+        Effect.map(Option.flatten),
         Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : closeCard)),
       );
       if (Option.isNone(answered)) yield* closeCard;