fix(transport): isolate stderr callback failures, continue reading lines (#932)

## Summary

Fixes #929.

`SubprocessCLITransport._handle_stderr` wrapped the entire `async for`
loop in a single `except Exception: pass`, so a raise from the
user-provided `options.stderr` callback was caught at the outer level —
the loop terminated and no further stderr lines were delivered for the
rest of the session. Silent: no log, no traceback.

The repro in #929 confirmed a callback that raises on the first line
dropped all subsequent lines (`callback_raised_count = 1` for a 2-line
stream). The contract on `stderr: Callable[[str], None]`
(`types.py:1741`) doesn't document any "must not raise" constraint, so
this is a bug, not user error.

## Changes

- `src/claude_agent_sdk/_internal/transport/subprocess_cli.py`: per-line
`try/except` around `self._options.stderr(line_str)` so a buggy callback
fails for that one line but the loop continues. The outer `except
Exception: pass` becomes `logger.debug(..., exc_info=True)` so
stream-read failures are at least visible at debug level. The `except
anyio.ClosedResourceError` for legitimate end-of-stream is preserved.
- `tests/test_transport.py`: regression test
`test_stderr_callback_raise_does_not_terminate_loop` — 3-line stream,
callback raises on line 1, asserts all 3 lines delivered.

## Test plan

- [x] `uv run pytest tests/test_transport.py` — 90 passed
- [x] `uv run mypy src/` — clean
- [x] `ruff check / ruff format` — clean
- [x] Manual repro from issue body now shows `count = 3` (was `count =
1` before fix)

Co-authored-by: Xian Zheng <xian.zheng@challenger.gauntletai.com>
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: 3 people

src/claude_agent_sdk/_internal/transport/subprocess_cli.py

@@ -526,13 +526,21 @@ async def _handle_stderr(self) -> None:
                 if not line_str:
                     continue
 
-                # Call the stderr callback if provided
+                # Call the stderr callback if provided. Isolate per-line so a
+                # raise in the user's callback doesn't terminate the loop and
+                # silently drop every subsequent line for the rest of the
+                # session.
                 if self._options.stderr:
-                    self._options.stderr(line_str)
+                    try:
+                        self._options.stderr(line_str)
+                    except Exception:
+                        logger.debug(
+                            "stderr callback raised; continuing", exc_info=True
+                        )
         except anyio.ClosedResourceError:
             pass  # Stream closed, exit normally
         except Exception:
-            pass  # Ignore other errors during stderr reading
+            logger.debug("stderr stream read failed", exc_info=True)
 
     async def close(self) -> None:
         """Close the transport and clean up resources."""

tests/test_transport.py

@@ -2,6 +2,7 @@
 
 import os
 import uuid
+from collections.abc import AsyncIterator
 from contextlib import nullcontext
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -2071,6 +2072,37 @@ async def _test():
 
         anyio.run(_test)
 
+    def test_stderr_callback_raise_does_not_terminate_loop(self) -> None:
+        """Regression for issue #929: a raise from ``options.stderr`` must not
+        kill the read loop. Previously the outer ``except Exception: pass``
+        caught it, exited the ``async for``, and silently dropped every
+        subsequent stderr line for the rest of the session."""
+
+        async def _test() -> None:
+            received: list[str] = []
+
+            def stderr_cb(line: str) -> None:
+                received.append(line)
+                if len(received) == 1:
+                    raise RuntimeError("simulated handler failure")
+
+            transport = SubprocessCLITransport(
+                prompt="x", options=ClaudeAgentOptions(stderr=stderr_cb)
+            )
+
+            async def mock_iter() -> AsyncIterator[str]:
+                yield "line 1"
+                yield "line 2"
+                yield "line 3"
+
+            transport._stderr_stream = mock_iter()  # type: ignore[assignment]
+            await transport._handle_stderr()
+
+            # All three lines must be delivered despite the first raise.
+            assert received == ["line 1", "line 2", "line 3"]
+
+        anyio.run(_test)
+
 
 class TestAtexitChildCleanup:
     """Tests for the atexit handler that terminates orphaned CLI subprocesses."""