diff --git a/packages/agents/src/core/turn.abort-timeout.test.ts b/packages/agents/src/core/turn.abort-timeout.test.ts
index 0ce5a5dbe7..d0ca23b1d8 100644
--- a/packages/agents/src/core/turn.abort-timeout.test.ts
+++ b/packages/agents/src/core/turn.abort-timeout.test.ts
@@ -10,7 +10,8 @@ import type { ServerAgentStreamEvent } from './turn.js';
 import { Turn, AgentEventType, DEFAULT_AGENT_ID } from './turn.js';
 import { reportError } from '@vybestack/llxprt-code-core/utils/errorReporting.js';
 import type { ChatSession } from './chatSession.js';
-import { StreamEventType } from './chatSession.js';
+import { StreamEventType, type StreamEvent } from './chatSession.js';
+import type { StreamLivenessListener } from '@vybestack/llxprt-code-core/utils/streamIdleTimeout.js';
 import type { ContentBlock } from '@vybestack/llxprt-code-core/services/history/IContent.js';
 import { type MockedChatInstance, mockChunk } from './turn-test-helpers.js';
 import { flushEventLoop } from '../test-utils/eventLoop.js';
@@ -387,3 +388,504 @@ describe('Turn run - abort and idle timeout', () => {
     }
   });
 });
