From bcb907d2785cc1ea897733a2099ba2ae123dacda Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Sat, 31 Oct 2026 02:10:49 +0100 Subject: [PATCH] perf(server): task waits skip unrelated sibling updates --- .../src/mcp/OrchestratorMcpService.test.ts | 151 ++++++++++++++++++ apps/server/src/mcp/OrchestratorMcpService.ts | 23 ++- 2 files changed, 168 insertions(+), 6 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 6450f9acac2cb..90dac31d66190 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -19,11 +19,13 @@ import { RunId, ThreadId, type OrchestrationV2ServerCommand, + type OrchestrationV2StoredEvent, type OrchestrationV2ThreadProjection, type ServerProvider, } from "@t3tools/contracts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; @@ -1128,6 +1130,155 @@ describe("OrchestratorMcpService provider resolution", () => { turnItems: [], } as unknown as OrchestrationV2ThreadProjection; + it.effect.each([ + "parent completion", + "nested child completion", + "child run completion", + "child question", + "child approval", + ] as const)("a task wait still returns on %s", (wake) => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const subscribed = yield* Deferred.make(); + const nestedId = NodeId.make("node:mcp-wait-nested"); + const childRunId = RunId.make("run:mcp-wait-child"); + const childRootId = NodeId.make("node:mcp-wait-child-root"); + const eventThreadId = wake === "parent completion" ? parentThreadId : childThreadId; + const eventType = + wake === "child run completion" + ? "run.updated" + : wake === "child question" || wake === "child approval" + ? "runtime-request.updated" + : "subagent.updated"; + const terminal = wake.endsWith("completion"); + let ready = false; + const task = () => ({ + id: taskId, + threadId: parentThreadId, + runId: parentRunId, + parentNodeId, + origin: "app_owned", + providerInstanceId: codexInstanceId, + driver: "codex", + childThreadId, + model: "gpt-5.4", + status: ready && terminal ? "completed" : "running", + result: ready && wake === "parent completion" ? "done" : null, + }); + const child = () => + ({ + ...childProjection, + runs: [ + { + id: childRunId, + ordinal: 1, + status: + wake === "nested child completion" || (ready && terminal) ? "completed" : "running", + startedAt: DateTime.makeUnsafe(0), + completedAt: ready && terminal ? DateTime.makeUnsafe(1) : null, + }, + ], + subagents: + wake === "nested child completion" + ? [{ id: nestedId, status: ready ? "completed" : "running" }] + : [], + nodes: [{ id: childRootId, runId: childRunId }], + runtimeRequests: + ready && !terminal + ? [ + { + id: "request:mcp-wait-answer", + nodeId: childRootId, + kind: wake === "child question" ? "user_input" : "command", + status: "pending", + responseCapability: { type: "live" }, + createdAt: DateTime.makeUnsafe(0), + }, + ] + : [], + }) as unknown as OrchestrationV2ThreadProjection; + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + assertProviderInstanceEnabled: () => Effect.void, + isProviderInstanceEnabledForProject: () => Effect.succeed(true), + getThreadRecords: (threadId) => + Effect.sync(() => (threadId === parentThreadId ? parentProjection([task()]) : child())), + getThreadEventSequence: () => Effect.succeed(0), + delegatedTaskResultPending: () => Effect.succeed(false), + streamStoredEventsFrom: (input) => + input?.threadId === eventThreadId && input.eventType === eventType + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + dispatch: () => + Effect.succeed({ + sequence: 1, + storedEvents: [{ event: { type: "subagent.updated", payload: task() } }], + } as never), + }), + ServerSettings.layerTest(), + providerRegistryLayer([ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ]), + adapterRegistryLayer([codexInstanceId]), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const waiting = yield* service + .delegateTask(scope, { task: "Review the change.", mode: "wait" }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + ready = true; + if (wake === "parent completion") { + yield* Queue.offer(events, { + sequence: 2, + event: { + type: "subagent.updated", + threadId: parentThreadId, + payload: { ...task(), id: nestedId }, + }, + } as unknown as OrchestrationV2StoredEvent); + } + yield* Queue.offer(events, { + sequence: 3, + event: { + type: eventType, + threadId: eventThreadId, + payload: { id: wake === "parent completion" ? taskId : nestedId }, + }, + } as unknown as OrchestrationV2StoredEvent); + const result = yield* Fiber.join(waiting); + assert.isFalse(result.waitTimedOut); + if (terminal) { + assert.equal(result.status, "completed"); + if (wake === "parent completion") assert.equal(result.summary, "done"); + } else { + assert.equal(result.status, "running"); + if (wake === "child question") assert.isTrue(result.hasPendingQuestion); + else { + assert.deepEqual(result.pendingApproval, { + requestId: "request:mcp-wait-answer", + kind: "command", + runId: childRunId, + }); + } + } + }).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 0eb2d0e8ea881..2d6fe80c211e6 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1756,12 +1756,23 @@ const make = Effect.gen(function* () { : TASK_WAKE_EVENTS; return yield* Stream.mergeAll( wakeEvents.map(({ thread, eventType }) => - threadManagement.streamStoredEventsFrom({ - threadId: - thread === "parent" || childThreadId === null ? scope.thread.threadId : childThreadId, - afterSequence, - eventType, - }), + threadManagement + .streamStoredEventsFrom({ + threadId: + thread === "parent" || childThreadId === null + ? scope.thread.threadId + : childThreadId, + afterSequence, + eventType, + }) + .pipe( + Stream.filter( + ({ event }) => + thread === "child" || + event.type !== "subagent.updated" || + event.payload.id === taskId, + ), + ), ), { concurrency: "unbounded" }, ).pipe(