fix(phase9 #97 D10.e): enforce §C.3 explicit error contract in decode_cursor

Per Weston msg=cc4a3ab0 二线 CR blocker: decode_cursor() previously
surfaced malformed / wrong-schema / expired wire payloads as bare
ValueError / KeyError, leaving every D10.c / D10.d caller to
re-derive the canonical mapping. That violates the §C.3
explicit-not-silent invariant by construction — any forgotten
mapping silently degrades into ValueError → tool error → first-page
restart, which is exactly the anti-pattern SILENT_RESET_FORBIDDEN
guards against.

Fix:
- decode_cursor() now raises CursorError directly with the right
  canonical code: cursor_invalid (malformed wire / base64 / json /
  missing field), cursor_schema_unsupported (unknown schema_version),
  cursor_expired (past issued_at + ttl_seconds clock).
- _decode_cursor_payload() preserved as a private structural-only
  decode for tests that need to craft expired / wrong-schema
  payloads to exercise the canonical error paths.
- 3 new canonical-code tests + 1 internal-decode escape hatch
  test added; old raw-error test deleted (pre-#1710 wire shape
  no longer reachable through public surface).

_payload() fixture's issued_at now defaults to current time so
round-trip tests stay green when run far from the fixture's
drafting date; tests that need expired / fixed payloads override
explicitly.

21/21 tests pass; ruff check + format clean.

Author: Bryce

aperag/mcp/cursor/codec.py

@@ -31,11 +31,14 @@
 from __future__ import annotations
 
 import base64
+import binascii
 import json
 import time
 from dataclasses import asdict, dataclass, field
 from typing import Any
 
+from aperag.mcp.cursor.errors import CursorError
+
 # CURSOR_SCHEMA_VERSION bumps when the on-wire payload shape changes
 # in an incompatible way. Decoders treat any `schema_version` they
 # don't know about as `cursor_schema_unsupported` (§C.3) — pending
@@ -81,26 +84,76 @@ def _b64url_decode(token: str) -> bytes:
     return base64.urlsafe_b64decode(token + pad)
 
 
-def decode_cursor(token: str) -> CursorPayload:
-    """Decode a wire cursor string back to a CursorPayload.
+def decode_cursor(token: str, *, now: int | None = None) -> CursorPayload:
+    """Decode a wire cursor string and enforce the §C.3 contract.
+
+    The decoder is the single chokepoint where every caller-facing
+    cursor failure becomes one of the six canonical
+    :class:`CursorError` codes — pushing the mapping out to each
+    tool would invite silent-reset drift (Weston msg=cc4a3ab0).
+
+    Raises:
+        CursorError: with code ``cursor_invalid`` (malformed wire /
+            base64 / JSON / missing field), ``cursor_schema_unsupported``
+            (unrecognised ``schema_version``), or ``cursor_expired``
+            (past ``issued_at + ttl_seconds`` clock).
+
+    Tool-boundary callers that need the raw structural decode
+    without the schema/expiry checks (typically only the test
+    surface) should use :func:`_decode_cursor_payload` directly.
+    """
 
-    On malformed / unsupported / expired input the caller is
-    responsible for raising the canonical CursorError code
-    (``cursor_invalid`` / ``cursor_schema_unsupported`` / etc) —
-    this codec only surfaces structural issues via ValueError /
-    KeyError; the canonical error mapping lives in
-    :mod:`aperag.mcp.cursor.errors` (pending spec amendment lock).
+    payload = _decode_cursor_payload(token)
+
+    if payload.schema_version != CURSOR_SCHEMA_VERSION:
+        raise CursorError(
+            "cursor_schema_unsupported",
+            f"cursor schema_version {payload.schema_version} not supported by this server",
+            details={
+                "received_schema_version": payload.schema_version,
+                "supported_schema_version": CURSOR_SCHEMA_VERSION,
+            },
+        )
+
+    if payload.is_expired(now=now):
+        raise CursorError(
+            "cursor_expired",
+            "cursor has passed its TTL window",
+            details={
+                "issued_at": payload.issued_at,
+                "ttl_seconds": payload.ttl_seconds,
+            },
+        )
+
+    return payload
+
+
+def _decode_cursor_payload(token: str) -> CursorPayload:
+    """Structural decode without §C.3 schema/expiry checks.
+
+    Internal helper — public callers should always go through
+    :func:`decode_cursor` so the explicit-not-silent contract is
+    enforced uniformly. This raw form exists so the test suite can
+    construct expired or wrong-schema payloads to exercise the
+    canonical error paths.
     """
 
-    raw = _b64url_decode(token)
-    obj = json.loads(raw)
-    return CursorPayload(
-        schema_version=obj["schema_version"],
-        sort_key=obj["sort_key"],
-        last_position=obj["last_position"],
-        invariant_hash=obj["invariant_hash"],
-        issued_at=obj["issued_at"],
-        ttl_seconds=obj.get("ttl_seconds", DEFAULT_TTL_SECONDS),
-        server_id=obj["server_id"],
-        extra=obj.get("extra", {}),
-    )
+    try:
+        raw = _b64url_decode(token)
+        obj = json.loads(raw)
+        return CursorPayload(
+            schema_version=obj["schema_version"],
+            sort_key=obj["sort_key"],
+            last_position=obj["last_position"],
+            invariant_hash=obj["invariant_hash"],
+            issued_at=obj["issued_at"],
+            ttl_seconds=obj.get("ttl_seconds", DEFAULT_TTL_SECONDS),
+            server_id=obj["server_id"],
+            extra=obj.get("extra", {}),
+        )
+    except (binascii.Error, ValueError, KeyError, TypeError) as exc:
+        raise CursorError(
+            "cursor_invalid",
+            "cursor wire payload could not be decoded",
+            details={"reason": str(exc) or exc.__class__.__name__},
+        ) from exc

tests/unit_test/mcp/test_cursor_contract.py

@@ -33,7 +33,11 @@
     decode_cursor,
     encode_cursor,
 )
-from aperag.mcp.cursor.codec import CURSOR_SCHEMA_VERSION, DEFAULT_TTL_SECONDS
+from aperag.mcp.cursor.codec import (
+    CURSOR_SCHEMA_VERSION,
+    DEFAULT_TTL_SECONDS,
+    _decode_cursor_payload,
+)
 from aperag.mcp.cursor.errors import (
     SILENT_RESET_FORBIDDEN,
     CursorError,
@@ -42,11 +46,17 @@
 
 
 def _payload(**overrides) -> CursorPayload:
+    # ``issued_at`` defaults to "now" so round-trip tests don't trip
+    # the §C.4 TTL window when the test is run far from the fixture's
+    # original drafting date. Tests that need a frozen / expired
+    # payload override ``issued_at`` explicitly.
+    import time as _time
+
     base = dict(
         sort_key="created_at",
         last_position={"created_at": "2026-04-26T03:00:00Z", "id": "doc-42"},
         invariant_hash="a" * 64,
-        issued_at=1761465600,
+        issued_at=int(_time.time()),
         server_id="srv-test",
     )
     base.update(overrides)
@@ -71,19 +81,40 @@ def test_default_schema_version_and_ttl_match_spec(self):
         assert payload.schema_version == CURSOR_SCHEMA_VERSION == 1
         assert payload.ttl_seconds == DEFAULT_TTL_SECONDS == 3600
 
-    def test_decode_rejects_garbage_via_value_error(self):
-        # codec layer surfaces structural failures as ValueError /
-        # JSONDecodeError; the canonical wire mapping into
-        # `cursor_invalid` happens at the tool boundary so the codec
-        # itself stays insulated from the error code naming.
-        with pytest.raises((ValueError, KeyError)):
+    def test_decode_garbage_raises_canonical_cursor_invalid(self):
+        # The decoder is the single chokepoint — every caller
+        # downstream sees `cursor_invalid` for a malformed wire
+        # without each tool reinventing the mapping.
+        with pytest.raises(CursorError) as excinfo:
             decode_cursor("not~~base64??")
+        assert excinfo.value.code == "cursor_invalid"
+
+    def test_decode_unknown_schema_version_raises_cursor_schema_unsupported(self):
+        future_token = encode_cursor(_payload(schema_version=CURSOR_SCHEMA_VERSION + 1))
+        with pytest.raises(CursorError) as excinfo:
+            decode_cursor(future_token)
+        assert excinfo.value.code == "cursor_schema_unsupported"
+        assert excinfo.value.details["received_schema_version"] == CURSOR_SCHEMA_VERSION + 1
+
+    def test_decode_past_ttl_raises_cursor_expired(self):
+        token = encode_cursor(_payload(issued_at=1000, ttl_seconds=60))
+        with pytest.raises(CursorError) as excinfo:
+            decode_cursor(token, now=2000)
+        assert excinfo.value.code == "cursor_expired"
 
     def test_is_expired_at_exact_ttl_boundary(self):
         payload = _payload(issued_at=1000, ttl_seconds=60)
         assert payload.is_expired(now=1059) is False
         assert payload.is_expired(now=1060) is True
 
+    def test_internal_decode_skips_schema_and_expiry_checks(self):
+        # _decode_cursor_payload is the test/debug escape hatch — it
+        # MUST NOT be called from production code, only from tests
+        # that need to craft wrong-schema or expired payloads.
+        future_token = encode_cursor(_payload(schema_version=999))
+        payload = _decode_cursor_payload(future_token)
+        assert payload.schema_version == 999
+
 
 class TestInvariantHash:
     def test_deterministic_across_dict_ordering(self):