test(errors,transport): regression coverage + CVE-2026-4539 fix (#122)

## Summary

Consolidates PR #111 and PR #116 into a single, clean PR with correct
authorship:

- **test(errors)**: Regression tests for remote JSON-RPC error mapping —
metadata parsing, recoverable vs fatal classification, taxonomy
fallback, and code slot normalization
- **test(transport)**: ASAP stream version-negotiation coverage —
comma-separated `ASAP-Version` header support and fail-closed
`VERSION_INCOMPATIBLE` behavior
- **build(deps)**: Override `pygments>=2.20.0` for CVE-2026-4539 and pin
`pymdown-extensions>=10.21` for compatibility

## Supersedes

- Closes #111
- Closes #116

## Test plan

- [x] `ruff check .` — no issues
- [x] `ruff format --check .` — all formatted
- [x] `mypy src/ scripts/ tests/` — no issues (342 files)
- [x] Full test suite: **2906 passed**, 7 skipped, **90.61% coverage**

Author: adriannoes

pyproject.toml

@@ -63,6 +63,7 @@ docs = [
     "ghp-import>=2.1.0",
     "mkdocs-material>=9.5",
     "mkdocstrings[python]>=0.27",
+    "pymdown-extensions>=10.21",
 ]
 dns-sd = [
     "zeroconf>=0.80",
@@ -110,6 +111,8 @@ Issues = "https://github.com/adriannoes/asap-protocol/issues"
 # - pillow>=12.2.0 for CVE-2026-40192 (transitive, e.g. pdf/vision stacks)
 # - pytest>=9.0.3 for CVE-2025-71176 (dev)
 # - python-multipart>=0.0.26 for CVE-2026-40347 (FastAPI stack)
+# - pygments>=2.20.0 for CVE-2026-4539
+# - langsmith>=0.7.31 for GHSA-rr7j-v2q5-chgv
 [tool.uv]
 override-dependencies = [
     "pyjwt>=2.12.0",
@@ -120,6 +123,8 @@ override-dependencies = [
     "pillow>=12.2.0",
     "pytest>=9.0.3",
     "python-multipart>=0.0.26",
+    "pygments>=2.20.0",
+    "langsmith>=0.7.31",
 ]
 
 [project.scripts]

tests/test_errors.py

@@ -2,14 +2,20 @@
 
 from asap.errors import (
     ASAPError,
+    RPC_CONNECTION_ERROR,
     InvalidTransitionError,
     MalformedEnvelopeError,
     RPC_REMOTE_GENERIC,
     RecoverableError,
+    RemoteFatalRPCError,
+    RemoteRecoverableRPCError,
     TaskNotFoundError,
     TaskAlreadyCompletedError,
     ThreadPoolExhaustedError,
     ASAPConnectionError,
+    is_asap_json_rpc_code,
+    jsonrpc_error_data_for_asap_exception,
+    remote_rpc_error_from_json,
 )
 
 
@@ -294,3 +300,101 @@ def test_thread_pool_exhausted_error_to_dict(self) -> None:
         assert result["details"]["max_threads"] == 15
         assert result["details"]["active_threads"] == 15
         assert result.get("retry_after_ms") is not None
+
+
+class TestJsonRpcErrorInterop:
+    """Tests for JSON-RPC interop and metadata shaping."""
+
+    def test_asap_json_rpc_code_boundaries(self) -> None:
+        """ASAP JSON-RPC helper accepts only reserved code range."""
+        assert is_asap_json_rpc_code(-32059) is True
+        assert is_asap_json_rpc_code(-32000) is True
+        assert is_asap_json_rpc_code(-31999) is False
+        assert is_asap_json_rpc_code(-32060) is False
+
+    def test_asap_error_rejects_out_of_range_rpc_code(self) -> None:
+        """Base ASAPError rejects JSON-RPC codes outside ASAP reserved slot."""
+        import pytest
+
+        with pytest.raises(ValueError, match="rpc_code must be in"):
+            ASAPError(code="asap:test/error", message="bad rpc", rpc_code=-32603)
+
+    def test_remote_rpc_error_from_json_returns_recoverable_and_strips_meta(self) -> None:
+        """Recoverable remote payload is typed and metadata fields are normalized."""
+        err = remote_rpc_error_from_json(
+            rpc_code=-32033,
+            message="remote timeout",
+            data={
+                "recoverable": True,
+                "retry_after_ms": 250,
+                "alternative_agents": ["urn:asap:agent:secondary"],
+                "fallback_action": "retry_with_backoff",
+                "asap_taxonomy_code": "asap:transport/timeout",
+                "rpc_code": -32033,
+                "upstream": "worker-1",
+            },
+        )
+
+        assert isinstance(err, RemoteRecoverableRPCError)
+        assert err.rpc_code == -32033
+        assert err.json_rpc_code == -32033
+        assert err.code == "asap:transport/timeout"
+        assert err.retry_after_ms == 250
+        assert err.alternative_agents == ["urn:asap:agent:secondary"]
+        assert err.fallback_action == "retry_with_backoff"
+        assert err.details == {"upstream": "worker-1"}
+
+    def test_remote_rpc_error_from_json_uses_code_field_fallback(self) -> None:
+        """A taxonomy-like `code` field in data overrides default taxonomy mapping."""
+        err = remote_rpc_error_from_json(
+            rpc_code=-32040,
+            message="resource exhausted",
+            data={
+                "recoverable": False,
+                "code": "asap:resource/exhausted",
+                "queue_depth": 11,
+            },
+        )
+
+        assert isinstance(err, RemoteFatalRPCError)
+        assert err.code == "asap:resource/exhausted"
+        assert err.details == {"queue_depth": 11}
+
+    def test_remote_rpc_error_preserves_wire_code_but_normalizes_internal_slot(self) -> None:
+        """Non-ASAP wire code is preserved while internal rpc_code uses remote-generic slot."""
+        err = remote_rpc_error_from_json(
+            rpc_code=-32603,
+            message="internal error",
+            data={"recoverable": False, "foo": "bar"},
+        )
+
+        assert isinstance(err, RemoteFatalRPCError)
+        assert err.json_rpc_code == -32603
+        assert err.rpc_code == RPC_REMOTE_GENERIC
+        assert err.details == {"foo": "bar"}
+
+    def test_jsonrpc_error_data_for_recoverable_exception(self) -> None:
+        """Server-side JSON-RPC data payload includes taxonomy and recoverability hints."""
+        err = ASAPConnectionError(
+            "connection refused",
+            url="https://agent.example.test/asap",
+            rpc_code=RPC_CONNECTION_ERROR,
+            retry_after_ms=500,
+            alternative_agents=["urn:asap:agent:fallback"],
+        )
+
+        payload = jsonrpc_error_data_for_asap_exception(err)
+        assert payload["code"] == "asap:transport/connection_error"
+        assert payload["rpc_code"] == RPC_CONNECTION_ERROR
+        assert payload["recoverable"] is True
+        assert payload["asap_taxonomy_code"] == "asap:transport/connection_error"
+        assert payload["retry_after_ms"] == 500
+        assert payload["alternative_agents"] == ["urn:asap:agent:fallback"]
+
+    def test_connection_error_message_not_duplicated_when_verify_text_present(self) -> None:
+        """Connection errors with existing troubleshooting guidance are not duplicated."""
+        err = ASAPConnectionError(
+            "Verify service availability before retrying.",
+            url="https://agent.example.test/asap",
+        )
+        assert str(err) == "Verify service availability before retrying."

tests/transport/test_streaming.py

@@ -9,12 +9,13 @@
 import pytest
 from httpx import ASGITransport
 
+from asap.models.constants import ASAP_DEFAULT_TRANSPORT_VERSION, ASAP_VERSION_HEADER
 from asap.models.entities import Manifest
 from asap.models.envelope import Envelope
 from asap.models.enums import TaskStatus
 from asap.models.payloads import TaskRequest, TaskStream
 from asap.transport.handlers import HandlerRegistry, create_echo_handler
-from asap.transport.jsonrpc import ASAP_METHOD
+from asap.transport.jsonrpc import ASAP_METHOD, VERSION_INCOMPATIBLE
 from asap.transport.server import create_app
 
 if TYPE_CHECKING:
@@ -119,3 +120,113 @@ async def test_asap_stream_sse_endpoint(
     assert events[0].payload_dict.get("final") is False
     assert events[-1].payload_dict.get("final") is True
     assert events[-1].payload_dict.get("status") == "completed"
+
+
+@pytest.mark.anyio
+async def test_asap_stream_negotiates_first_supported_wire_version(
+    sample_manifest: Manifest,
+    isolated_rate_limiter: ASAPRateLimiter | None,
+) -> None:
+    """`/asap/stream` uses the first supported ASAP-Version token and echoes it."""
+    registry = HandlerRegistry()
+    registry.register("task.request", create_echo_handler())
+    registry.register_streaming_handler("task.request", _word_stream_handler)
+    app = create_app(sample_manifest, registry, rate_limit="999999/minute")
+    if isolated_rate_limiter is not None:
+        app.state.limiter = isolated_rate_limiter
+
+    env = Envelope(
+        asap_version="0.1",
+        sender="urn:asap:agent:client",
+        recipient=sample_manifest.id,
+        payload_type="task.request",
+        payload=TaskRequest(
+            conversation_id="conv-version-stream-ok",
+            skill_id="echo",
+            input={"text": "one two"},
+        ).model_dump(),
+        correlation_id="corr-version-stream-ok",
+    )
+    rpc = {
+        "jsonrpc": "2.0",
+        "method": ASAP_METHOD,
+        "params": {"envelope": env.model_dump(mode="json")},
+        "id": 2,
+    }
+
+    transport = ASGITransport(app=app)
+    async with (
+        httpx.AsyncClient(transport=transport, base_url="http://test") as client,
+        client.stream(
+            "POST",
+            "/asap/stream",
+            json=rpc,
+            headers={ASAP_VERSION_HEADER: "2.2, 2.1"},
+        ) as response,
+    ):
+        assert response.status_code == 200
+        assert response.headers.get(ASAP_VERSION_HEADER) == "2.2"
+        assert response.headers["content-type"].startswith("text/event-stream")
+
+        buf = ""
+        events: list[Envelope] = []
+        async for chunk in response.aiter_text():
+            buf += chunk
+            while "\n\n" in buf:
+                raw_event, buf = buf.split("\n\n", 1)
+                for line in raw_event.split("\n"):
+                    stripped = line.strip()
+                    if stripped.startswith("data:"):
+                        payload_json = stripped[5:].strip()
+                        events.append(Envelope.model_validate(json.loads(payload_json)))
+
+    assert len(events) == 2
+    assert events[-1].payload_dict.get("final") is True
+
+
+@pytest.mark.anyio
+async def test_asap_stream_rejects_incompatible_wire_version(
+    sample_manifest: Manifest,
+    isolated_rate_limiter: ASAPRateLimiter | None,
+) -> None:
+    """`/asap/stream` returns JSON-RPC VERSION_INCOMPATIBLE when header has no match."""
+    registry = HandlerRegistry()
+    registry.register("task.request", create_echo_handler())
+    registry.register_streaming_handler("task.request", _word_stream_handler)
+    app = create_app(sample_manifest, registry, rate_limit="999999/minute")
+    if isolated_rate_limiter is not None:
+        app.state.limiter = isolated_rate_limiter
+
+    env = Envelope(
+        asap_version="0.1",
+        sender="urn:asap:agent:client",
+        recipient=sample_manifest.id,
+        payload_type="task.request",
+        payload=TaskRequest(
+            conversation_id="conv-version-stream-bad",
+            skill_id="echo",
+            input={"text": "one two"},
+        ).model_dump(),
+        correlation_id="corr-version-stream-bad",
+    )
+    rpc = {
+        "jsonrpc": "2.0",
+        "method": ASAP_METHOD,
+        "params": {"envelope": env.model_dump(mode="json")},
+        "id": 3,
+    }
+
+    transport = ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/asap/stream",
+            json=rpc,
+            headers={ASAP_VERSION_HEADER: "9.9"},
+        )
+
+    assert response.status_code == 200
+    assert response.headers.get(ASAP_VERSION_HEADER) == ASAP_DEFAULT_TRANSPORT_VERSION
+    assert response.headers["content-type"].startswith("application/json")
+    payload = response.json()
+    assert payload.get("error", {}).get("code") == VERSION_INCOMPATIBLE
+    assert payload.get("error", {}).get("data", {}).get("requested") == "9.9"