From 710c60d03ae05a2bb48acfa5a6664846c748ace5 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 17:02:46 +0000 Subject: [PATCH 1/4] perf(server): request_secret waits on thread events instead of polling request_secret re-read the thread's runs and secret cards every 500 ms until the user answered, so a card left open for its 10 minute default cost about 1,200 projection reads. It now reads once after recording the card, then wakes only on stored turn-item.updated events for its card and run.updated events for its run, the same way task waits already do. Saved, declined, run-ended, timed-out and interrupted outcomes are unchanged. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 95 +++++++++++++++++++ apps/server/src/mcp/OrchestratorMcpService.ts | 76 +++++++++------ 2 files changed, 144 insertions(+), 27 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 47a6bda166ce1..546552c97eb71 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -18,6 +18,7 @@ import { ProviderThreadId, ProviderTurnId, RunId, + SecretRef, ThreadId, type OrchestrationV2ServerCommand, type OrchestrationV2StoredEvent, @@ -34,6 +35,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"; @@ -1280,6 +1282,99 @@ describe("OrchestratorMcpService provider resolution", () => { }), ); + it.effect.each([ + "saved", + "declined", + "the run ending", + "an answer before the wait", + "no answer", + ] as const)("a secret request returns on %s", (outcome) => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const subscribed = yield* Deferred.make(); + 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; + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: () => + Effect.sync(() => { + const projection = parentProjection([]); + return { + ...projection, + runs: projection.runs.map((run) => ({ ...run, status: runStatus })), + turnItems: card === undefined ? [] : [{ ...card, type: "secret_request" }], + } as unknown as OrchestrationV2ThreadProjection; + }), + getThreadEventSequence: () => Effect.succeed(0), + streamStoredEventsFrom: (input) => + input?.eventType === eventType + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + 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: 1, 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* Queue.offer(events, { + sequence: 2, + event: { type: "turn-item.updated", threadId: parentThreadId, payload: card }, + } as unknown as OrchestrationV2StoredEvent); + } else if (outcome === "the run ending") { + runStatus = "interrupted"; + yield* Queue.offer(events, { + sequence: 2, + event: { type: "run.updated", threadId: parentThreadId, payload: { id: parentRunId } }, + } as unknown as OrchestrationV2StoredEvent); + } else if (outcome === "no answer") { + yield* TestClock.adjust("10 minutes"); + } + const result = yield* Fiber.join(waiting); + assert.deepEqual( + result, + outcome === "saved" || outcome === "an answer before the wait" + ? { status: "saved", secretRef } + : outcome === "declined" + ? { status: "declined" } + : { status: outcome === "no answer" ? "timed_out" : "cancelled" }, + ); + 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", () => diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index b124a03ea2c40..d2d2d21c51560 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -131,7 +131,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; @@ -2151,41 +2150,64 @@ const make = Effect.gen(function* () { ), ); + const readError = (error: unknown) => + failure("orchestration_error", `Unable to read 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 Option.some(item.secretStatus); + } + const current = projection.runs.find((candidate) => candidate.id === runId); + if (current === undefined || ThreadManagementService.isTerminalRunStatus(current.status)) { + yield* record("cancelled"); + return Option.some("cancelled" as const); + } + return Option.none(); + }); + 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(readError)); + const initial = yield* readAnswer; + if (Option.isSome(initial)) return initial; + const next = 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(readError), + Stream.mapEffect(() => readAnswer), + Stream.filter(Option.isSome), + Stream.runHead, + ); + return Option.flatten(next); }).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; From 3c72824a60e57aadda106560925aa87edf86aa21 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 17:12:06 +0000 Subject: [PATCH 2/4] refactor(server): name secret request watch errors and drop nested options Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 23 +++++++++++-------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index d2d2d21c51560..570655e1add07 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -79,6 +79,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"; @@ -2152,6 +2153,11 @@ 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"], { @@ -2161,24 +2167,24 @@ const make = Effect.gen(function* () { .pipe(Effect.mapError(readError)); const item = projection.turnItems.find((candidate) => candidate.id === turnItemId); if (item?.type === "secret_request" && item.secretStatus !== "pending") { - return Option.some(item.secretStatus); + return item.secretStatus; } const current = projection.runs.find((candidate) => candidate.id === runId); if (current === undefined || ThreadManagementService.isTerminalRunStatus(current.status)) { yield* record("cancelled"); - return Option.some("cancelled" as const); + return "cancelled" as const; } - return Option.none(); + return undefined; }); const answered = yield* Effect.gen(function* () { // Taken after the pending record: earlier answers show in the first read, later ones replay. const afterSequence = yield* threadManagement .getThreadEventSequence(threadId) - .pipe(Effect.mapError(readError)); + .pipe(Effect.mapError(watchError)); const initial = yield* readAnswer; - if (Option.isSome(initial)) return initial; - const next = yield* Stream.merge( + if (initial !== undefined) return Option.some(initial); + return yield* Stream.merge( threadManagement .streamStoredEventsFrom({ threadId, afterSequence, eventType: "turn-item.updated" }) .pipe( @@ -2195,12 +2201,11 @@ const make = Effect.gen(function* () { ), ), ).pipe( - Stream.mapError(readError), + Stream.mapError(watchError), Stream.mapEffect(() => readAnswer), - Stream.filter(Option.isSome), + Stream.filter(Predicate.isNotUndefined), Stream.runHead, ); - return Option.flatten(next); }).pipe( Effect.timeoutOption( Duration.millis( From a8a15a024850b6abe6226cad9043d7e76591b0e7 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 17:43:30 +0000 Subject: [PATCH 3/4] test(server): cover a secret answer that lands between the first read and the wait The secret request mock now numbers its events and replays only those after the wait's cursor. A new row saves the answer right after the first read and before the wait subscribes. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 47 +++++++++++++------ 1 file changed, 32 insertions(+), 15 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 546552c97eb71..32f089f56c340 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -26,6 +26,7 @@ import { 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"; @@ -1287,35 +1288,52 @@ describe("OrchestratorMcpService provider resolution", () => { "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(); + const events = yield* Queue.unbounded(); const subscribed = yield* Deferred.make(); 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, Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadRecords: () => - Effect.sync(() => { + Effect.gen(function* () { const projection = parentProjection([]); - return { + 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.succeed(0), + 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.never, + : Stream.empty, dispatch: (command) => Effect.sync(() => { if (command.type === "secret_request.record") { @@ -1325,7 +1343,7 @@ describe("OrchestratorMcpService provider resolution", () => { outcome === "an answer before the wait" ? "saved" : command.secretStatus, }; } - return { sequence: 1, storedEvents: [] } as never; + return { sequence: ++sequence, storedEvents: [] } as never; }), }), ServerSettings.layerTest(), @@ -1346,23 +1364,22 @@ describe("OrchestratorMcpService provider resolution", () => { if (outcome !== "an answer before the wait") yield* Deferred.await(subscribed); if (outcome === "saved" || outcome === "declined") { card = { id: card!.id, secretStatus: outcome }; - yield* Queue.offer(events, { - sequence: 2, - event: { type: "turn-item.updated", threadId: parentThreadId, payload: card }, - } as unknown as OrchestrationV2StoredEvent); + yield* commit("turn-item.updated", card); } else if (outcome === "the run ending") { runStatus = "interrupted"; - yield* Queue.offer(events, { - sequence: 2, - event: { type: "run.updated", threadId: parentThreadId, payload: { id: parentRunId } }, - } as unknown as OrchestrationV2StoredEvent); + 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); assert.deepEqual( result, - outcome === "saved" || outcome === "an answer before the wait" + outcome === "saved" || + outcome === "an answer before the wait" || + outcome === "an answer during the first read" ? { status: "saved", secretRef } : outcome === "declined" ? { status: "declined" } From b68ea2ff3de53821a32d30eafb0e7d614061f679 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 18:09:17 +0000 Subject: [PATCH 4/4] test(server): update request_secret test mocks for main's crypto layer and unsaved notes The secret request table now provides NodeCrypto.layer and expects the unsaved note main added to declined, cancelled and timed-out results. The activity test's ThreadManagementService mock answers getThreadEventSequence and streamStoredEventsFrom, which the event-driven wait calls. Co-Authored-By: Claude Opus 5.5 --- .../OrchestratorMcpService.activity.test.ts | 3 +++ .../src/mcp/OrchestratorMcpService.test.ts | 22 +++++++++++++------ 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index c35b605dca4fb..6ee55cf7d8019 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/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), ServerSettings.layerTest(), diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index d9547a881e128..0302be2f157a1 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -21,6 +21,7 @@ import { ProviderThreadId, ProviderTurnId, RunId, + SECRET_REQUEST_UNSAVED_NOTE, SecretRef, ThreadId, type OrchestrationV2ServerCommand, @@ -1413,6 +1414,7 @@ describe("OrchestratorMcpService provider resolution", () => { } as unknown as OrchestrationV2StoredEvent); const dependencies = Layer.mergeAll( NodeServices.layer, + NodeCrypto.layer, Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadRecords: () => Effect.gen(function* () { @@ -1482,15 +1484,21 @@ describe("OrchestratorMcpService provider resolution", () => { 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, - outcome === "saved" || - outcome === "an answer before the wait" || - outcome === "an answer during the first read" - ? { status: "saved", secretRef } - : outcome === "declined" - ? { status: "declined" } - : { status: outcome === "no answer" ? "timed_out" : "cancelled" }, + status === "saved" + ? { status, secretRef } + : { status, note: SECRET_REQUEST_UNSAVED_NOTE[status] }, ); if (outcome === "the run ending" || outcome === "no answer") { assert.equal(card?.secretStatus, "cancelled");