From 06a909da2a1bbfdf44b5f0e35e42f17c32cfc78d Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 29 Oct 2026 18:04:51 +0100 Subject: [PATCH 1/3] perf(server): skip unused transfers in parentless thread reads --- apps/server/src/mcp/OrchestratorMcpService.ts | 20 +++++++++++------ ...OrchestratorMcpToolkit.integration.test.ts | 22 ++++++++++++++----- 2 files changed, 30 insertions(+), 12 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index d8dacb35a0617..d2f33dc69ca8c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1201,13 +1201,19 @@ const make = Effect.gen(function* () { `Thread ${threadId} is no longer available.`, ); } - const target = yield* threadManagement - .getProjectThreadRecords({ projectId: shell.projectId, threadId }, [ - "runs", - "runtimeRequests", - "contextTransfers", - ]) - .pipe(Effect.mapError(threadManagementFailure)); + const input = { projectId: shell.projectId, threadId }; + // Transfers are needed only to acknowledge a caller's child-task result. + const records = + parent === undefined + ? threadManagement + .getProjectThreadRecords(input, ["runs", "runtimeRequests"]) + .pipe(Effect.map((target) => ({ ...target, contextTransfers: [] }))) + : threadManagement.getProjectThreadRecords(input, [ + "runs", + "runtimeRequests", + "contextTransfers", + ]); + const target = yield* records.pipe(Effect.mapError(threadManagementFailure)); return { parent, target, shell } as const; }); diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index ec8c9434bec43..b51e20a418f39 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -1,3 +1,4 @@ +import * as NodeURL from "node:url"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import { @@ -96,9 +97,7 @@ const readDelegatedTaskStatusTranscript = Effect.fn("readDelegatedTaskStatusTran followupResult: string, ) { const fs = yield* FileSystem.FileSystem; - const text = yield* fs.readFileString( - decodeURIComponent(delegatedTaskStatusTranscriptFile.pathname), - ); + const text = yield* fs.readFileString(NodeURL.fileURLToPath(delegatedTaskStatusTranscriptFile)); const transcript = yield* decodeProviderReplayNdjson( text.replaceAll(queuedFollowupResult, followupResult), ); @@ -315,11 +314,11 @@ describe("orchestrator MCP toolkit", () => { capabilities: new Set(["orchestration"]), issuedAt: 1, }; - const invoke = (name: string, args: Record) => + const invoke = (name: string, args: Record, scope = invocation) => server .callTool({ name, arguments: args }) .pipe( - Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), Effect.provideService(McpSchema.McpServerClient, client), ); @@ -596,11 +595,24 @@ describe("orchestrator MCP toolkit", () => { expect(finalStatus.latestTerminalResultContextTransferId).not.toBe( delegated.resultContextTransferId, ); + const parentlessRead = yield* invoke( + "t3_thread_read", + { threadId: delegated.childThreadId, limit: 100 }, + { ...invocation, thread: undefined }, + ); + expect(parentlessRead.isError).not.toBe(true); + const afterParentlessRead = yield* orchestrator.getThreadProjection(parentThreadId); + expect( + afterParentlessRead.subagents.find((task) => task.id === queuedFollowup.taskId) + ?.completionDelivery?.state, + ).toBe("claimed"); + const read = yield* invoke("t3_thread_read", { threadId: delegated.childThreadId, limit: 100, }); expect(read.isError).not.toBe(true); + expect(read.structuredContent).toEqual(parentlessRead.structuredContent); const afterRead = yield* orchestrator.getThreadProjection(parentThreadId); expect( afterRead.subagents.find((task) => task.id === queuedFollowup.taskId) From bdac9ed20c6366b15fdce51a7a2e8ea70937b7cc Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 29 Oct 2026 22:38:01 +0100 Subject: [PATCH 2/3] perf(server): skip unused caller prompts during thread reads --- .../OrchestratorMcpService.activity.test.ts | 40 ++++++++++++++----- apps/server/src/mcp/OrchestratorMcpService.ts | 10 +++-- 2 files changed, 37 insertions(+), 13 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index 45561e7670411..b290ad124c0f2 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -25,6 +25,7 @@ import * as SecretRequests from "../secrets/SecretRequests.ts"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as RemoteDelegation from "../peer/RemoteDelegation.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import type * as McpInvocationContext from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; @@ -116,7 +117,8 @@ function makeRun(input: { }; } -it("readThread prefers activity-run status over a newer cancelled queued run", async () => { +it("readThread preserves caller access errors and prefers activity-run status over a newer cancelled queued run", async () => { + let callerUnavailable = false; const projection = { thread: baseThread({ threadId: parentThreadId, @@ -142,15 +144,24 @@ it("readThread prefers activity-run status over a newer cancelled queued run", a Layer.mock(ThreadManagementService.ThreadManagementService)({ getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }), getThreadRecords: (threadId) => - threadId === parentThreadId - ? Effect.succeed(projection) - : Effect.die(`unexpected thread ${threadId}`), + callerUnavailable + ? Effect.fail( + new Orchestrator.OrchestratorProjectionError({ + threadId, + cause: "Caller unavailable", + }), + ) + : threadId === parentThreadId + ? Effect.succeed(projection) + : Effect.die(`unexpected thread ${threadId}`), getThreadShell: (threadId) => - Effect.succeed( - threadId === parentThreadId - ? (projection.thread as unknown as OrchestrationV2ThreadShell) - : null, - ), + callerUnavailable + ? Effect.die("A failed caller read must not reach the target") + : Effect.succeed( + threadId === parentThreadId + ? (projection.thread as unknown as OrchestrationV2ThreadShell) + : null, + ), getProjectThreadRecords: (input) => input.threadId === parentThreadId ? Effect.succeed(projection) @@ -180,6 +191,17 @@ it("readThread prefers activity-run status over a newer cancelled queued run", a expect(result.thread.status).toBe("running"); expect(result.thread.latestRunId).toBe(cancelledRunId); expect(result.thread.activeRunId).toBe(activeRunId); + + callerUnavailable = true; + const denied = yield* service + .readThread({ ...makeScope(), capabilities: new Set() }, { threadId: parentThreadId }) + .pipe(Effect.flip); + expect(denied.code).toBe("capability_denied"); + const missingCaller = yield* service + .readThread(makeScope(), { threadId: parentThreadId }) + .pipe(Effect.flip); + expect(missingCaller.code).toBe("orchestration_error"); + expect(missingCaller.message).toContain(`Unable to read thread ${parentThreadId}:`); }).pipe(Effect.provide(layer), Effect.runPromise); }); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index d2f33dc69ca8c..5b85443773388 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1009,7 +1009,7 @@ const make = Effect.gen(function* () { ), ); - const loadProjection = (threadId: ThreadId) => + const loadProjection = (threadId: ThreadId, includeUserMessages = true) => threadManagement .getThreadRecords( threadId, @@ -1023,7 +1023,7 @@ const make = Effect.gen(function* () { "contextTransfers", "turnItems", ], - { turnItemTypes: [], messageRoles: ["user"] }, + { turnItemTypes: [], messageRoles: includeUserMessages ? ["user"] : [] }, ) .pipe( Effect.mapError((error) => @@ -1073,6 +1073,7 @@ const make = Effect.gen(function* () { const loadCaller = Effect.fn("OrchestratorMcpService.loadCaller")(function* ( scope: McpInvocationScope, + includeUserMessages = true, ) { yield* requireCapability(scope); if (scope.thread === undefined) { @@ -1081,7 +1082,7 @@ const make = Effect.gen(function* () { limits: clientModeCeiling(scope.client), } as const; } - const parent = yield* loadProjection(scope.thread.threadId); + const parent = yield* loadProjection(scope.thread.threadId, includeUserMessages); return { parent, limits: { @@ -1190,7 +1191,8 @@ const make = Effect.gen(function* () { scope: McpInvocationScope, threadId: ThreadId, ) { - const { parent } = yield* loadCaller(scope); + // Caller prompts are not used to read a timeline or acknowledge a child result. + const { parent } = yield* loadCaller(scope, false); const shell = yield* threadManagement .getThreadShell(threadId) .pipe(Effect.mapError(threadManagementFailure)); From 2d96946a29d45a6545138e2a673cbb8f82abef85 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 29 Oct 2026 22:50:08 +0100 Subject: [PATCH 3/3] test(server): guard parentless reads against unused transfer queries --- ...OrchestratorMcpToolkit.integration.test.ts | 25 +++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index b51e20a418f39..e53a4d85583e4 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -23,6 +23,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as Tracer from "effect/Tracer"; import { McpSchema, McpServer } from "effect/ai"; import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts"; @@ -595,12 +596,31 @@ describe("orchestrator MCP toolkit", () => { expect(finalStatus.latestTerminalResultContextTransferId).not.toBe( delegated.resultContextTransferId, ); + const transferReads: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const query = span.attributes.get("db.query.text"); + if ( + typeof query === "string" && + /SELECT[\s\S]*FROM orchestration_v2_projection_context_transfers/i.test(query) + ) { + transferReads.push(query); + } + }; + return span; + }, + }); const parentlessRead = yield* invoke( "t3_thread_read", { threadId: delegated.childThreadId, limit: 100 }, { ...invocation, thread: undefined }, - ); + ).pipe(Effect.withTracer(tracer)); expect(parentlessRead.isError).not.toBe(true); + expect(transferReads).toHaveLength(0); const afterParentlessRead = yield* orchestrator.getThreadProjection(parentThreadId); expect( afterParentlessRead.subagents.find((task) => task.id === queuedFollowup.taskId) @@ -610,8 +630,9 @@ describe("orchestrator MCP toolkit", () => { const read = yield* invoke("t3_thread_read", { threadId: delegated.childThreadId, limit: 100, - }); + }).pipe(Effect.withTracer(tracer)); expect(read.isError).not.toBe(true); + expect(transferReads.length).toBeGreaterThan(0); expect(read.structuredContent).toEqual(parentlessRead.structuredContent); const afterRead = yield* orchestrator.getThreadProjection(parentThreadId); expect(