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

`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. The failure was silent: no log, no traceback.

A reproducer at the regression test confirms a callback that raises on
the first line previously dropped lines 2 and 3; with the fix all three
lines are delivered.

Move the ``try/except`` inside the loop and log at debug level so a
buggy callback fails per-line but doesn't disable stderr piping. Also
log (instead of silently swallow) at the outer level so a stream-read
failure is at least visible at debug level.

Closes #929

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: Xian Zheng

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."""