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 f52ac95174277..0302be2f157a1 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -21,6 +21,8 @@ import { ProviderThreadId, ProviderTurnId, RunId, + SECRET_REQUEST_UNSAVED_NOTE, + SecretRef, ThreadId, type OrchestrationV2ServerCommand, type OrchestrationV2StoredEvent, @@ -28,6 +30,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"; @@ -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(); + 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, + 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", () => diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 9d15408483a3e..d96ae2a686ef9 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/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;