test: add targeted unit/state tests to reach 90% coverage

Add 10 new test files targeting previously uncovered branches:

Unit tests:
- tests/unit/test_lease.py — validate_lease_shape, validate_lease_op,
  is_lease_subset, assert_lease_subset, canonicalize_target,
  initial_budget_from_lease, _parse_iso_utc helpers (~45 tests)
- tests/unit/test_job_unit.py — Job.apply_cost_metric, emit_result,
  emit_error, JobContext property accessors and all emitters (~50 tests)
- tests/unit/test_job_handle.py — JobHandle property accessors and
  streaming (chunks, collect_chunks, done awaitable) (13 tests)
- tests/unit/test_client_dispatch.py — dispatch handlers: session.ping,
  session.error, session.bye, job.event, job.result, job.error (14 tests)

State/integration tests (in-memory transport):
- tests/state/test_heartbeat.py — heartbeat_loop: cancellation, loss
  detection, ping enqueue, done-future guard (5 tests)
- tests/state/test_result_stream.py — ResultStream open/write/close (9 tests)
- tests/state/test_list_jobs_filters.py — list_jobs state/agent filters (5 tests)
- tests/state/test_subscribe_history.py — replay history on subscribe (4 tests)
- tests/state/test_cancel_and_unsubscribe.py — cancel/unsubscribe paths (5 tests)
- tests/state/test_ping_pong.py — session.ping/pong round-trip (2 tests)

Result: 304 passed, 3 skipped; total coverage 90.25% (fail-under=90 ✓)

Author: nficano

tests/state/test_cancel_and_unsubscribe.py

