diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index f26397040f0ff..921e212a516b0 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -825,6 +825,150 @@ describe("OrchestratorMcpService", () => { }), ); + // Task C's queued turn is promoted when another task's turn ends, before C's queued cancel lands. + const cancelPromotedTask = Effect.fn("cancelPromotedTask")(function* (input: { + readonly name: string; + // C's turn also ends, and D's queued turn starts, before C's interrupt lands. + readonly endsBeforeInterrupt: boolean; + }) { + const parentThreadId = ThreadId.make(`thread:${input.name}-parent`); + const childThreadId = ThreadId.make(`thread:${input.name}-child`); + const otherTaskId = NodeId.make(`node:${input.name}-other-task`); + const taskId = NodeId.make(`node:${input.name}-task`); + const laterTaskId = NodeId.make(`node:${input.name}-later-task`); + const otherRunId = RunId.make(`run:${input.name}-other`); + const runId = RunId.make(`run:${input.name}-run`); + const laterRunId = RunId.make(`run:${input.name}-later`); + const dispatched = yield* Ref.make>([]); + const parentProjection = { + thread: { id: parentThreadId }, + runs: [], + providerTurns: [], + contextTransfers: [], + subagents: [ + { + id: taskId, + threadId: parentThreadId, + origin: "app_owned", + childThreadId, + driver: "codex", + model: "gpt-5.6-terra", + result: null, + completionDelivery: { state: "pending" }, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const childProjection = (statuses: readonly [string, string, string]) => + ({ + thread: { id: childThreadId, title: "Child" }, + runs: [ + { id: otherRunId, ordinal: 2, status: statuses[0], delegatedTaskId: otherTaskId }, + { id: runId, ordinal: 3, status: statuses[1], delegatedTaskId: taskId }, + { id: laterRunId, ordinal: 4, status: statuses[2], delegatedTaskId: laterTaskId }, + ], + contextTransfers: [], + messages: [], + subagents: [], + providerThreads: [], + providerTurns: [], + runtimeRequests: [], + }) as unknown as OrchestrationV2ThreadProjection; + const child = yield* Ref.make(childProjection(["running", "queued", "queued"])); + const descendantStops = yield* Ref.make>([]); + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: (threadId) => + threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), + getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), + listAppOwnedChildThreadIds: () => Effect.succeed([]), + stopDelegatedTasks: (stop) => Ref.update(descendantStops, (runs) => [...runs, stop.runId]), + dispatch: (command) => + Effect.gen(function* () { + yield* Ref.update(dispatched, (commands) => [...commands, command]); + if (command.type === "queued-run.cancel") { + yield* Ref.set(child, childProjection(["completed", "running", "queued"])); + return yield* Effect.fail(new Error("Run is not queued.") as never); + } + if (command.type === "run.interrupt" && input.endsBeforeInterrupt) { + yield* Ref.set(child, childProjection(["completed", "completed", "running"])); + return yield* Effect.fail(new Error("Run is not interruptible.") as never); + } + return {} as never; + }), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make(`environment:${input.name}`), + requestNamespace: `provider-session:${input.name}`, + thread: { + threadId: parentThreadId, + providerSessionId: `provider-session:${input.name}`, + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + const result = yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.cancelTask(scope, { taskId, clientRequestId: `${input.name}-cancel` }); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + return { + status: result.status, + runId, + descendantStops: yield* Ref.get(descendantStops), + commands: (yield* Ref.get(dispatched)).map((command) => ({ + type: command.type, + runId: "runId" in command ? command.runId : undefined, + })), + }; + }); + + it.effect("stops a queued follow-up task whose turn starts while it is cancelled", () => + Effect.gen(function* () { + const cancelled = yield* cancelPromotedTask({ + name: "mcp-cancel-promoted", + endsBeforeInterrupt: false, + }); + assert.equal(cancelled.status, "cancel_requested"); + assert.deepEqual(cancelled.commands, [ + { type: "queued-run.cancel", runId: cancelled.runId }, + { type: "run.interrupt", runId: cancelled.runId }, + { type: "delegated_task.completion-delivery.dispose", runId: undefined }, + ]); + // Only the tasks the promoted turn delegated, not the other task's. + assert.deepEqual(cancelled.descendantStops, [cancelled.runId]); + }), + ); + + it.effect("never stops the turn the queue starts after a cancelled task's turn ends", () => + Effect.gen(function* () { + const cancelled = yield* cancelPromotedTask({ + name: "mcp-cancel-promoted-ended", + endsBeforeInterrupt: true, + }); + assert.equal(cancelled.status, "cancel_requested"); + assert.deepEqual(cancelled.commands, [ + { type: "queued-run.cancel", runId: cancelled.runId }, + { type: "run.interrupt", runId: cancelled.runId }, + { type: "delegated_task.completion-delivery.dispose", runId: undefined }, + ]); + // Only the tasks the promoted turn delegated, not the other task's. + assert.deepEqual(cancelled.descendantStops, [cancelled.runId]); + }), + ); + it.effect("refuses to cancel a task whose child now runs above the parent's modes", () => Effect.gen(function* () { const parentThreadId = ThreadId.make("thread:mcp-cancel-dispose-failed-parent"); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 066722ea6625f..c6d27fbce0764 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -63,6 +63,7 @@ import { type ServerProviderModel, ThreadId, isOrchestrationV2WorkActive, + delegatedTaskYieldsLiveRun, latestProviderTurnForAttempt, resolveProjectProviderInstanceEnabled, SECRET_REQUEST_UNSAVED_NOTE, @@ -2642,36 +2643,92 @@ const make = Effect.gen(function* () { const stopChild = Effect.gen(function* () { const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" }); const reason = agentStopReason(input.reason); - yield* threadManagement - .dispatch({ - type: "thread.stop", - commandId, - threadId: childThreadId, - stopDelegatedTasks: false, - reason, - }) - .pipe( - Effect.mapError((error) => - failure( - "task_not_cancellable", - `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`, + const readRuns = threadManagement.getThreadRecords(childThreadId, ["runs"]).pipe( + Effect.map(({ runs }) => runs), + Effect.mapError(threadManagementFailure), + ); + const readRun = (runId: RunId) => + Effect.map(readRuns, (runs) => runs.find((run) => run.id === runId)); + // By its id: a thread-wide stop could reach another task's turn the queue promoted + // meanwhile. A turn that ended before the interrupt landed needs nothing more, but the + // tasks it delegated still go, and only those. + const stopOwnRun = Effect.fn("OrchestratorMcpService.stopOwnRun")(function* ( + run: OrchestrationV2Run, + ) { + const runCommandId = stableCommandId({ + scope, + requestKey: key, + operation: `cancel-task-run:${run.id}`, + }); + if (ThreadManagementService.isActiveRun(run)) { + yield* threadManagement + .dispatch({ + type: "run.interrupt", + commandId: runCommandId, + threadId: childThreadId, + runId: run.id, + reason, + }) + .pipe( + Effect.mapError(threadManagementFailure), + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest !== undefined && ThreadManagementService.isActiveRun(latest) + ? Effect.fail(error) + : Effect.void, + ), + ), + ); + } + yield* threadManagement + .stopDelegatedTasks({ + threadId: childThreadId, + commandId: runCommandId, + reason, + runId: run.id, + }) + .pipe( + Effect.mapError((error) => + failure( + "task_not_cancellable", + `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, + ), ), - ), - ); - yield* threadManagement - .stopDelegatedTasks({ threadId: childThreadId, commandId, reason }) - .pipe( - Effect.mapError((error) => - failure( - "task_not_cancellable", - `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, + ); + }); + // One snapshot for the check and the sweep: a turn promoted between two reads would + // be neither cancelled nor interrupted. + const runs = yield* readRuns; + // The child's live turn belongs to another task, so only this task's queued turns go. + if (!delegatedTaskYieldsLiveRun(runs, input.taskId)) { + yield* threadManagement + .dispatch({ + type: "thread.stop", + commandId, + threadId: childThreadId, + stopDelegatedTasks: false, + reason, + }) + .pipe( + Effect.mapError((error) => + failure( + "task_not_cancellable", + `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`, + ), ), - ), - ); - const stopped = yield* threadManagement - .getThreadRecords(childThreadId, ["runs"]) - .pipe(Effect.mapError(threadManagementFailure)); - for (const run of stopped.runs) { + ); + yield* threadManagement + .stopDelegatedTasks({ threadId: childThreadId, commandId, reason }) + .pipe( + Effect.mapError((error) => + failure( + "task_not_cancellable", + `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, + ), + ), + ); + } + for (const run of runs) { if ( run.status !== "queued" || (run.delegatedTaskId !== input.taskId && run.id !== current.childRunId) @@ -2690,7 +2747,19 @@ const make = Effect.gen(function* () { threadId: childThreadId, runId: run.id, }) - .pipe(Effect.mapError(threadManagementFailure)); + .pipe( + Effect.mapError(threadManagementFailure), + // A turn that started meanwhile is no longer queued. + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest?.status === "queued" + ? Effect.fail(error) + : latest === undefined + ? Effect.void + : stopOwnRun(latest), + ), + ), + ); } }); // Checked above; a user raising one of these threads meanwhile is @@ -2741,7 +2810,7 @@ const make = Effect.gen(function* () { return subagents.some((task) => task.id === input.taskId && task.status === "cancelled"); }).pipe(Effect.mapError(threadManagementFailure)); if (isTerminalTaskStatus(current.status)) { - // The child can still run work for another turn, so it is stopped either way. + // The child can still run work for another turn; another task's live turn is left alone. yield* stopWithinLimit; const cancelled = yield* cancelOrphan; yield* disposeCompletionDelivery; diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index bbeac142dfebb..05944aa8f6c39 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -194,264 +194,275 @@ const layerUnusedScheduledTaskStub = Layer.succeed( }), ); -describe("orchestrator MCP toolkit", () => { - it.live.each([ - { label: "distinct", result: queuedFollowupResult, latestSummary: queuedFollowupResult }, - { label: "identical", result: delegatedResult, latestSummary: null }, - { label: "emoji", result: "😀", latestSummary: "😀" }, - ])("reports $label child results", ({ result, latestSummary }) => - Effect.scoped( +const makeQueuedFollowupScenario = Effect.fn("makeQueuedFollowupScenario")(function* ( + result: string, +) { + const cwd = yield* checkpointWorkspace("delegated-task-status"); + const rawTranscript = yield* readDelegatedTaskStatusTranscript(result); + const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( + materializeReplayTranscriptWorkspace(rawTranscript, cwd), + ); + const layerOrchestrator = ProviderReplayHarness.layerProviderReplay( + { + name: "delegated-task-status/codex", + transcript, + commands: [], + runtimePolicyOverride: { cwd }, + }, + CodexOrchestratorReplayHarness, + ); + // SQL spans carry no bound parameters, so this records which thread each tool read. + const threadRecordReads: Array<{ + readonly threadId: ThreadId; + readonly fields: ReadonlyArray; + }> = []; + const layerOrchestration = Layer.merge( + layerOrchestrator, + Layer.effect( + ThreadManagementService.ThreadManagementService, Effect.gen(function* () { - const cwd = yield* checkpointWorkspace("delegated-task-status"); - const rawTranscript = yield* readDelegatedTaskStatusTranscript(result); - const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( - materializeReplayTranscriptWorkspace(rawTranscript, cwd), - ); - const layerOrchestrator = ProviderReplayHarness.layerProviderReplay( - { - name: "delegated-task-status/codex", - transcript, - commands: [], - runtimePolicyOverride: { cwd }, - }, - CodexOrchestratorReplayHarness, - ); - // SQL spans carry no bound parameters, so this records which thread each tool read. - const threadRecordReads: Array<{ - readonly threadId: ThreadId; - readonly fields: ReadonlyArray; - }> = []; - const layerOrchestration = Layer.merge( - layerOrchestrator, - Layer.effect( - ThreadManagementService.ThreadManagementService, - Effect.gen(function* () { - const service = yield* ThreadManagementService.ThreadManagementService; - return ThreadManagementService.ThreadManagementService.of({ - ...service, - getThreadRecords: (threadId, fields, filter) => - Effect.sync(() => threadRecordReads.push({ threadId, fields })).pipe( - Effect.andThen(service.getThreadRecords(threadId, fields, filter)), - ), - }); - }), - ).pipe( - Layer.provide( - ThreadManagementService.layer.pipe( - Layer.provide(layerOrchestrator), - Layer.provide(ServerSettings.layerTest()), - Layer.provide( - CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), - ), - ), + const service = yield* ThreadManagementService.ThreadManagementService; + return ThreadManagementService.ThreadManagementService.of({ + ...service, + getThreadRecords: (threadId, fields, filter) => + Effect.sync(() => threadRecordReads.push({ threadId, fields })).pipe( + Effect.andThen(service.getThreadRecords(threadId, fields, filter)), ), - ), - ); - const layerProviderRegistry = ProviderRegistryMock.layer([ - makeProviderSnapshot({ - instanceId: codexInstanceId, - driver: ProviderDriverKind.make("codex"), - model: codexModel, - }), - ]); - const layerTest = layerOrchestratorToolkit.pipe( - Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), - Layer.provideMerge(layerOrchestration), - Layer.provide( - CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript).pipe( - Layer.provide(McpProviderSessions.layer), - ), - ), - Layer.provide(layerProviderRegistry), + }); + }), + ).pipe( + Layer.provide( + ThreadManagementService.layer.pipe( + Layer.provide(layerOrchestrator), Layer.provide(ServerSettings.layerTest()), - Layer.provide(layerUnusedScheduledTaskStub), - Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), - Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), - Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), - Layer.provide(Layer.mock(ProjectService.ProjectService)({})), - Layer.provideMerge( - SecretRequests.layer.pipe( - Layer.provide(layerMemorySecretStore), - Layer.provide(Layer.mock(SecretResolver.SecretResolver)({})), - Layer.provide(layerOrchestration), - ), + Layer.provide( + CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), ), - Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), - Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), - ); + ), + ), + ), + ); + const layerProviderRegistry = ProviderRegistryMock.layer([ + makeProviderSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: codexModel, + }), + ]); + const layerTest = layerOrchestratorToolkit.pipe( + Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), + Layer.provideMerge(layerOrchestration), + Layer.provide( + CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript).pipe( + Layer.provide(McpProviderSessions.layer), + ), + ), + Layer.provide(layerProviderRegistry), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerUnusedScheduledTaskStub), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), + Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), + Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ProjectService.ProjectService)({})), + Layer.provideMerge( + SecretRequests.layer.pipe( + Layer.provide(layerMemorySecretStore), + Layer.provide(Layer.mock(SecretResolver.SecretResolver)({})), + Layer.provide(layerOrchestration), + ), + ), + Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), + Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const server = yield* McpServer.McpServer; - const parentCreate = yield* orchestrator.dispatch({ - type: "thread.create", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("command:mcp-replay-parent:create"), - threadId: parentThreadId, - projectId, - title: "MCP replay parent", - modelSelection: codexSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: cwd, - }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("command:mcp-replay-parent:start"), - threadId: parentThreadId, - messageId: MessageId.make("message:mcp-replay-parent:start"), - text: parentPrompt, - attachments: [], - modelSelection: codexSelection, - dispatchMode: { type: "start_immediately" }, - }); - yield* orchestrator - .streamStoredEventsFrom({ - threadId: parentThreadId, - afterSequence: parentCreate.sequence, - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "provider-turn.updated" && - stored.event.payload.status === "running", - ), - Stream.runHead, - Effect.flatMap( - Option.match({ - onNone: () => Effect.die("Parent provider turn did not start."), - onSome: () => Effect.void, - }), - ), - ); + const start = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const server = yield* McpServer.McpServer; + const parentCreate = yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-replay-parent:create"), + threadId: parentThreadId, + projectId, + title: "MCP replay parent", + modelSelection: codexSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-replay-parent:start"), + threadId: parentThreadId, + messageId: MessageId.make("message:mcp-replay-parent:start"), + text: parentPrompt, + attachments: [], + modelSelection: codexSelection, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: parentCreate.sequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "provider-turn.updated" && + stored.event.payload.status === "running", + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Parent provider turn did not start."), + onSome: () => Effect.void, + }), + ), + ); - const invocation: McpInvocationContext.McpInvocationScope = { - environmentId: EnvironmentId.make("environment:mcp-replay"), - requestNamespace: "mcp-provider-session-replay-parent", - thread: { - threadId: parentThreadId, - providerSessionId: "mcp-provider-session-replay-parent", - providerInstanceId: codexInstanceId, - }, - client: undefined, - capabilities: new Set(["orchestration"]), - issuedAt: 1, - }; - const invoke = (name: string, args: Record, scope = invocation) => - server - .callTool({ name, arguments: args }) - .pipe( - Effect.provideService(McpInvocationContext.McpInvocationContext, scope), - Effect.provideService(McpSchema.McpServerClient, client), - ); + const invocation: McpInvocationContext.McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-replay"), + requestNamespace: "mcp-provider-session-replay-parent", + thread: { + threadId: parentThreadId, + providerSessionId: "mcp-provider-session-replay-parent", + providerInstanceId: codexInstanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + const invoke = (name: string, args: Record, scope = invocation) => + server + .callTool({ name, arguments: args }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); - const delegationStartSequence = - yield* orchestrator.getThreadEventSequence(parentThreadId); - const delegatedCall = yield* invoke("delegate_task", { - task: delegatedPrompt, - target: { - providerInstanceId: codexInstanceId, - model: codexModel, - }, - mode: "async", - clientRequestId: "delegate-codex-replay-1", - }); - const delegatedStart = yield* decodeDelegateTaskResult( - delegatedCall.structuredContent, - ).pipe(Effect.orDie); - expect(delegatedStart.childRunId).not.toBeNull(); - yield* orchestrator - .streamStoredEventsFrom({ - threadId: parentThreadId, - afterSequence: delegationStartSequence, - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "context-transfer.created" && - stored.event.payload.type === "subagent_result" && - stored.event.payload.sourceThreadId === delegatedStart.childThreadId && - stored.event.payload.sourcePoint.runId === delegatedStart.childRunId, - ), - Stream.runHead, - Effect.flatMap( - Option.match({ - onNone: () => Effect.die("Delegated result transfer was not created."), - onSome: () => Effect.void, - }), - ), - ); - const delegatedStatusCall = yield* invoke("task_status", { - taskId: delegatedStart.taskId, - }); - const delegated = yield* decodeDelegateTaskResult( - delegatedStatusCall.structuredContent, - ).pipe(Effect.orDie); - expect(delegated).toMatchObject({ - status: "completed", - summary: delegatedResult, - providerInstanceId: codexInstanceId, - hasPendingChildRuns: false, - latestTerminalRunId: delegated.childRunId, - latestTerminalStatus: "completed", - latestTerminalSummary: null, - latestTerminalResultContextTransferId: null, - }); - expect(delegated.resultContextTransferId).not.toBeNull(); + const delegationStartSequence = yield* orchestrator.getThreadEventSequence(parentThreadId); + const delegatedCall = yield* invoke("delegate_task", { + task: delegatedPrompt, + target: { + providerInstanceId: codexInstanceId, + model: codexModel, + }, + mode: "async", + clientRequestId: "delegate-codex-replay-1", + }); + const delegatedStart = yield* decodeDelegateTaskResult(delegatedCall.structuredContent).pipe( + Effect.orDie, + ); + expect(delegatedStart.childRunId).not.toBeNull(); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: delegationStartSequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "context-transfer.created" && + stored.event.payload.type === "subagent_result" && + stored.event.payload.sourceThreadId === delegatedStart.childThreadId && + stored.event.payload.sourcePoint.runId === delegatedStart.childRunId, + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Delegated result transfer was not created."), + onSome: () => Effect.void, + }), + ), + ); + const delegatedStatusCall = yield* invoke("task_status", { + taskId: delegatedStart.taskId, + }); + const delegated = yield* decodeDelegateTaskResult(delegatedStatusCall.structuredContent).pipe( + Effect.orDie, + ); + expect(delegated).toMatchObject({ + status: "completed", + summary: delegatedResult, + providerInstanceId: codexInstanceId, + hasPendingChildRuns: false, + latestTerminalRunId: delegated.childRunId, + latestTerminalStatus: "completed", + latestTerminalSummary: null, + latestTerminalResultContextTransferId: null, + }); + expect(delegated.resultContextTransferId).not.toBeNull(); - const delegatedChild = yield* orchestrator.getThreadProjection( - localChild(delegated.childThreadId), - ); - expect(delegatedChild.thread.lineage.relationshipToParent).toBe("subagent"); - expect(isProviderNativeSubagentThread(delegatedChild.thread)).toBe(false); - const followupStartSequence = yield* orchestrator.getThreadEventSequence( - localChild(delegated.childThreadId), - ); - const runningFollowupCall = yield* invoke("t3_thread_send", { - threadId: localChild(delegated.childThreadId), - message: cancellationPrompt, - clientRequestId: "delegated-child-replay-running-1", - }); - const runningFollowup = yield* decodeThreadSendResult( - runningFollowupCall.structuredContent, - ).pipe(Effect.orDie); - yield* orchestrator - .streamStoredEventsFrom({ - threadId: localChild(delegated.childThreadId), - afterSequence: followupStartSequence, - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "provider-turn.updated" && - stored.event.payload.status === "running", - ), - Stream.runHead, - Effect.flatMap( - Option.match({ - onNone: () => Effect.die("Follow-up provider turn did not start."), - onSome: () => Effect.void, - }), - ), - ); + const delegatedChild = yield* orchestrator.getThreadProjection( + localChild(delegated.childThreadId), + ); + expect(delegatedChild.thread.lineage.relationshipToParent).toBe("subagent"); + expect(isProviderNativeSubagentThread(delegatedChild.thread)).toBe(false); + const followupStartSequence = yield* orchestrator.getThreadEventSequence( + localChild(delegated.childThreadId), + ); + const runningFollowupCall = yield* invoke("t3_thread_send", { + threadId: localChild(delegated.childThreadId), + message: cancellationPrompt, + clientRequestId: "delegated-child-replay-running-1", + }); + const runningFollowup = yield* decodeThreadSendResult( + runningFollowupCall.structuredContent, + ).pipe(Effect.orDie); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: localChild(delegated.childThreadId), + afterSequence: followupStartSequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "provider-turn.updated" && + stored.event.payload.status === "running", + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Follow-up provider turn did not start."), + onSome: () => Effect.void, + }), + ), + ); - const queuedFollowupCall = yield* invoke("t3_thread_send", { - threadId: localChild(delegated.childThreadId), - message: queuedFollowupPrompt, - mode: "queue", - clientRequestId: "delegated-child-replay-queued-1", - }); - const queuedFollowup = yield* decodeThreadSendResult( - queuedFollowupCall.structuredContent, - ).pipe(Effect.orDie); - expect(queuedFollowup).toMatchObject({ - status: "queued", - delivery: "queued", - }); + const queuedFollowupCall = yield* invoke("t3_thread_send", { + threadId: localChild(delegated.childThreadId), + message: queuedFollowupPrompt, + mode: "queue", + clientRequestId: "delegated-child-replay-queued-1", + }); + const queuedFollowup = yield* decodeThreadSendResult(queuedFollowupCall.structuredContent).pipe( + Effect.orDie, + ); + expect(queuedFollowup).toMatchObject({ + status: "queued", + delivery: "queued", + }); + return { orchestrator, invoke, invocation, delegated, runningFollowup, queuedFollowup }; + }); + return { layer: layerTest, start, threadRecordReads }; +}); + +describe("orchestrator MCP toolkit", () => { + it.live.each([ + { label: "distinct", result: queuedFollowupResult, latestSummary: queuedFollowupResult }, + { label: "identical", result: delegatedResult, latestSummary: null }, + { label: "emoji", result: "😀", latestSummary: "😀" }, + ])("reports $label child results", ({ result, latestSummary }) => + Effect.scoped( + Effect.gen(function* () { + const scenario = yield* makeQueuedFollowupScenario(result); + yield* Effect.gen(function* () { + const { orchestrator, invoke, invocation, delegated, runningFollowup, queuedFollowup } = + yield* scenario.start; const pendingProjection = yield* orchestrator.getThreadProjection( localChild(delegated.childThreadId), @@ -758,6 +769,7 @@ describe("orchestrator MCP toolkit", () => { expect(new Set(projectionTableReads.slice(tableReadsBeforeCapabilities))).toEqual( new Set(["threads"]), ); + const { threadRecordReads } = scenario; const runReadsBeforeStatus = runReads.length; const transferReadsBeforeStatus = transferReads.length; const threadRecordReadsBeforeStatus = threadRecordReads.length; @@ -776,7 +788,43 @@ describe("orchestrator MCP toolkit", () => { ) .map(({ threadId }) => threadId), ).toEqual([delegated.childThreadId]); - }).pipe(Effect.provide(layerTest)); + }).pipe(Effect.provide(scenario.layer)); + }), + ), + ); + + it.live("cancelling a queued follow-up task leaves another task's running turn alone", () => + Effect.scoped( + Effect.gen(function* () { + const scenario = yield* makeQueuedFollowupScenario(queuedFollowupResult); + yield* Effect.gen(function* () { + const { orchestrator, invoke, delegated, runningFollowup, queuedFollowup } = + yield* scenario.start; + const childThreadId = localChild(delegated.childThreadId); + + const cancelQueued = invoke("task_cancel", { + taskId: queuedFollowup.taskId, + reason: "Drop only the queued follow-up.", + clientRequestId: "cancel-queued-followup-task-1", + }); + expect((yield* cancelQueued).isError).not.toBe(true); + // A retry finds the task already settled and must still leave the other task's turn. + expect((yield* cancelQueued).isError).not.toBe(true); + + const child = yield* orchestrator.getThreadProjection(childThreadId); + expect({ + running: child.runs.find((run) => run.id === runningFollowup.runId)?.status, + queued: child.runs.find((run) => run.id === queuedFollowup.runId)?.status, + interruptRequestsForRunningTurn: child.turnItems.filter( + (item) => + item.runId === runningFollowup.runId && item.type === "run_interrupt_request", + ).length, + }).toEqual({ + running: "running", + queued: "cancelled", + interruptRequestsForRunningTurn: 0, + }); + }).pipe(Effect.provide(scenario.layer)); }), ), ); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index de12768ac7b17..bcebdce3b246f 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -487,6 +487,8 @@ export interface ThreadManagementServiceShape { readonly threadId: ThreadId; readonly commandId: CommandId; readonly reason?: string | undefined; + /** Only the tasks this run delegated, with their whole subtrees. */ + readonly runId?: RunId | undefined; }) => Effect.Effect< void, Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError @@ -1477,6 +1479,7 @@ const make = Effect.gen(function* () { const stoppedChildThreadIds = new Set(); for (const task of subagents) { if (task.origin !== "app_owned") continue; + if (input.runId !== undefined && task.runId !== input.runId) continue; if (task.childThreadId === null) { // A task in a linked environment ends here; its follower then stops it there. // One with its result keeps it, but its thread there may run follow-up work. @@ -1523,7 +1526,7 @@ const make = Effect.gen(function* () { stopDelegatedTasks: false, ...(input.reason === undefined ? {} : { reason: input.reason }), }).pipe( - Effect.andThen(stopDelegatedTasks({ ...input, threadId })), + Effect.andThen(stopDelegatedTasks({ ...input, threadId, runId: undefined })), Effect.catch((error) => Effect.logWarning("Unable to stop a delegated task", { parentThreadId: input.threadId, diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx index 102d9734ded92..02996acfe259a 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx @@ -662,9 +662,16 @@ export function ThreadRelationshipsPanel(props: { ) => { if (stoppingId !== null) return; setStoppingId(childThreadId); + // The row shows the child's live work, which an earlier task than its latest one can own. + const owner = + projected.findLast( + (subagent) => + subagent.childThreadId === childThreadId && + (subagent.status === "running" || subagent.status === "waiting"), + ) ?? agent; const result = await interruptTurn({ environmentId: props.environmentId, - input: { threadId: agent.threadId, subagentId: agent.id }, + input: { threadId: owner.threadId, subagentId: owner.id }, }); setStoppingId(null); if (result._tag === "Failure") { diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 821a92721d357..a4a9847d6b5d3 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -15,6 +15,7 @@ import { TurnItemId, WS_METHODS, type OrchestrationV2Command, + OrchestrationV2DispatchCommandError, type OrchestrationV2ThreadLaunchInput, type OrchestrationV2ThreadProjection, type ProjectMutation, @@ -60,21 +61,27 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct readonly projects: ProjectMutation[]; readonly launches?: OrchestrationV2ThreadLaunchInput[]; readonly projection?: OrchestrationV2ThreadProjection; + readonly childProjections?: ReadonlyMap; readonly projectionRequests?: ThreadId[]; readonly advertiseThreadStop?: boolean; + readonly onCommand?: ( + command: OrchestrationV2Command, + ) => Effect.Effect; }) { const client = { [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => Effect.sync(() => { input.commands.push(command); return { sequence: input.commands.length }; - }), + }).pipe(Effect.tap(() => input.onCommand?.(command) ?? Effect.void)), [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (requestInput: { readonly threadId: ThreadId; }) => Effect.sync(() => { input.projectionRequests?.push(requestInput.threadId); - return input.projection ?? v2Projection; + return ( + input.childProjections?.get(requestInput.threadId) ?? input.projection ?? v2Projection + ); }), [ORCHESTRATION_V2_WS_METHODS.launchThread]: (launchInput: OrchestrationV2ThreadLaunchInput) => Effect.sync(() => { @@ -531,6 +538,190 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); + // Task B's turn runs on the child thread while tasks C and D are queued behind it. + const stopQueuedFollowupTask = Effect.fn("stopQueuedFollowupTask")(function* ( + // C's turn had started, or also ended, when its queued cancel landed. + promoted?: "running" | "completed", + ) { + const childThreadId = ThreadId.make("app-owned-child-thread"); + const runningTaskId = NodeId.make("running-follow-up-task"); + const queuedTaskId = NodeId.make("queued-follow-up-task"); + const laterTaskId = NodeId.make("later-follow-up-task"); + const queuedRunId = RunId.make("queued-follow-up-run"); + const grandchildThreadId = ThreadId.make("grandchild-thread"); + const task = (id: NodeId, status: "running" | "pending") => ({ + id, + threadId: v2ThreadId, + runId: RunId.make("parent-run"), + parentNodeId: NodeId.make("owner-node"), + origin: "app_owned" as const, + createdBy: "agent" as const, + driver: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Work", + title: null, + model: null, + result: null, + status, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + }); + const childRun = ( + id: RunId, + ordinal: number, + status: "running" | "queued" | "completed", + taskId: NodeId, + ) => ({ + id, + threadId: childThreadId, + ordinal, + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: v2Projection.thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message-${id}`), + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: v2Now, + startedAt: status === "queued" ? null : v2Now, + completedAt: status === "completed" ? v2Now : null, + checkpointId: null, + contextHandoffId: null, + delegatedTaskId: taskId, + }); + const childProjection = ( + running: "running" | "completed", + queued: "running" | "queued" | "completed", + ) => ({ + ...v2Projection, + thread: { ...v2Projection.thread, id: childThreadId }, + runs: [ + childRun(RunId.make("running-follow-up-run"), 2, running, runningTaskId), + childRun(queuedRunId, 3, queued, queuedTaskId), + childRun(RunId.make("later-follow-up-run"), 4, "queued", laterTaskId), + ], + // Once promoted, the queued task's turn delegated a task of its own, beside the other's. + subagents: + queued === "queued" + ? [] + : [ + { + ...task(NodeId.make("other-grandchild-task"), "running"), + threadId: childThreadId, + runId: RunId.make("running-follow-up-run"), + childThreadId: ThreadId.make("other-grandchild-thread"), + }, + { + ...task(NodeId.make("grandchild-task"), "running"), + threadId: childThreadId, + runId: queuedRunId, + childThreadId: grandchildThreadId, + }, + ], + }); + const childProjections = new Map([[childThreadId, childProjection("running", "queued")]]); + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ + commands, + projects: [], + advertiseThreadStop: true, + projection: { + ...v2Projection, + subagents: [ + task(runningTaskId, "running"), + task(queuedTaskId, "pending"), + task(laterTaskId, "pending"), + ], + }, + childProjections, + onCommand: (command) => { + if (command.type !== "queued-run.cancel" || promoted === undefined) return Effect.void; + childProjections.set(childThreadId, childProjection("completed", promoted)); + return Effect.fail( + new OrchestrationV2DispatchCommandError({ + commandId: command.commandId, + commandType: command.type, + message: "Run is not queued.", + }), + ); + }, + }); + + yield* interruptThreadTurn({ threadId: v2ThreadId, subagentId: queuedTaskId }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + ); + return { + childThreadId, + queuedRunId, + grandchildThreadId, + commands: commands.filter( + (command) => "threadId" in command && command.threadId !== v2ThreadId, + ), + }; + }); + + it.effect("stopping a queued follow-up task leaves another task's running child turn alone", () => + Effect.gen(function* () { + const stopped = yield* stopQueuedFollowupTask(); + expect(stopped.commands).toEqual([ + { + type: "queued-run.cancel", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, + }, + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); + + it.effect("stopping a follow-up task whose turn just started interrupts only that turn", () => + Effect.gen(function* () { + const stopped = yield* stopQueuedFollowupTask("running"); + expect(stopped.commands).toEqual([ + { + type: "queued-run.cancel", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, + }, + { + type: "run.interrupt", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, + }, + { + type: "thread.stop", + commandId: expect.any(String), + threadId: stopped.grandchildThreadId, + }, + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); + + it.effect("stopping a follow-up task whose turn already ended holds no other task's turn", () => + Effect.gen(function* () { + const stopped = yield* stopQueuedFollowupTask("completed"); + expect(stopped.commands).toEqual([ + { + type: "queued-run.cancel", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, + }, + { + type: "thread.stop", + commandId: expect.any(String), + threadId: stopped.grandchildThreadId, + }, + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); + it.effect("stops a thread without relying on background work in the history window", () => Effect.gen(function* () { const commands: OrchestrationV2Command[] = []; diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index f813ece95dc45..24a2d22ba60f4 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -24,9 +24,13 @@ import { type RuntimeMode, type RuntimeRequestId, type ThreadId, + type OrchestrationV2Run, type ThreadEnvMode, type UploadChatAttachment, type GuestMessageId, + delegatedTaskYieldsLiveRun, + isOrchestrationV2RunActive, + isOrchestrationV2WorkActive, } from "@t3tools/contracts"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; @@ -804,7 +808,80 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea subagentId: subagent.id, }).pipe(Effect.ignore); } - threadId = subagent.childThreadId; + const childThreadId = subagent.childThreadId; + const child = yield* getProjection(childThreadId); + // The child's live turn belongs to another task, so only this task's queued turns go. + if (delegatedTaskYieldsLiveRun(child.runs, subagent.id)) { + const readRun = (runId: RunId) => + Effect.map(getProjection(childThreadId), (latest) => + latest.runs.find((run) => run.id === runId), + ); + const settled = { sequence: 0 }; + // By its id: a thread-wide stop could reach, or hold, another task's turn. A turn that + // ended before the interrupt landed needs nothing more, but the tasks it delegated + // still go, and only those. + const stopOwnRun = Effect.fn("EnvironmentCommands.stopOwnRun")(function* ( + run: OrchestrationV2Run, + ) { + let result = settled; + if (isOrchestrationV2RunActive(run.status)) { + result = yield* dispatch({ + type: "run.interrupt", + commandId: yield* allocateCommandId({}), + threadId: childThreadId, + runId: run.id, + }).pipe( + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest !== undefined && isOrchestrationV2RunActive(latest.status) + ? Effect.fail(error) + : Effect.succeed(settled), + ), + ), + ); + } + // A child thread's stop also stops the tasks it delegated, all the way down. + const { subagents } = yield* getProjection(childThreadId); + for (const task of subagents) { + if ( + task.runId !== run.id || + task.origin !== "app_owned" || + task.childThreadId === null || + !isOrchestrationV2WorkActive(task.status) + ) + continue; + result = yield* dispatch({ + type: "thread.stop", + commandId: yield* allocateCommandId({}), + threadId: task.childThreadId, + }); + } + return result; + }); + let result = settled; + for (const run of child.runs) { + if (run.status !== "queued" || run.delegatedTaskId !== subagent.id) continue; + result = yield* dispatch({ + type: "queued-run.cancel", + commandId: yield* allocateCommandId({}), + threadId: childThreadId, + runId: run.id, + }).pipe( + // A turn that started meanwhile is no longer queued. + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest?.status === "queued" + ? Effect.fail(error) + : latest === undefined + ? Effect.succeed(settled) + : stopOwnRun(latest), + ), + ), + ); + } + return result; + } + threadId = childThreadId; } else { if (subagent?.runId == null) return { sequence: 0 }; return yield* dispatch({ @@ -832,13 +909,7 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea } if (runId === undefined) { const projection = yield* getProjection(threadId); - runId = projection.runs.findLast( - (run) => - run.status === "preparing" || - run.status === "starting" || - run.status === "running" || - run.status === "waiting", - )?.id; + runId = projection.runs.findLast((run) => isOrchestrationV2RunActive(run.status))?.id; if (runId === undefined) { const latestRun = projection.runs.at(-1); if ( diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index fc38ffd4da9bb..bbb913c1560a5 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -941,6 +941,28 @@ export function isOrchestrationV2WorkActive( return status === "pending" || status === "running" || status === "waiting"; } +export function isOrchestrationV2RunActive(status: OrchestrationV2Run["status"]): boolean { + return ( + status === "preparing" || status === "starting" || status === "running" || status === "waiting" + ); +} + +/** + * Whether the child's live run belongs to another delegated task. Stopping this task then + * cancels only its own queued runs instead of stopping the child thread. + */ +export function delegatedTaskYieldsLiveRun( + runs: ReadonlyArray, + taskId: NodeId, +): boolean { + return runs.some( + (run) => + isOrchestrationV2RunActive(run.status) && + run.delegatedTaskId !== undefined && + run.delegatedTaskId !== taskId, + ); +} + export const OrchestrationV2CheckpointScope = Schema.Struct({ id: CheckpointScopeId, threadId: ThreadId,