fix(tui): cancel sibling stream reader when a git reader fails

The concurrent stdout/stderr drain in _run_git used asyncio.gather, which
leaves the sibling _read_bounded coroutine running after the first reader
raises — only the process was killed, so the other reader could keep draining
a closing pipe. Move the concurrent read into _read_streams, which uses
explicit tasks and cancels+awaits both readers on any failure (or cancellation)
before propagating. Adds a regression test for a reader-exception path.

Author: elkaix

src/pythinker_code/ui/shell/prompting/completion/workspace.py

@@ -414,10 +414,7 @@ async def _run_git(self, root: HostPath, *args: str) -> str | None:
                 # Drain stdout and stderr concurrently: reading them
                 # sequentially can deadlock if git fills its stderr pipe buffer
                 # (e.g. submodule warnings) while we are still draining stdout.
-                stdout, _stderr = await asyncio.gather(
-                    self._read_bounded(process.stdout),
-                    self._read_bounded(process.stderr),
-                )
+                stdout = await self._read_streams(process)
                 returncode = await process.wait()
         except asyncio.CancelledError:
             if process is not None:
@@ -444,6 +441,24 @@ async def _run_git(self, root: HostPath, *args: str) -> str | None:
         encoding = "utf-8"
         return stdout.decode(encoding=encoding, errors="replace")
 
+    async def _read_streams(self, process: HostProcess) -> bytes:
+        """Drain stdout and stderr concurrently and return stdout.
+
+        Uses explicit tasks so that if either reader raises (or this coroutine
+        is cancelled), the sibling reader is cancelled and awaited rather than
+        left running against a closing pipe.
+        """
+        stdout_reader = asyncio.create_task(self._read_bounded(process.stdout))
+        stderr_reader = asyncio.create_task(self._read_bounded(process.stderr))
+        try:
+            stdout, _stderr = await asyncio.gather(stdout_reader, stderr_reader)
+        except BaseException:
+            for reader in (stdout_reader, stderr_reader):
+                reader.cancel()
+            await asyncio.gather(stdout_reader, stderr_reader, return_exceptions=True)
+            raise
+        return stdout
+
     @staticmethod
     async def _read_bounded(stream: AsyncReadable) -> bytes:
         chunks = bytearray()

tests/ui_and_conv/test_workspace_index.py

@@ -207,6 +207,48 @@ async def test_git_failure_uses_degraded_fallback() -> None:
     await lifecycle.aclose()
 
 
+@pytest.mark.asyncio
+async def test_read_streams_cancels_sibling_when_one_reader_fails() -> None:
+    # When one stream reader raises, the sibling reader must be cancelled and
+    # awaited rather than left draining a closing pipe.
+    host = _FakeHost()
+    index, lifecycle = _index(host, HostPath("/workspace"))
+
+    class _Blocking:
+        def __init__(self) -> None:
+            self.started = asyncio.Event()
+            self.cancelled = False
+
+        async def read(self, _n: int = -1) -> bytes:
+            self.started.set()
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                self.cancelled = True
+                raise
+            return b""
+
+    class _Raising:
+        def __init__(self, gate: asyncio.Event) -> None:
+            self._gate = gate
+
+        async def read(self, _n: int = -1) -> bytes:
+            await self._gate.wait()  # ensure the sibling is mid-read first
+            raise RuntimeError("stdout reader boom")
+
+    blocking = _Blocking()
+
+    class _Proc:
+        stdout = _Raising(blocking.started)
+        stderr = blocking
+
+    with pytest.raises(RuntimeError, match="boom"):
+        await index._read_streams(cast(Any, _Proc()))
+
+    assert blocking.cancelled is True
+    await lifecycle.aclose()
+
+
 @pytest.mark.asyncio
 async def test_results_are_truncated_at_one_thousand_entries() -> None:
     host = _FakeHost()