test(connection): harden unhandled-handler-exception regression tests

Rewrites the request/notification error-logging tests to be deterministic
and precise instead of only asserting that "some record has exc_info".

- Drive Connection._run_request / _run_notification directly with a
  recording sender (mirroring test_core's TrackingSender) so assertions
  cover the exact wire frame: a request emits one -32603 error carrying
  {"details": ...} and re-raises RequestError; a notification emits
  nothing (deterministic, no timeout read).
- Assert the logged record is the original RuntimeError (type + message)
  under the expected method=..., not merely that a record exists.
- Add an end-to-end test (via the connect fixture, like
  test_cancel_prompt_flow) proving a real agent cancel handler that raises
  is logged and, crucially, does not tear down the connection: a
  follow-up initialize still round-trips.

Confirmed the suite fails on the pre-fix behaviour (silent suppress / no
request log) and passes with it.

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>

Author: michelTho

tests/test_request_error_logging.py

@@ -1,78 +1,136 @@
+"""Unhandled RPC handler exceptions must be logged instead of silently swallowed.
+
+Requests already returned a JSON-RPC -32603 error but discarded the original
+traceback; notifications suppressed the exception entirely. Both now log the
+underlying exception so integrators (e.g. Sentry via its logging integration)
+can see server-side handler crashes.
+"""
+
+from __future__ import annotations
+
 import asyncio
-import json
 import logging
-from typing import cast
+from typing import Any
+from unittest.mock import MagicMock
 
 import pytest
 
-from acp import Agent
-from acp.core import AgentSideConnection
+from acp.connection import Connection, MethodHandler
+from acp.exceptions import RequestError
 from tests.conftest import TestAgent
 
 
+class _RecordingSender:
+    """Duck-typed MessageSender that records outgoing frames instead of writing them."""
+
+    def __init__(self, writer: asyncio.StreamWriter, supervisor: Any) -> None:
+        self.sent: list[dict[str, Any]] = []
+
+    async def send(self, payload: dict[str, Any]) -> None:
+        self.sent.append(payload)
+
+    async def close(self) -> None:
+        pass
+
+
+def _make_connection(handler: MethodHandler) -> tuple[Connection, _RecordingSender]:
+    captured: dict[str, _RecordingSender] = {}
+
+    def sender_factory(writer: asyncio.StreamWriter, supervisor: Any) -> _RecordingSender:
+        captured["sender"] = _RecordingSender(writer, supervisor)
+        return captured["sender"]
+
+    conn = Connection(handler, MagicMock(), MagicMock(), sender_factory=sender_factory, listening=False)
+    return conn, captured["sender"]
+
+
+async def _raising_handler(method: str, params: Any, is_notification: bool) -> Any:
+    raise RuntimeError("kaboom")
+
+
+def _assert_logged_runtime_error(caplog: pytest.LogCaptureFixture, method: str) -> None:
+    records = [
+        record
+        for record in caplog.records
+        if record.levelno == logging.ERROR and record.exc_info and f"method={method}" in record.getMessage()
+    ]
+    assert len(records) == 1, f"expected exactly one logged error for method={method}"
+    exc_info = records[0].exc_info
+    assert exc_info is not None
+    logged = exc_info[1]
+    assert isinstance(logged, RuntimeError)
+    assert str(logged) == "kaboom"
+
+
 @pytest.mark.asyncio
-async def test_unexpected_handler_error_is_logged_and_returns_internal_error(server, caplog):
-    class _RaisingAgent(TestAgent):
-        __test__ = False
-
-        async def initialize(
-            self,
-            protocol_version,
-            client_capabilities=None,
-            client_info=None,
-            **kwargs,
-        ):
-            raise RuntimeError("boom")
-
-    AgentSideConnection(
-        cast(Agent, _RaisingAgent()),
-        server.server_writer,
-        server.server_reader,
-        listening=True,
-    )
-
-    req = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": 1}}
-    with caplog.at_level(logging.ERROR):
-        server.client_writer.write((json.dumps(req) + "\n").encode())
-        await server.client_writer.drain()
-        line = await asyncio.wait_for(server.client_reader.readline(), timeout=1)
+async def test_run_request_unhandled_exception_is_logged_and_returned_as_internal_error(caplog):
+    conn, sender = _make_connection(_raising_handler)
+    request = {"jsonrpc": "2.0", "id": 7, "method": "explode", "params": None}
+
+    try:
+        with caplog.at_level(logging.ERROR), pytest.raises(RequestError) as exc_info:
+            await conn._run_request(request)
+    finally:
+        await conn.close()
+
+    # The handler exception is re-raised as a JSON-RPC internal error...
+    raised = exc_info.value
+    assert isinstance(raised, RequestError)
+    assert raised.code == -32603
+    assert raised.data == {"details": "kaboom"}
 
-    resp = json.loads(line)
-    assert resp["id"] == 1
-    assert resp["error"]["code"] == -32603  # internal error
+    # ...and exactly one error frame carrying the handler's message is written to the peer.
+    assert len(sender.sent) == 1
+    response = sender.sent[0]
+    assert response["id"] == 7
+    assert "result" not in response
+    assert response["error"] == {"code": -32603, "message": "Internal error", "data": {"details": "kaboom"}}
 
-    # The original exception (with its traceback) must be logged, not silently
-    # swallowed into the JSON-RPC response.
-    assert any(record.exc_info for record in caplog.records)
+    # The original exception is logged, not discarded by `raise err from None`.
+    _assert_logged_runtime_error(caplog, "explode")
 
 
 @pytest.mark.asyncio
-async def test_unexpected_notification_error_is_logged_and_not_swallowed(server, caplog):
-    class _RaisingCancelAgent(TestAgent):
-        __test__ = False
-
-        def __init__(self) -> None:
-            super().__init__()
-            self.cancel_called = asyncio.Event()
-
-        async def cancel(self, session_id, **kwargs):
-            self.cancel_called.set()
-            raise RuntimeError("boom")
-
-    agent = _RaisingCancelAgent()
-    AgentSideConnection(
-        cast(Agent, agent),
-        server.server_writer,
-        server.server_reader,
-        listening=True,
-    )
-
-    note = {"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": "s1"}}
+async def test_run_notification_unhandled_exception_is_logged_and_not_answered(caplog):
+    conn, sender = _make_connection(_raising_handler)
+    notification = {"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": "s1"}}
+
+    try:
+        with caplog.at_level(logging.ERROR):
+            result = await conn._run_notification(notification)
+    finally:
+        await conn.close()
+
+    # A notification has no response: the error is neither raised nor written to the wire.
+    assert result is None
+    assert sender.sent == []
+
+    # It must still be logged — previously contextlib.suppress dropped it silently.
+    _assert_logged_runtime_error(caplog, "session/cancel")
+
+
+class _RaisingCancelAgent(TestAgent):
+    __test__ = False
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.cancel_called = asyncio.Event()
+
+    async def cancel(self, session_id: str, **kwargs: Any) -> None:
+        self.cancel_called.set()
+        raise RuntimeError("kaboom")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("agent", [_RaisingCancelAgent()])
+async def test_agent_notification_handler_error_is_logged_and_connection_survives(connect, agent, caplog):
+    _, client_conn = connect()
+
     with caplog.at_level(logging.ERROR):
-        server.client_writer.write((json.dumps(note) + "\n").encode())
-        await server.client_writer.drain()
+        await client_conn.cancel(session_id="sess-1")
         await asyncio.wait_for(agent.cancel_called.wait(), timeout=1)
+        # The crashing handler must not tear down the connection: a follow-up request still round-trips.
+        response = await asyncio.wait_for(client_conn.initialize(protocol_version=1), timeout=1)
 
-    # A notification handler that raises must be logged, not silently swallowed,
-    # so integrators can surface it (e.g. to Sentry).
-    assert any(record.exc_info for record in caplog.records)
+    assert response.protocol_version == 1
+    _assert_logged_runtime_error(caplog, "session/cancel")