@@ -0,0 +1,229 @@
+"""§8.6 job.cancel + §10.3 job.unsubscribe handler coverage."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+
+import pytest
+
+from arcp import (
+    Capabilities,
+    ClientInfo,
+    InvalidRequestError,
+    JobNotFoundError,
+    PermissionDeniedError,
+    RuntimeInfo,
+    pair_memory_transports,
+)
+from arcp._envelope import Envelope
+from arcp._messages.execution import JobCancelPayload
+from arcp._ulid import new_envelope_id
+from arcp.client import ARCPClient
+from arcp.runtime import ARCPRuntime, StaticBearerVerifier
+
+
+def _make_rt() -> ARCPRuntime:
+    return ARCPRuntime(
+        runtime=RuntimeInfo(name="r", version="1"),
+        bearer=StaticBearerVerifier({"tok": "p1"}),
+        heartbeat_interval_sec=None,
+    )
+
+
+async def test_cancel_unknown_job_raises_not_found() -> None:
+    from arcp._runtime._handlers import handle_cancel
+
+    rt = _make_rt()
+    server_t, client_t = pair_memory_transports()
+    accept_task = asyncio.create_task(rt.accept(server_t))
+
+    client = ARCPClient(
+        client=ClientInfo(name="c", version="1"),
+        token="tok",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    welcome = await client.connect(client_t)
+    ctx = rt._sessions[welcome.session_id]
+
+    env = Envelope(
+        id=new_envelope_id(),
+        type="job.cancel",
+        session_id=welcome.session_id,
+        job_id="nonexistent-job",
+        payload=JobCancelPayload(reason="test").model_dump(mode="json"),
+    )
+    with pytest.raises(JobNotFoundError):
+        await handle_cancel(rt, ctx, env)
+
+    await client.close()
+    accept_task.cancel()
+    with contextlib.suppress(asyncio.CancelledError, Exception):
+        await accept_task
+    await rt.close()
+
+
+async def test_cancel_missing_job_id_raises() -> None:
+    from arcp._runtime._handlers import handle_cancel
+
+    rt = _make_rt()
+    server_t, client_t = pair_memory_transports()
+    accept_task = asyncio.create_task(rt.accept(server_t))
+
+    client = ARCPClient(
+        client=ClientInfo(name="c", version="1"),
+        token="tok",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    welcome = await client.connect(client_t)
+    ctx = rt._sessions[welcome.session_id]
+
+    env = Envelope(
+        id=new_envelope_id(),
+        type="job.cancel",
+        session_id=welcome.session_id,
+        job_id=None,  # missing job_id
+        payload=JobCancelPayload(reason="test").model_dump(mode="json"),
+    )
+    with pytest.raises(InvalidRequestError, match="job_id"):
+        await handle_cancel(rt, ctx, env)
+
+    await client.close()
+    accept_task.cancel()
+    with contextlib.suppress(asyncio.CancelledError, Exception):
+        await accept_task
+    await rt.close()
+
+
+async def test_cancel_other_principals_job_raises_permission_denied() -> None:
+    from arcp._runtime._handlers import handle_cancel
+
+    rt = ARCPRuntime(
+        runtime=RuntimeInfo(name="r", version="1"),
+        bearer=StaticBearerVerifier({"tok": "p1", "other": "p2"}),
+        heartbeat_interval_sec=None,
+    )
+
+    long_running_started = asyncio.Event()
+
+    async def slow_agent(input_value, ctx):
+        long_running_started.set()
+        await asyncio.sleep(10)
+
+    rt.register_agent("slow", slow_agent)
+
+    # Connect as p1 and submit
+    server_a, client_a = pair_memory_transports()
+    server_b, client_b = pair_memory_transports()
+    task_a = asyncio.create_task(rt.accept(server_a))
+    task_b = asyncio.create_task(rt.accept(server_b))
+
+    ca = ARCPClient(
+        client=ClientInfo(name="a", version="1"),
+        token="tok",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    cb = ARCPClient(
+        client=ClientInfo(name="b", version="1"),
+        token="other",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    welcome_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",
+        session_id=welcome_b.session_id,
+        job_id=handle.job_id,
+        payload=JobCancelPayload(reason="hostile").model_dump(mode="json"),
+    )
+
+    with pytest.raises(PermissionDeniedError):
+        await handle_cancel(rt, ctx_b, env)
+
+    for cli in (ca, cb):
+        await cli.close()
+    for t in (task_a, task_b):
+        t.cancel()
+        with contextlib.suppress(asyncio.CancelledError, Exception):
+            await t
+    await rt.close()
+
+
+async def test_cancel_own_job_cancels_task() -> None:
+    """Cancelling a running job via client.cancel_job cancels the underlying task."""
+    rt = _make_rt()
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def slow_agent(input_value, ctx):
+        started.set()
+        try:
+            await asyncio.sleep(10)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    rt.register_agent("slow", slow_agent)
+
+    server_t, client_t = pair_memory_transports()
+    accept_task = asyncio.create_task(rt.accept(server_t))
+
+    client = ARCPClient(
+        client=ClientInfo(name="c", version="1"),
+        token="tok",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    await client.connect(client_t)
+
+    handle = await client.submit(agent="slow")
+    await asyncio.wait_for(started.wait(), timeout=2.0)
+    await client.cancel_job(handle.job_id)
+    await asyncio.wait_for(cancelled.wait(), timeout=2.0)
+    assert cancelled.is_set()
+
+    await client.close()
+    accept_task.cancel()
+    with contextlib.suppress(asyncio.CancelledError, Exception):
+        await accept_task
+    await rt.close()
+
+
+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
+
+    rt = _make_rt()
+    server_t, client_t = pair_memory_transports()
+    accept_task = asyncio.create_task(rt.accept(server_t))
+
+    client = ARCPClient(
+        client=ClientInfo(name="c", version="1"),
+        token="tok",
+        capabilities=Capabilities(features=rt.capabilities.features),
+    )
+    welcome = await client.connect(client_t)
+    ctx = rt._sessions[welcome.session_id]
+
+    env = Envelope(
+        id=new_envelope_id(),
+        type="job.unsubscribe",
+        session_id=welcome.session_id,
+        payload=JobUnsubscribePayload(job_id="ghost-job").model_dump(mode="json"),
+    )
+    # Must not raise
+    await handle_unsubscribe(rt, ctx, env)
+
+    await client.close()
+    accept_task.cancel()
+    with contextlib.suppress(asyncio.CancelledError, Exception):
+        await accept_task
+    await rt.close()

tests/state/test_heartbeat.py

@@ -0,0 +1,131 @@
+"""§6.4 heartbeat_loop — ping cycle, cancellation, and loss detection coverage."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from datetime import UTC, datetime
+from unittest.mock import MagicMock
+
+import pytest
+
+from arcp._errors import HeartbeatLostError
+from arcp._runtime.session import SessionContext, SessionState, heartbeat_loop
+from arcp._transport.base import Transport
+
+
+def _make_ctx() -> SessionContext:
+    """Construct a minimal SessionContext for heartbeat testing."""
+    state = SessionState(
+        session_id="heartbeat-test-session",
+        resume_token="tok",
+        principal="p1",
+        negotiated_features=(),
+        heartbeat_interval_sec=None,
+        resume_window_sec=60,
+        accepted_at=datetime.now(UTC),
+    )
+    transport = MagicMock(spec=Transport)
+    return SessionContext(
+        transport=transport,
+        state=state,
+        send_queue=asyncio.Queue(),
+    )
+
+
+async def test_heartbeat_cancelled_returns_cleanly() -> None:
+    """CancelledError during sleep causes heartbeat_loop to exit without setting exception."""
+    ctx = _make_ctx()
+    loop = asyncio.get_running_loop()
+    ctx.heartbeat_outcome = loop.create_future()
+
+    task = asyncio.create_task(heartbeat_loop(ctx, interval=60.0))
+    await asyncio.sleep(0.01)  # let the task enter asyncio.sleep
+    task.cancel()
+
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+    # heartbeat_outcome must not have been set
+    assert not ctx.heartbeat_outcome.done()
+    ctx.heartbeat_outcome.cancel()  # clean up the future
+
+
+async def test_heartbeat_lost_sets_exception_on_outcome() -> None:
+    """Gap >= interval * miss_threshold sets HeartbeatLostError on heartbeat_outcome."""
+    ctx = _make_ctx()
+    loop = asyncio.get_running_loop()
+    ctx.heartbeat_outcome = loop.create_future()
+
+    # Back-date last_inbound_at so gap is always enormous
+    ctx._last_inbound_at = datetime(2000, 1, 1, tzinfo=UTC)
+
+    # interval=0.05 s, miss_threshold=1 → threshold=0.05 s; gap >> threshold
+    await asyncio.wait_for(
+        heartbeat_loop(ctx, interval=0.05, miss_threshold=1),
+        timeout=1.0,
+    )
+
+    assert ctx.heartbeat_outcome.done()
+    with pytest.raises(HeartbeatLostError):
+        ctx.heartbeat_outcome.result()
+
+
+async def test_heartbeat_loss_without_outcome_future_is_safe() -> None:
+    """heartbeat_loop handles heartbeat_outcome=None gracefully when loss fires."""
+    ctx = _make_ctx()
+    ctx.heartbeat_outcome = None  # no future attached
+
+    # Back-date to trigger loss on first check
+    ctx._last_inbound_at = datetime(2000, 1, 1, tzinfo=UTC)
+
+    # Should complete without AttributeError or TypeError
+    await asyncio.wait_for(
+        heartbeat_loop(ctx, interval=0.05, miss_threshold=1),
+        timeout=1.0,
+    )
+    # No assertion needed — reaching here means no exception was raised
+
+
+async def test_heartbeat_sends_ping_and_invokes_on_ping_callback() -> None:
+    """Normal interval enqueues session.ping and calls on_ping with the nonce."""
+    ctx = _make_ctx()
+    received_nonces: list[str] = []
+
+    # Use large miss_threshold so we never trigger the lost-heartbeat branch
+    task = asyncio.create_task(
+        heartbeat_loop(ctx, interval=0.05, miss_threshold=20, on_ping=received_nonces.append)
+    )
+    # Wait long enough for at least one complete iteration
+    await asyncio.sleep(0.13)
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+    assert len(received_nonces) >= 1
+
+    # The send queue must contain at least one session.ping envelope
+    found_ping = False
+    while not ctx._send_queue.empty():
+        item = ctx._send_queue.get_nowait()
+        if item is not None and item.type == "session.ping":
+            found_ping = True
+            break
+    assert found_ping, "expected at least one session.ping in the send queue"
+
+
+async def test_heartbeat_loss_with_done_outcome_does_not_raise() -> None:
+    """If heartbeat_outcome is already done, set_exception is skipped (guarded by .done())."""
+    ctx = _make_ctx()
+    loop = asyncio.get_running_loop()
+    fut: asyncio.Future[None] = loop.create_future()
+    fut.cancel()  # mark future as done (cancelled)
+    ctx.heartbeat_outcome = fut
+
+    ctx._last_inbound_at = datetime(2000, 1, 1, tzinfo=UTC)
+
+    # Should complete without InvalidStateError (future already done guard)
+    await asyncio.wait_for(
+        heartbeat_loop(ctx, interval=0.05, miss_threshold=1),
+        timeout=1.0,
+    )