+
+// ─── Issue #3236: abort settles provider reads that ignore the signal ──────
+
+function captureProcessFailures(): { captured: unknown[]; stop: () => void } {
+  const captured: unknown[] = [];
+  const onUnhandledRejection = (reason: unknown): void => {
+    captured.push(reason);
+  };
+  const onUncaughtException = (error: unknown): void => {
+    captured.push(error);
+  };
+  process.on('unhandledRejection', onUnhandledRejection);
+  process.on('uncaughtException', onUncaughtException);
+  return {
+    captured,
+    stop: () => {
+      process.off('unhandledRejection', onUnhandledRejection);
+      process.off('uncaughtException', onUncaughtException);
+    },
+  };
+}
+
+interface StallingStreamHarness {
+  readonly stream: AsyncIterable<StreamEvent>;
+  /** Resolves once the never-settling read has been entered. */
+  readonly stallEntered: Promise<void>;
+  readonly returnSpy: ReturnType<typeof vi.fn>;
+  resolveStalledRead(result: IteratorResult<StreamEvent>): void;
+  rejectStalledRead(reason?: unknown): void;
+}
+
+/**
+ * A provider iterator whose read after the first `chunksBeforeStall` chunks
+ * NEVER settles on its own — it ignores the abort signal exactly like the
+ * non-cooperative transports behind issue #3236. The test settles the
+ * abandoned read explicitly afterwards to prove a late settlement can
+ * neither escape as an unhandled rejection nor emit additional events.
+ */
+function createStallingStream(
+  chunksBeforeStall: number,
+): StallingStreamHarness {
+  let readCount = 0;
+  let resolveStalled:
+    | ((result: IteratorResult<StreamEvent>) => void)
+    | undefined;
+  let rejectStalled: ((reason?: unknown) => void) | undefined;
+  let markStallEntered: () => void = () => {};
+  const stallEntered = new Promise<void>((resolve) => {
+    markStallEntered = resolve;
+  });
+  const returnSpy = vi.fn(
+    (): Promise<IteratorResult<StreamEvent>> =>
+      Promise.resolve({ value: undefined, done: true }),
+  );
+  const iterator: AsyncIterator<StreamEvent> = {
+    next: (): Promise<IteratorResult<StreamEvent>> => {
+      readCount += 1;
+      if (readCount <= chunksBeforeStall) {
+        return Promise.resolve({
+          done: false,
+          value: {
+            type: StreamEventType.CHUNK,
+            value: mockChunk({ text: `part ${readCount}` }),
+          },
+        });
+      }
+      markStallEntered();
+      return new Promise<IteratorResult<StreamEvent>>((resolve, reject) => {
+        resolveStalled = resolve;
+        rejectStalled = reject;
+      });
+    },
+    return: returnSpy,
+  };
+  return {
+    stream: { [Symbol.asyncIterator]: () => iterator },
+    stallEntered,
+    returnSpy,
+    resolveStalledRead: (result) => resolveStalled?.(result),
+    rejectStalledRead: (reason) => rejectStalled?.(reason),
+  };
+}
+
+function runTurnCollecting(
+  turn: Turn,
+  signal: AbortSignal,
+): { events: ServerAgentStreamEvent[]; done: Promise<void> } {
+  const events: ServerAgentStreamEvent[] = [];
+  const done = (async () => {
+    for await (const event of turn.run([{ text: 'test query' }], signal)) {
+      events.push(event);
+    }
+  })();
+  return { events, done };
+}
+
+function singleCancelledEventCount(events: ServerAgentStreamEvent[]): number {
+  return events.filter((event) => event.type === AgentEventType.UserCancelled)
+    .length;
+}
+
+/**
+ * Wraps an AbortSignal's addEventListener/removeEventListener so tests can
+ * observe how many 'abort' listeners are currently registered.
+ */
+function instrumentAbortListeners(signal: AbortSignal): {
+  pendingAbortListenerCount: () => number;
+} {
+  const pending = new Set<EventListener>();
+  // Bound AbortSignal.addEventListener is generic over event-map keys; the
+  // plain-string delegate below needs the degenerately-typed binding.
+  const originalAdd = signal.addEventListener.bind(signal) as (
+    type: string,
+    listener: EventListenerOrEventListenerObject | null,
+    options?: boolean | AddEventListenerOptions,
+  ) => void;
+  const originalRemove = signal.removeEventListener.bind(signal) as (
+    type: string,
+    listener: EventListenerOrEventListenerObject | null,
+    options?: boolean | AddEventListenerOptions,
+  ) => void;
+  signal.addEventListener = (
+    type: string,
+    listener: EventListenerOrEventListenerObject | null,
+    options?: boolean | AddEventListenerOptions,
+  ): void => {
+    if (type === 'abort' && typeof listener === 'function') {
+      pending.add(listener);
+    }
+    originalAdd(type, listener, options);
+  };
+  signal.removeEventListener = (
+    type: string,
+    listener: EventListenerOrEventListenerObject | null,
+    options?: boolean | AddEventListenerOptions,
+  ): void => {
+    if (type === 'abort' && typeof listener === 'function') {
+      pending.delete(listener);
+    }
+    originalRemove(type, listener, options);
+  };
+  return { pendingAbortListenerCount: () => pending.size };
+}
+
+/**
+ * Builds the issue #3236 test Turn over the shared stream mocks, with an
+ * optional ephemeral-settings source. Keeps config-key and Turn-constructor
+ * drift in one place across the tests below.
+ */
+function makeTurnWithConfig(
+  getEphemeralSetting?: (key: string) => unknown,
+): Turn {
+  const chatInstance: MockedChatInstance = {
+    sendMessageStream: mockSendMessageStream,
+    getHistory: mockGetHistory,
+    getConfig: () =>
+      getEphemeralSetting === undefined ? undefined : { getEphemeralSetting },
+  };
+  return new Turn(
+    chatInstance as unknown as ChatSession,
+    'prompt-id-1',
+    DEFAULT_AGENT_ID,
+    'test',
+  );
+}
+
+describe('Turn run — abort settles provider reads that ignore the abort signal (issue #3236)', () => {
+  let turn: Turn;
+
+  beforeEach(() => {
+    vi.resetAllMocks();
+    turn = makeTurnWithConfig();
+    mockGetHistory.mockReturnValue([]);
+    mockSendMessageStream.mockResolvedValue((async function* () {})());
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it('settles a default-config abort mid-read with exactly one UserCancelled, and a late chunk changes nothing', async () => {
+    const failures = captureProcessFailures();
+    try {
+      const abortController = new AbortController();
+      const stalled = createStallingStream(1);
+      mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+      const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+      await stalled.stallEntered;
+      // Default config: the inter-chunk watchdog is disabled, so this read is
+      // unbounded unless the turn races it against abort. No timers advance.
+      abortController.abort();
+
+      await done;
+
+      expect(events).toStrictEqual([
+        { type: AgentEventType.Content, value: 'part 1', traceId: undefined },
+        { type: AgentEventType.UserCancelled },
+      ]);
+      expect(singleCancelledEventCount(events)).toBe(1);
+
+      // The abandoned read settles late with another chunk: no additional
+      // events, no unhandled rejection, turn state unchanged.
+      stalled.resolveStalledRead({
+        done: false,
+        value: {
+          type: StreamEventType.CHUNK,
+          value: mockChunk({ text: 'late chunk' }),
+        },
+      });
+      await flushEventLoop();
+      expect(events).toHaveLength(2);
+      expect(singleCancelledEventCount(events)).toBe(1);
+      expect(failures.captured).toHaveLength(0);
+    } finally {
+      failures.stop();
+    }
+  });
+
+  it('emits no unhandled rejection when the abandoned read later rejects', async () => {
+    const failures = captureProcessFailures();
+    try {
+      const abortController = new AbortController();
+      const stalled = createStallingStream(1);
+      mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+      const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+      await stalled.stallEntered;
+      abortController.abort();
+      await done;
+
+      expect(events).toStrictEqual([
+        { type: AgentEventType.Content, value: 'part 1', traceId: undefined },
+        { type: AgentEventType.UserCancelled },
+      ]);
+
+      stalled.rejectStalledRead(new Error('late provider failure'));
+      await flushEventLoop();
+      expect(events).toHaveLength(2);
+      expect(singleCancelledEventCount(events)).toBe(1);
+      expect(failures.captured).toHaveLength(0);
+    } finally {
+      failures.stop();
+    }
+  });
+
+  it('abort wins the watchdog-active read without waiting for the inter-chunk guard to fire', async () => {
+    turn = makeTurnWithConfig((key) =>
+      key === 'stream-idle-timeout-ms' ? 30_000 : undefined,
+    );
+
+    const abortController = new AbortController();
+    const stalled = createStallingStream(1);
+    mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+    const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+    await stalled.stallEntered;
+    // Real timers, deliberately NOT advanced: with the 30s inter-chunk guard
+    // armed, completion can only come from the abort racing the pending
+    // read. Had the watchdog fired instead, the terminal event would be
+    // StreamIdleTimeout, not UserCancelled.
+    abortController.abort();
+    await done;
+
+    expect(events).toStrictEqual([
+      { type: AgentEventType.Content, value: 'part 1', traceId: undefined },
+      { type: AgentEventType.UserCancelled },
+    ]);
+    expect(singleCancelledEventCount(events)).toBe(1);
+  });
+
+  it('aborts the first unbounded read when the watchdog is fully disabled', async () => {
+    const failures = captureProcessFailures();
+    try {
+      turn = makeTurnWithConfig((key) =>
+        key === 'stream-first-response-timeout-ms' ? 0 : undefined,
+      );
+
+      const abortController = new AbortController();
+      const stalled = createStallingStream(0);
+      mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+      const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+      await stalled.stallEntered;
+      // First-response and inter-chunk guards are both disabled, so the very
+      // first read is unbounded unless the turn races it against abort.
+      abortController.abort();
+      await done;
+
+      expect(events).toStrictEqual([{ type: AgentEventType.UserCancelled }]);
+
+      stalled.rejectStalledRead(new DOMException('Aborted', 'AbortError'));
+      await flushEventLoop();
+      expect(failures.captured).toHaveLength(0);
+    } finally {
+      failures.stop();
+    }
+  });
+
+  it('aborts the default-config watchdog-active FIRST read with exactly one UserCancelled', async () => {
+    const failures = captureProcessFailures();
+    try {
+      const abortController = new AbortController();
+      // chunksBeforeStall 0: the very first next() never settles, so the
+      // acquisition sits in the watchdog-active branch under the default
+      // 5-minute first-response guard.
+      const stalled = createStallingStream(0);
+      mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+      const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+      await stalled.stallEntered;
+      // No timers advance: with the first-response guard armed, only the
+      // parent abort can settle the acquisition race.
+      abortController.abort();
+      await done;
+
+      expect(events).toStrictEqual([{ type: AgentEventType.UserCancelled }]);
+      expect(singleCancelledEventCount(events)).toBe(1);
+
+      stalled.rejectStalledRead(new Error('late provider failure'));
+      await flushEventLoop();
+      expect(singleCancelledEventCount(events)).toBe(1);
+      expect(failures.captured).toHaveLength(0);
+    } finally {
+      failures.stop();
+    }
+  });
+
+  it('abort still settles when provider liveness disarms the first-response guard mid-acquisition', async () => {
+    const abortController = new AbortController();
+    const stalled = createStallingStream(0);
+    mockSendMessageStream.mockImplementation(async (params) => {
+      const config = params as {
+        config?: { onStreamLiveness?: StreamLivenessListener };
+      };
+      // Liveness before any chunk disarms phase A; the inter-chunk guard
+      // stays disabled under default config, so nothing but the parent
+      // abort can ever settle the first read.
+      config.config?.onStreamLiveness?.({
+        sourceEvent: 'response.created',
+        sseObserved: true,
+      });
+      return stalled.stream;
+    });
+
+    const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+    await stalled.stallEntered;
+    abortController.abort();
+    await done;
+
+    expect(events).toStrictEqual([{ type: AgentEventType.UserCancelled }]);
+    expect(singleCancelledEventCount(events)).toBe(1);
+  });
+
+  it('settles immediately when abort fires during acquisition, before the first read is raced', async () => {
+    const failures = captureProcessFailures();
+    try {
+      const abortController = new AbortController();
+      const stalled = createStallingStream(0);
+      mockSendMessageStream.mockImplementation(async () => {
+        // Cancel while the sendMessageStream handshake is still pending: the
+        // very first read is never raced against anything, so only the
+        // pre-aborted fast path inside raceReadWithAbort can settle the turn
+        // (issue #3236 production window: user cancels before first chunk).
+        abortController.abort();
+        return stalled.stream;
+      });
+
+      const { events, done } = runTurnCollecting(turn, abortController.signal);
+      await done;
+
+      expect(events).toStrictEqual([{ type: AgentEventType.UserCancelled }]);
+      expect(singleCancelledEventCount(events)).toBe(1);
+
+      // The sunk first read settles late with a rejection: no unhandled
+      // rejection, no additional events.
+      await stalled.stallEntered;
+      stalled.rejectStalledRead(new Error('late provider failure'));
+      await flushEventLoop();
+      expect(singleCancelledEventCount(events)).toBe(1);
+      expect(failures.captured).toHaveLength(0);
+    } finally {
+      failures.stop();
+    }
+  });
+
+  it('emits exactly one UserCancelled when the provider rejects the read right after abort wins', async () => {
+    const failures = captureProcessFailures();
+    try {
+      const abortController = new AbortController();
+      const stalled = createStallingStream(1);
+      mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+      const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+      await stalled.stallEntered;
+      abortController.abort();
+      // The transport rejects the pending read a microtask after the abort
+      // race wins — this must neither duplicate the terminal event nor
+      // surface an error event.
+      queueMicrotask(() =>
+        stalled.rejectStalledRead(new DOMException('Aborted', 'AbortError')),
+      );
+
+      await done;
+
+      expect(events).toStrictEqual([
+        { type: AgentEventType.Content, value: 'part 1', traceId: undefined },
+        { type: AgentEventType.UserCancelled },
+      ]);
+      await flushEventLoop();
+      expect(failures.captured).toHaveLength(0);
+    } finally {
+      failures.stop();
+    }
+  });
+
+  it('still closes the abort-ignoring iterator via return() after the abort race wins', async () => {
+    const abortController = new AbortController();
+    const stalled = createStallingStream(1);
+    mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+    const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+    await stalled.stallEntered;
+    abortController.abort();
+    await done;
+
+    expect(events).toContainEqual({ type: AgentEventType.UserCancelled });
+    expect(stalled.returnSpy).toHaveBeenCalled();
+  });
+
+  it('a many-chunk stream still aborts with exactly one UserCancelled', async () => {
+    const abortController = new AbortController();
+    const stalled = createStallingStream(5);
+    mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+    const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+    await stalled.stallEntered;
+    abortController.abort();
+    await done;
+
+    expect(singleCancelledEventCount(events)).toBe(1);
+    expect(
+      events.filter((event) => event.type === AgentEventType.Content),
+    ).toHaveLength(5);
+    expect(events[events.length - 1]).toStrictEqual({
+      type: AgentEventType.UserCancelled,
+    });
+  });
+
+  it('a clean stream completes normally through the abort race', async () => {
+    const abortController = new AbortController();
+    const stalled = createStallingStream(3);
+    mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+    const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+    await stalled.stallEntered;
+    stalled.resolveStalledRead({ value: undefined, done: true });
+    await done;
+
+    expect(events).toStrictEqual([
+      { type: AgentEventType.Content, value: 'part 1', traceId: undefined },
+      { type: AgentEventType.Content, value: 'part 2', traceId: undefined },
+      { type: AgentEventType.Content, value: 'part 3', traceId: undefined },
+    ]);
+    expect(singleCancelledEventCount(events)).toBe(0);
+  });
+
+  it('removes per-read abort listeners once a clean multi-chunk stream completes', async () => {
+    const abortController = new AbortController();
+    const instrumented = instrumentAbortListeners(abortController.signal);
+    const stalled = createStallingStream(3);
+    mockSendMessageStream.mockResolvedValue(stalled.stream);
+
+    const { events, done } = runTurnCollecting(turn, abortController.signal);
+
+    await stalled.stallEntered;
+    // Reads 1-3 settled cleanly: only the turn's parent hook plus the ONE
+    // listener for the in-flight read remain registered — earlier reads
+    // removed theirs.
+    expect(instrumented.pendingAbortListenerCount()).toBe(2);
+
+    stalled.resolveStalledRead({ value: undefined, done: true });
+    await done;
+
+    expect(events).toHaveLength(3);
+    expect(singleCancelledEventCount(events)).toBe(0);
+    // Clean completion removed every 'abort' listener — no leak, even
+    // though the signal never fired.
+    expect(instrumented.pendingAbortListenerCount()).toBe(0);
+  });
+});
diff --git a/packages/agents/src/core/turn.ts b/packages/agents/src/core/turn.ts
index e038fc7403..ba5dcb7530 100644
--- a/packages/agents/src/core/turn.ts
+++ b/packages/agents/src/core/turn.ts
@@ -62,6 +62,12 @@ import {
 } from './turnJsonUtils.js';
 import { buildCitationEvent } from './turnCitations.js';
 import { buildErrorReportContext } from './turnErrorReportContext.js';
+import {
+  beginWatchdogBoundedAcquisition,
+  closeLateAcquiredIterator,
+  formatStreamIdleTimeoutMessage,
+  raceReadWithAbort,
+} from './turnStreamGuards.js';
 import {
   DEFAULT_AGENT_ID,
   AgentEventType,
@@ -74,20 +80,6 @@ type TurnRequest = string | object | readonly unknown[];
 /** @deprecated Use DEFAULT_STREAM_IDLE_TIMEOUT_MS from streamIdleTimeout.js instead */
 export const TURN_STREAM_IDLE_TIMEOUT_MS = DEFAULT_STREAM_IDLE_TIMEOUT_MS;
 
-function formatStreamIdleTimeoutMessage(
-  fire: StreamWatchdogFire,
-  livenessObserved: boolean,
-): string {
-  const guardLabel =
-    fire.guard === 'first-response'
-      ? 'First-response'
-      : 'Inter-chunk stream-idle';
-  const livenessPart = livenessObserved
-    ? '; provider liveness was observed before the timeout'
-    : '';
-  return `${guardLabel} timeout: no response received within the allowed time (threshold ${fire.thresholdMs}ms) from ${fire.configSource}${livenessPart}.`;
-}
-
 interface IdleFlag {
   timedOut: boolean;
   fire: StreamWatchdogFire | undefined;
@@ -412,17 +404,23 @@ export class Turn {
         // watchdog in run()); consume it directly, then clear the pending slot.
         result = pendingResult;
         pendingResult = undefined;
-      } else if (watchdog.isActive) {
-        // The watchdog governs the whole stream: its inter-chunk guard is
-        // rearmed by both provider liveness pings and semantic events, so a
-        // healthy stream never false-trips regardless of chunk cadence.
-        result = await Promise.race([
-          streamIterator.next(),
-          watchdog.timeoutPromise,
-        ]);
       } else {
-        // Watchdog disabled: call iterator.next() directly
-        result = await streamIterator.next();
+        // The watchdog governs the whole stream when active: its inter-chunk
+        // guard is rearmed by both provider liveness pings and semantic
+        // events, so a healthy stream never false-trips regardless of chunk
+        // cadence. Transports that ignore the abort signal (issue #3236)
+        // would leave the read pending until the guard fires — or forever
+        // when the watchdog is disabled — so every read also races the
+        // parent abort signal.
+        const read = watchdog.isActive
+          ? Promise.race([streamIterator.next(), watchdog.timeoutPromise])
+          : streamIterator.next();
+        const outcome = await raceReadWithAbort(read, signal);
+        if (outcome.aborted) {
+          yield { type: AgentEventType.UserCancelled };
+          return;
+        }
+        result = outcome.value;
       }
       if (result.done === true) {
         break;
@@ -665,6 +663,7 @@ export class Turn {
       try {
         const { iterator, firstResult } = await this.acquireFirstStreamEvent(
           req,
+          signal,
           timeoutSignal,
           watchdog,
           idleFlag,
@@ -743,6 +742,7 @@ export class Turn {
 
   private async acquireFirstStreamEvent(
     req: TurnRequest,
+    signal: AbortSignal,
     timeoutSignal: AbortSignal,
     watchdog: StreamWatchdog,
     idleFlag: IdleFlag,
@@ -760,51 +760,50 @@ export class Turn {
         onStreamLiveness,
       );
       try {
-        const firstResult = await iterator.next();
-        return { iterator, firstResult };
+        const outcome = await raceReadWithAbort(iterator.next(), signal);
+        if (outcome.aborted) {
+          // Throwing AbortError routes cancellation through run()'s catch →
+          // handleRunError, which maps it to UserCancelled because the
+          // parent signal is aborted.
+          throw new DOMException('Aborted', 'AbortError');
+        }
+        return { iterator, firstResult: outcome.value };
       } catch (error) {
         await closeIteratorBounded(iterator, timeoutSignal);
         throw error;
       }
     }
 
-    let acquiredIterator: AsyncIterator<StreamEvent> | undefined;
     const acquisitionPromise = this.openResponseStreamIterator(
       req,
       timeoutSignal,
       onProviderError,
       onStreamLiveness,
     );
-    acquisitionPromise
-      .then((iterator) => {
-        acquiredIterator = iterator;
-        return iterator;
-      })
-      .catch(() => undefined);
-    const firstEventPromise = acquisitionPromise.then(async (iterator) => {
-      const firstResult = await iterator.next();
-      return { iterator, firstResult };
-    });
-    firstEventPromise.catch(() => {});
+    const acquisition = beginWatchdogBoundedAcquisition(acquisitionPromise);
 
     try {
-      const result = await Promise.race([
-        firstEventPromise,
-        watchdog.timeoutPromise,
-      ]);
-      return result;
+      const outcome = await raceReadWithAbort(
+        Promise.race([acquisition.firstEventPromise, watchdog.timeoutPromise]),
+        signal,
+      );
+      if (outcome.aborted) {
+        // Same routing as the unbounded branch: AbortError → handleRunError
+        // maps an aborted parent signal to UserCancelled. The catch below
+        // still sinks/closes the abandoned acquisition.
+        throw new DOMException('Aborted', 'AbortError');
+      }
+      return outcome.value;
     } catch (error) {
       watchdog.cancel();
-      const iteratorAtCatch = acquiredIterator;
+      const iteratorAtCatch = acquisition.acquiredIterator();
       await closeIteratorBounded(iteratorAtCatch, timeoutSignal);
       // Close a late-acquired iterator without waiting for its first next().
-      acquisitionPromise
-        .then((lateIterator) =>
-          lateIterator === iteratorAtCatch
-            ? undefined
-            : closeIteratorBounded(lateIterator, timeoutSignal),
-        )
-        .catch(() => undefined);
+      closeLateAcquiredIterator(
+        acquisitionPromise,
+        iteratorAtCatch,
+        timeoutSignal,
+      );
       if (idleFlag.fire !== undefined) {
         throw new Error(
           formatStreamIdleTimeoutMessage(
diff --git a/packages/agents/src/core/turnStreamGuards.ts b/packages/agents/src/core/turnStreamGuards.ts
new file mode 100644
index 0000000000..14933031cd
--- /dev/null
+++ b/packages/agents/src/core/turnStreamGuards.ts
@@ -0,0 +1,164 @@
+/**
+ * @license
+ * Copyright 2026 Vybestack LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * @fileoverview Bounded-settlement helpers for Turn's provider stream reads.
+ *
+ * {@link raceReadWithAbort} lets a pending read settle immediately when the
+ * turn's parent abort signal fires, even for transports that ignore the
+ * signal (issue #3236), while ensuring the abandoned read can never surface
+ * as an unhandled rejection.
+ *
+ * {@link formatStreamIdleTimeoutMessage} renders the user-facing diagnostic
+ * for a watchdog fire. Relocated verbatim from Turn so that file stays under
+ * the lint `max-lines` budget; behaviour is unchanged.
+ */
+
+import type { StreamWatchdogFire } from '@vybestack/llxprt-code-core/utils/streamWatchdog.js';
+import type { StreamEvent } from './chatSession.js';
+import { closeIteratorBounded } from './iteratorCleanup.js';
+
+export function formatStreamIdleTimeoutMessage(
+  fire: StreamWatchdogFire,
+  livenessObserved: boolean,
+): string {
+  const guardLabel =
+    fire.guard === 'first-response'
+      ? 'First-response'
+      : 'Inter-chunk stream-idle';
+  const livenessPart = livenessObserved
+    ? '; provider liveness was observed before the timeout'
+    : '';
+  return `${guardLabel} timeout: no response received within the allowed time (threshold ${fire.thresholdMs}ms) from ${fire.configSource}${livenessPart}.`;
+}
+
+/**
+ * Attaches no-op handlers to an abandoned read so its eventual settlement —
+ * resolve or reject — is observed and can never surface as an unhandled
+ * rejection (issue #3236).
+ */
+function sinkAbandonedRead<T>(read: Promise<T>): void {
+  read.then(
+    () => undefined,
+    () => undefined,
+  );
+}
+
+/**
+ * Races a pending provider read (or first-event acquisition) against the
+ * turn's parent abort signal.
+ *
+ * Some transports ignore the abort signal entirely (issue #3236), so an
+ * in-flight read can stay pending forever after cancellation — even when it
+ * is already raced against the stream watchdog. Racing it against the parent
+ * signal lets the turn settle immediately; the abandoned read is sunk so its
+ * eventual settlement can never surface as an unhandled rejection.
+ */
+export function raceReadWithAbort<T>(
+  read: Promise<T>,
+  signal: AbortSignal,
+): Promise<{ aborted: true } | { aborted: false; value: T }> {
+  if (signal.aborted) {
+    sinkAbandonedRead(read);
+    return Promise.resolve({ aborted: true } as const);
+  }
+  return new Promise((resolve, reject) => {
+    let settled = false;
+    const onAbort = (): void => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      // No sink needed here: the read.then handlers below already observe
+      // the abandoned read's eventual settlement (they attach synchronously
+      // in this executor, before any abort event can dispatch). Only the
+      // already-aborted fast path above skips that attachment.
+      resolve({ aborted: true });
+    };
+    signal.addEventListener('abort', onAbort, { once: true });
+    read.then(
+      (value: T) => {
+        if (settled) {
+          return;
+        }
+        settled = true;
+        signal.removeEventListener('abort', onAbort);
+        resolve({ aborted: false, value });
+      },
+      (error: unknown) => {
+        if (settled) {
+          return;
+        }
+        settled = true;
+        signal.removeEventListener('abort', onAbort);
+        // Propagate the provider error so Turn's error handling preserves
+        // provider-error semantics, including the already-aborted →
+        // UserCancelled mapping.
+        reject(error);
+      },
+    );
+  });
+}
+
+/** The resolved shape of a first provider stream event acquisition. */
+export interface AcquiredFirstEvent {
+  readonly iterator: AsyncIterator<StreamEvent>;
+  readonly firstResult: IteratorResult<StreamEvent>;
+}
+
+/**
+ * Live view over a watchdog-bounded first-event acquisition started by Turn.
+ * Kept here (not in turn.ts) so Turn stays within its lint `max-lines`
+ * budgets; behaviour is unchanged from the inline original.
+ */
+export interface WatchdogBoundedAcquisition {
+  /** Resolves with the iterator and its first read; sunk on rejection. */
+  readonly firstEventPromise: Promise<AcquiredFirstEvent>;
+  /** Latest iterator handed back by the (still-settling) acquisition. */
+  readonly acquiredIterator: () => AsyncIterator<StreamEvent> | undefined;
+}
+
+/**
+ * Wires the side-effect sinks Turn needs around a first-event acquisition:
+ * records the iterator as soon as the acquisition resolves (so an error path
+ * can close it) and derives the first-read promise whose rejection is sunk.
+ */
+export function beginWatchdogBoundedAcquisition(
+  acquisitionPromise: Promise<AsyncIterator<StreamEvent>>,
+): WatchdogBoundedAcquisition {
+  let acquired: AsyncIterator<StreamEvent> | undefined;
+  acquisitionPromise.then(
+    (iterator) => {
+      acquired = iterator;
+    },
+    () => undefined,
+  );
+  const firstEventPromise = acquisitionPromise.then(async (iterator) => {
+    const firstResult = await iterator.next();
+    return { iterator, firstResult };
+  });
+  firstEventPromise.catch(() => undefined);
+  return { firstEventPromise, acquiredIterator: () => acquired };
+}
+
+/**
+ * Closes an iterator the acquisition hands back AFTER Turn's error path has
+ * already run, without waiting for its first next() (issue #3236: transports
+ * whose reads never settle must not block cleanup).
+ */
+export function closeLateAcquiredIterator(
+  acquisitionPromise: Promise<AsyncIterator<StreamEvent>>,
+  exclude: AsyncIterator<StreamEvent> | undefined,
+  timeoutSignal: AbortSignal,
+): void {
+  acquisitionPromise
+    .then((lateIterator) =>
+      lateIterator === exclude
+        ? undefined
+        : closeIteratorBounded(lateIterator, timeoutSignal),
+    )
+    .catch(() => undefined);
+}
diff --git a/packages/agents/src/internals.ts b/packages/agents/src/internals.ts
index 44523b8559..0fbef18c2f 100644
--- a/packages/agents/src/internals.ts
+++ b/packages/agents/src/internals.ts
@@ -42,6 +42,8 @@ export {
   InvalidStreamError,
   type StreamEvent,
 } from './core/chatSession.js';
+export { MessageStreamOrchestrator } from './core/MessageStreamOrchestrator.js';
+export { TodoContinuationService } from './core/TodoContinuationService.js';
 export * from './core/ChatSessionFactory.js';
 export { CoreToolScheduler } from './core/coreToolScheduler.js';
 export { executeToolCall } from './core/nonInteractiveToolExecutor.js';
diff --git a/packages/cli/src/__tests__/configBridgeGuard.test.ts b/packages/cli/src/__tests__/configBridgeGuard.test.ts
index 49ec37e358..591b814688 100644
--- a/packages/cli/src/__tests__/configBridgeGuard.test.ts
+++ b/packages/cli/src/__tests__/configBridgeGuard.test.ts
@@ -40,13 +40,17 @@ function walk(dir: string, acc: string[] = []): string[] {
   return acc;
 }
 
+const TEST_FILE_SUFFIXES = [
+  '.test.ts',
+  '.test.tsx',
+  '.spec.ts',
+  '.spec.tsx',
+  '.bun.ts',
+  '.bun.tsx',
+] as const;
+
 function isTestFile(path: string): boolean {
-  return (
-    path.endsWith('.test.ts') ||
-    path.endsWith('.test.tsx') ||
-    path.endsWith('.spec.ts') ||
-    path.endsWith('.spec.tsx')
-  );
+  return TEST_FILE_SUFFIXES.some((suffix) => path.endsWith(suffix));
 }
 
 function read(rel: string): string {
diff --git a/packages/cli/src/ui/hooks/agentStream/__tests__/useSubmitQuery.providerIgnoreCancel.bun.tsx b/packages/cli/src/ui/hooks/agentStream/__tests__/useSubmitQuery.providerIgnoreCancel.bun.tsx
new file mode 100644
index 0000000000..2bc47830ec
--- /dev/null
+++ b/packages/cli/src/ui/hooks/agentStream/__tests__/useSubmitQuery.providerIgnoreCancel.bun.tsx
@@ -0,0 +1,915 @@
+/**
+ * @license
+ * Copyright 2026 Vybestack LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * End-to-end CLI regression for issue #3236 — "Cancelled turn whose provider
+ * read never settles blocks follow-up prompts."
+ *
+ * Composition under test (real engine, one controlled seam):
+ *
+ *   REAL useSubmitQuery + REAL useQueuedSubmissions + REAL useCancellation
+ *     → REAL useAgentEventStream.runStream
+ *       → REAL createAgenticLoop + mapLoopStream (inside agent.stream)
+ *         → REAL MessageStreamOrchestrator (with a REAL TodoContinuationService
+ *           over a seeded on-disk task store — the reported repro context)
+ *           → REAL Turn
+ *             → controlled chat seam: turn A streams one content chunk, then
+ *               its next provider read NEVER settles — and ignores the abort
+ *               signal. B/C turns answer cleanly.
+ *
+ * The only deferred CLI-side boundary is `recordingIntegration
+ * .flushAtTurnBoundary()` for turn A (ancillary persistence, a real injected
+ * dep) so the test can deterministically observe the post-`done` window in
+ * which turn A has settled but still owns `activeTurnRef`.
+ *
+ * Pinned behavior:
+ *  - Escape cancels turn A; the REAL engine chain still terminates (the
+ *    generator emits done{reason:'aborted'}; the CLI router then drops that
+ *    final event via its break-on-abort in iterateAgentStream) even though
+ *    A's provider read stays parked forever. The CLI-observable proof is A's
+ *    turn-boundary recording flush running, NOT a routed aborted-done event;
+ *  - while A still owns the turn, fresh prompt B front-enqueues via the #3169
+ *    resume branch (suppression cleared, nothing starts) and C appends;
+ *  - once A's CLI lifecycle finishes, B and C drain automatically, exactly
+ *    once each, in order, with no concurrent turns; final state is Idle with
+ *    an empty queue — and the provider read never settled.
+ */
+
+import { describe, it, expect, vi, beforeEach, afterEach } from 'bun:test';
+import React, { act, useRef, type Dispatch, type SetStateAction } from 'react';
+import { renderHook } from '../../../../test-utils/render.js';
+// Act-aware waitFor: the plain poll in test-utils/render.js lets React state
+// updates land outside act(), which floods CI output with act() warnings for
+// this test's long post-release drain sequence.
+import { waitFor } from '../../../../test-utils/async.js';
+import { useSubmitQuery, type UseSubmitQueryDeps } from '../useSubmitQuery.js';
+import {
+  useAgentEventStream,
+  type AgentEventRouter,
+} from '../useAgentEventStream.js';
+import { useCancellation } from '../useAgentStreamLifecycle.js';
+import { useQueuedSubmissions } from '../useQueuedSubmissions.js';
+import { StreamingState, type HistoryItemWithoutId } from '../../../types.js';
+import { KeypressProvider } from '../../../contexts/KeypressContext.js';
+import { PendingResponseBuffer } from '../pendingResponseBuffer.js';
+import { createStreamRuntimeForTest } from './streamRuntimeTestHelper.js';
+import { createDeferred } from './createDeferred.js';
+import {
+  createLoadedSettings,
+  createMockOverrides,
+} from './submitQueryTestFixtures.js';
+import type { RecordingIntegration } from '@vybestack/llxprt-code-core';
+import type { AgentRequestInput } from '@vybestack/llxprt-code-core/core/clientContract.js';
+import { StreamEventType } from '@vybestack/llxprt-code-core/core/chatSessionTypes.js';
+import type { QueuedSubmission } from '../types.js';
+import { DebugLogger } from '@vybestack/llxprt-code-core/debug/index.js';
+import { DEFAULT_AGENT_ID } from '@vybestack/llxprt-code-core/core/turn.js';
+import { LocalTodoStore } from '@vybestack/llxprt-code-tools';
+import type { Todo, ToolRegistry } from '@vybestack/llxprt-code-tools';
+import { LoopDetectionService } from '@vybestack/llxprt-code-core/services/loopDetectionService.js';
+import { ComplexityAnalyzer } from '@vybestack/llxprt-code-core/services/complexity-analyzer.js';
+import { TodoReminderService } from '@vybestack/llxprt-code-core/services/todo-reminder-service.js';
+import { MessageBus } from '@vybestack/llxprt-code-core/confirmation-bus/message-bus.js';
+import { PolicyEngine } from '@vybestack/llxprt-code-core/policy/policy-engine.js';
+import { PolicyDecision } from '@vybestack/llxprt-code-core/policy/types.js';
+import {
+  ApprovalMode,
+  DEFAULT_IMAGE_PAYLOAD_BUDGET_BYTES,
+} from '@vybestack/llxprt-code-core/config/configTypes.js';
+import {
+  getOrCreateScheduler,
+  disposeScheduler,
+  clearAllSchedulers,
+} from '@vybestack/llxprt-code-core/config/schedulerSingleton.js';
+import type {
+  Config,
+  Config as AgentsConfig,
+} from '@vybestack/llxprt-code-core/config/config.js';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import {
+  createAgenticLoop,
+  createToolScheduler,
+  mapLoopStream,
+  type Agent,
+  type AgentInput,
+  type AgentClientContract,
+  type AgentEvent,
+} from '@vybestack/llxprt-code-agents';
+// Engine internals come through the sanctioned low-level subpath barrel
+// (@vybestack/llxprt-code-agents/internals.js), never raw cross-package
+// relative paths — see the precedent in src/integration-tests/test-utils.ts.
+import {
+  MessageStreamOrchestrator,
+  TodoContinuationService,
+  type ChatSession,
+  type StreamEvent,
+} from '@vybestack/llxprt-code-agents/internals.js';
+
+// ─── Module mocks (UI-side only; the engine below is real) ──────────────────
+
+const prepareQueryForAgentMock = vi
+  .fn()
+  .mockImplementation(async (query: AgentRequestInput) => ({
+    queryToSend: query,
+    shouldProceed: true,
+  }));
+
+const handleContentEventMock = vi
+  .fn()
+  .mockImplementation((text: string, buffer: string) => buffer + text);
+
+void vi.mock('../useStreamEventHandlers.js', () => ({
+  useStreamEventHandlers: () => ({
+    displayUserMessage: vi.fn(),
+    prepareQueryForAgent: prepareQueryForAgentMock,
+    handleLoopDetectedEvent: vi.fn(),
+    handleContentEvent: handleContentEventMock,
+    handleFinishedNotice: vi.fn(),
+  }),
+}));
+
+void vi.mock('../../../contexts/SessionContext.js', () => ({
+  useSessionStats: () => ({
+    startNewPrompt: vi.fn(),
+    getPromptCount: () => 0,
+  }),
+}));
+
+void vi.mock('../turnPreparation.js', () => ({
+  prepareTurnForQuery: vi.fn().mockResolvedValue(undefined),
+}));
+
+void vi.mock('../streamUtils.js', () => ({
+  handleSubmissionError: vi.fn(),
+  processSlashCommandResult: vi.fn(),
+}));
+
+// ─── Controlled chat seam (the ONLY controlled engine boundary) ─────────────
+
+const PROMPT_A_CONTENT = 'A partial answer before the transport hang';
+type ChunkStreamEvent = Extract<
+  StreamEvent,
+  { type: typeof StreamEventType.CHUNK }
+>;
+
+function chunkEvent(
+  text: string,
+  finishReason?: string,
+): { type: typeof StreamEventType.CHUNK; value: unknown } {
+  return {
+    type: StreamEventType.CHUNK,
+    value: {
+      content: { speaker: 'ai', blocks: [{ type: 'text', text }] },
+      ...(finishReason !== undefined
+        ? { finishReason, rawStopReason: finishReason }
+        : {}),
+    },
+  };
+}
+
+class ControlledChatSeam {
+  mode: 'turnA' | 'clean' = 'turnA';
+  private aReadCount = 0;
+  private abortObserved = false;
+  private readSettled = false;
+  /** Resolves when the parked second read has registered its abort listener. */
+  readonly parkedReadA = createDeferred<void>();
+  /** The abort signal Turn hands the provider via config.abortSignal. */
+  private turnAbortSignal: AbortSignal | undefined;
+  cleanRequests = 0;
+
+  abortObservedByProvider(): boolean {
+    return this.abortObserved;
+  }
+
+  providerReadSettled(): boolean {
+    return this.readSettled;
+  }
+
+  /** Turn shape used by Turn.openResponseStreamIterator. */
+  asChatSession(config: AgentsConfig): ChatSession {
+    // ChatSession is a class with private state, so it is nominally typed —
+    // a structural double cannot satisfy it without this cast. Drift risk is
+    // accepted here deliberately: the double feeds the REAL Turn, whose
+    // runtime consumption of these three members is the contract under test.
+    return {
+      sendMessageStream: async (req: unknown) => {
+        const reqConfig = (req as { config?: { abortSignal?: AbortSignal } })
+          .config;
+        this.turnAbortSignal = reqConfig?.abortSignal;
+        if (this.mode === 'turnA') {
+          return { [Symbol.asyncIterator]: () => this.turnAIterator() };
+        }
+        return { [Symbol.asyncIterator]: () => this.cleanIterator() };
+      },
+      getHistory: () => [],
+      getConfig: () => config,
+      addHistory: () => {},
+    } as unknown as ChatSession;
+  }
+
+  /** #3236 transport: one chunk, then a read that parks forever and ignores abort. */
+  private turnAIterator(): AsyncIterator<
+    ChunkStreamEvent | { type: typeof StreamEventType.CHUNK; value: unknown }
+  > {
+    return {
+      next: () => {
+        this.aReadCount += 1;
+        if (this.aReadCount === 1) {
+          return Promise.resolve({
+            done: false,
+            value: chunkEvent(PROMPT_A_CONTENT),
+          } as IteratorResult<never>);
+        }
+        // Second read: parked forever. Observe (but ignore) the turn's own
+        // abort signal — the #3236 provider-ignores-abort transport model.
+        this.turnAbortSignal?.addEventListener('abort', () => {
+          this.abortObserved = true;
+        });
+        const parked = new Promise<IteratorResult<never>>(() => {});
+        void parked.then(
+          () => void (this.readSettled = true),
+          () => void (this.readSettled = true),
+        );
+        this.parkedReadA.resolve();
+        return parked;
+      },
+      // Cleanup is intentionally uncooperative too; closeIteratorBounded's
+      // internal bound is what caps this.
+      return: () => new Promise<IteratorResult<never>>(() => {}),
+    };
+  }
+
+  private cleanIterator(): AsyncIterator<{
+    type: typeof StreamEventType.CHUNK;
+    value: unknown;
+  }> {
+    this.cleanRequests += 1;
+    let served = false;
+    return {
+      next: async () => {
+        if (served) return { done: true, value: undefined };
+        served = true;
+        return { done: false, value: chunkEvent('clean answer', 'stop') };
+      },
+      return: async () => ({ done: true, value: undefined }),
+    };
+  }
+}
+
+// ─── Real engine construction ───────────────────────────────────────────────
+
+interface EngineEnv {
+  agent: Agent;
+  chat: ControlledChatSeam;
+  startedPrompts: string[];
+  maxConcurrent: () => number;
+  routedEvents: AgentEvent[];
+}
+
+function createOrchestratorConfig(sessionId: string): AgentsConfig {
+  return {
+    getEphemeralSetting: () => undefined,
+    getMaxSessionTurns: () => 100,
+    getIdeMode: () => false,
+    getContinueOnFailedApiCall: () => false,
+    getSettingsService: () => ({
+      getCurrentProfileName: () => null,
+      get: () => undefined,
+    }),
+    getSessionId: () => sessionId,
+  } as unknown as AgentsConfig;
+}
+
+function createEmptyToolRegistry(): ToolRegistry {
+  return {
+    getToolByName: () => null,
+    getFunctionDeclarations: () => [],
+    getTools: () => [],
+    discoverTools: async () => {},
+    getAllTools: () => [],
+    getAllToolNames: () => [],
+    getToolsByServer: () => [],
+    registerTool: () => {},
+    getToolByDisplayName: () => null,
+    tools: new Map(),
+    discovery: {},
+  } as unknown as ToolRegistry;
+}
+
+function createLoopConfig(options: {
+  messageBus: MessageBus;
+  toolRegistry: ToolRegistry;
+  policyEngine: PolicyEngine;
+}): Config {
+  const { messageBus, toolRegistry, policyEngine } = options;
+  const fixture: Record<string, unknown> = {
+    getSessionId: () => 'issue3236-loop',
+    getUsageStatisticsEnabled: () => false,
+    getDebugMode: () => false,
+    getImagePayloadBudgetBytes: () => DEFAULT_IMAGE_PAYLOAD_BUDGET_BYTES,
+    getApprovalMode: () => ApprovalMode.YOLO,
+    getEphemeralSettings: () => ({}),
+    getEphemeralSetting: () => undefined,
+    getAllowedTools: () => [],
+    getExcludeTools: () => [],
+    getContentGeneratorConfig: () => ({ model: 'test-model' }),
+    getModel: () => 'test-model',
+    getToolRegistry: () => toolRegistry,
+    getMessageBus: () => messageBus,
+    getPolicyEngine: () => policyEngine,
+    getTelemetryLogPromptsEnabled: () => false,
+    isInteractive: () => true,
+    getNonInteractive: () => false,
+    getToolSchedulerFactory: () => createToolScheduler,
+    getOrCreateScheduler: (
+      sessionId: string,
+      callbacks: Parameters<Config['getOrCreateScheduler']>[1],
+      schedulerOptions: Parameters<Config['getOrCreateScheduler']>[2],
+      deps: Parameters<Config['getOrCreateScheduler']>[3],
+    ) => {
+      const schedulerMessageBus = deps?.messageBus;
+      if (!schedulerMessageBus)
+        throw new Error('Test config requires deps.messageBus');
+      return getOrCreateScheduler(
+        fixture as unknown as Config,
+        sessionId,
+        callbacks,
+        schedulerOptions,
+        {
+          messageBus: schedulerMessageBus,
+          toolRegistry: deps.toolRegistry ?? toolRegistry,
+        },
+      );
+    },
+    disposeScheduler: (sessionId: string) => disposeScheduler(sessionId),
+  };
+  return fixture as unknown as Config;
+}
+
+function createEngineEnv(options: {
+  sessionId: string;
+  todoDataDir: string;
+}): EngineEnv {
+  const chat = new ControlledChatSeam();
+  const orchestratorConfig = createOrchestratorConfig(options.sessionId);
+  const chatSession = chat.asChatSession(orchestratorConfig);
+
+  const orchestrator = new MessageStreamOrchestrator({
+    config: orchestratorConfig,
+    getChat: () => chatSession,
+    logger: new DebugLogger('issue3236:providerIgnoreCancel'),
+    loopDetector: new LoopDetectionService(orchestratorConfig),
+    todoContinuationService: new TodoContinuationService({
+      config: orchestratorConfig,
+      todoReminderService: new TodoReminderService(),
+      complexitySuggestionCooldown: 300000,
+      todoDataDirResolver: () => options.todoDataDir,
+    }),
+    ideContextTracker: {
+      getContextParts: () => ({ contextParts: [], newIdeContext: undefined }),
+      recordSentContext: () => {},
+    } as never,
+    agentHookManager: {
+      cleanupOldHookState: () => {},
+      fireBeforeAgentHookSafe: async () => undefined,
+      fireAfterAgentHookSafe: async () => undefined,
+    } as never,
+    getEffectiveModelIdentity: () => ({
+      providerName: 'test',
+      model: 'test-model',
+    }),
+    getHistory: async () => [],
+    getSessionTurnCount: () => 1,
+    incrementSessionTurnCount: () => {},
+    lazyInitialize: async () => {},
+    startChat: async () => {
+      throw new Error('startChat must not run');
+    },
+    getPreviousHistory: () => undefined,
+    setChat: () => {},
+    hasChat: () => true,
+    complexityAnalyzer: new ComplexityAnalyzer(),
+    getLastPromptId: () => undefined,
+    setLastPromptId: () => {},
+    resetCurrentSequenceModel: () => {},
+    updateTelemetryTokenCount: () => {},
+    async *sendMessageStream(): AsyncGenerator<never> {},
+  });
+
+  const agentClient = {
+    async initialize() {},
+    isInitialized: () => true,
+    hasChatInitialized: () => true,
+    getChat: () => chatSession,
+    async getHistory() {
+      return [];
+    },
+    getHistoryService: () => null,
+    storeHistoryServiceForReuse: () => {},
+    storeHistoryForLaterUse: () => {},
+    addHistory: async () => {},
+    async *sendMessageStream(
+      req: AgentRequestInput,
+      signal: AbortSignal,
+      promptId: string,
+    ): AsyncGenerator<unknown> {
+      yield* orchestrator.execute(req, signal, promptId, 25, false);
+    },
+  } as unknown as AgentClientContract;
+
+  const policyEngine = new PolicyEngine({
+    rules: [],
+    defaultDecision: PolicyDecision.ALLOW,
+    nonInteractive: false,
+  });
+  const messageBus = new MessageBus(policyEngine, false);
+  const loopConfig = createLoopConfig({
+    messageBus,
+    toolRegistry: createEmptyToolRegistry(),
+    policyEngine,
+  });
+
+  const startedPrompts: string[] = [];
+  let active = 0;
+  let maxConcurrent = 0;
+  const agent = {
+    async chat() {
+      return { text: '', toolCalls: [], finishReason: 'stop' };
+    },
+    async *stream(
+      input: AgentInput,
+      streamOpts?: {
+        readonly signal?: AbortSignal;
+        readonly promptId?: string;
+      },
+    ): AsyncIterable<AgentEvent> {
+      const prompt = promptTextOf(input);
+      startedPrompts.push(prompt);
+      active += 1;
+      if (active > maxConcurrent) maxConcurrent = active;
+      try {
+        const loop = createAgenticLoop({
+          agentClient,
+          config: loopConfig,
+          messageBus,
+          interactiveMode: true,
+          displayCallbacks: {},
+        });
+        yield* mapLoopStream(
+          loop.run(
+            input as never,
+            streamOpts?.signal ?? new AbortController().signal,
+            streamOpts?.promptId ?? 'issue3236',
+          ),
+        );
+      } finally {
+        active -= 1;
+      }
+    },
+    getProvider: () => 'test',
+    async setProvider() {
+      return {
+        changed: false,
+        previousProvider: 'test',
+        nextProvider: 'test',
+        infoMessages: [],
+      };
+    },
+    getProviderStatus: () => ({
+      provider: 'test',
+      model: 'test-model',
+      authStatus: 'authenticated',
+    }),
+    getModel: () => 'test-model',
+    async setModel() {},
+    getCurrentSequenceModel: () => null,
+    getApprovalMode: () => ApprovalMode.DEFAULT,
+    setApprovalMode: () => {},
+    getRuntimeId: () => 'issue3236-agent',
+    getEphemeralSetting: () => undefined,
+    setEphemeralSetting: () => {},
+    getEphemeralSettings: () => ({}),
+    getModelParams: () => ({}),
+    setModelParam: () => {},
+    clearModelParam: () => {},
+    getUserTier: () => undefined,
+    tools: {
+      list: () => [],
+      get: () => undefined,
+      async setEnabled() {},
+      onConfirmationRequest: () => () => {},
+      respondToConfirmation: () => {},
+      onToolUpdate: () => () => {},
+      setEditorCallbacks: () => {},
+      setDisplayCallbacks: () => {},
+      recordCompletedToolCalls: () => {},
+    },
+    async getHistory() {
+      return [];
+    },
+    async setHistory() {},
+    async addHistory() {},
+    async restoreHistory() {},
+    async resetChat() {},
+    async updateSystemInstruction() {},
+    async addDirectoryContext() {},
+    async compress() {
+      return { status: 'skipped' };
+    },
+    getStats: () => ({
+      promptTokens: 0,
+      candidateTokens: 0,
+      totalTokens: 0,
+      cachedTokens: 0,
+      contextWindowSize: 0,
+      contextWindowUsed: 0,
+      turnCount: 0,
+    }),
+    onStats: () => () => {},
+    async generate() {
+      return '';
+    },
+    async generateJson() {
+      return {};
+    },
+    async generateEmbedding() {
+      return [];
+    },
+    listProviders: () => [],
+    listTools: () => [],
+    async dispose() {},
+  } as unknown as Agent;
+
+  return {
+    agent,
+    chat,
+    startedPrompts,
+    maxConcurrent: () => maxConcurrent,
+    routedEvents: [],
+  };
+}
+
+// ─── Render harness (REAL queue store + REAL event-stream runner) ───────────
+
+function createMockSetState(
+  calls: boolean[],
+): Dispatch<SetStateAction<boolean>> {
+  return (value) => {
+    if (typeof value === 'boolean') calls.push(value);
+  };
+}
+
+interface TestHandles {
+  setIsRespondingCalls: boolean[];
+  setIsResponding: Dispatch<SetStateAction<boolean>>;
+  abortControllerRef: React.MutableRefObject<AbortController | null>;
+  addItem: ReturnType<typeof vi.fn>;
+  flushPendingHistoryItem: ReturnType<typeof vi.fn>;
+  setPendingHistoryItem: ReturnType<typeof vi.fn>;
+  setLastAgentActivityTime: ReturnType<typeof vi.fn>;
+  pendingHistoryItemRef: React.MutableRefObject<HistoryItemWithoutId | null>;
+}
+
+function createTestHandles(): TestHandles {
+  const setIsRespondingCalls: boolean[] = [];
+  return {
+    setIsRespondingCalls,
+    setIsResponding: createMockSetState(setIsRespondingCalls),
+    abortControllerRef: { current: null },
+    addItem: vi.fn().mockReturnValue(1),
+    flushPendingHistoryItem: vi.fn(),
+    setPendingHistoryItem: vi.fn(),
+    setLastAgentActivityTime: vi.fn(),
+    pendingHistoryItemRef: { current: null },
+  };
+}
+
+function renderHarness(options: {
+  env: EngineEnv;
+  handles: TestHandles;
+  recordingIntegration: RecordingIntegration;
+}) {
+  const { env, handles } = options;
+  const turnCancelledRef: React.MutableRefObject<boolean> = { current: false };
+  const drainSuppressedRef: React.MutableRefObject<boolean> = {
+    current: false,
+  };
+
+  const hook = renderHook(
+    ({ streamingState }: { streamingState: StreamingState }) => {
+      const queue = useQueuedSubmissions();
+      const processAgentEventRef = useRef<AgentEventRouter | null>(null);
+      const eventStream = useAgentEventStream({
+        agent: env.agent,
+        addItem: handles.addItem,
+        processAgentEventRef,
+        flushPendingHistoryItem: handles.flushPendingHistoryItem,
+        clearPendingHistoryItem: vi.fn(),
+        performMemoryRefresh: vi.fn().mockResolvedValue(undefined),
+        markToolsAsDisplayCleared: vi.fn(),
+        onToolCallsUpdate: vi.fn(),
+        outputUpdateHandler: vi.fn(),
+        getPreferredEditor: vi.fn(),
+        onEditorOpen: vi.fn(),
+        onEditorClose: vi.fn(),
+      });
+      const runStreamRef = useRef(eventStream.runStream);
+      runStreamRef.current = eventStream.runStream;
+
+      const submitDeps: UseSubmitQueryDeps = {
+        runtime: createStreamRuntimeForTest({}, createMockOverrides()),
+        agent: env.agent,
+        addItem: handles.addItem,
+        removeItems: vi.fn(),
+        settings: createLoadedSettings(),
+        onDebugMessage: vi.fn(),
+        onCancelSubmit: vi.fn(),
+        onAuthError: vi.fn(),
+        recordingIntegration: options.recordingIntegration,
+        sanitizeContent: (text: string) => ({ text, blocked: false }),
+        flushPendingHistoryItem: handles.flushPendingHistoryItem,
+        pendingResponse: new PendingResponseBuffer(undefined),
+        pendingHistoryItemRef: handles.pendingHistoryItemRef,
+        thinkingBlocksRef: { current: [] },
+        turnCancelledRef,
+        setTurnCancelled: (v: boolean) => void (turnCancelledRef.current = v),
+        drainSuppressedRef,
+        queuedSubmissionsRef: queue.queuedSubmissionsRef,
+        enqueueSubmission: queue.enqueueSubmission,
+        enqueueSubmissionFirst: queue.enqueueSubmissionFirst,
+        requeueSubmission: queue.requeueSubmission,
+        dequeueSubmission: queue.dequeueSubmission,
+        clearSubmissions: queue.clearSubmissions,
+        tryReserveDrain: queue.tryReserveDrain,
+        releaseDrain: queue.releaseDrain,
+        setPendingHistoryItem: handles.setPendingHistoryItem,
+        setIsResponding: handles.setIsResponding,
+        setInitError: vi.fn(),
+        setThought: vi.fn(),
+        setLastAgentActivityTime: handles.setLastAgentActivityTime,
+        scheduleToolCalls: vi.fn(),
+        abortActiveStream: vi.fn(),
+        handleShellCommand: vi.fn().mockReturnValue(false),
+        handleSlashCommand: vi.fn().mockResolvedValue(false),
+        logger: null,
+        shellModeActive: false,
+        loopDetectedRef: { current: false },
+        lastProfileNameRef: { current: undefined },
+        lastModelInfoRef: { current: null },
+        lastModelIdentityRef: { current: null },
+        abortControllerRef: handles.abortControllerRef,
+        runStreamRef,
+        submitQueryRef: { current: null },
+        isResponding: false,
+        streamingState,
+      };
+
+      const submission = useSubmitQuery(submitDeps);
+      const cancellation = useCancellation(
+        streamingState,
+        turnCancelledRef,
+        (v: boolean) => void (turnCancelledRef.current = v),
+        handles.abortControllerRef,
+        vi.fn(),
+        handles.pendingHistoryItemRef,
+        handles.flushPendingHistoryItem,
+        handles.addItem,
+        handles.setPendingHistoryItem,
+        vi.fn(),
+        handles.setIsResponding,
+        vi.fn(),
+        drainSuppressedRef,
+      );
+      processAgentEventRef.current = (event, timestamp, signal) => {
+        env.routedEvents.push(event);
+        submission.processAgentEvent(event, timestamp, signal);
+      };
+      return {
+        ...submission,
+        ...cancellation,
+        queue,
+        turnCancelledRef,
+        drainSuppressedRef,
+      };
+    },
+    {
+      initialProps: { streamingState: StreamingState.Idle },
+      wrapper: ({ children }: React.PropsWithChildren) => (
+        <KeypressProvider>{children}</KeypressProvider>
+      ),
+    },
+  );
+  return { ...hook, turnCancelledRef, drainSuppressedRef };
+}
+
+// ─── Utilities ──────────────────────────────────────────────────────────────
+
+function promptTextOf(input: AgentInput | QueuedSubmission['query']): string {
+  if (typeof input === 'string') {
+    return input;
+  }
+  if (Array.isArray(input)) {
+    const first = input[0] as { text?: unknown } | undefined;
+    if (first !== undefined && 'text' in first) {
+      return String(first.text);
+    }
+  }
+  return '';
+}
+
+function queueTexts(queue: ReturnType<typeof useQueuedSubmissions>): string[] {
+  return queue.queuedSubmissionsRef.current.map((s) => promptTextOf(s.query));
+}
+
+function stopDoneCount(events: AgentEvent[]): number {
+  return events.filter(
+    (e): e is Extract<AgentEvent, { type: 'done' }> =>
+      e.type === 'done' && e.reason === 'stop',
+  ).length;
+}
+
+// ─── Test ───────────────────────────────────────────────────────────────────
+
+describe('useSubmitQuery — cancelled turn whose provider read never settles (issue #3236)', () => {
+  beforeEach(() => {
+    // Module-level mock call histories must not leak between tests, or a
+    // second test's waitFor(...).toHaveBeenCalledWith gates would pass
+    // vacuously on stale history (sibling useAgentEventStream.bun.tsx
+    // convention).
+    vi.clearAllMocks();
+    clearAllSchedulers();
+  });
+  afterEach(() => {
+    clearAllSchedulers();
+  });
+
+  it('ends turn A via the abort race, then drains B and C exactly once, in order', async () => {
+    const dataDir = mkdtempSync(join(tmpdir(), 'issue3236-todos-'));
+    const sessionId = 'issue3236-cli-session';
+    try {
+      // Reported repro context: a real active task list on disk.
+      const todoStore = new LocalTodoStore(
+        sessionId,
+        { dataDirResolver: () => dataDir },
+        DEFAULT_AGENT_ID,
+      );
+      const seededTodo: Todo = {
+        id: 'todo-3236-1',
+        content: 'Diagnose the #3236 cancel deadlock',
+        status: 'in_progress',
+      };
+      await todoStore.writeTodos([seededTodo]);
+
+      const env = createEngineEnv({ sessionId, todoDataDir: dataDir });
+      const handles = createTestHandles();
+
+      // Real CLI dep boundary: A's turn-boundary recording flush is deferred
+      // so the post-done/pre-release ownership window is deterministic. The
+      // latch proves A's runStream settled while the provider read is parked.
+      const flushGateA = createDeferred<void>();
+      let flushEntered = false;
+      let flushMode: 'turnA' | 'immediate' = 'turnA';
+      const recordingIntegration = {
+        flushAtTurnBoundary: async (): Promise<void> => {
+          flushEntered = true;
+          if (flushMode === 'turnA') await flushGateA.promise;
+        },
+      } as unknown as RecordingIntegration;
+
+      const { result, rerender, unmount } = renderHarness({
+        env,
+        handles,
+        recordingIntegration,
+      });
+
+      // 1. Prompt A starts through the real submit path and real engine; its
+      //    stream emits one content event, then the provider read parks.
+      const turnAPromiseRef: { current: Promise<void> | null } = {
+        current: null,
+      };
+      act(() => {
+        turnAPromiseRef.current = result.current.submitQuery('A');
+      });
+      // Promise-based latch instead of a polling waitFor: resolves the
+      // first time the real chain routes A's content event, so the wait is
+      // driven by the event itself (no timer drift; a broken chain fails
+      // fast at the test runner's per-test deadline instead).
+      const contentEventALatch = createDeferred<void>();
+      handleContentEventMock.mockImplementation(
+        (text: string, buffer: string) => {
+          if (text === PROMPT_A_CONTENT) contentEventALatch.resolve();
+          return buffer + text;
+        },
+      );
+      await contentEventALatch.promise;
+      // ESC must land AFTER the second read parks: the seam's abort listener
+      // (and thus the "provider observed abort" observation under test) is
+      // registered by the parked read itself. Cancelling before the park
+      // exercises the already-aborted fast path instead of the #3236
+      // read-ignores-abort path.
+      await env.chat.parkedReadA.promise;
+      expect(handleContentEventMock).toHaveBeenCalledWith(
+        PROMPT_A_CONTENT,
+        '',
+        expect.any(Number),
+      );
+      expect(env.startedPrompts).toStrictEqual(['A']);
+      rerender({ streamingState: StreamingState.Responding });
+
+      // 2. Escape through the real useCancellation path.
+      const turnASignal = handles.abortControllerRef.current?.signal;
+      await act(async () => {
+        result.current.cancelOngoingRequest();
+      });
+      expect(result.current.turnCancelledRef.current).toBe(true);
+      expect(result.current.drainSuppressedRef.current).toBe(true);
+      expect(turnASignal?.aborted).toBe(true);
+
+      // 3. THE #3236 invariant: the real chain terminates on abort even
+      //    though the provider read stays parked and the provider observed
+      //    (and ignored) the abort. iterateAgentStream drops the aborted
+      //    turn's final done event (break-on-abort), so the CLI-observable
+      //    proof is A's lifecycle reaching its real turn-boundary flush.
+      await waitFor(() => expect(flushEntered).toBe(true), { timeout: 5000 });
+      expect(env.chat.abortObservedByProvider()).toBe(true);
+      expect(env.chat.providerReadSettled()).toBe(false);
+      // A's CLI lifecycle is still inside its deferred turn-boundary flush,
+      // so ownership is retained: the queue must not drain.
+      expect(env.startedPrompts).toStrictEqual(['A']);
+
+      // 4. Fresh prompt B while A still owns the turn: #3169 resume branch
+      //    front-enqueues it and releases suppression; nothing starts.
+      rerender({ streamingState: StreamingState.Idle });
+      await act(async () => {
+        await result.current.submitQuery('B');
+      });
+      expect(result.current.drainSuppressedRef.current).toBe(false);
+      expect(queueTexts(result.current.queue)).toStrictEqual(['B']);
+
+      // 5. Fresh prompt C appends behind B; still nothing starts.
+      await act(async () => {
+        await result.current.submitQuery('C');
+      });
+      expect(queueTexts(result.current.queue)).toStrictEqual(['B', 'C']);
+      expect(env.startedPrompts).toStrictEqual(['A']);
+
+      // 6. A's CLI lifecycle completes (flush released) → B drains exactly
+      //    once. The queue may then drain C immediately after B, so mid-state
+      //    snapshots are not asserted here; order, exactly-once, and
+      //    serialization are proven from the final state below.
+      env.chat.mode = 'clean';
+      flushMode = 'immediate';
+      await act(async () => {
+        flushGateA.resolve();
+      });
+      await waitFor(() => expect(env.startedPrompts).toContain('B'), {
+        timeout: 5000,
+      });
+      expect(handles.abortControllerRef.current?.signal).not.toBe(turnASignal);
+      expect(result.current.turnCancelledRef.current).toBe(false);
+
+      // 7-8. C drains automatically after B, in order, exactly once. Final
+      //      state: Idle, empty queue, never concurrent, and the provider
+      //      read never settled — the CLI recovered without it.
+      await waitFor(() => expect(env.startedPrompts).toContain('C'), {
+        timeout: 5000,
+      });
+      await waitFor(
+        () => expect(stopDoneCount(env.routedEvents)).toBeGreaterThanOrEqual(2),
+        { timeout: 5000 },
+      );
+      await waitFor(
+        () => expect(queueTexts(result.current.queue)).toStrictEqual([]),
+        { timeout: 5000 },
+      );
+      // "Final state is Idle": the real lifecycle's last responding
+      // transition must have settled back to false after the C drain.
+      const respondingTransitions = handles.setIsRespondingCalls;
+      expect(respondingTransitions[respondingTransitions.length - 1]).toBe(
+        false,
+      );
+      expect(env.startedPrompts).toStrictEqual(['A', 'B', 'C']);
+      expect(env.maxConcurrent()).toBe(1);
+      expect(env.chat.cleanRequests).toBeGreaterThanOrEqual(2);
+      expect(env.chat.providerReadSettled()).toBe(false);
+
+      const turnAPromise = turnAPromiseRef.current;
+      if (turnAPromise !== null) {
+        await act(async () => {
+          await turnAPromise;
+        });
+      }
+      await act(async () => {
+        unmount();
+      });
+    } finally {
+      rmSync(dataDir, { recursive: true, force: true });
+    }
+  });
+});
diff --git a/project-plans/issue3236/PLAN.md b/project-plans/issue3236/PLAN.md
new file mode 100644
index 0000000000..930d37a7f1
--- /dev/null
+++ b/project-plans/issue3236/PLAN.md
@@ -0,0 +1,242 @@
+# Plan: Make Provider Stream Reads Cancellation-Responsive at the Turn Layer (Issue #3236)
+
+Plan ID: PLAN-20260815-ISSUE3236
+Generated: 2026-08-15
+Issue: #3236
+
+## Problem statement
+
+Cancelling an active turn with Escape can leave the CLI permanently unable to
+process another prompt. The UI immediately presents Idle (Escape synchronously
+clears `isResponding` and cancels tools), but the interactive submission gate
+(`activeTurnRef` in `useSubmitQuery.ts`) is released only when the full
+provider chain settles. No layer between the CLI and the provider transport
+races the pending provider read against the abort signal:
+
+- `Turn.consumeStreamEvents` (`packages/agents/src/core/turn.ts:415-426`) does
+  a direct unbounded `await streamIterator.next()` when the watchdog is
+  inactive, and even the watchdog-active branch races only the timeout
+  promise — abort is not a contender.
+- The watchdog disarms after the first liveness ping or semantic chunk
+  (`streamWatchdog.ts:189-192`) because the inter-chunk idle timeout defaults
+  to disabled (`DEFAULT_STREAM_IDLE_TIMEOUT_MS = 0`), so every post-first-event
+  read is unbounded with default config.
+- `MessageStreamOrchestrator._processStreamIteration`, `AgenticLoop.
+  streamAndCollect` (`AgenticLoop.ts:487-497`), and the CLI's
+  `iterateAgentStream` (`useAgentEventStream.ts:344-345`) are plain `for await`
+  loops that can only observe abort when the next event arrives or the stream
+  ends.
+
+If the provider transport's pending `next()` does not reject when the signal
+aborts (SDK-internal retry/buffer queues, SSE pump state, proxy/keep-alive
+edges, fetch abort quirks), `Turn.run` never finishes, no public `done` is
+synthesized, `runStream` never settles, the submission `finally` never clears
+`activeTurnRef`, and every subsequent prompt is queued forever. Restart is the
+only recovery. The failure is intermittent because the abort signal is
+correctly plumbed into all audited provider fetches; only transport-level
+non-settlement triggers the hang, and nothing in this codebase can recover
+from it.
+
+Existing regression coverage cannot catch this: `turn.abort-timeout.test.ts`
+uses `rejectWhenAborted` — abort-honoring iterators only.
+
+## Preflight findings
+
+1. `Turn.consumeStreamEvents` is the single owner of provider iteration for
+   agent turns; both read branches await settlement with no abort contender.
+2. The abort check at `turn.ts:432-435` runs only after `next()` resolves, so
+   it cannot help a pending read.
+3. `closeIteratorBounded` (`iteratorCleanup.ts`) is already bounded (1s cap)
+   and returns immediately when the signal is already aborted, so the existing
+   `cleanupStreamResources` finally-block cannot hang once consumption exits.
+4. `TurnProcessor.sendMessageStream` (`TurnProcessor.ts:204-226`) already
+   force-resolves its send-serialization gate on abort and documents this
+   deadlock class, but only for the gate — not the read path.
+5. `acquireFirstStreamEvent` has an unbounded `await iterator.next()` in its
+   watchdog-inactive branch (`turn.ts:763`); the watchdog-active branch is
+   bounded only by the first-response timeout (300s default), so it self-heals
+   with an error rather than hanging silently. Both branches should still be
+   abort-responsive.
+6. The post-read abort check and the new abort-race win are mutually
+   exclusive by construction: the race yields `UserCancelled` only when abort
+   wins (result discarded), and the post-read check fires only when the result
+   arrived before abort. Exactly one `UserCancelled` is emitted either way.
+7. Watchdog semantics are orthogonal and must not change: first-response and
+   inter-chunk guards keep their current arm/disarm/fire behavior.
+
+## Requirements and behavior
+
+### REQ-3236-1: Abort settles a pending provider read
+
+**Full text:** Every `streamIterator.next()` await owned by `Turn` — in both
+`consumeStreamEvents` branches (watchdog active and inactive) and the
+`acquireFirstStreamEvent` watchdog-inactive first read — must settle promptly
+when the turn's parent AbortSignal fires, regardless of provider/SDK
+cooperation.
+
+- GIVEN default config (inter-chunk idle timeout disabled) and a provider
+  iterator that yielded at least one semantic chunk
+- WHEN the turn signal aborts while the next read is pending and the iterator
+  ignores the signal
+- THEN the read await settles via the abort race without any timer advancing
+- AND `Turn.run` completes and emits exactly one `UserCancelled`
+- GIVEN the watchdog-active branch (idle timeout configured)
+- WHEN the turn signal aborts while the watchdog race is pending
+- THEN abort wins without waiting for the watchdog fire
+- AND the existing watchdog timeout behavior is unchanged when no abort occurs
+
+### REQ-3236-2: Terminal-path and cleanup invariants
+
+**Full text:** The abort-race win path must emit exactly one `UserCancelled`,
+return from stream consumption so `cleanupStreamResources` runs, and leave the
+bounded iterator cleanup (`closeIteratorBounded`) as the final owner-facing
+close.
+
+- GIVEN an abort-ignoring iterator abandoned mid-read
+- WHEN abort wins the race
+- THEN exactly one `UserCancelled` is emitted (no duplicate from the post-read
+  abort check)
+- AND the iterator's `return()` is still invoked via the existing cleanup path
+- AND `Turn.run` resolves rather than throwing (cancellation is not an error)
+
+### REQ-3236-3: Abandoned-read safety
+
+**Full text:** A provider read abandoned by the abort race must never produce
+an unhandled rejection and must never mutate the completed turn when it
+settles late.
+
+- GIVEN the abort race won and the abandoned `next()` promise later rejects
+- WHEN the microtask queue drains
+- THEN no unhandled rejection escapes to the process
+- GIVEN the abandoned `next()` promise later resolves with another chunk
+- WHEN the turn has already emitted `UserCancelled`
+- THEN no additional event is emitted and turn state is unchanged
+
+### REQ-3236-4: Listener hygiene
+
+**Full text:** The per-read abort listener must be removed when either side of
+the race wins, so long streams do not accumulate listeners on the turn signal.
+
+- GIVEN a stream that yields many chunks without abort
+- WHEN each read completes
+- THEN the abort listener installed for that read is removed
+- GIVEN abort fires
+- THEN the listener (registered `{ once: true }`) is not retained
+
+### REQ-3236-5: CLI queue drain after provider-ignored cancel
+
+**Full text:** With the Turn-level fix in place, the real CLI submission and
+queue paths must recover from a provider-ignored cancellation: the cancelled
+turn's lifecycle completes, `activeTurnRef` is released, and queued prompts
+drain automatically, exactly once, in order, without a second Enter.
+
+- GIVEN a real `useSubmitQuery` + `useQueuedSubmissions` + real cancellation
+  wiring around a controlled Agent stream whose provider read never settles
+  after one chunk
+- WHEN Escape-style cancellation runs and prompts B then C are submitted
+- THEN B and C drain automatically exactly once in order after the cancelled
+  turn releases ownership
+- AND final streaming state is Idle with an empty queue
+- AND no concurrent `AgenticLoop.run()` executions occur
+
+## Design constraints
+
+1. The fix is local to `packages/agents/src/core/turn.ts` (plus tests). Do not
+   alter watchdog semantics, YOLO approval flow, queue presentation, or any
+   provider package.
+2. Do not solve with polling, retry loops, forced default idle timeouts, or
+   provider-specific assumptions.
+3. The provider signal (`timeoutController.signal`) continues to be passed to
+   the provider unchanged; the race is an additional settlement path owned by
+   `Turn`, not a replacement for provider cooperation.
+4. Preserve existing public event contracts: exactly one `UserCancelled`
+   terminal event for a cancelled turn; successful streams are byte-for-byte
+   compatible.
+5. No new `eslint-disable*`, `@ts-ignore`, `@ts-expect-error`, `@ts-nocheck`,
+   severity downgrades, threshold increases, or ignore blocks.
+6. TypeScript strict: no `any`, no type assertions where a predicate works.
+
+## Implementation outline
+
+### Phase 1 — Turn-level abort race (test-first)
+
+1. Extend `packages/agents/src/core/turn.abort-timeout.test.ts` (existing
+   describe or a new describe block in the same file) with behavioral tests:
+   - REQ-3236-1: unresolved second read after first semantic chunk, default
+     config (no `stream-idle-timeout-ms`), abort mid-read → `Turn.run`
+     completes with exactly one `UserCancelled`; use real timers with a
+     deterministic "second read entered" barrier; resolve/reject the abandoned
+     promise only in teardown.
+   - REQ-3236-1: watchdog-active variant — idle timeout configured, abort
+     mid-read, no timer advance needed to settle.
+   - REQ-3236-2: `return()` still called on the abort-ignoring iterator.
+   - REQ-3236-3: late-then-rejects and late-then-resolves abandoned reads
+     produce no unhandled rejection and no extra events.
+   - REQ-3236-4: many-chunk stream does not accumulate listeners (observable
+     via `signal.listenerCount` or equivalent if available in the runtime;
+     otherwise assert via behavior — long clean stream plus abort still
+     settles exactly once).
+   - Existing tests in the file must remain green unchanged (they pin
+     watchdog semantics and post-read abort behavior).
+2. Implement in `turn.ts`:
+   - A private helper that wraps a pending read (`Promise<IteratorResult>`)
+     with a turn-signal abort race: settles `{ aborted: true }` on abort (and
+     attaches a no-op rejection/resolve sink to the abandoned read), otherwise
+     settles with the read outcome; removes the abort listener on either win.
+   - Use it in `consumeStreamEvents` for both the watchdog-race branch (wrap
+     the `Promise.race` result) and the direct-await branch.
+   - Use it for the first read in `acquireFirstStreamEvent`'s
+     watchdog-inactive branch.
+   - On abort win: `yield { type: AgentEventType.UserCancelled }` and return.
+3. Run the Phase 1 suite plus the agents package tests.
+
+### Phase 2 — CLI-level integration regression
+
+1. New file `packages/cli/src/ui/hooks/agentStream/__tests__/
+   useSubmitQuery.providerIgnoreCancel.bun.tsx` following the existing
+   `useSubmitQuery.*.bun.tsx` harness patterns: real `useSubmitQuery`, real
+   `useQueuedSubmissions` queue primitives, real cancellation hook wiring, and
+   a controlled Agent stream boundary (the agent boundary may be controlled;
+   the hook/queue/cancellation code under test must be real).
+2. Scenarios per REQ-3236-5: cancelled turn with never-settling provider read
+   (post-fix this settles at the Turn boundary — drive the same abort race
+   through the controlled stream), prompts B and C queue then drain exactly
+   once in order, final Idle + empty queue, no concurrent runs.
+3. If — and only if — a concrete missing guard is found in the CLI layer,
+   make the minimal production change in `useSubmitQuery.ts` and pin it with a
+   test; otherwise CLI production code is untouched.
+
+### Phase 3 — regression retention
+
+- Run the existing regression suites referenced by the issue: #2259, #2296,
+  #2882, #2954, #3048, #3169 (`useSubmitQuery.cancelResumeRace.bun.tsx`,
+  `useSubmitQuery.doublecancel.bun.tsx`, `useSubmitQuery.terminalError.bun.tsx`,
+  `useQueuedSubmissions.test.ts`, and their neighbors).
+
+## Verification
+
+The implementer must satisfy the full issue-workflow cycle:
+
+- `npm run test`
+- `npm run lint`
+- `npm run typecheck`
+- `npm run format`
+- `npm run build`
+- `bun scripts/start.ts --profile-load stepfun-37 "write me a haiku and nothing else"`
+
+Policy invariance: no suppressions, no severity downgrades, no threshold
+increases, no new ignores.
+
+## Risks and mitigations
+
+- **Double `UserCancelled`:** impossible by construction (race win discards
+  the result; post-read check only runs when the result arrived first); pinned
+  by tests in both orders.
+- **Unhandled rejections from abandoned reads:** sink both outcomes of the
+  abandoned promise; pinned by late-settlement tests with a process-level
+  rejection capture in the test file.
+- **Watchdog behavior drift:** watchdog code is untouched; existing watchdog
+  tests must pass unchanged.
+- **Bun async-generator quirks (1.3.14):** the race does not rely on
+  `return()` propagation through `yield*`; cleanup uses the existing bounded
+  close.