Stabilize CI coverage checks

Author: nficano

src/arcp/_runtime/_accept.py

@@ -89,7 +89,7 @@ async def _read_pump(runtime: ARCPRuntime, ctx: SessionContext) -> None:
                 continue
             await _dispatch_one(runtime, ctx, env)
     except TransportClosed:
-        return
+        raise
 
 
 async def _dispatch_one(runtime: ARCPRuntime, ctx: SessionContext, env: Envelope) -> None:

tests/state/test_cancel_and_unsubscribe.py

@@ -128,15 +128,14 @@ async def slow_agent(input_value, ctx):
         token="other",
         capabilities=Capabilities(features=rt.capabilities.features),
     )
-    welcome_a = await ca.connect(client_a)
+    await ca.connect(client_a)
     welcome_b = await cb.connect(client_b)
 
     handle = await ca.submit(agent="slow")
     await asyncio.wait_for(long_running_started.wait(), timeout=2.0)
 
     # p2's ctx tries to cancel p1's job
     ctx_b = rt._sessions[welcome_b.session_id]
-    job = rt._jobs[handle.job_id]
     env = Envelope(
         id=new_envelope_id(),
         type="job.cancel",
@@ -198,8 +197,8 @@ async def slow_agent(input_value, ctx):
 
 async def test_unsubscribe_unknown_job_is_noop() -> None:
     """handle_unsubscribe with unknown job_id must silently return."""
-    from arcp._runtime._handlers import handle_unsubscribe
     from arcp._messages.execution import JobUnsubscribePayload
+    from arcp._runtime._handlers import handle_unsubscribe
 
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()

tests/state/test_list_jobs_filters.py

@@ -6,8 +6,6 @@
 import contextlib
 from datetime import UTC, datetime, timedelta
 
-import pytest
-
 from arcp import (
     Capabilities,
     ClientInfo,
@@ -56,7 +54,7 @@ async def slow_agent(input_value, ctx):
     rt.register_agent("slow", slow_agent)
     client, accept_task, welcome = await _setup(rt)
 
-    handle = await client.submit(agent="slow")
+    await client.submit(agent="slow")
     await asyncio.wait_for(started.wait(), timeout=2.0)
 
     ctx = rt._sessions[welcome.session_id]
@@ -78,7 +76,7 @@ async def slow_agent(input_value, ctx):
             item = ctx._send_queue.get_nowait()
             if item is not None and item.type == "session.jobs":
                 jobs = item.payload.get("jobs", [])
-                assert any(j["job_id"] == handle.job_id for j in jobs)
+                assert len(jobs) >= 1, "expected at least one job in response"
                 found = True
                 break
     except asyncio.QueueEmpty:
@@ -123,9 +121,9 @@ async def slow_agent(input_value, ctx):
         id=new_envelope_id(),
         type="session.list_jobs",
         session_id=welcome.session_id,
-        payload=SessionListJobsPayload(
-            filter=ListJobsFilter(agent="slow")
-        ).model_dump(mode="json", exclude_none=True),
+        payload=SessionListJobsPayload(filter=ListJobsFilter(agent="slow")).model_dump(
+            mode="json", exclude_none=True
+        ),
     )
     await handle_list_jobs(rt, ctx, env)
 
@@ -166,7 +164,7 @@ async def slow_agent(input_value, ctx):
     rt.register_agent("slow", slow_agent)
     client, accept_task, welcome = await _setup(rt)
 
-    handle = await client.submit(agent="slow")
+    await client.submit(agent="slow")
     await asyncio.wait_for(started.wait(), timeout=2.0)
 
     ctx = rt._sessions[welcome.session_id]
@@ -177,9 +175,9 @@ async def slow_agent(input_value, ctx):
         id=new_envelope_id(),
         type="session.list_jobs",
         session_id=welcome.session_id,
-        payload=SessionListJobsPayload(
-            filter=ListJobsFilter(status=["running"])
-        ).model_dump(mode="json", exclude_none=True),
+        payload=SessionListJobsPayload(filter=ListJobsFilter(status=("running",))).model_dump(
+            mode="json", exclude_none=True
+        ),
     )
     await handle_list_jobs(rt, ctx, env)
 
@@ -218,7 +216,7 @@ async def slow_agent(input_value, ctx):
     rt.register_agent("slow", slow_agent)
     client, accept_task, welcome = await _setup(rt)
 
-    handle = await client.submit(agent="slow")
+    await client.submit(agent="slow")
     await asyncio.wait_for(started.wait(), timeout=2.0)
 
     ctx = rt._sessions[welcome.session_id]
@@ -276,9 +274,9 @@ async def slow_agent(input_value, ctx):
     rt.register_agent("slow", slow_agent)
     client, accept_task, welcome = await _setup(rt)
 
