fix: allow CLI to flush session transcript before terminating subprocess

src/claude_agent_sdk/_internal/transport/subprocess_cli.py

@@ -463,14 +463,18 @@ async def close(self) -> None:
                 await self._stderr_stream.aclose()
             self._stderr_stream = None
 
-        # Terminate and wait for process
+        # Wait for process to exit gracefully after stdin EOF,
+        # giving the CLI time to flush the session transcript.
+        # Only resort to SIGTERM if it doesn't exit in time.
         if self._process.returncode is None:
-            with suppress(ProcessLookupError):
-                self._process.terminate()
-                # Wait for process to finish with timeout
-                with suppress(Exception):
-                    # Just try to wait, but don't block if it fails
+            try:
+                with anyio.fail_after(10):
                     await self._process.wait()
+            except TimeoutError:
+                with suppress(ProcessLookupError):
+                    self._process.terminate()
+                    with suppress(Exception):
+                        await self._process.wait()
 
         self._process = None
         self._stdout_stream = None

tests/test_transport.py

@@ -213,7 +213,12 @@ async def _test():
                 mock_process = MagicMock()
                 mock_process.returncode = None
                 mock_process.terminate = MagicMock()
-                mock_process.wait = AsyncMock()
+
+                # Simulate graceful exit: wait() sets returncode to 0
+                async def _graceful_wait():
+                    mock_process.returncode = 0
+
+                mock_process.wait = AsyncMock(side_effect=_graceful_wait)
                 mock_process.stdout = MagicMock()
                 mock_process.stderr = MagicMock()
 
@@ -235,7 +240,9 @@ async def _test():
                 assert transport.is_ready()
 
                 await transport.close()
-                mock_process.terminate.assert_called_once()
+                # Process should exit gracefully without needing SIGTERM
+                mock_process.terminate.assert_not_called()
+                mock_process.wait.assert_called()
 
         anyio.run(_test)