fix(server): stop actually stops a runaway Claude turn

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );
 

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -95,8 +95,17 @@ class FakeClaudeQuery implements AsyncIterable<SDKMessage> {
  }
  }
 
+ /**
+ * Models the real SDK's interrupt, which is a control request that can block
+ * until the in-flight tool call returns. Set to make it never answer.
+ */
+ public interruptNeverResolves = false;
+
  readonly interrupt = async (): Promise<void> => {
  this.interruptCalls.push(undefined);
+ if (this.interruptNeverResolves) {
+ await new Promise<never>(() => {});
+ }
  };
 
  readonly stopTask = async (taskId: string): Promise<void> => {
@@ -1665,6 +1674,97 @@ describe("ClaudeAdapterLive", () => {
  );
  });
 
+ it.effect("interruptTurn stops the session when the interrupt never lands", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnStartedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.started"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+ const exitedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "session.exited"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work forever",
+ attachments: [],
+ });
+ yield* Fiber.join(turnStartedFiber);
+
+ // The SDK's interrupt() blocks behind an in-flight tool call and never
+ // answers. Stop must not block on it, and must still stop the turn.
+ harness.query.interruptNeverResolves = true;
+ const interruptFiber = yield* adapter.interruptTurn(session.threadId).pipe(Effect.forkChild);
+
+ yield* TestClock.adjust("5 seconds");
+ yield* Fiber.join(interruptFiber);
+ assert.equal(harness.query.interruptCalls.length, 1);
+
+ const exitEvents = Array.from(yield* Fiber.join(exitedFiber));
+ assert.equal(exitEvents.length, 1);
+ assert.equal(harness.query.closeCalls, 1);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
+ it.effect("interruptTurn leaves the session open when the turn actually ends", () => {
+ const harness = makeHarness();
+ return Effect.gen(function* () {
+ const adapter = yield* ClaudeAdapter;
+
+ const turnCompletedFiber = yield* adapter.streamEvents.pipe(
+ Stream.filter((event) => event.type === "turn.completed"),
+ Stream.take(1),
+ Stream.runCollect,
+ Effect.forkChild,
+ );
+
+ const session = yield* adapter.startSession({
+ threadId: THREAD_ID,
+ provider: ProviderDriverKind.make("claudeAgent"),
+ runtimeMode: "full-access",
+ });
+ yield* adapter.sendTurn({
+ threadId: session.threadId,
+ input: "work",
+ attachments: [],
+ });
+ yield* adapter.interruptTurn(session.threadId);
+
+ harness.query.emit({
+ type: "result",
+ subtype: "success",
+ is_error: false,
+ errors: [],
+ session_id: "sdk-session",
+ uuid: "result-interrupted",
+ } as unknown as SDKMessage);
+ yield* Fiber.join(turnCompletedFiber);
+
+ // The turn ended, so the watchdog must leave the session alone.
+ yield* TestClock.adjust("30 seconds");
+ assert.equal(harness.query.closeCalls, 0);
+ }).pipe(
+ Effect.provideService(Random.Random, makeDeterministicRandomService()),
+ Effect.provide(harness.layer),
+ );
+ });
+
  it.effect("workflow member coalescing: identical snapshots suppress, changes emit", () => {
  const harness = makeHarness();
  return Effect.gen(function* () {

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -99,6 +99,21 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonStri
 const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown));
 
 const PROVIDER = ProviderDriverKind.make("claudeAgent");
+
+/**
+ * How long the SDK gets to acknowledge an interrupt control request. Its
+ * `interrupt()` can block until the in-flight tool call returns, which stalls
+ * every other thread command behind it on the provider reactor's queue.
+ */
+const INTERRUPT_ACK_TIMEOUT = "5 seconds";
+
+/**
+ * How long the interrupted turn gets to wind down before the session is torn
+ * down. An acknowledged interrupt is not proof the turn stopped, so Stop
+ * escalates rather than trusting the provider.
+ */
+const INTERRUPT_HARD_STOP_DELAY = "5 seconds";
+
 type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
 type ClaudeToolResultStreamKind = Extract<
  RuntimeContentStreamKind,
@@ -1689,6 +1704,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  }) as ClaudeQueryRuntime);
 
  const sessions = new Map<ThreadId, ClaudeSessionContext>();
+ // Detached forks that must outlive the request that started them (the Stop
+ // watchdog): the caller returns as soon as the interrupt is requested.
+ const adapterContext = yield* Effect.context<never>();
+ const runAdapterFork = Effect.runForkWith(adapterContext);
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
 
  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
@@ -4518,10 +4537,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
  { concurrency: 8, discard: true },
  ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
  }
+ // Stop has to actually stop. interrupt() is only a control request: the
+ // SDK can take minutes to answer it while a tool call is in flight, and
+ // an answer still does not prove the turn ended. Arm the teardown first
+ // so a slow ack cannot delay it, then ask nicely.
+ const interruptedTurnId = context.turnState?.turnId;
+ if (interruptedTurnId !== undefined) {
+ runAdapterFork(
+ Effect.sleep(INTERRUPT_HARD_STOP_DELAY).pipe(
+ Effect.flatMap(() =>
+ context.stopped || context.turnState?.turnId !== interruptedTurnId
+ ? Effect.void
+ : Effect.logWarning("Claude turn ignored its interrupt; stopping the session.", {
+ threadId,
+ turnId: interruptedTurnId,
+ }).pipe(Effect.andThen(stopSessionInternal(context, { emitExitEvent: true }))),
+ ),
+ Effect.catchCause((cause) =>
+ Effect.logError("Failed to force-stop the Claude session.", { threadId, cause }),
+ ),
+ ),
+ );
+ }
+
  yield* Effect.tryPromise({
  try: () => context.query.interrupt(),
  catch: (cause) => toRequestError(threadId, "turn/interrupt", cause),
- });
+ }).pipe(Effect.timeoutOption(INTERRUPT_ACK_TIMEOUT));
  },
  );