diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 10f6a8c7627edb5996b93def2ebfa94164d5ab88..83d5b11aef6ace9de1b74be46b0f3b0b2d3bb591 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2130,6 +2130,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { const makeWakeHarnessWithOptions = (options?: { readonly close?: (sdkMessages: Queue.Queue) => Effect.Effect; readonly interrupt?: Effect.Effect; + readonly stopTask?: (taskId: string) => Effect.Effect; readonly environment?: NodeJS.ProcessEnv; readonly settings?: ClaudeSettings; // A CLI process opened after the first streams from its own queue, so the @@ -2155,6 +2156,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { const permissionModeChanges: Array = []; const continuationRequests: Array = []; + const subagentReceipts = + yield* Queue.unbounded< + Extract + >(); const terminalReceipts = yield* Queue.unbounded< Extract @@ -2227,6 +2232,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { Effect.sync(() => { permissionModeChanges.push(mode); }), + ...(options?.stopTask === undefined ? {} : { stopTask: options.stopTask }), interrupt: options?.interrupt ?? Effect.void, close: options?.close?.(sdkMessages) ?? Effect.void, }; @@ -2253,6 +2259,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { Stream.runForEach((event) => Effect.gen(function* () { events.push(event); + if (event.type === "subagent.updated") yield* Queue.offer(subagentReceipts, event); if (event.type === "turn.terminal") { yield* Queue.offer(terminalReceipts, event); } @@ -2292,6 +2299,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { continuationRequests, events, terminalReceipts, + subagentReceipts, systemNoticeReceipts, getOpenedOptions: () => openedOptions, terminalEvents, @@ -8076,6 +8084,107 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each([false, true])( + "stops one native subagent while its sibling keeps running, owner settled %s", + (ownerSettled) => + Effect.scoped( + Effect.gen(function* () { + const stopped: string[] = []; + let closes = 0; + const harness = yield* makeWakeHarnessWithOptions({ + stopTask: (taskId) => + Effect.sync(() => { + stopped.push(taskId); + }), + interrupt: Effect.die("A child stop must not interrupt its owner"), + close: () => + Effect.sync(() => { + closes++; + }), + }); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-native-child-stop"), + text: "Start two agents.", + attachments: [], + }), + ); + for (const [taskId, toolUseId, uuid] of [ + ["agent-stop", "toolu-stop", "00000000-0000-4000-8000-000000000801"], + ["agent-keep", "toolu-keep", "00000000-0000-4000-8000-000000000802"], + ]) { + yield* harness.offerAndWait( + makeSubagentTaskStartedFrame({ taskId: taskId!, toolUseId: toolUseId!, uuid: uuid! }), + ); + yield* Queue.take(harness.subagentReceipts); + } + if (ownerSettled) { + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000803", + result: "Agents are working.", + }), + ); + yield* Queue.take(harness.terminalReceipts); + } + const stop = harness.runtime.stopSubagent; + assert.isDefined(stop); + yield* stop!({ providerThread: harness.providerThread, nativeTaskId: "agent-stop" }); + assert.deepEqual(stopped, ["agent-stop"]); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...makeSubagentNotificationFrame({ + taskId: "agent-stop", + toolUseId: "toolu-stop", + summary: "Stopped by user", + uuid: "00000000-0000-4000-8000-000000000804", + }), + status: "stopped", + }), + ); + if (ownerSettled) { + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-native-child-stop-wake"), + text: "Background task stopped.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + } + const stoppedEvent = yield* Queue.take(harness.subagentReceipts); + assert.equal(stoppedEvent.subagent.status, "cancelled"); + const tasks = harness.events.filter((event) => event.type === "subagent.updated"); + assert.equal( + tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-stop") + ?.subagent.status, + "cancelled", + ); + assert.equal( + tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-keep") + ?.subagent.status, + "running", + ); + assert.lengthOf(harness.terminalEvents(), ownerSettled ? 1 : 0); + yield* stop!({ providerThread: harness.providerThread, nativeTaskId: "agent-stop" }); + assert.deepEqual(stopped, ["agent-stop"]); + assert.equal(closes, 0); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect.each(["requested", "observed-before", "observed-after", "inherit", "unknown"] as const)( "records the subagent model from %s without inheriting the parent override", (source) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index caf63f2811b5940854e7eae0c82d2b1207f3e696..6251e698b2ad6caa50e9a1dedb0830755b8f1dd0 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -342,6 +342,7 @@ export interface ClaudeAgentSdkQuerySession { readonly setPermissionMode: ( mode: PermissionMode, ) => Effect.Effect; + readonly stopTask?: (taskId: string) => Effect.Effect; readonly interrupt: Effect.Effect; readonly close: Effect.Effect; } @@ -734,6 +735,11 @@ export const layerQueryRunner: Layer.Layer< }), ), ), + stopTask: (taskId) => + Effect.tryPromise({ + try: () => queryRuntime.stopTask(taskId), + catch: (cause) => queryRunnerError(cause, "stopTask"), + }), interrupt: Effect.tryPromise({ try: () => queryRuntime.interrupt(), catch: (cause) => queryRunnerError(cause, "interrupt"), @@ -8092,6 +8098,34 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }), steerTurn, interruptTurn, + stopSubagent: Effect.fn("ClaudeAdapterV2.stopSubagent")(function* (input) { + const existing = yield* Ref.get(queryContext); + const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId; + const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); + if (subagent === undefined || subagent.task.status !== "running") return; + if ( + existing === null || + existing.nativeThreadId !== nativeThreadId || + subagent.task.threadId !== input.providerThread.appThreadId || + existing.subagentsFromEarlierProcesses.has(subagent) || + existing.query.stopTask === undefined + ) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CLAUDE_PROVIDER, + detail: "Claude subagent has no live owning query with task stopping support.", + }); + } + yield* existing.query.stopTask(input.nativeTaskId).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CLAUDE_PROVIDER, + detail: "Claude native subagent stop failed.", + cause, + }), + ), + ); + }), respondToRuntimeRequest: Effect.fn("ClaudeAdapterV2.respondToRuntimeRequest")( function* (requestInput) { const pending = (yield* Ref.get(pendingRuntimeRequests)).get( diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index d9b21038cf71ec5139f45c9480a03b824479bdc4..0cb3de2e0012f1bfaf3368bcf13929acab555938 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -3,6 +3,7 @@ import { CheckpointScopeId, CommandId, MessageId, + NodeId, ProviderSessionId, RunAttemptId, ProviderApprovalDecision, @@ -48,6 +49,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ providerSessionId: ProviderSessionId, providerThreadId: ProviderThreadId, providerTurnId: ProviderTurnId, + subagent: Schema.optional(Schema.Struct({ id: NodeId, nativeTaskId: Schema.String })), }), Schema.Struct({ type: Schema.Literal("provider-turn.steer"), diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index d9bd8689f092259165bd8f056aef09ebbca3d980..230f4d4b77751bca240e5e81a6b4ed1ff36570b1 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -1,6 +1,7 @@ import { assert, it } from "@effect/vitest"; import { CommandId, + NodeId, ProviderSessionId, ProviderThreadId, ProviderTurnId, @@ -837,3 +838,32 @@ it.effect("settles a delegated child once its restart continuation fails for goo }).pipe(Effect.provide(layer)); }), ); + +it.effect("a native subagent stop does not settle its owner's background work", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const events = yield* Ref.make>([]); + const layer = layerExecutorFor({ + events, + interrupt: (input) => { + assert.equal(input.subagent?.nativeTaskId, "native-task"); + return Ref.update(events, (current) => [...current, "stop-child"]); + }, + threads: { dispatch: () => Effect.die("The owner's background work must stay running") }, + }); + yield* Effect.gen(function* () { + const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; + yield* executor.execute({ + ...restartEffect(now, { type: "detach" }), + request: { + type: "provider-turn.interrupt", + providerSessionId: oldSessionId, + providerThreadId, + providerTurnId, + subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" }, + }, + }); + }).pipe(Effect.provide(layer)); + assert.deepEqual(yield* Ref.get(events), ["stop-child"]); + }), +); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 9ae94910b7c8d5d287ee27cd19f7120193f7ca8d..ddfc54b533205bea832e1a32d9d6069e53ae692e 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -169,6 +169,9 @@ export const layerExecutor: Layer.Layer< providerSessionId: effect.request.providerSessionId, providerThreadId: effect.request.providerThreadId, providerTurnId: effect.request.providerTurnId, + ...(effect.request.subagent === undefined + ? {} + : { subagent: effect.request.subagent }), }) .pipe( Effect.catch((cause) => @@ -185,13 +188,15 @@ export const layerExecutor: Layer.Layer< // One Stop can interrupt several provider threads, so the // settle is keyed by effect, not by the Stop command. Effect.andThen( - threads.dispatch({ - type: "thread.background-work.settle", - commandId: CommandId.make(`${effect.id}:background-work-settled`), - threadId: effect.threadId, - providerThreadId: effect.request.providerThreadId, - providerTurnId: effect.request.providerTurnId, - }), + effect.request.subagent !== undefined + ? Effect.void + : threads.dispatch({ + type: "thread.background-work.settle", + commandId: CommandId.make(`${effect.id}:background-work-settled`), + threadId: effect.threadId, + providerThreadId: effect.request.providerThreadId, + providerTurnId: effect.request.providerTurnId, + }), ), Effect.mapError( (cause) => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 94560f62f6f5ce16612764d8fd116e2d234b28de..d1e1dd1e7a27b0a097860b39797dfd4ecac770f6 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -463,6 +463,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "prepared-run.progress": case "prepared-run.fail": case "prepared-run.retry": + case "subagent.stop": case "run.interrupt": case "queued-message.promote-to-steer": case "queue.resume": @@ -8821,6 +8822,89 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + const dispatchSubagentStop = ( + command: Extract, + events: Ref.Ref>, + effects: Ref.Ref>, + ) => + Effect.gen(function* () { + const projection = yield* loadProjectionForCommand(command, [ + "runs", + "subagents", + "providerThreads", + "providerTurns", + ]); + const run = projection.runs.find((candidate) => candidate.id === command.runId); + const subagent = projection.subagents.find( + (candidate) => candidate.id === command.subagentId, + ); + if ( + subagent === undefined || + subagent.runId !== command.runId || + subagent.origin !== "provider_native" || + subagent.driver !== "claudeAgent" || + subagent.nativeTaskRef?.strength !== "strong" || + subagent.nativeTaskRef.nativeId === null + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The requested native subagent cannot be stopped on this run.", + }); + } + if (!["pending", "running", "waiting"].includes(subagent.status)) return; + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === run?.providerThreadId, + ); + const providerTurn = projection.providerTurns.findLast( + (candidate) => candidate.providerThreadId === providerThread?.id, + ); + if ( + providerThread?.providerSessionId == null || + providerTurn === undefined || + providerThread.driver !== subagent.driver || + providerThread.providerInstanceId !== subagent.providerInstanceId + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "The requested native subagent has no owning provider session.", + }); + } + const providerSessionId = providerThread.providerSessionId; + const nativeTaskId = subagent.nativeTaskRef.nativeId; + const now = yield* DateTime.now; + // Record the request with its effect; the provider notification supplies the outcome. + yield* emit( + events, + command, + )({ + type: "subagent.updated", + threadId: command.threadId, + runId: command.runId, + nodeId: subagent.id, + providerInstanceId: subagent.providerInstanceId, + occurredAt: now, + payload: { ...subagent, updatedAt: now }, + }); + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:subagent.stop:${subagent.id}`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "provider-turn.interrupt", + providerSessionId, + providerThreadId: providerThread.id, + providerTurnId: providerTurn.id, + subagent: { id: subagent.id, nativeTaskId }, + }, + } satisfies PendingOrchestrationEffectV2, + ]); + return; + }); + const dispatchRunInterrupt = ( command: Extract, events: Ref.Ref>, @@ -10526,6 +10610,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.user-input.dismiss": yield* dispatchThreadUserInputDismiss(command, events, effects); break; + case "subagent.stop": + yield* dispatchSubagentStop(command, events, effects); + break; case "run.interrupt": cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects); // Stop also stops every delegated task under the thread once it commits. @@ -10739,7 +10826,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // its expected outcome, not a failure. planned.events.length > 0 || command.type === "thread.background-work.settle" || - command.type === "thread.stop" + command.type === "thread.stop" || + command.type === "subagent.stop" ? Effect.succeed(planned) : Effect.fail( new OrchestratorDispatchError({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index 4421da12da6ff4a8694aec587bb991d75964dd83..eaf1a9f940d0d0a9774e315392a538bca7925961 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -1,6 +1,7 @@ import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import { MessageId, + type NodeId, ProviderSessionId, ProviderThreadId, ProviderTurnId, @@ -43,6 +44,7 @@ const isProviderTurnControlError = Schema.is(ProviderTurnControlError); export interface ProviderTurnControlServiceV2Shape { readonly interrupt: (input: { + readonly subagent?: { readonly id: NodeId; readonly nativeTaskId: string }; readonly threadId: ThreadId; readonly providerSessionId: ProviderSessionId; readonly providerThreadId: ProviderThreadId; @@ -177,6 +179,21 @@ export const layer: Layer.Layer< ? loaded.session : yield* sessions.get(input.providerSessionId); if (Option.isNone(session)) return; + if (input.subagent !== undefined) { + if (session.value.stopSubagent === undefined) { + return yield* new ProviderTurnControlError({ + threadId: input.threadId, + operation: "interrupt", + providerTurnId: input.providerTurnId, + cause: "The provider does not support stopping a native subagent.", + }); + } + yield* session.value.stopSubagent({ + providerThread: loaded.providerThread, + nativeTaskId: input.subagent.nativeTaskId, + }); + return; + } // A settled turn reaches its adapter too: only the adapter knows // whether it still runs work for the thread, and each one either // stops it or reports there is nothing left to stop. Background work diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 6c5874cb75f2c1cedbad47a8a276e3f4a11d06bd..6b058f25a7e882ee8c85a11b542e787d46aa704d 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -7,6 +7,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, ProviderTurnId, ThreadId, TurnItemId, @@ -20,6 +21,7 @@ import * as SqlClient from "effect/sql/SqlClient"; import * as TestClock from "effect/testing/TestClock"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { CLAUDE_PROVIDER } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -582,3 +584,120 @@ it.effect("thread.stop marks a turn it cannot interrupt so a late agent watch is assert.deepEqual(yield* threadState(threadId), { runs: ["running"], watched: [] }); }).pipe(Effect.provide(layerTest)), ); + +it.effect( + "a native child stop persists only its targeted effect and leaves its owner's queue and watch alone", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:native-child-stop"); + yield* createWatchingThread(threadId, 20); + yield* send(threadId, "owner", "start_immediately"); + yield* delegate(threadId, "native-child-fixture"); + const before = yield* orchestrator.getThreadProjection(threadId); + const run = before.runs[0]!; + const now = yield* DateTime.now; + const subagent = { + ...before.subagents[0]!, + origin: "provider_native" as const, + driver: CLAUDE_PROVIDER, + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + nativeTaskRef: { + driver: CLAUDE_PROVIDER, + nativeId: "claude-task", + strength: "strong" as const, + }, + status: "running" as const, + startedAt: now, + }; + yield* projections.apply({ + id: EventId.make("event:native-child-stop:agent"), + type: "subagent.updated", + threadId, + runId: run.id, + nodeId: subagent.id, + occurredAt: now, + payload: subagent, + }); + const providerThread = before.providerThreads.find( + (thread) => thread.id === run.providerThreadId, + )!; + yield* projections.apply({ + id: EventId.make("event:native-child-stop:provider-thread"), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { + ...providerThread, + driver: CLAUDE_PROVIDER, + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + providerSessionId: ProviderSessionId.make("session:native-child-stop"), + }, + }); + const providerTurnId = ProviderTurnId.make("turn:native-child-stop"); + yield* projections.apply({ + id: EventId.make("event:native-child-stop:provider-turn"), + type: "provider-turn.updated", + threadId, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId: providerThread.id, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId!, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }); + // The owner has finished but its child and queued prompt remain. + yield* projections.apply({ + id: EventId.make("event:native-child-stop:owner-completed"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }); + yield* send(threadId, "queued", "queue_after_active"); + const stateBefore = yield* threadState(threadId); + const commandId = CommandId.make("stop-native-child"); + yield* orchestrator.dispatch({ + type: "subagent.stop", + commandId, + threadId, + runId: run.id, + subagentId: subagent.id, + }); + assert.deepEqual(yield* threadState(threadId), stateBefore); + const effects = yield* sql<{ payload_json: string }>` + SELECT payload_json FROM orchestration_v2_effect_outbox WHERE command_id = ${commandId} + `; + assert.equal(effects.length, 1); + assert.deepEqual(JSON.parse(effects[0]!.payload_json), { + type: "provider-turn.interrupt", + providerSessionId: "session:native-child-stop", + providerThreadId: providerThread.id, + providerTurnId, + subagent: { id: subagent.id, nativeTaskId: "claude-task" }, + }); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).subagents[0]?.status, + "running", + ); + const unrelated = yield* Effect.exit( + orchestrator.dispatch({ + type: "subagent.stop", + commandId: CommandId.make("stop-unknown-child"), + threadId, + runId: run.id, + subagentId: NodeId.make("unknown-child"), + }), + ); + assert.isTrue(Exit.isFailure(unrelated)); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index 3d8ef589050e68d583ddf542add8591e8086294f..c43c17e012946911cc5c6345742aa96936c4761a 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -166,6 +166,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray { state.projection = null; }); -it.each(["codex", "claudeAgent"])( - "stops only active app-owned %s subagents without opening their thread", - async (driver) => { +it.each([ + { driver: "codex", origin: "app_owned" }, + { driver: "claudeAgent", origin: "app_owned" }, + { driver: "claudeAgent", origin: "provider_native" }, +])( + "stops active $origin $driver subagents without opening their thread", + async ({ driver, origin }) => { vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); const parent = { id: "parent", @@ -70,7 +74,9 @@ it.each(["codex", "claudeAgent"])( const agent = { id: "agent", childThreadId: "child", - origin: "app_owned", + origin, + threadId: "parent", + nativeTaskRef: { driver, nativeId: "claude-task", strength: "strong" }, driver, providerInstanceId: "codex", title: "Worker", @@ -106,7 +112,10 @@ it.each(["codex", "claudeAgent"])( await act(async () => stopButton().props.onClick()); expect(state.command).toHaveBeenCalledWith({ environmentId: "test", - input: { threadId: "child" }, + input: + origin === "provider_native" + ? { threadId: "parent", subagentId: "agent" } + : { threadId: "child" }, }); expect(state.navigate).not.toHaveBeenCalled(); @@ -132,6 +141,12 @@ it.each(["codex", "claudeAgent"])( ); state.projection = { ...projection, subagents: [{ ...agent, status: "completed" }] }; await act(async () => renderer.update(cloneElement(panel))); + if (origin === "provider_native") { + expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength( + 0, + ); + continue; + } await act(async () => stopButton().props.onClick()); expect(state.command).toHaveBeenCalledTimes(1); expect(state.command).toHaveBeenLastCalledWith({ @@ -152,7 +167,7 @@ it.each(["codex", "claudeAgent"])( expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(0); state.projection = { ...projection, - subagents: [{ ...agent, origin: "provider_native", driver: "claudeAgent" }], + subagents: [{ ...agent, origin: "provider_native", driver: "codex" }], }; await act(async () => renderer.update(cloneElement(panel))); expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(0); diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx index 4c637d13c05d75b865a9b4dd8a29572c2b3a1b3a..d7f6471eea3c6c7fda51e60e24aeb586b6b826d1 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx @@ -24,7 +24,12 @@ import { canDetachThreadProviderSession, resolveLatestMergeBackRun, } from "@t3tools/client-runtime/state/thread-workflows"; -import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts"; +import type { + EnvironmentId, + OrchestrationV2Subagent, + OrchestrationV2ThreadShell, + ThreadId, +} from "@t3tools/contracts"; import { deriveSubagentElapsedMs } from "@t3tools/shared/orchestrationTiming"; import { groupBy } from "effect/Array"; import * as DateTime from "effect/DateTime"; @@ -221,6 +226,10 @@ export function ThreadRelationshipsPanel(props: { subagent.childThreadId, { ...projectedSubagentsToRuntime([subagent])[0]!, + id: subagent.id, + threadId: subagent.threadId, + nativeTaskRef: subagent.nativeTaskRef, + nativeStatus: subagent.status, driver: subagent.driver, providerInstanceId: subagent.providerInstanceId, origin: subagent.origin, @@ -328,12 +337,18 @@ export function ThreadRelationshipsPanel(props: { setBusyAction(null); }; - const stopSubagent = async (childThreadId: ThreadId) => { + const stopSubagent = async ( + childThreadId: ThreadId, + agent: Pick, + ) => { if (stoppingThreadId !== null) return; setStoppingThreadId(childThreadId); const result = await interruptTurn({ environmentId: props.environmentId, - input: { threadId: childThreadId }, + input: + agent.origin === "provider_native" + ? { threadId: agent.threadId, subagentId: agent.id } + : { threadId: childThreadId }, }); setStoppingThreadId(null); if (result._tag === "Failure") { @@ -398,7 +413,13 @@ export function ThreadRelationshipsPanel(props: { ); const failed = status === "failed" || status === "error"; const canStop = - agent?.origin === "app_owned" && + agent && + (agent.origin === "app_owned" || + (agent.origin === "provider_native" && + agent.driver === "claudeAgent" && + agent.nativeTaskRef?.strength === "strong" && + agent.nativeTaskRef.nativeId !== null && + ["pending", "running", "waiting"].includes(agent.nativeStatus))) && agent.startedAt && ["pending", "running", "waiting"].includes(agent.status); const trailingVisibilityClass = canStop @@ -566,7 +587,7 @@ export function ThreadRelationshipsPanel(props: { tone="destructive" aria-label={`Stop subagent ${threadTitle}`} disabled={stoppingThreadId !== null} - onClick={() => void stopSubagent(threadId)} + onClick={() => void stopSubagent(threadId, agent)} /> } > diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 52e08fe6e74ee85e3fafece4fd772a1eb54018ad..2c38e18f3e5c8f29924db461f7987bdf82dccc88 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -10,6 +10,7 @@ import { PlanId, ProjectId, ProviderInstanceId, + ProviderDriverKind, RunId, RuntimeRequestId, ThreadId, @@ -995,6 +996,60 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestCrypto)), ); + it.effect( + "a native subagent stop targets its owning run rather than holding the parent queue", + () => + Effect.gen(function* () { + const subagentId = NodeId.make("native-subagent"); + const runId = RunId.make("settled-owner-run"); + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ + commands, + projects: [], + projection: { + ...v2Projection, + subagents: [ + { + id: subagentId, + threadId: v2ThreadId, + runId, + parentNodeId: NodeId.make("owner-node"), + origin: "provider_native", + createdBy: "agent", + driver: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + providerThreadId: null, + childThreadId: ThreadId.make("native-child-thread"), + nativeTaskRef: null, + prompt: "Work", + title: null, + model: null, + result: null, + status: "running", + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + }, + ], + }, + }); + yield* interruptThreadTurn({ + commandId: CommandId.make("native-stop"), + threadId: v2ThreadId, + subagentId, + }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor)); + expect(commands).toEqual([ + { + type: "subagent.stop", + commandId: "native-stop", + threadId: v2ThreadId, + runId, + subagentId, + }, + ]); + }).pipe(Effect.provide(layerTestCrypto)), + ); + it.effect("interrupts a known run without fetching the full projection", () => 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 84f4a2ad3cff83e85b5e5d5d938fac5a7abe9178..aec86014071e530bd0fbf82e30eaa5817f5a24df 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -10,6 +10,7 @@ import { type ChatAttachment, type MessageId, type ModelSelection, + type NodeId, type OrchestrationV2Command, type OrchestrationV2CreationSource, type PlanId, @@ -183,6 +184,7 @@ export interface StartThreadTurnInput extends ThreadCommandInput { export interface InterruptThreadTurnInput extends ThreadCommandInput { readonly runId?: RunId; + readonly subagentId?: NodeId; /** Temporary caller compatibility while UI naming moves from turns to runs. */ readonly turnId?: string; } @@ -780,6 +782,18 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThreadTurn")(function* ( input: InterruptThreadTurnInput, ) { + if (input.subagentId !== undefined) { + const projection = yield* getProjection(input.threadId); + const subagent = projection.subagents.find((agent) => agent.id === input.subagentId); + if (subagent?.runId == null) return { sequence: 0 }; + return yield* dispatch({ + type: "subagent.stop", + commandId: yield* allocateCommandId(input), + threadId: input.threadId, + runId: subagent.runId, + subagentId: subagent.id, + }); + } let runId = input.runId ?? (input.turnId as RunId | undefined); if (runId === undefined) { const projection = yield* getProjection(input.threadId); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 7aa8c74fd80a5e69c67d5503cf9496dd427fe18a..865077ef9d2bfb9b31bc7e8a999563bc68fe9c9f 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2915,6 +2915,14 @@ export const OrchestrationV2Command = Schema.Union([ threadId: ThreadId, runId: RunId, }), + /** Stops one provider-native child while its owning run keeps working. */ + Schema.Struct({ + type: Schema.Literal("subagent.stop"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + subagentId: NodeId, + }), Schema.Struct({ type: Schema.Literal("run.interrupt"), commandId: CommandId, diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index ac5cd5fe8ab32d5e9b728ed7fe6a5a6c8e6f53b3..9094e8989fa46f05110aee9319804ac740b869fc 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -598,6 +598,11 @@ export interface ProviderAdapterV2SessionRuntime { readonly interruptTurn: ( input: ProviderAdapterV2InterruptInput, ) => Effect.Effect; + /** Stops one native child through its owning runtime; the owner keeps running. */ + readonly stopSubagent?: (input: { + readonly providerThread: OrchestrationV2ProviderThread; + readonly nativeTaskId: string; + }) => Effect.Effect; /** * Lets a runtime shared by several app threads unload one provider thread's * native state (and its MCP servers) when that app thread detaches, while