From 1c67719e0c4fee83cd9115e884c24c4c82d8d446 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Fri, 30 Oct 2026 22:41:13 +0100 Subject: [PATCH 1/2] perf(server): skip unused parent history in linked task status --- .../src/mcp/OrchestratorMcpService.test.ts | 68 ++++++++++++++----- apps/server/src/mcp/OrchestratorMcpService.ts | 10 ++- apps/server/src/peer/RemoteDelegation.test.ts | 37 ++++++++++ 3 files changed, 92 insertions(+), 23 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 6450f9acac2cb..70601d63ea084 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -2,6 +2,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; import { CommandId, + ContextTransferId, EnvironmentId, MessageId, NodeId, @@ -78,12 +79,21 @@ describe("OrchestratorMcpService", () => { const childThreadId = ThreadId.make("thread:mcp-ack-child"); const childRunId = RunId.make("run:mcp-ack-child"); const taskId = NodeId.make("node:mcp-ack-task"); + const resultTransferId = ContextTransferId.make("transfer:mcp-ack-result"); const acknowledgementCommandIds = yield* Ref.make>([]); const acknowledgementAttempts = yield* Ref.make(0); const parentProjection = { thread: { id: parentThreadId }, runs: [], - contextTransfers: [], + contextTransfers: [ + { + id: resultTransferId, + type: "subagent_result", + sourceThreadId: childThreadId, + targetThreadId: parentThreadId, + sourcePoint: { runId: childRunId }, + }, + ], subagents: [ { id: taskId, @@ -116,37 +126,56 @@ describe("OrchestratorMcpService", () => { }, ], contextTransfers: [], - messages: [], + messages: [ + { + id: MessageId.make("message:mcp-ack-result"), + runId: childRunId, + role: "assistant", + text: "terminal result", + createdAt: DateTime.makeUnsafe("2026-10-03T10:10:00Z"), + updatedAt: DateTime.makeUnsafe("2026-10-03T10:10:00Z"), + }, + ], + turnItems: [], subagents: [], providerThreads: [], runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; let hasNestedWork = true; + let settleAfterTaskRead = false; const layerDependencies = Layer.mergeAll( NodeServices.layer, Layer.mock(ThreadManagementService.ThreadManagementService)({ getThreadRecords: (threadId) => - Effect.succeed( - threadId === parentThreadId - ? hasNestedWork - ? { - ...parentProjection, - subagents: parentProjection.subagents.map((task) => ({ - ...task, - result: null, - status: "running" as const, - })), - } - : parentProjection - : hasNestedWork + Effect.sync(() => { + if (threadId !== parentThreadId) { + return hasNestedWork ? { ...childProjection, subagents: [ { ...parentProjection.subagents[0]!, status: "running" as const }, ], } - : childProjection, - ), + : childProjection; + } + const snapshot = hasNestedWork + ? { + ...parentProjection, + contextTransfers: [], + subagents: parentProjection.subagents.map((task) => ({ + ...task, + result: null, + status: "running" as const, + })), + } + : parentProjection; + if (settleAfterTaskRead) { + settleAfterTaskRead = false; + hasNestedWork = false; + } + return snapshot; + }), + delegatedTaskResultPending: () => Effect.succeed(false), dispatch: (command) => Ref.update(acknowledgementCommandIds, (commandIds) => [ ...commandIds, @@ -189,16 +218,21 @@ describe("OrchestratorMcpService", () => { assert.equal(pending.status, "running"); assert.equal(pending.workState, "waiting_for_children"); assert.isNull(pending.summary); + assert.isNull(pending.resultContextTransferId); assert.equal(yield* Ref.get(acknowledgementAttempts), 0); hasNestedWork = false; const error = yield* service.taskStatus(scope, taskId).pipe(Effect.flip); assert.equal(error.code, "orchestration_error"); + // Completion lands after task selection, before the remaining parent/child reads. + hasNestedWork = true; + settleAfterTaskRead = true; const result = yield* service.taskStatus(scope, taskId); assert.equal(result.status, "completed"); assert.equal(result.summary, "terminal result"); assert.equal(result.latestTerminalRunId, childRunId); assert.equal(result.latestTerminalStatus, "completed"); + assert.equal(result.resultContextTransferId, resultTransferId); const commandIds = yield* Ref.get(acknowledgementCommandIds); assert.equal(commandIds.length, 2); assert.notEqual(commandIds[0], commandIds[1]); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 02f63741728b9..c52e0e8e81756 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1059,11 +1059,9 @@ const make = Effect.gen(function* () { // Task waits read on every wake of a parent that may hold thousands of // tasks, so they read only the task they watch and what its status needs. - const loadTaskParent = (threadId: ThreadId, taskId: NodeId) => + const loadTaskParent = (threadId: ThreadId) => threadManagement - .getThreadRecords(threadId, ["runs", "subagents", "contextTransfers"], { - subagentIds: [taskId], - }) + .getThreadRecords(threadId, ["runs", "contextTransfers"]) .pipe( Effect.mapError((error) => failure( @@ -1424,8 +1422,7 @@ const make = Effect.gen(function* () { acknowledgementOperation = "task-status-acknowledge", ): Effect.fn.Return { yield* requireCapability(scope); - const parentProjection = yield* loadTaskParent(scope.thread.threadId, taskId); - const task = parentProjection.subagents.find( + const task = (yield* loadTasks(scope.thread.threadId, [taskId])).find( (candidate) => candidate.id === taskId && candidate.origin === "app_owned" && @@ -1440,6 +1437,7 @@ const make = Effect.gen(function* () { if (task.childThreadId === null) { return yield* readRemoteTask(scope, task, waitTimedOut, acknowledgeTerminal); } + const parentProjection = yield* loadTaskParent(scope.thread.threadId); const childControls = yield* threadManagement .getThreadRecords( task.childThreadId, diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 7672251288d8c..47c8574c7dfae 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -30,6 +30,7 @@ import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as Tracer from "effect/Tracer"; import type * as SqlClient from "effect/sql/SqlClient"; import { McpSchema, McpServer } from "effect/ai"; @@ -826,6 +827,42 @@ it.effect("a task delegated to the box wakes the laptop's parent when it ends th expect(retried.taskId).toBe(delegated.taskId); expect((yield* a.orchestrator.getThreadProjection(parentThreadId)).subagents).toHaveLength(1); + const historyReads: 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" && + /\bSELECT\s+payload_json\s+FROM orchestration_v2_projection_(?:runs|context_transfers)\b/i.test( + query, + ) + ) { + historyReads.push(query); + } + }; + return span; + }, + }); + const running = yield* a + .call("task_status", { taskId: delegated.taskId }) + .pipe(Effect.withTracer(tracer)); + expect(running.structuredContent).toMatchObject({ + taskId: delegated.taskId, + childThreadId: null, + remoteChild: delegated.remoteChild, + status: "running", + workState: "working", + summary: null, + hasPendingQuestion: false, + pendingApproval: null, + }); + // A running task there needs its local row, not the parent's run or transfer history. + expect(historyReads).toEqual([]); + yield* Ref.set(thread.reply, "All 412 tests pass on the box."); yield* Ref.set(thread.status, "completed"); yield* Deferred.succeed(thread.finished, undefined); From b5ba9c666bc96f03d9794447808d9152df59184d Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 2 Nov 2026 01:25:20 +0000 Subject: [PATCH 2/2] fix(server): reconcile tracer bindings after main merge --- apps/server/src/peer/RemoteDelegation.test.ts | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 5e029667320b5..7ccb7727960c0 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -29,7 +29,6 @@ import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; -import * as Tracer from "effect/Tracer"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as Tracer from "effect/Tracer"; import type * as SqlClient from "effect/sql/SqlClient"; @@ -831,7 +830,7 @@ it.effect("a task delegated to the box wakes the laptop's parent when it ends th expect((yield* a.orchestrator.getThreadProjection(parentThreadId)).subagents).toHaveLength(1); const historyReads: Array = []; - const tracer = Tracer.make({ + const historyTracer = Tracer.make({ span(options) { const span = new Tracer.NativeSpan(options); const end = span.end.bind(span); @@ -852,7 +851,7 @@ it.effect("a task delegated to the box wakes the laptop's parent when it ends th }); const running = yield* a .call("task_status", { taskId: delegated.taskId }) - .pipe(Effect.withTracer(tracer)); + .pipe(Effect.withTracer(historyTracer)); expect(running.structuredContent).toMatchObject({ taskId: delegated.taskId, childThreadId: null,