fix(core): restore asyncio.StreamReader inheritance on ACP stdio wrappers (#152)

* fix(core): restore asyncio.StreamReader inheritance on ACP stdio wrappers

ClientSideConnection.__init__ in the upstream ACP SDK enforces

    isinstance(input_stream, asyncio.StreamWriter) and
    isinstance(output_stream, asyncio.StreamReader)

before constructing the JSON-RPC connection. The earlier branch commit
1207f27 ("refactor(core): P1+P2-A/B dead code") rewrote
JsonRpcObjectStreamReader and _ByteCountingStreamReader as plain
composition classes on the assumption that the inheritance was
decorative. It wasn't — it was load-bearing for that isinstance gate.
Released as 0.19.0b34, the wrappers no longer match the gate and the
TUI fails on first ACP turn with:

    Agent session failed: ClientSideConnection requires asyncio
    StreamWriter/StreamReader

Restore the inheritance on both wrappers without calling super().__init__
(state is fully delegated to the wrapped reader). Document the contract
in the class docstring so a future cleanup pass keeps the inheritance.

Lock the contract with a unit regression test in tests/unit/test_acp_session.py
that constructs a real ClientSideConnection over each wrapper and asserts
the isinstance gate accepts it. The previous unit suite stubbed the
connection out with _FakeConnection / _FakeProcess, so the production
isinstance check was never exercised — by testing.md this is exactly the
"contract acceptance tests don't reach" scenario where a unit test earns
its place.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test: add real-stdio ACP integration harness with hermetic echo agent

Adopts patterns from the ACP reference suite (anthropics/agent-protocol):

- tests/helpers/echo_agent.py — vendored examples/echo_agent.py. A real ACP
  agent in 80 lines that we spawn via sys.executable. No provider binary
  needed.
- tests/helpers/acp_loopback.py — TCP-loopback fixture that yields real
  asyncio.StreamReader/StreamWriter pairs (asyncio.start_server +
  asyncio.open_connection). Lets tests construct the real
  acp.client.connection.ClientSideConnection without a subprocess.
- tests/integration/acp_real/test_stream_wrappers.py — drives both wrappers
  through the real SDK and the full handshake → session → prompt →
  notification → teardown roundtrip via spawn_filtered_agent_process.
  Fails with the production error message when inheritance is removed:
    TypeError: ClientSideConnection requires asyncio StreamWriter/StreamReader

The previous always-on suite stubbed the SDK out entirely (_FakeConnection,
_FakeProcess) so no test ever exercised the production isinstance gate.
That's why 0.19.0b34 shipped a runtime regression with a green CI bar.

docs/internal/testing.md gains a "Real-stdio integration tests" section
explaining the new layer and when to add to it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(core): call super().__init__() on ACP stream wrappers

Greptile review on PR 152 flagged that both wrappers subclass
asyncio.StreamReader but skip super().__init__(), leaving base
attributes (notably _exception) uninitialised — wrapper.exception()
would AttributeError on any error path.

Today's SDK only calls readline() on the wrapped reader, so the gap
is unreachable, but it's a footgun for future SDK upgrades and the
parent constructor is cheap (allocates a small unused buffer).

Call super().__init__() in both wrappers and update docstrings to
reflect the new design. Extend the integration regression test to
assert wrapper.exception() returns None — verified to fail with
"AttributeError: object has no attribute '_exception'" if super()
is skipped.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(acp): assert wrapper.exception() in unit regression test too

Mirrors the assertion added to the integration test in the previous
commit. Greptile's review thread on tests/unit/test_acp_session.py:238
explicitly requested this — the unit test stopped at construction so
the un-overridden exception() / set_exception() were never exercised.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(core): delegate exception()/set_exception() to wrapped reader

Greptile review on PR 152 (iter 2) flagged that calling super().__init__()
gives the wrapper its own _exception field that is never written to. The
asyncio transport calls set_exception() on the *wrapped* reader when the
process dies; the wrapper's own _exception stays None, so wrapper.exception()
returns None even after a transport failure.

Override exception() and set_exception() on both wrappers to delegate to
self._reader, so:
- A transport failure on the underlying reader is visible via the wrapper.
- An SDK-level set_exception on the wrapper propagates to the underlying
  reader (and any blocked readline()/read() on it).

Add a unit regression test that calls underlying.set_exception(...) and
asserts wrapper.exception() returns the same exception, plus the inverse
direction. Verified the test fails (assert is boom) without the overrides.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: aorumbayev

docs/internal/testing.md

@@ -50,6 +50,32 @@ tests exercise but don't assert on directly, or when it tests **platform-depende
 
 ______________________________________________________________________
 
+## Real-stdio integration tests
+
+Tests in `tests/integration/` exercise the *real* third-party SDK constructors and stdio path
+without depending on a provider binary. They sit between unit and the `KAGAN_INTEGRATION_TESTS=1`
+end-to-end suite, and they catch contracts unit-stubs falsify by accepting too much.
+
+The current suite (`tests/integration/acp_real/`) drives the real `acp.client.connection.ClientSideConnection`
+over either:
+
+1. **TCP-loopback streams** (`tests/helpers/acp_loopback.py`) — yields real `asyncio.StreamReader`
+   / `asyncio.StreamWriter` pairs via `asyncio.start_server` + `asyncio.open_connection`. Catches
+   `isinstance(asyncio.StreamReader)` gates and read-method shape contracts in milliseconds.
+2. **A hermetic echo subprocess** (`tests/helpers/echo_agent.py`, vendored from the ACP SDK
+   examples) invoked via `sys.executable`. Speaks the real ACP wire protocol, exercises
+   handshake → session → prompt → notification → teardown end-to-end.
+
+Add a test here whenever a regression escapes through stubbed SDK fakes. Example: 0.19.0b34
+shipped a `ClientSideConnection requires asyncio StreamWriter/StreamReader` runtime error
+because the always-on suite stubbed `_FakeConnection` and never constructed the real SDK
+type. Both checks now live in `acp_real/test_stream_wrappers.py`.
+
+- Files use `pytestmark = [pytest.mark.integration]` and run on every PR (no env-var gate).
+- They must not require any agent binary on PATH; spawn-based tests use `sys.executable`.
+
+______________________________________________________________________
+
 ## Isolation
 
 Each test creates its own universe — fresh `tmp_path`, fresh SQLite, fresh git repo.
@@ -140,7 +166,12 @@ tests/
 │   └── test_web_ui.py                   # SPA static file serving
 ├── integrations/                        # kagan.core.integrations (behavioral)
 │   └── test_github.py                    # GitHub sync: preflight, preview, create, skip, re-import, labels
+├── integration/                          # real-stdio integration (real SDK, hermetic agent)
+│   └── acp_real/
+│       └── test_stream_wrappers.py      # Stream wrappers + spawn pipeline against the echo agent
 └── helpers/                             # DSL: KaganDriver, FakeAgent, fixtures
+    ├── acp_loopback.py                   # TCP-loopback fixture yielding real asyncio.StreamReader/Writer
+    └── echo_agent.py                     # Vendored ACP echo agent (run as subprocess via sys.executable)
 ```
 
 Name tests as specs: `test_<behavior>_<expected_outcome>`. Each file has 2-6 tests,

pyproject.toml

@@ -133,6 +133,9 @@ max-complexity = 20
 "src/kagan/server/_system_routes.py" = [
   "C901",
 ] # Route registration nests all handlers inside one function
+"tests/helpers/echo_agent.py" = [
+  "TC002",
+] # Vendored ACP example script run as subprocess — imports must resolve at runtime
 
 [tool.ruff.format]
 docstring-code-format = true

src/kagan/core/_acp_streams.py

@@ -2,25 +2,30 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from loguru import logger
 
-if TYPE_CHECKING:
-    import asyncio
 
-
-class JsonRpcObjectStreamReader:
+class JsonRpcObjectStreamReader(asyncio.StreamReader):
     """Drop non-JSON-RPC stdout lines before the ACP SDK parses them.
 
     ACP transports are line-delimited JSON-RPC. Some Windows agent launchers can
     emit terminal/control output on stdout before or between JSON-RPC frames.
     The upstream ACP SDK logs a full traceback for every such line, so we filter
     non-object JSON here and leave valid frames untouched.
+
+    Subclasses ``asyncio.StreamReader`` because ``ClientSideConnection.__init__``
+    enforces ``isinstance(output_stream, asyncio.StreamReader)``. ``super().__init__``
+    *is* called so base attributes (``_exception``, ``_buffer``, ``_loop``) are
+    populated — methods like ``exception()`` keep working even though we delegate
+    every read to ``self._reader``.
     """
 
     def __init__(self, reader: asyncio.StreamReader, *, backend_name: str) -> None:
+        super().__init__()
         self._reader = reader
         self._backend_name = backend_name
         self._dropped = 0
@@ -59,6 +64,12 @@ async def readexactly(self, n: int) -> bytes:
     def at_eof(self) -> bool:
         return self._reader.at_eof()
 
+    def exception(self) -> BaseException | None:
+        return self._reader.exception()
+
+    def set_exception(self, exc: BaseException) -> None:
+        self._reader.set_exception(exc)
+
     def _record_drop(self, line: bytes, *, reason: str) -> None:
         self._dropped += 1
         if self._dropped > 1:

src/kagan/core/_agent.py

@@ -866,13 +866,19 @@ async def spawn_agent(
 _MAX_CUMULATIVE_BYTES: Final[int] = 500 * 1024 * 1024  # 500 MB total per session
 
 
-class _ByteCountingStreamReader:
+class _ByteCountingStreamReader(asyncio.StreamReader):
     """Composition wrapper that enforces a cumulative byte cap on reads.
 
     The ACP JSON-RPC read loop calls ``readline()`` or ``read()`` on the
     underlying reader.  This wrapper counts every byte returned and terminates
     the associated process when the cumulative limit is exceeded, preventing
     unbounded memory growth.
+
+    Subclasses ``asyncio.StreamReader`` because ``ClientSideConnection.__init__``
+    enforces ``isinstance(output_stream, asyncio.StreamReader)``. ``super().__init__``
+    *is* called so base attributes (``_exception``, ``_buffer``, ``_loop``) are
+    populated — methods like ``exception()`` keep working even though we delegate
+    every read to ``self._reader``.
     """
 
     def __init__(
@@ -881,6 +887,7 @@ def __init__(
         process: asyncio.subprocess.Process,
         cumulative_limit: int = _MAX_CUMULATIVE_BYTES,
     ) -> None:
+        super().__init__()
         self._reader = reader
         self._process = process
         self._cumulative_bytes = 0
@@ -924,6 +931,12 @@ async def readexactly(self, n: int) -> bytes:
     def at_eof(self) -> bool:
         return self._reader.at_eof()
 
+    def exception(self) -> BaseException | None:
+        return self._reader.exception()
+
+    def set_exception(self, exc: BaseException) -> None:
+        self._reader.set_exception(exc)
+
 
 async def spawn_agent_via_acp(
     backend_name: str,

tests/helpers/acp_loopback.py

@@ -0,0 +1,81 @@
+"""TCP-loopback fixture yielding real ``asyncio`` stream pairs.
+
+Ported from ``references/acp/tests/conftest.py::_Server`` (anthropics/agent-protocol).
+Lets tests construct the real ``acp.client.connection.ClientSideConnection`` over
+honest ``asyncio.StreamReader``/``asyncio.StreamWriter`` instances without
+spawning a subprocess. Catches contracts the SDK enforces at construction time
+(``isinstance`` gates) and at runtime (read-method shape) that hand-rolled
+fakes routinely under-specify.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+
+
+class AcpLoopback:
+    """Async context manager that yields two end-to-end connected stream pairs.
+
+    On enter, opens a TCP server bound to ``127.0.0.1:0`` and a client
+    connection to it. The "server" side acts as the agent (writes responses,
+    reads requests); the "client" side is what Kagan would normally hand to
+    ``ClientSideConnection``.
+    """
+
+    def __init__(self) -> None:
+        self._server: asyncio.AbstractServer | None = None
+        self._server_reader: asyncio.StreamReader | None = None
+        self._server_writer: asyncio.StreamWriter | None = None
+        self._client_reader: asyncio.StreamReader | None = None
+        self._client_writer: asyncio.StreamWriter | None = None
+
+    async def __aenter__(self) -> AcpLoopback:
+        ready = asyncio.Event()
+
+        async def handle(
+            reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+        ) -> None:
+            self._server_reader = reader
+            self._server_writer = writer
+            ready.set()
+
+        self._server = await asyncio.start_server(handle, host="127.0.0.1", port=0)
+        host, port = self._server.sockets[0].getsockname()[:2]
+        self._client_reader, self._client_writer = await asyncio.open_connection(
+            host, port
+        )
+        await asyncio.wait_for(ready.wait(), timeout=2.0)
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        del exc_type, exc, tb
+        for writer in (self._client_writer, self._server_writer):
+            if writer is None:
+                continue
+            writer.close()
+            with contextlib.suppress(Exception):
+                await writer.wait_closed()
+        if self._server is not None:
+            self._server.close()
+            await self._server.wait_closed()
+
+    @property
+    def client_reader(self) -> asyncio.StreamReader:
+        assert self._client_reader is not None
+        return self._client_reader
+
+    @property
+    def client_writer(self) -> asyncio.StreamWriter:
+        assert self._client_writer is not None
+        return self._client_writer
+
+    @property
+    def server_reader(self) -> asyncio.StreamReader:
+        assert self._server_reader is not None
+        return self._server_reader
+
+    @property
+    def server_writer(self) -> asyncio.StreamWriter:
+        assert self._server_writer is not None
+        return self._server_writer

tests/helpers/echo_agent.py

@@ -0,0 +1,105 @@
+"""Hermetic ACP echo agent for stdio integration tests.
+
+Vendored from ``references/acp/examples/echo_agent.py`` (anthropics/agent-protocol).
+The agent speaks the real ACP wire protocol and only depends on the ``acp``
+package we already pin. Tests spawn this script as a subprocess via
+``spawn_filtered_agent_process`` to exercise the full handshake → prompt →
+notification → teardown path with no provider binary on PATH.
+"""
+
+# /// script
+# requires-python = ">=3.10,<3.15"
+# dependencies = [
+#     "agent-client-protocol",
+# ]
+# ///
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from uuid import uuid4
+
+from acp import (
+    Agent,
+    InitializeResponse,
+    NewSessionResponse,
+    PromptResponse,
+    run_agent,
+    text_block,
+    update_agent_message,
+)
+from acp.interfaces import Client
+from acp.schema import (
+    AudioContentBlock,
+    ClientCapabilities,
+    EmbeddedResourceContentBlock,
+    HttpMcpServer,
+    ImageContentBlock,
+    Implementation,
+    McpServerStdio,
+    ResourceContentBlock,
+    SseMcpServer,
+    TextContentBlock,
+)
+
+
+class EchoAgent(Agent):
+    _conn: Client
+
+    def on_connect(self, conn: Client) -> None:
+        self._conn = conn
+
+    async def initialize(
+        self,
+        protocol_version: int,
+        client_capabilities: ClientCapabilities | None = None,
+        client_info: Implementation | None = None,
+        **kwargs: Any,
+    ) -> InitializeResponse:
+        del client_capabilities, client_info, kwargs
+        return InitializeResponse(protocol_version=protocol_version)
+
+    async def new_session(
+        self,
+        cwd: str,
+        mcp_servers: list[HttpMcpServer | SseMcpServer | McpServerStdio],
+        **kwargs: Any,
+    ) -> NewSessionResponse:
+        del cwd, mcp_servers, kwargs
+        return NewSessionResponse(session_id=uuid4().hex)
+
+    async def prompt(
+        self,
+        prompt: list[
+            TextContentBlock
+            | ImageContentBlock
+            | AudioContentBlock
+            | ResourceContentBlock
+            | EmbeddedResourceContentBlock
+        ],
+        session_id: str,
+        **kwargs: Any,
+    ) -> PromptResponse:
+        del kwargs
+        for block in prompt:
+            text = (
+                block.get("text", "")
+                if isinstance(block, dict)
+                else getattr(block, "text", "")
+            )
+            chunk = update_agent_message(text_block(text))
+            chunk.field_meta = {"echo": True}
+            chunk.content.field_meta = {"echo": True}
+
+            await self._conn.session_update(
+                session_id=session_id, update=chunk, source="echo_agent"
+            )
+        return PromptResponse(stop_reason="end_turn")
+
+
+async def main() -> None:
+    await run_agent(EchoAgent())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())