authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-07 00:15:59-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-06 21:15:59-07:00
log6fa457607886caf096e7871b67e447f76d3772f6
treea70398db559e411d558b46a7a1b6e04569d8cf66
parentae7b27de824e890f2cdfc85018fc9301e7d45022
signature Signed by PGP key B5690EEEBB952194

fix(server): settle stopped Claude subagents (#5568)

2 files changed, 53 insertions(+), 6 deletions(-)

apps/server/src/provider/Layers/ClaudeAdapter.test.ts+18-1
...@@ -1511,7 +1511,7 @@ describe("ClaudeAdapterLive", () => {...@@ -1511,7 +1511,7 @@ describe("ClaudeAdapterLive", () => {
1511 );1511 );
1512 });1512 });
15131513
1514 it.effect("interruptTurn stops every live task before interrupting the turn", () => {1514 it.effect("interruptTurn settles every acknowledged live task before interrupting", () => {
1515 const harness = makeHarness();1515 const harness = makeHarness();
1516 return Effect.gen(function* () {1516 return Effect.gen(function* () {
1517 const adapter = yield* ClaudeAdapter;1517 const adapter = yield* ClaudeAdapter;
...@@ -1568,11 +1568,28 @@ describe("ClaudeAdapterLive", () => {...@@ -1568,11 +1568,28 @@ describe("ClaudeAdapterLive", () => {
15681568
1569 yield* Fiber.join(taskEventsFiber);1569 yield* Fiber.join(taskEventsFiber);
15701570
1571 const stoppedTaskEventFiber = yield* adapter.streamEvents.pipe(
1572 Stream.filter((event) => event.type === "task.completed"),
1573 Stream.take(1),
1574 Stream.runCollect,
1575 Effect.forkChild,
1576 );
1571 yield* adapter.interruptTurn(session.threadId);1577 yield* adapter.interruptTurn(session.threadId);
15721578
1573 // Only the still-live task is stopped; interrupt always fires after.1579 // Only the still-live task is stopped; interrupt always fires after.
1574 assert.deepEqual(harness.query.stopTaskCalls, ["task-live"]);1580 assert.deepEqual(harness.query.stopTaskCalls, ["task-live"]);
1575 assert.equal(harness.query.interruptCalls.length, 1);1581 assert.equal(harness.query.interruptCalls.length, 1);
1582
1583 const stoppedTaskEvents = Array.from(yield* Fiber.join(stoppedTaskEventFiber));
1584 assert.equal(stoppedTaskEvents.length, 1);
1585 const stoppedTaskEvent = stoppedTaskEvents[0];
1586 assert.equal(stoppedTaskEvent?.type, "task.completed");
1587 if (stoppedTaskEvent?.type === "task.completed") {
1588 assert.equal(String(stoppedTaskEvent.payload.taskId), "task-live");
1589 assert.equal(stoppedTaskEvent.payload.status, "stopped");
1590 assert.equal(stoppedTaskEvent.payload.taskType, "local_agent");
1591 assert.equal(stoppedTaskEvent.payload.title, "Agent A");
1592 }
1576 }).pipe(1593 }).pipe(
1577 Effect.provideService(Random.Random, makeDeterministicRandomService()),1594 Effect.provideService(Random.Random, makeDeterministicRandomService()),
1578 Effect.provide(harness.layer),1595 Effect.provide(harness.layer),
apps/server/src/provider/Layers/ClaudeAdapter.ts+35-5
...@@ -65,6 +65,7 @@ import * as Effect from "effect/Effect";...@@ -65,6 +65,7 @@ import * as Effect from "effect/Effect";
65import * as Exit from "effect/Exit";65import * as Exit from "effect/Exit";
66import * as FileSystem from "effect/FileSystem";66import * as FileSystem from "effect/FileSystem";
67import * as Fiber from "effect/Fiber";67import * as Fiber from "effect/Fiber";
68import * as Option from "effect/Option";
68import * as Path from "effect/Path";69import * as Path from "effect/Path";
69import * as Queue from "effect/Queue";70import * as Queue from "effect/Queue";
70import * as Ref from "effect/Ref";71import * as Ref from "effect/Ref";
...@@ -4419,11 +4420,40 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -4419,11 +4420,40 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
4419 yield* Effect.forEach(4420 yield* Effect.forEach(
4420 liveIds,4421 liveIds,
4421 (taskId) =>4422 (taskId) =>
4422 Effect.tryPromise({4423 Effect.gen(function* () {
4423 // Invoke through the query object: SDK methods rely on `this`.4424 const stopAcknowledged = yield* Effect.tryPromise({
4424 try: () => context.query.stopTask!(taskId),4425 // Invoke through the query object: SDK methods rely on `this`.
4425 catch: () => undefined,4426 try: () => context.query.stopTask!(taskId),
4426 }).pipe(Effect.timeoutOption("3 seconds"), Effect.ignore),4427 catch: () => undefined,
4428 }).pipe(
4429 Effect.timeoutOption("3 seconds"),
4430 Effect.orElseSucceed(() => Option.none()),
4431 );
4432 if (Option.isNone(stopAcknowledged) || !context.liveTaskIds.delete(taskId)) {
4433 return;
4434 }
4435
4436 // stopTask only acknowledges the control request. Its separate
4437 // task_notification can lose the race with interrupt(), so make
4438 // the acknowledged stop authoritative for the durable UI state.
4439 const stamp = yield* makeEventStamp();
4440 yield* offerRuntimeEvent({
4441 type: "task.completed",
4442 eventId: stamp.eventId,
4443 provider: PROVIDER,
4444 createdAt: stamp.createdAt,
4445 threadId: context.session.threadId,
4446 ...(context.turnState
4447 ? { turnId: asCanonicalTurnId(context.turnState.turnId) }
4448 : {}),
4449 payload: {
4450 taskId: RuntimeTaskId.make(taskId),
4451 status: "stopped",
4452 ...taskLinkageFor(context.taskAgents, taskId),
4453 },
4454 providerRefs: nativeProviderRefs(context),
4455 });
4456 }).pipe(Effect.ignore),
4427 { concurrency: 8, discard: true },4457 { concurrency: 8, discard: true },
4428 ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);4458 ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
4429 }4459 }