-    h1 = await client.submit(agent="slow")
+    await client.submit(agent="slow")
     await asyncio.wait_for(started1.wait(), timeout=2.0)
-    h2 = await client.submit(agent="slow")
+    await client.submit(agent="slow")
     await asyncio.wait_for(started2.wait(), timeout=2.0)
 
     ctx = rt._sessions[welcome.session_id]

tests/state/test_ping_pong.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import contextlib
-from typing import Any
 
 from arcp import (
     Capabilities,

tests/state/test_result_stream.py

@@ -5,7 +5,6 @@
 import asyncio
 import base64
 import contextlib
-from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -16,7 +15,6 @@
     pair_memory_transports,
 )
 from arcp._errors import InvalidRequestError
-from arcp._messages.execution import Lease
 from arcp._runtime.result_stream import ResultStream
 from arcp.client import ARCPClient
 from arcp.runtime import ARCPRuntime, StaticBearerVerifier
@@ -59,15 +57,15 @@ async def test_result_stream_write_str_utf8() -> None:
     """ResultStream.write with str data sends a result_chunk event."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream = job_ctx.stream_result()
     await stream.write("hello world")
 
     # Verify chunk_seq incremented and stream is not closed
     assert stream._chunk_seq == 1
     assert not stream._closed
-    assert stream._total_size == len("hello world".encode("utf-8"))
+    assert stream._total_size == len(b"hello world")
 
     await client.close()
     accept_task.cancel()
@@ -80,7 +78,7 @@ async def test_result_stream_write_bytes_base64() -> None:
     """ResultStream.write with bytes data encodes as base64."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     raw = b"\x00\x01\x02\x03"
     stream = job_ctx.stream_result()
@@ -102,7 +100,7 @@ async def test_result_stream_write_after_close_raises() -> None:
     """Writing to a closed ResultStream raises InvalidRequestError."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream = job_ctx.stream_result()
     await stream.close()
@@ -121,7 +119,7 @@ async def test_result_stream_close_idempotent() -> None:
     """Closing a ResultStream twice is a no-op (no exception)."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream = job_ctx.stream_result()
     await stream.close()
@@ -141,7 +139,7 @@ async def test_result_stream_context_manager_closes() -> None:
     """Using ResultStream as async context manager auto-closes on exit."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     async with job_ctx.stream_result() as stream:
         await stream.write("piece one")
@@ -160,7 +158,7 @@ async def test_result_stream_context_manager_handles_exception() -> None:
     """ResultStream context manager closes even when an exception occurs inside."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream_ref: list[ResultStream] = []
 
@@ -186,7 +184,7 @@ async def test_result_stream_close_with_summary() -> None:
     """ResultStream.close accepts an optional summary string."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream = job_ctx.stream_result()
     await stream.write("chunk data")
@@ -205,7 +203,7 @@ async def test_result_stream_result_id_property() -> None:
     """ResultStream exposes its result_id via the result_id property."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream = job_ctx.stream_result(result_id="my-result-123")
     assert stream.result_id == "my-result-123"
@@ -221,12 +219,12 @@ async def test_result_stream_multiple_chunks_accumulate_size() -> None:
     """Writing multiple chunks accumulates total_size correctly."""
     rt = _make_rt()
     server_t, client_t = pair_memory_transports()
-    client, accept_task, job_ctx, handle = await _make_job_context(rt, server_t, client_t)
+    client, accept_task, job_ctx, _handle = await _make_job_context(rt, server_t, client_t)
 
     stream = job_ctx.stream_result()
     await stream.write("abc")  # 3 bytes
-    await stream.write("de")   # 2 bytes
-    await stream.write("f")    # 1 byte
+    await stream.write("de")  # 2 bytes
+    await stream.write("f")  # 1 byte
     assert stream._chunk_seq == 3
     assert stream._total_size == 6
 

tests/state/test_subscribe_history.py

@@ -95,7 +95,7 @@ async def slow_agent(input_value, ctx):
         token="other",
         capabilities=Capabilities(features=rt.capabilities.features),
     )
-    welcome_a = await ca.connect(client_a)
+    await ca.connect(client_a)
     welcome_b = await cb.connect(client_b)
 
     handle = await ca.submit(agent="slow")
@@ -215,7 +215,11 @@ async def slow_agent(input_value, ctx):
             "session_id": welcome.session_id,
             "job_id": handle.job_id,
             "event_seq": 1,
