History

fix(server): stop kills lingering Claude work (#5891)

Theo Browne committed 2 months ago

Commit 4d12e521 parent 3db38b82 files · +241 −125Browse files

Unified · Split

2 files
  1. apps/server/src/provider/Layers/ClaudeAdapter.test.ts +177 −14
  2. apps/server/src/provider/Layers/ClaudeAdapter.ts +64 −111
apps/server/src/provider/Layers/ClaudeAdapter.test.ts+177−14
@@ -54,12 +54,11 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> { · 53 unmodified lines
5454 private done = false;
5555 private failure: unknown | undefined;
5656
57− public readonly interruptCalls: Array<void> = [];
58− public readonly stopTaskCalls: Array<string> = [];
5957 public readonly setModelCalls: Array<string | undefined> = [];
6058 public readonly setPermissionModeCalls: Array<string> = [];
6159 public readonly setMaxThinkingTokensCalls: Array<number | null> = [];
6260 public closeCalls = 0;
61+ public closeError: unknown | undefined;
6362
6463 emit(message: SDKMessage): void {
6564 if (this.done) {
@@ -95,14 +94,6 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> { · 29 unmodified lines
9594 }
9695 }
9796
98− readonly interrupt = async (): Promise<void> => {
99− this.interruptCalls.push(undefined);
100− };
101−
102− readonly stopTask = async (taskId: string): Promise<void> => {
103− this.stopTaskCalls.push(taskId);
104− };
105−
10697 readonly setModel = async (model?: string): Promise<void> => {
10798 this.setModelCalls.push(model);
10899 };
@@ -117,6 +108,9 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> { · 8 unmodified lines
117108
118109 readonly close = (): void => {
119110 this.closeCalls += 1;
111+ if (this.closeError !== undefined) {
112+ throw this.closeError;
113+ }
120114 this.finish();
121115 };
122116
@@ -1580,7 +1574,7 @@ describe("ClaudeAdapterLive", () => { · 1457 unmodified lines
15801574 );
15811575 });
15821576
1583− it.effect("interruptTurn settles every acknowledged live task before interrupting", () => {
1577+ it.effect("interruptTurn settles live tasks and closes the provider session", () => {
15841578 const harness = makeHarness();
15851579 return Effect.gen(function* () {
15861580 const adapter = yield* ClaudeAdapter;
@@ -1645,9 +1639,12 @@ describe("ClaudeAdapterLive", () => { · 58 unmodified lines
16451639 );
16461640 yield* adapter.interruptTurn(session.threadId);
16471641
1648− // Only the still-live task is stopped; interrupt always fires after.
1649− assert.deepEqual(harness.query.stopTaskCalls, ["task-live"]);
1650− assert.equal(harness.query.interruptCalls.length, 1);
1642+ // Closing the session is the hard stop because SDK interrupt can leave
1643+ // resumed background work alive.
1644+ assert.equal(harness.query.closeCalls, 1);
1645+
1646+ const sessions = yield* adapter.listSessions();
1647+ assert.equal(sessions.length, 0);
16511648
16521649 const stoppedTaskEvents = Array.from(yield* Fiber.join(stoppedTaskEventFiber));
16531650 assert.equal(stoppedTaskEvents.length, 1);
@@ -1665,6 +1662,172 @@ describe("ClaudeAdapterLive", () => { · 11 unmodified lines
16651662 );
16661663 });
16671664
1665+ it.effect("keeps the session available when process close fails", () => {
1666+ const harness = makeHarness();
1667+ return Effect.gen(function* () {
1668+ const adapter = yield* ClaudeAdapter;
1669+ const session = yield* adapter.startSession({
1670+ threadId: THREAD_ID,
1671+ provider: ProviderDriverKind.make("claudeAgent"),
1672+ runtimeMode: "full-access",
1673+ });
1674+ harness.query.closeError = new Error("close failed");
1675+
1676+ const result = yield* adapter.interruptTurn(session.threadId).pipe(Effect.result);
1677+
1678+ assert.equal(result._tag, "Failure");
1679+ if (result._tag === "Failure") {
1680+ assert.equal(result.failure._tag, "ProviderAdapterProcessError");
1681+ }
1682+ assert.equal(harness.query.closeCalls, 1);
1683+ assert.equal(yield* adapter.hasSession(session.threadId), true);
1684+ assert.equal((yield* adapter.listSessions())[0]?.status, "ready");
1685+ }).pipe(
1686+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
1687+ Effect.provide(harness.layer),
1688+ );
1689+ });
1690+
1691+ it.effect("stopAll attempts every session when one process close fails", () => {
1692+ const queries: FakeClaudeQuery[] = [];
1693+ const layer = Layer.effect(
1694+ ClaudeAdapter,
1695+ Effect.gen(function* () {
1696+ const claudeConfig = decodeClaudeSettings({});
1697+ return yield* makeClaudeAdapter(claudeConfig, {
1698+ createQuery: () => {
1699+ const query = new FakeClaudeQuery();
1700+ queries.push(query);
1701+ return query;
1702+ },
1703+ });
1704+ }),
1705+ ).pipe(
1706+ Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")),
1707+ Layer.provideMerge(ServerSettingsService.layerTest()),
1708+ Layer.provideMerge(NodeServices.layer),
1709+ );
1710+
1711+ return Effect.gen(function* () {
1712+ const adapter = yield* ClaudeAdapter;
1713+ yield* adapter.startSession({
1714+ threadId: THREAD_ID,
1715+ provider: ProviderDriverKind.make("claudeAgent"),
1716+ runtimeMode: "full-access",
1717+ });
1718+ yield* adapter.startSession({
1719+ threadId: RESUME_THREAD_ID,
1720+ provider: ProviderDriverKind.make("claudeAgent"),
1721+ runtimeMode: "full-access",
1722+ });
1723+ const firstQuery = queries[0];
1724+ if (!firstQuery) {
1725+ return;
1726+ }
1727+ firstQuery.closeError = new Error("close failed");
1728+
1729+ const result = yield* adapter.stopAll().pipe(Effect.result);
1730+
1731+ assert.equal(result._tag, "Failure");
1732+ assert.equal(queries[0]?.closeCalls, 1);
1733+ assert.equal(queries[1]?.closeCalls, 1);
1734+ assert.equal(yield* adapter.hasSession(THREAD_ID), true);
1735+ assert.equal(yield* adapter.hasSession(RESUME_THREAD_ID), false);
1736+ }).pipe(
1737+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
1738+ Effect.provide(layer),
1739+ );
1740+ });
1741+
1742+ it.effect("keeps a resumed replacement session during slow stop cleanup", () => {
1743+ const queries: FakeClaudeQuery[] = [];
1744+ let signalUsageStarted: () => void = () => undefined;
1745+ const usageStarted = new Promise<void>((resolve) => {
1746+ signalUsageStarted = resolve;
1747+ });
1748+ const layer = Layer.effect(
1749+ ClaudeAdapter,
1750+ Effect.gen(function* () {
1751+ const claudeConfig = decodeClaudeSettings({});
1752+ return yield* makeClaudeAdapter(claudeConfig, {
1753+ createQuery: () => {
1754+ const query = new FakeClaudeQuery();
1755+ if (queries.length === 0) {
1756+ Object.assign(query, {
1757+ getContextUsage: async () => {
1758+ signalUsageStarted();
1759+ return await new Promise<never>(() => undefined);
1760+ },
1761+ });
1762+ }
1763+ queries.push(query);
1764+ return query;
1765+ },
1766+ });
1767+ }),
1768+ ).pipe(
1769+ Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")),
1770+ Layer.provideMerge(ServerSettingsService.layerTest()),
1771+ Layer.provideMerge(NodeServices.layer),
1772+ );
1773+
1774+ return Effect.gen(function* () {
1775+ const adapter = yield* ClaudeAdapter;
1776+ const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 8).pipe(
1777+ Stream.runCollect,
1778+ Effect.forkChild,
1779+ );
1780+ const firstSession = yield* adapter.startSession({
1781+ threadId: THREAD_ID,
1782+ provider: ProviderDriverKind.make("claudeAgent"),
1783+ runtimeMode: "full-access",
1784+ });
1785+ yield* adapter.sendTurn({
1786+ threadId: firstSession.threadId,
1787+ input: "hello",
1788+ attachments: [],
1789+ });
1790+
1791+ const interruptFiber = yield* adapter
1792+ .interruptTurn(firstSession.threadId)
1793+ .pipe(Effect.forkChild);
1794+ yield* Effect.promise(() => usageStarted);
1795+ assert.equal(queries[0]?.closeCalls, 1);
1796+
1797+ const replacement = yield* adapter.startSession({
1798+ threadId: THREAD_ID,
1799+ provider: ProviderDriverKind.make("claudeAgent"),
1800+ runtimeMode: "full-access",
1801+ resumeCursor: firstSession.resumeCursor,
1802+ });
1803+ yield* TestClock.adjust("1 second");
1804+ yield* Fiber.join(interruptFiber);
1805+
1806+ const activeSessions = yield* adapter.listSessions();
1807+ const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber));
1808+ assert.equal(queries.length, 2);
1809+ assert.equal(queries[1]?.closeCalls, 0);
1810+ assert.equal(activeSessions.length, 1);
1811+ assert.deepEqual(activeSessions[0]?.resumeCursor, replacement.resumeCursor);
1812+ assert.deepEqual(
1813+ runtimeEvents
1814+ .filter((event) => event.type.startsWith("session."))
1815+ .map((event) => event.type),
1816+ [
1817+ "session.started",
1818+ "session.configured",
1819+ "session.state.changed",
1820+ "session.started",
1821+ "session.configured",
1822+ "session.state.changed",
1823+ ],
1824+ );
1825+ }).pipe(
1826+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
1827+ Effect.provide(layer),
1828+ );
1829+ });
1830+
16681831 it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
16691832 const harness = makeHarness();
16701833 return Effect.gen(function* () {
apps/server/src/provider/Layers/ClaudeAdapter.ts+64−111
@@ -282,9 +282,6 @@ interface ClaudeSessionContext { · 281 unmodified lines
282282 }
283283
284284 interface ClaudeQueryRuntime extends AsyncIterable<SDKMessage> {
285− readonly interrupt: () => Promise<void>;
286− /** SDK Query.stopTask — present on real queries; optional for test doubles. */
287− readonly stopTask?: (taskId: string) => Promise<void>;
288285 readonly setModel: (model?: string) => Promise<void>;
289286 readonly setPermissionMode: (mode: PermissionMode) => Promise<void>;
290287 readonly setMaxThinkingTokens: (maxThinkingTokens: number | null) => Promise<void>;
@@ -2106,13 +2103,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 1815 unmodified lines
21062103 } catch {
21072104 return undefined;
21082105 }
2109− });
2110− if (!usage) {
2106+ }).pipe(Effect.timeoutOption("1 second"));
2107+ if (Option.isNone(usage) || !usage.value) {
21112108 return undefined;
21122109 }
21132110
2114− context.lastKnownContextWindow = usage.maxTokens;
2115− return normalizeClaudeContextUsageApiSnapshot(usage, totalProcessedTokens);
2111+ context.lastKnownContextWindow = usage.value.maxTokens;
2112+ return normalizeClaudeContextUsageApiSnapshot(usage.value, totalProcessedTokens);
21162113 });
21172114
21182115 const emitProposedPlanCompleted = Effect.fn("emitProposedPlanCompleted")(function* (
@@ -3636,8 +3633,42 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 1517 unmodified lines
36363633 ) {
36373634 if (context.stopped) return;
36383635
3636+ // Schedule process termination before any cleanup that can wait on the
3637+ // provider. The SDK closes stdin, then escalates from SIGTERM to SIGKILL.
3638+ yield* Effect.try({
3639+ try: () => context.query.close(),
3640+ catch: (cause) =>
3641+ new ProviderAdapterProcessError({
3642+ provider: PROVIDER,
3643+ threadId: context.session.threadId,
3644+ detail: "Failed to close Claude runtime query.",
3645+ cause,
3646+ }),
3647+ });
3648+
36393649 context.stopped = true;
36403650
3651+ for (const taskId of Array.from(context.liveTaskIds)) {
3652+ if (!context.liveTaskIds.delete(taskId)) {
3653+ continue;
3654+ }
3655+ const stamp = yield* makeEventStamp();
3656+ yield* offerRuntimeEvent({
3657+ type: "task.completed",
3658+ eventId: stamp.eventId,
3659+ provider: PROVIDER,
3660+ createdAt: stamp.createdAt,
3661+ threadId: context.session.threadId,
3662+ ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}),
3663+ payload: {
3664+ taskId: RuntimeTaskId.make(taskId),
3665+ status: "stopped",
3666+ ...taskLinkageFor(context.taskAgents, taskId),
3667+ },
3668+ providerRefs: nativeProviderRefs(context),
3669+ });
3670+ }
3671+
36413672 for (const [requestId, pending] of context.pendingApprovals) {
36423673 yield* Deferred.succeed(pending.decision, "cancel");
36433674 const stamp = yield* makeEventStamp();
@@ -3676,26 +3707,6 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 32 unmodified lines
36763707 yield* Fiber.interrupt(streamFiber);
36773708 }
36783709
3679− yield* Effect.try({
3680− try: () => context.query.close(),
3681− catch: (cause) =>
3682− new ProviderAdapterProcessError({
3683− provider: PROVIDER,
3684− threadId: context.session.threadId,
3685− detail: "Failed to close Claude runtime query.",
3686− cause,
3687− }),
3688− }).pipe(
3689− Effect.catch((error) =>
3690− emitRuntimeError(context, "Failed to close Claude runtime query.", {
3691− errorTag: error._tag,
3692− provider: error.provider,
3693− threadId: error.threadId,
3694− detail: error.detail,
3695− }),
3696− ),
3697− );
3698−
36993710 const updatedAt = yield* nowIso;
37003711 context.session = {
37013712 ...context.session,
@@ -3704,7 +3715,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 2 unmodified lines
37043715 updatedAt,
37053716 };
37063717
3707− if (options?.emitExitEvent !== false) {
3718+ if (options?.emitExitEvent !== false && sessions.get(context.session.threadId) === context) {
37083719 const stamp = yield* makeEventStamp();
37093720 yield* offerRuntimeEvent({
37103721 type: "session.exited",
@@ -3720,7 +3731,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 9 unmodified lines
37203731 });
37213732 }
37223733
3723− sessions.delete(context.session.threadId);
3734+ if (sessions.get(context.session.threadId) === context) {
3735+ sessions.delete(context.session.threadId);
3736+ }
37243737 });
37253738
37263739 const requireSession = (
@@ -3765,16 +3778,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 38 unmodified lines
37653778 });
37663779 yield* stopSessionInternal(existingContext, {
37673780 emitExitEvent: false,
3768− }).pipe(
3769− // Replacement cleanup is best-effort: never block the new session on
3770− // either typed failures or unexpected defects from tearing down the old one.
3771− Effect.catchCause((cause) =>
3772− Effect.logWarning("claude.session.replace.stop-failed", {
3773− threadId: input.threadId,
3774− cause,
3775− }),
3776− ),
3777− );
3781+ });
37783782 }
37793783
37803784 const startedAt = yield* nowIso;
@@ -4466,62 +4470,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 685 unmodified lines
44664470 const interruptTurn: ClaudeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")(
44674471 function* (threadId, _turnId) {
44684472 const context = yield* requireSession(threadId);
4469− // Stop-everything semantics: users reach for Stop precisely when a
4470− // fleet ran away. interrupt() alone only ends the parent turn —
4471− // background subagents/shells keep running and keep burning tokens.
4472− // Stop every live task first (best-effort per task: one refusal must
4473− // not strand the rest or block the turn interrupt), then interrupt.
4474− if (context.query.stopTask && context.liveTaskIds.size > 0) {
4475− const liveIds = Array.from(context.liveTaskIds);
4476− // Bounded: a wedged child's stopTask promise may never settle
4477− // (Effect.ignore handles rejection, not non-resolution), and the
4478− // parent interrupt below MUST still run — Stop matters most during
4479− // runaway fleets (review finding). Per-task timeout keeps one hung
4480− // child from consuming the whole budget.
4481− yield* Effect.forEach(
4482− liveIds,
4483− (taskId) =>
4484− Effect.gen(function* () {
4485− const stopAcknowledged = yield* Effect.tryPromise({
4486− // Invoke through the query object: SDK methods rely on `this`.
4487− try: () => context.query.stopTask!(taskId),
4488− catch: () => undefined,
4489− }).pipe(
4490− Effect.timeoutOption("3 seconds"),
4491− Effect.orElseSucceed(() => Option.none()),
4492− );
4493− if (Option.isNone(stopAcknowledged) || !context.liveTaskIds.delete(taskId)) {
4494− return;
4495− }
4496−
4497− // stopTask only acknowledges the control request. Its separate
4498− // task_notification can lose the race with interrupt(), so make
4499− // the acknowledged stop authoritative for the durable UI state.
4500− const stamp = yield* makeEventStamp();
4501− yield* offerRuntimeEvent({
4502− type: "task.completed",
4503− eventId: stamp.eventId,
4504− provider: PROVIDER,
4505− createdAt: stamp.createdAt,
4506− threadId: context.session.threadId,
4507− ...(context.turnState
4508− ? { turnId: asCanonicalTurnId(context.turnState.turnId) }
4509− : {}),
4510− payload: {
4511− taskId: RuntimeTaskId.make(taskId),
4512− status: "stopped",
4513− ...taskLinkageFor(context.taskAgents, taskId),
4514− },
4515− providerRefs: nativeProviderRefs(context),
4516− });
4517− }).pipe(Effect.ignore),
4518− { concurrency: 8, discard: true },
4519− ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
4520− }
4521− yield* Effect.tryPromise({
4522− try: () => context.query.interrupt(),
4523− catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
4524− });
4473+ // interrupt() can acknowledge while resumed background tasks keep the
4474+ // CLI alive. Stop is a hard session boundary for Claude, so close the
4475+ // query and let the SDK escalate to SIGKILL when graceful exit fails.
4476+ yield* stopSessionInternal(context);
45254477 },
45264478 );
45274479
@@ -4594,25 +4546,26 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 66 unmodified lines
45944546 return context !== undefined && !context.stopped;
45954547 });
45964548
4597− const stopAll: ClaudeAdapterShape["stopAll"] = () =>
4598− Effect.forEach(
4599− sessions,
4600− ([, context]) =>
4601− stopSessionInternal(context, {
4602− emitExitEvent: true,
4603− }),
4604− { discard: true },
4549+ const stopSessions = Effect.fn("stopSessions")(function* (
4550+ contexts: ReadonlyArray<ClaudeSessionContext>,
4551+ emitExitEvent: boolean,
4552+ ) {
4553+ const results = yield* Effect.forEach(contexts, (context) =>
4554+ stopSessionInternal(context, { emitExitEvent }).pipe(Effect.result),
46054555 );
46064556
4557+ for (const result of results) {
4558+ if (result._tag === "Failure") {
4559+ return yield* Effect.fail(result.failure);
4560+ }
4561+ }
4562+ });
4563+
4564+ const stopAll: ClaudeAdapterShape["stopAll"] = () =>
4565+ stopSessions(Array.from(sessions.values()), true);
4566+
46074567 yield* Effect.addFinalizer(() =>
4608− Effect.forEach(
4609− sessions,
4610− ([, context]) =>
4611− stopSessionInternal(context, {
4612− emitExitEvent: false,
4613− }),
4614− { discard: true },
4615− ).pipe(
4568+ stopSessions(Array.from(sessions.values()), false).pipe(
46164569 Effect.catch((cause) =>
46174570 Effect.logError("Failed to emit Claude session shutdown event.", { cause }),
46184571 ),