authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-10 09:50:11-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 09:50:11-05:00
log8c777fbfade340f2092c011483c7fb0cde44c675
tree0279d4930624b58090a5c827559dfd16d393c0a3
parent98beed1a226c42b85c52ff5ee9d5dbb67d776a77
signature Signed by PGP key B5690EEEBB952194

feat(server): stop Claude subagents without stopping their owner (#17826)

15 files changed, 542 insertions(+), 19 deletions(-)

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts+109
......@@ -2130,6 +2130,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
21302130 const makeWakeHarnessWithOptions = (options?: {
21312131 readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>;
21322132 readonly interrupt?: Effect.Effect<void>;
2133 readonly stopTask?: (taskId: string) => Effect.Effect<void>;
21332134 readonly environment?: NodeJS.ProcessEnv;
21342135 readonly settings?: ClaudeSettings;
21352136 // A CLI process opened after the first streams from its own queue, so the
......@@ -2155,6 +2156,10 @@ describe("ClaudeAdapterV2 background wake turns", () => {
21552156 const permissionModeChanges: Array<string> = [];
21562157 const continuationRequests: Array<ProviderContinuationRequests.ProviderContinuationRequest> =
21572158 [];
2159 const subagentReceipts =
2160 yield* Queue.unbounded<
2161 Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "subagent.updated" }>
2162 >();
21582163 const terminalReceipts =
21592164 yield* Queue.unbounded<
21602165 Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "turn.terminal" }>
......@@ -2227,6 +2232,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
22272232 Effect.sync(() => {
22282233 permissionModeChanges.push(mode);
22292234 }),
2235 ...(options?.stopTask === undefined ? {} : { stopTask: options.stopTask }),
22302236 interrupt: options?.interrupt ?? Effect.void,
22312237 close: options?.close?.(sdkMessages) ?? Effect.void,
22322238 };
......@@ -2253,6 +2259,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
22532259 Stream.runForEach((event) =>
22542260 Effect.gen(function* () {
22552261 events.push(event);
2262 if (event.type === "subagent.updated") yield* Queue.offer(subagentReceipts, event);
22562263 if (event.type === "turn.terminal") {
22572264 yield* Queue.offer(terminalReceipts, event);
22582265 }
......@@ -2292,6 +2299,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
22922299 continuationRequests,
22932300 events,
22942301 terminalReceipts,
2302 subagentReceipts,
22952303 systemNoticeReceipts,
22962304 getOpenedOptions: () => openedOptions,
22972305 terminalEvents,
......@@ -8076,6 +8084,107 @@ describe("ClaudeAdapterV2 background wake turns", () => {
80768084 ),
80778085 );
80788086
8087 it.effect.each([false, true])(
8088 "stops one native subagent while its sibling keeps running, owner settled %s",
8089 (ownerSettled) =>
8090 Effect.scoped(
8091 Effect.gen(function* () {
8092 const stopped: string[] = [];
8093 let closes = 0;
8094 const harness = yield* makeWakeHarnessWithOptions({
8095 stopTask: (taskId) =>
8096 Effect.sync(() => {
8097 stopped.push(taskId);
8098 }),
8099 interrupt: Effect.die("A child stop must not interrupt its owner"),
8100 close: () =>
8101 Effect.sync(() => {
8102 closes++;
8103 }),
8104 });
8105 yield* harness.runtime.startTurn(
8106 makeClaudeTestTurnInput({
8107 threadId: harness.threadId,
8108 providerThread: harness.providerThread,
8109 now: yield* DateTime.now,
8110 attemptId: RunAttemptId.make("attempt-native-child-stop"),
8111 text: "Start two agents.",
8112 attachments: [],
8113 }),
8114 );
8115 for (const [taskId, toolUseId, uuid] of [
8116 ["agent-stop", "toolu-stop", "00000000-0000-4000-8000-000000000801"],
8117 ["agent-keep", "toolu-keep", "00000000-0000-4000-8000-000000000802"],
8118 ]) {
8119 yield* harness.offerAndWait(
8120 makeSubagentTaskStartedFrame({ taskId: taskId!, toolUseId: toolUseId!, uuid: uuid! }),
8121 );
8122 yield* Queue.take(harness.subagentReceipts);
8123 }
8124 if (ownerSettled) {
8125 yield* harness.offerAndWait(
8126 makeResultFrame({
8127 uuid: "00000000-0000-4000-8000-000000000803",
8128 result: "Agents are working.",
8129 }),
8130 );
8131 yield* Queue.take(harness.terminalReceipts);
8132 }
8133 const stop = harness.runtime.stopSubagent;
8134 assert.isDefined(stop);
8135 yield* stop!({ providerThread: harness.providerThread, nativeTaskId: "agent-stop" });
8136 assert.deepEqual(stopped, ["agent-stop"]);
8137 yield* harness.offerAndWait(
8138 claudeSdkFrame({
8139 ...makeSubagentNotificationFrame({
8140 taskId: "agent-stop",
8141 toolUseId: "toolu-stop",
8142 summary: "Stopped by user",
8143 uuid: "00000000-0000-4000-8000-000000000804",
8144 }),
8145 status: "stopped",
8146 }),
8147 );
8148 if (ownerSettled) {
8149 yield* harness.runtime.startTurn(
8150 makeClaudeTestTurnInput({
8151 threadId: harness.threadId,
8152 providerThread: harness.providerThread,
8153 now: yield* DateTime.now,
8154 attemptId: RunAttemptId.make("attempt-native-child-stop-wake"),
8155 text: "Background task stopped.",
8156 attachments: [],
8157 providerTurnOrdinal: 2,
8158 messageCreatedBy: "agent",
8159 messageCreationSource: "provider",
8160 }),
8161 );
8162 }
8163 const stoppedEvent = yield* Queue.take(harness.subagentReceipts);
8164 assert.equal(stoppedEvent.subagent.status, "cancelled");
8165 const tasks = harness.events.filter((event) => event.type === "subagent.updated");
8166 assert.equal(
8167 tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-stop")
8168 ?.subagent.status,
8169 "cancelled",
8170 );
8171 assert.equal(
8172 tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-keep")
8173 ?.subagent.status,
8174 "running",
8175 );
8176 assert.lengthOf(harness.terminalEvents(), ownerSettled ? 1 : 0);
8177 yield* stop!({ providerThread: harness.providerThread, nativeTaskId: "agent-stop" });
8178 assert.deepEqual(stopped, ["agent-stop"]);
8179 assert.equal(closes, 0);
8180 }).pipe(
8181 Effect.provide(
8182 Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
8183 ),
8184 ),
8185 ),
8186 );
8187
80798188 it.effect.each(["requested", "observed-before", "observed-after", "inherit", "unknown"] as const)(
80808189 "records the subagent model from %s without inheriting the parent override",
80818190 (source) =>
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+34
......@@ -342,6 +342,7 @@ export interface ClaudeAgentSdkQuerySession {
342342 readonly setPermissionMode: (
343343 mode: PermissionMode,
344344 ) => Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
345 readonly stopTask?: (taskId: string) => Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
345346 readonly interrupt: Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
346347 readonly close: Effect.Effect<void, ClaudeAgentSdkQueryRunnerError>;
347348}
......@@ -734,6 +735,11 @@ export const layerQueryRunner: Layer.Layer<
734735 }),
735736 ),
736737 ),
738 stopTask: (taskId) =>
739 Effect.tryPromise({
740 try: () => queryRuntime.stopTask(taskId),
741 catch: (cause) => queryRunnerError(cause, "stopTask"),
742 }),
737743 interrupt: Effect.tryPromise({
738744 try: () => queryRuntime.interrupt(),
739745 catch: (cause) => queryRunnerError(cause, "interrupt"),
......@@ -8092,6 +8098,34 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
80928098 }),
80938099 steerTurn,
80948100 interruptTurn,
8101 stopSubagent: Effect.fn("ClaudeAdapterV2.stopSubagent")(function* (input) {
8102 const existing = yield* Ref.get(queryContext);
8103 const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId;
8104 const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId);
8105 if (subagent === undefined || subagent.task.status !== "running") return;
8106 if (
8107 existing === null ||
8108 existing.nativeThreadId !== nativeThreadId ||
8109 subagent.task.threadId !== input.providerThread.appThreadId ||
8110 existing.subagentsFromEarlierProcesses.has(subagent) ||
8111 existing.query.stopTask === undefined
8112 ) {
8113 return yield* new ProviderAdapter.ProviderAdapterProtocolError({
8114 driver: CLAUDE_PROVIDER,
8115 detail: "Claude subagent has no live owning query with task stopping support.",
8116 });
8117 }
8118 yield* existing.query.stopTask(input.nativeTaskId).pipe(
8119 Effect.mapError(
8120 (cause) =>
8121 new ProviderAdapter.ProviderAdapterProtocolError({
8122 driver: CLAUDE_PROVIDER,
8123 detail: "Claude native subagent stop failed.",
8124 cause,
8125 }),
8126 ),
8127 );
8128 }),
80958129 respondToRuntimeRequest: Effect.fn("ClaudeAdapterV2.respondToRuntimeRequest")(
80968130 function* (requestInput) {
80978131 const pending = (yield* Ref.get(pendingRuntimeRequests)).get(
apps/server/src/orchestration-v2/EffectOutbox.ts+2
......@@ -3,6 +3,7 @@ import {
33 CheckpointScopeId,
44 CommandId,
55 MessageId,
6 NodeId,
67 ProviderSessionId,
78 RunAttemptId,
89 ProviderApprovalDecision,
......@@ -48,6 +49,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
4849 providerSessionId: ProviderSessionId,
4950 providerThreadId: ProviderThreadId,
5051 providerTurnId: ProviderTurnId,
52 subagent: Schema.optional(Schema.Struct({ id: NodeId, nativeTaskId: Schema.String })),
5153 }),
5254 Schema.Struct({
5355 type: Schema.Literal("provider-turn.steer"),
apps/server/src/orchestration-v2/EffectWorker.test.ts+30
......@@ -1,6 +1,7 @@
11import { assert, it } from "@effect/vitest";
22import {
33 CommandId,
4 NodeId,
45 ProviderSessionId,
56 ProviderThreadId,
67 ProviderTurnId,
......@@ -837,3 +838,32 @@ it.effect("settles a delegated child once its restart continuation fails for goo
837838 }).pipe(Effect.provide(layer));
838839 }),
839840);
841
842it.effect("a native subagent stop does not settle its owner's background work", () =>
843 Effect.gen(function* () {
844 const now = yield* DateTime.now;
845 const events = yield* Ref.make<ReadonlyArray<string>>([]);
846 const layer = layerExecutorFor({
847 events,
848 interrupt: (input) => {
849 assert.equal(input.subagent?.nativeTaskId, "native-task");
850 return Ref.update(events, (current) => [...current, "stop-child"]);
851 },
852 threads: { dispatch: () => Effect.die("The owner's background work must stay running") },
853 });
854 yield* Effect.gen(function* () {
855 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
856 yield* executor.execute({
857 ...restartEffect(now, { type: "detach" }),
858 request: {
859 type: "provider-turn.interrupt",
860 providerSessionId: oldSessionId,
861 providerThreadId,
862 providerTurnId,
863 subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" },
864 },
865 });
866 }).pipe(Effect.provide(layer));
867 assert.deepEqual(yield* Ref.get(events), ["stop-child"]);
868 }),
869);
apps/server/src/orchestration-v2/EffectWorker.ts+12-7
......@@ -169,6 +169,9 @@ export const layerExecutor: Layer.Layer<
169169 providerSessionId: effect.request.providerSessionId,
170170 providerThreadId: effect.request.providerThreadId,
171171 providerTurnId: effect.request.providerTurnId,
172 ...(effect.request.subagent === undefined
173 ? {}
174 : { subagent: effect.request.subagent }),
172175 })
173176 .pipe(
174177 Effect.catch((cause) =>
......@@ -185,13 +188,15 @@ export const layerExecutor: Layer.Layer<
185188 // One Stop can interrupt several provider threads, so the
186189 // settle is keyed by effect, not by the Stop command.
187190 Effect.andThen(
188 threads.dispatch({
189 type: "thread.background-work.settle",
190 commandId: CommandId.make(`${effect.id}:background-work-settled`),
191 threadId: effect.threadId,
192 providerThreadId: effect.request.providerThreadId,
193 providerTurnId: effect.request.providerTurnId,
194 }),
191 effect.request.subagent !== undefined
192 ? Effect.void
193 : threads.dispatch({
194 type: "thread.background-work.settle",
195 commandId: CommandId.make(`${effect.id}:background-work-settled`),
196 threadId: effect.threadId,
197 providerThreadId: effect.request.providerThreadId,
198 providerTurnId: effect.request.providerTurnId,
199 }),
195200 ),
196201 Effect.mapError(
197202 (cause) =>
apps/server/src/orchestration-v2/Orchestrator.ts+89-1
......@@ -463,6 +463,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId {
463463 case "prepared-run.progress":
464464 case "prepared-run.fail":
465465 case "prepared-run.retry":
466 case "subagent.stop":
466467 case "run.interrupt":
467468 case "queued-message.promote-to-steer":
468469 case "queue.resume":
......@@ -8821,6 +8822,89 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
88218822 });
88228823 });
88238824
8825 const dispatchSubagentStop = (
8826 command: Extract<OrchestrationV2Command, { readonly type: "subagent.stop" }>,
8827 events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
8828 effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
8829 ) =>
8830 Effect.gen(function* () {
8831 const projection = yield* loadProjectionForCommand(command, [
8832 "runs",
8833 "subagents",
8834 "providerThreads",
8835 "providerTurns",
8836 ]);
8837 const run = projection.runs.find((candidate) => candidate.id === command.runId);
8838 const subagent = projection.subagents.find(
8839 (candidate) => candidate.id === command.subagentId,
8840 );
8841 if (
8842 subagent === undefined ||
8843 subagent.runId !== command.runId ||
8844 subagent.origin !== "provider_native" ||
8845 subagent.driver !== "claudeAgent" ||
8846 subagent.nativeTaskRef?.strength !== "strong" ||
8847 subagent.nativeTaskRef.nativeId === null
8848 ) {
8849 return yield* new OrchestratorDispatchError({
8850 commandId: command.commandId,
8851 commandType: command.type,
8852 cause: "The requested native subagent cannot be stopped on this run.",
8853 });
8854 }
8855 if (!["pending", "running", "waiting"].includes(subagent.status)) return;
8856 const providerThread = projection.providerThreads.find(
8857 (candidate) => candidate.id === run?.providerThreadId,
8858 );
8859 const providerTurn = projection.providerTurns.findLast(
8860 (candidate) => candidate.providerThreadId === providerThread?.id,
8861 );
8862 if (
8863 providerThread?.providerSessionId == null ||
8864 providerTurn === undefined ||
8865 providerThread.driver !== subagent.driver ||
8866 providerThread.providerInstanceId !== subagent.providerInstanceId
8867 ) {
8868 return yield* new OrchestratorDispatchError({
8869 commandId: command.commandId,
8870 commandType: command.type,
8871 cause: "The requested native subagent has no owning provider session.",
8872 });
8873 }
8874 const providerSessionId = providerThread.providerSessionId;
8875 const nativeTaskId = subagent.nativeTaskRef.nativeId;
8876 const now = yield* DateTime.now;
8877 // Record the request with its effect; the provider notification supplies the outcome.
8878 yield* emit(
8879 events,
8880 command,
8881 )({
8882 type: "subagent.updated",
8883 threadId: command.threadId,
8884 runId: command.runId,
8885 nodeId: subagent.id,
8886 providerInstanceId: subagent.providerInstanceId,
8887 occurredAt: now,
8888 payload: { ...subagent, updatedAt: now },
8889 });
8890 yield* Ref.update(effects, (existing) => [
8891 ...existing,
8892 {
8893 id: `effect:${command.commandId}:subagent.stop:${subagent.id}`,
8894 commandId: command.commandId,
8895 threadId: command.threadId,
8896 request: {
8897 type: "provider-turn.interrupt",
8898 providerSessionId,
8899 providerThreadId: providerThread.id,
8900 providerTurnId: providerTurn.id,
8901 subagent: { id: subagent.id, nativeTaskId },
8902 },
8903 } satisfies PendingOrchestrationEffectV2,
8904 ]);
8905 return;
8906 });
8907
88248908 const dispatchRunInterrupt = (
88258909 command: Extract<OrchestrationV2Command, { readonly type: "run.interrupt" }>,
88268910 events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
......@@ -10526,6 +10610,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1052610610 case "thread.user-input.dismiss":
1052710611 yield* dispatchThreadUserInputDismiss(command, events, effects);
1052810612 break;
10613 case "subagent.stop":
10614 yield* dispatchSubagentStop(command, events, effects);
10615 break;
1052910616 case "run.interrupt":
1053010617 cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects);
1053110618 // Stop also stops every delegated task under the thread once it commits.
......@@ -10739,7 +10826,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1073910826 // its expected outcome, not a failure.
1074010827 planned.events.length > 0 ||
1074110828 command.type === "thread.background-work.settle" ||
10742 command.type === "thread.stop"
10829 command.type === "thread.stop" ||
10830 command.type === "subagent.stop"
1074310831 ? Effect.succeed(planned)
1074410832 : Effect.fail(
1074510833 new OrchestratorDispatchError({
apps/server/src/orchestration-v2/ProviderTurnControlService.ts+17
......@@ -1,6 +1,7 @@
11import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences";
22import {
33 MessageId,
4 type NodeId,
45 ProviderSessionId,
56 ProviderThreadId,
67 ProviderTurnId,
......@@ -43,6 +44,7 @@ const isProviderTurnControlError = Schema.is(ProviderTurnControlError);
4344
4445export interface ProviderTurnControlServiceV2Shape {
4546 readonly interrupt: (input: {
47 readonly subagent?: { readonly id: NodeId; readonly nativeTaskId: string };
4648 readonly threadId: ThreadId;
4749 readonly providerSessionId: ProviderSessionId;
4850 readonly providerThreadId: ProviderThreadId;
......@@ -177,6 +179,21 @@ export const layer: Layer.Layer<
177179 ? loaded.session
178180 : yield* sessions.get(input.providerSessionId);
179181 if (Option.isNone(session)) return;
182 if (input.subagent !== undefined) {
183 if (session.value.stopSubagent === undefined) {
184 return yield* new ProviderTurnControlError({
185 threadId: input.threadId,
186 operation: "interrupt",
187 providerTurnId: input.providerTurnId,
188 cause: "The provider does not support stopping a native subagent.",
189 });
190 }
191 yield* session.value.stopSubagent({
192 providerThread: loaded.providerThread,
193 nativeTaskId: input.subagent.nativeTaskId,
194 });
195 return;
196 }
180197 // A settled turn reaches its adapter too: only the adapter knows
181198 // whether it still runs work for the thread, and each one either
182199 // stops it or reports there is nothing left to stop. Background work
apps/server/src/orchestration-v2/ThreadStop.test.ts+119
......@@ -7,6 +7,7 @@ import {
77 ProjectId,
88 ProviderDriverKind,
99 ProviderInstanceId,
10 ProviderSessionId,
1011 ProviderTurnId,
1112 ThreadId,
1213 TurnItemId,
......@@ -20,6 +21,7 @@ import * as SqlClient from "effect/sql/SqlClient";
2021import * as TestClock from "effect/testing/TestClock";
2122
2223import * as SqlitePersistence from "../persistence/Sqlite.ts";
24import { CLAUDE_PROVIDER } from "./Adapters/ClaudeAdapterV2.ts";
2325import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
2426import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts";
2527import * 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
582584 assert.deepEqual(yield* threadState(threadId), { runs: ["running"], watched: [] });
583585 }).pipe(Effect.provide(layerTest)),
584586);
587
588it.effect(
589 "a native child stop persists only its targeted effect and leaves its owner's queue and watch alone",
590 () =>
591 Effect.gen(function* () {
592 const orchestrator = yield* Orchestrator.OrchestratorV2;
593 const projections = yield* ProjectionStore.ProjectionStoreV2;
594 const sql = yield* SqlClient.SqlClient;
595 const threadId = ThreadId.make("thread:native-child-stop");
596 yield* createWatchingThread(threadId, 20);
597 yield* send(threadId, "owner", "start_immediately");
598 yield* delegate(threadId, "native-child-fixture");
599 const before = yield* orchestrator.getThreadProjection(threadId);
600 const run = before.runs[0]!;
601 const now = yield* DateTime.now;
602 const subagent = {
603 ...before.subagents[0]!,
604 origin: "provider_native" as const,
605 driver: CLAUDE_PROVIDER,
606 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
607 nativeTaskRef: {
608 driver: CLAUDE_PROVIDER,
609 nativeId: "claude-task",
610 strength: "strong" as const,
611 },
612 status: "running" as const,
613 startedAt: now,
614 };
615 yield* projections.apply({
616 id: EventId.make("event:native-child-stop:agent"),
617 type: "subagent.updated",
618 threadId,
619 runId: run.id,
620 nodeId: subagent.id,
621 occurredAt: now,
622 payload: subagent,
623 });
624 const providerThread = before.providerThreads.find(
625 (thread) => thread.id === run.providerThreadId,
626 )!;
627 yield* projections.apply({
628 id: EventId.make("event:native-child-stop:provider-thread"),
629 type: "provider-thread.updated",
630 threadId,
631 occurredAt: now,
632 payload: {
633 ...providerThread,
634 driver: CLAUDE_PROVIDER,
635 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
636 providerSessionId: ProviderSessionId.make("session:native-child-stop"),
637 },
638 });
639 const providerTurnId = ProviderTurnId.make("turn:native-child-stop");
640 yield* projections.apply({
641 id: EventId.make("event:native-child-stop:provider-turn"),
642 type: "provider-turn.updated",
643 threadId,
644 occurredAt: now,
645 payload: {
646 id: providerTurnId,
647 providerThreadId: providerThread.id,
648 nodeId: run.rootNodeId!,
649 runAttemptId: run.activeAttemptId!,
650 nativeTurnRef: null,
651 ordinal: 1,
652 status: "completed",
653 startedAt: now,
654 completedAt: now,
655 },
656 });
657 // The owner has finished but its child and queued prompt remain.
658 yield* projections.apply({
659 id: EventId.make("event:native-child-stop:owner-completed"),
660 type: "run.updated",
661 threadId,
662 runId: run.id,
663 occurredAt: now,
664 payload: { ...run, status: "completed", startedAt: now, completedAt: now },
665 });
666 yield* send(threadId, "queued", "queue_after_active");
667 const stateBefore = yield* threadState(threadId);
668 const commandId = CommandId.make("stop-native-child");
669 yield* orchestrator.dispatch({
670 type: "subagent.stop",
671 commandId,
672 threadId,
673 runId: run.id,
674 subagentId: subagent.id,
675 });
676 assert.deepEqual(yield* threadState(threadId), stateBefore);
677 const effects = yield* sql<{ payload_json: string }>`
678 SELECT payload_json FROM orchestration_v2_effect_outbox WHERE command_id = ${commandId}
679 `;
680 assert.equal(effects.length, 1);
681 assert.deepEqual(JSON.parse(effects[0]!.payload_json), {
682 type: "provider-turn.interrupt",
683 providerSessionId: "session:native-child-stop",
684 providerThreadId: providerThread.id,
685 providerTurnId,
686 subagent: { id: subagent.id, nativeTaskId: "claude-task" },
687 });
688 assert.equal(
689 (yield* orchestrator.getThreadProjection(threadId)).subagents[0]?.status,
690 "running",
691 );
692 const unrelated = yield* Effect.exit(
693 orchestrator.dispatch({
694 type: "subagent.stop",
695 commandId: CommandId.make("stop-unknown-child"),
696 threadId,
697 runId: run.id,
698 subagentId: NodeId.make("unknown-child"),
699 }),
700 );
701 assert.isTrue(Exit.isFailure(unrelated));
702 }).pipe(Effect.provide(layerTest)),
703);
apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts+1
......@@ -166,6 +166,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray<Thread
166166 case "prepared-run.progress":
167167 case "prepared-run.fail":
168168 case "prepared-run.retry":
169 case "subagent.stop":
169170 case "run.interrupt":
170171 case "queued-message.promote-to-steer":
171172 case "queue.resume":
apps/web/src/components/chat/ThreadRelationshipsControl.agents.test.tsx+21-6
......@@ -53,9 +53,13 @@ afterEach(async () => {
5353 state.projection = null;
5454});
5555
56it.each(["codex", "claudeAgent"])(
57 "stops only active app-owned %s subagents without opening their thread",
58 async (driver) => {
56it.each([
57 { driver: "codex", origin: "app_owned" },
58 { driver: "claudeAgent", origin: "app_owned" },
59 { driver: "claudeAgent", origin: "provider_native" },
60])(
61 "stops active $origin $driver subagents without opening their thread",
62 async ({ driver, origin }) => {
5963 vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true);
6064 const parent = {
6165 id: "parent",
......@@ -70,7 +74,9 @@ it.each(["codex", "claudeAgent"])(
7074 const agent = {
7175 id: "agent",
7276 childThreadId: "child",
73 origin: "app_owned",
77 origin,
78 threadId: "parent",
79 nativeTaskRef: { driver, nativeId: "claude-task", strength: "strong" },
7480 driver,
7581 providerInstanceId: "codex",
7682 title: "Worker",
......@@ -106,7 +112,10 @@ it.each(["codex", "claudeAgent"])(
106112 await act(async () => stopButton().props.onClick());
107113 expect(state.command).toHaveBeenCalledWith({
108114 environmentId: "test",
109 input: { threadId: "child" },
115 input:
116 origin === "provider_native"
117 ? { threadId: "parent", subagentId: "agent" }
118 : { threadId: "child" },
110119 });
111120 expect(state.navigate).not.toHaveBeenCalled();
112121
......@@ -132,6 +141,12 @@ it.each(["codex", "claudeAgent"])(
132141 );
133142 state.projection = { ...projection, subagents: [{ ...agent, status: "completed" }] };
134143 await act(async () => renderer.update(cloneElement(panel)));
144 if (origin === "provider_native") {
145 expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(
146 0,
147 );
148 continue;
149 }
135150 await act(async () => stopButton().props.onClick());
136151 expect(state.command).toHaveBeenCalledTimes(1);
137152 expect(state.command).toHaveBeenLastCalledWith({
......@@ -152,7 +167,7 @@ it.each(["codex", "claudeAgent"])(
152167 expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(0);
153168 state.projection = {
154169 ...projection,
155 subagents: [{ ...agent, origin: "provider_native", driver: "claudeAgent" }],
170 subagents: [{ ...agent, origin: "provider_native", driver: "codex" }],
156171 };
157172 await act(async () => renderer.update(cloneElement(panel)));
158173 expect(renderer.root.findAllByProps({ "aria-label": "Stop subagent Worker" })).toHaveLength(0);
apps/web/src/components/chat/ThreadRelationshipsControl.tsx+26-5
......@@ -24,7 +24,12 @@ import {
2424 canDetachThreadProviderSession,
2525 resolveLatestMergeBackRun,
2626} from "@t3tools/client-runtime/state/thread-workflows";
27import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts";
27import type {
28 EnvironmentId,
29 OrchestrationV2Subagent,
30 OrchestrationV2ThreadShell,
31 ThreadId,
32} from "@t3tools/contracts";
2833import { deriveSubagentElapsedMs } from "@t3tools/shared/orchestrationTiming";
2934import { groupBy } from "effect/Array";
3035import * as DateTime from "effect/DateTime";
......@@ -221,6 +226,10 @@ export function ThreadRelationshipsPanel(props: {
221226 subagent.childThreadId,
222227 {
223228 ...projectedSubagentsToRuntime([subagent])[0]!,
229 id: subagent.id,
230 threadId: subagent.threadId,
231 nativeTaskRef: subagent.nativeTaskRef,
232 nativeStatus: subagent.status,
224233 driver: subagent.driver,
225234 providerInstanceId: subagent.providerInstanceId,
226235 origin: subagent.origin,
......@@ -328,12 +337,18 @@ export function ThreadRelationshipsPanel(props: {
328337 setBusyAction(null);
329338 };
330339
331 const stopSubagent = async (childThreadId: ThreadId) => {
340 const stopSubagent = async (
341 childThreadId: ThreadId,
342 agent: Pick<OrchestrationV2Subagent, "id" | "origin" | "threadId">,
343 ) => {
332344 if (stoppingThreadId !== null) return;
333345 setStoppingThreadId(childThreadId);
334346 const result = await interruptTurn({
335347 environmentId: props.environmentId,
336 input: { threadId: childThreadId },
348 input:
349 agent.origin === "provider_native"
350 ? { threadId: agent.threadId, subagentId: agent.id }
351 : { threadId: childThreadId },
337352 });
338353 setStoppingThreadId(null);
339354 if (result._tag === "Failure") {
......@@ -398,7 +413,13 @@ export function ThreadRelationshipsPanel(props: {
398413 );
399414 const failed = status === "failed" || status === "error";
400415 const canStop =
401 agent?.origin === "app_owned" &&
416 agent &&
417 (agent.origin === "app_owned" ||
418 (agent.origin === "provider_native" &&
419 agent.driver === "claudeAgent" &&
420 agent.nativeTaskRef?.strength === "strong" &&
421 agent.nativeTaskRef.nativeId !== null &&
422 ["pending", "running", "waiting"].includes(agent.nativeStatus))) &&
402423 agent.startedAt &&
403424 ["pending", "running", "waiting"].includes(agent.status);
404425 const trailingVisibilityClass = canStop
......@@ -566,7 +587,7 @@ export function ThreadRelationshipsPanel(props: {
566587 tone="destructive"
567588 aria-label={`Stop subagent ${threadTitle}`}
568589 disabled={stoppingThreadId !== null}
569 onClick={() => void stopSubagent(threadId)}
590 onClick={() => void stopSubagent(threadId, agent)}
570591 />
571592 }
572593 >
packages/client-runtime/src/operations/commands.test.ts+55
......@@ -10,6 +10,7 @@ import {
1010 PlanId,
1111 ProjectId,
1212 ProviderInstanceId,
13 ProviderDriverKind,
1314 RunId,
1415 RuntimeRequestId,
1516 ThreadId,
......@@ -995,6 +996,60 @@ describe("V2 environment commands", () => {
995996 }).pipe(Effect.provide(layerTestCrypto)),
996997 );
997998
999 it.effect(
1000 "a native subagent stop targets its owning run rather than holding the parent queue",
1001 () =>
1002 Effect.gen(function* () {
1003 const subagentId = NodeId.make("native-subagent");
1004 const runId = RunId.make("settled-owner-run");
1005 const commands: OrchestrationV2Command[] = [];
1006 const supervisor = yield* makeSupervisor({
1007 commands,
1008 projects: [],
1009 projection: {
1010 ...v2Projection,
1011 subagents: [
1012 {
1013 id: subagentId,
1014 threadId: v2ThreadId,
1015 runId,
1016 parentNodeId: NodeId.make("owner-node"),
1017 origin: "provider_native",
1018 createdBy: "agent",
1019 driver: ProviderDriverKind.make("claudeAgent"),
1020 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
1021 providerThreadId: null,
1022 childThreadId: ThreadId.make("native-child-thread"),
1023 nativeTaskRef: null,
1024 prompt: "Work",
1025 title: null,
1026 model: null,
1027 result: null,
1028 status: "running",
1029 startedAt: v2Now,
1030 completedAt: null,
1031 updatedAt: v2Now,
1032 },
1033 ],
1034 },
1035 });
1036 yield* interruptThreadTurn({
1037 commandId: CommandId.make("native-stop"),
1038 threadId: v2ThreadId,
1039 subagentId,
1040 }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
1041 expect(commands).toEqual([
1042 {
1043 type: "subagent.stop",
1044 commandId: "native-stop",
1045 threadId: v2ThreadId,
1046 runId,
1047 subagentId,
1048 },
1049 ]);
1050 }).pipe(Effect.provide(layerTestCrypto)),
1051 );
1052
9981053 it.effect("interrupts a known run without fetching the full projection", () =>
9991054 Effect.gen(function* () {
10001055 const commands: OrchestrationV2Command[] = [];
packages/client-runtime/src/operations/commands.ts+14
......@@ -10,6 +10,7 @@ import {
1010 type ChatAttachment,
1111 type MessageId,
1212 type ModelSelection,
13 type NodeId,
1314 type OrchestrationV2Command,
1415 type OrchestrationV2CreationSource,
1516 type PlanId,
......@@ -183,6 +184,7 @@ export interface StartThreadTurnInput extends ThreadCommandInput {
183184
184185export interface InterruptThreadTurnInput extends ThreadCommandInput {
185186 readonly runId?: RunId;
187 readonly subagentId?: NodeId;
186188 /** Temporary caller compatibility while UI naming moves from turns to runs. */
187189 readonly turnId?: string;
188190}
......@@ -780,6 +782,18 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")(
780782export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThreadTurn")(function* (
781783 input: InterruptThreadTurnInput,
782784) {
785 if (input.subagentId !== undefined) {
786 const projection = yield* getProjection(input.threadId);
787 const subagent = projection.subagents.find((agent) => agent.id === input.subagentId);
788 if (subagent?.runId == null) return { sequence: 0 };
789 return yield* dispatch({
790 type: "subagent.stop",
791 commandId: yield* allocateCommandId(input),
792 threadId: input.threadId,
793 runId: subagent.runId,
794 subagentId: subagent.id,
795 });
796 }
783797 let runId = input.runId ?? (input.turnId as RunId | undefined);
784798 if (runId === undefined) {
785799 const projection = yield* getProjection(input.threadId);
packages/contracts/src/orchestrationV2.ts+8
......@@ -2915,6 +2915,14 @@ export const OrchestrationV2Command = Schema.Union([
29152915 threadId: ThreadId,
29162916 runId: RunId,
29172917 }),
2918 /** Stops one provider-native child while its owning run keeps working. */
2919 Schema.Struct({
2920 type: Schema.Literal("subagent.stop"),
2921 commandId: CommandId,
2922 threadId: ThreadId,
2923 runId: RunId,
2924 subagentId: NodeId,
2925 }),
29182926 Schema.Struct({
29192927 type: Schema.Literal("run.interrupt"),
29202928 commandId: CommandId,
packages/provider-core/src/server/ProviderAdapter.ts+5
......@@ -598,6 +598,11 @@ export interface ProviderAdapterV2SessionRuntime {
598598 readonly interruptTurn: (
599599 input: ProviderAdapterV2InterruptInput,
600600 ) => Effect.Effect<void, ProviderAdapterV2Error>;
601 /** Stops one native child through its owning runtime; the owner keeps running. */
602 readonly stopSubagent?: (input: {
603 readonly providerThread: OrchestrationV2ProviderThread;
604 readonly nativeTaskId: string;
605 }) => Effect.Effect<void, ProviderAdapterV2Error>;
601606 /**
602607 * Lets a runtime shared by several app threads unload one provider thread's
603608 * native state (and its MCP servers) when that app thread detaches, while