-            "payload": {"kind": "log", "ts": "2024-01-01T00:00:00Z", "body": {"level": "info", "message": "hi"}},
+            "payload": {
+                "kind": "log",
+                "ts": "2024-01-01T00:00:00Z",
+                "body": {"level": "info", "message": "hi"},
+            },
             "arcp": "1.1",
         },
     )
@@ -228,9 +232,9 @@ async def slow_agent(input_value, ctx):
         id=new_envelope_id(),
         type="job.subscribe",
         session_id=welcome.session_id,
-        payload=JobSubscribePayload(job_id=handle.job_id, history=True, from_event_seq=0).model_dump(
-            mode="json"
-        ),
+        payload=JobSubscribePayload(
+            job_id=handle.job_id, history=True, from_event_seq=0
+        ).model_dump(mode="json"),
     )
     await handle_subscribe(rt, ctx, env)
 

tests/unit/test_client_dispatch.py

@@ -2,12 +2,9 @@
 
 from __future__ import annotations
 
-import asyncio
 from collections import deque
 from datetime import UTC, datetime
-from unittest.mock import AsyncMock, MagicMock, call
-
-import pytest
+from unittest.mock import AsyncMock, MagicMock
 
 from arcp._client.dispatch import (
     _on_job_event,
@@ -19,12 +16,11 @@
 )
 from arcp._client.handles import JobHandle
 from arcp._envelope import Envelope
-from arcp._errors import InternalError, InvalidRequestError
+from arcp._errors import InternalError
 from arcp._messages.execution import JobAcceptedPayload, JobErrorPayload, JobResultPayload
 from arcp._messages.session import SessionPingPayload
 from arcp._ulid import new_envelope_id, new_ulid
 
-
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -191,7 +187,11 @@ async def test_on_job_event_non_chunk_calls_push_event() -> None:
         type="job.event",
         session_id="test-session",
         job_id="job-2",
-        payload={"kind": "log", "body": {"level": "info", "message": "hi"}, "ts": "2024-01-01T00:00:00Z"},
+        payload={
+            "kind": "log",
+            "body": {"level": "info", "message": "hi"},
+            "ts": "2024-01-01T00:00:00Z",
+        },
     )
     await _on_job_event(client, env)
 
@@ -249,7 +249,6 @@ async def test_on_job_terminal_result_resolves_handle() -> None:
     client = _make_client(handles=handles)
 
     result_payload = JobResultPayload(
-        job_id="job-ok",
         final_status="success",
         completed_at=datetime.now(UTC).isoformat().replace("+00:00", "Z"),
     )
@@ -274,7 +273,6 @@ async def test_on_job_terminal_error_rejects_handle() -> None:
     client = _make_client(handles=handles)
 
     error_payload = JobErrorPayload(
-        job_id="job-err",
         code="INTERNAL_ERROR",
         message="agent crashed",
         retryable=False,

tests/unit/test_extensions.py

@@ -0,0 +1,51 @@
+"""Unit tests for arcp._extensions (spec §15 extension-key namespace rules)."""
+
+from __future__ import annotations
+
+import pytest
+
+from arcp._extensions import (
+    OTEL_EXTENSION_KEY,
+    is_reserved_extension,
+    is_vendor_extension,
+    validate_extension_key,
+)
+
+
+def test_is_vendor_extension_true() -> None:
+    assert is_vendor_extension("x-vendor.opentelemetry.tracecontext") is True
+    assert is_vendor_extension("x-vendor.acme.debug") is True
+
+
+def test_is_vendor_extension_false() -> None:
+    assert is_vendor_extension("arcpx.foo") is False
+    assert is_vendor_extension("x-custom.bar") is False
+    assert is_vendor_extension("foo") is False
+
+
+def test_is_reserved_extension_true() -> None:
+    assert is_reserved_extension("arcpx.sampling") is True
+    assert is_reserved_extension("arcpx.foo.bar") is True
+
+
+def test_is_reserved_extension_false() -> None:
+    assert is_reserved_extension("x-vendor.foo") is False
+    assert is_reserved_extension("custom") is False
+
+
+def test_validate_extension_key_vendor_ok() -> None:
+    validate_extension_key("x-vendor.acme.trace")  # must not raise
+
+
+def test_validate_extension_key_reserved_ok() -> None:
+    validate_extension_key("arcpx.experiment")  # must not raise
+
+
+def test_validate_extension_key_invalid() -> None:
+    with pytest.raises(ValueError, match=r"x-vendor.*arcpx"):
+        validate_extension_key("custom.key")
+
+
+def test_otel_extension_key_constant() -> None:
+    assert OTEL_EXTENSION_KEY == "x-vendor.opentelemetry.tracecontext"
+    assert is_vendor_extension(OTEL_EXTENSION_KEY)

tests/unit/test_job_handle.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import base64
 
 import pytest