From c4ca682e680feaf0675baf680de09dad4af2f76c Mon Sep 17 00:00:00 2001 From: SAPHID Date: Mon, 12 Oct 2026 22:49:19 +0000 Subject: [PATCH] perf(server): stop each delegated child conversation once Deduplicate historical delegated task rows by child thread before dispatching Stop and walking descendants. Co-Authored-By: GPT-6.1 Sol --- .../ThreadManagementService.ts | 6 +- .../src/orchestration-v2/ThreadStop.test.ts | 71 +++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index a3d6f7b852609..328545e341f12 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -432,7 +432,7 @@ export interface ThreadManagementServiceShape { input: OrchestrationV2StopWorkflowInput, ) => Effect.Effect; /** - * Sends `thread.stop` to every delegated task under a thread, depth first. Their command IDs + * Sends `thread.stop` to every distinct delegated child thread, depth first. Their command IDs * derive from `commandId`, so a retry repeats nothing that already stopped. A task that * cannot be stopped does not keep the others running: all are tried, then it fails. */ @@ -1075,9 +1075,13 @@ const make = Effect.gen(function* () { const failures: Array< Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError > = []; + const stoppedChildThreadIds = new Set(); for (const task of subagents) { if (task.origin !== "app_owned" || task.childThreadId === null) continue; const threadId = task.childThreadId; + // Follow-up tasks reuse a child thread; walk its subtree only once. + if (stoppedChildThreadIds.has(threadId)) continue; + stoppedChildThreadIds.add(threadId); yield* dispatch({ type: "thread.stop", commandId: CommandId.make(`${input.commandId}:stop:${threadId}`), diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index c0732ba2a54cf..e608b74bce472 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -16,6 +16,7 @@ import * as DateTime from "effect/DateTime"; 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 Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; @@ -219,6 +220,76 @@ it.effect("Stop ends watches, holds queues, and stops the delegated tasks under }).pipe(Effect.provide(layerTest)), ); +it.effect.each([1, 2, 8])( + "Stop visits each descendant once with %s historical tasks per child", + (tasksPerChild) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2; + const rootThreadId = ThreadId.make("thread:stop-history"); + yield* createWatchingThread(rootThreadId, 30); + yield* send(rootThreadId, "work", "start_immediately"); + const descendants: Array = []; + let parentThreadId = rootThreadId; + for (let depth = 0; depth < 3; depth++) { + const childThreadId = yield* delegate(parentThreadId, `history child ${depth}`); + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + // Published tasks and later follow-ups all retain the same child thread. + for (let history = 1; history < tasksPerChild; history++) { + const taskId = NodeId.make(`node:stop-history:${depth}:${history}`); + yield* projections.apply({ + id: EventId.make(`event:stop-history:${depth}:${history}`), + type: "subagent.updated", + threadId: parentThreadId, + runId: task.runId!, + nodeId: taskId, + occurredAt: yield* DateTime.now, + payload: { ...task, id: taskId, status: "completed" }, + }); + } + descendants.push(childThreadId); + parentThreadId = childThreadId; + } + + const stopped: Array = []; + const read: Array = []; + const measuredOrchestrator = Orchestrator.OrchestratorV2.of({ + ...orchestrator, + dispatch: (command) => + Effect.sync(() => { + if (command.type === "thread.stop") stopped.push(command.threadId); + }).pipe(Effect.andThen(orchestrator.dispatch(command))), + getThreadRecords: (threadId, fields, filter) => + Effect.sync(() => read.push(threadId)).pipe( + Effect.andThen(orchestrator.getThreadRecords(threadId, fields, filter)), + ), + }); + const commandId = CommandId.make("stop-history"); + yield* Effect.gen(function* () { + const threads = yield* ThreadManagementService.ThreadManagementService; + yield* threads.stopDelegatedTasks({ threadId: rootThreadId, commandId }); + }).pipe( + Effect.provide( + Layer.fresh(ThreadManagementService.layer).pipe( + Layer.provide(Layer.succeed(Orchestrator.OrchestratorV2, measuredOrchestrator)), + ), + ), + ); + + // Check real persisted state and receipts before asserting traversal cost. + for (const threadId of descendants) { + assert.deepEqual((yield* threadState(threadId)).runs, ["interrupted"]); + const receipt = yield* receipts.getByCommandId( + CommandId.make(`${commandId}:stop:${threadId}`), + ); + assert.isTrue(Option.isSome(receipt)); + } + assert.deepEqual(stopped, descendants); + assert.deepEqual(read, [rootThreadId, ...descendants]); + }).pipe(Effect.provide(layerTest)), +); + it.effect( "thread.stop ends an idle thread's watches and accepts a thread with nothing to stop", () =>