tests: type the split-token tests against the real production types

The mypy gate on tests/ was added in 0.26.0 with tests/ clean, but a run of
commits then landed without CI ever executing, so 21 errors accumulated
unseen. They are not one mistake repeated; they are three, and each one is
the test having drifted from production rather than production being wrong.

Structural stand-ins for nominal types. test_split_token.py built a local
_Auth class with the three attributes _identity_tail reads, and a local _Ctx
with a bare `auth = None`. Both work at runtime because Python does not
check, but AuthContext and CallContext are nominal types with real
constructors, and every other test in the tree (test_worker, test_auth,
test_aggregate_function) already builds them directly. Replaced with
AuthContext(domain=..., authenticated=True, principal=...) and the
`CallContext(auth=AuthContext.anonymous(), ...)` idiom the rest of the suite
uses. Behaviourally identical: _identity_tail maps both `None` and an
unauthenticated context to the same anonymous tail.

Borrowing an unbound method instead of inheriting it. Three test-local
worker doubles did `_stamp_split_tokens = Worker._stamp_split_tokens` on a
class that was not a Worker, which mypy correctly flags as an invalid self
argument -- the double satisfied the method's needs by accident, and would
keep compiling if _stamp_split_tokens grew a dependency on any other Worker
member. They are now Worker subclasses overriding _current_split_anchor,
which is the seam the method actually documents. Worker is constructible
with no arguments, so this costs nothing.

Wire types passed as their serialized shape. `arguments=b""` and
`function_type="TABLE"` are what those fields become on the wire, not what
BindRequest declares (Arguments and the FunctionType enum). Every other
BindRequest in the suite already passes the declared types; these three were
written before anything checked.

Also narrowed PlanResponse.splits at its two use sites. It is genuinely
`list[ScanSplit] | list[bytes]` because a worker may hand back either, but
stamping always returns the serialized form -- so the new _stamped() helper
asserts that rather than casting it away, which pins that half of the
contract while it narrows.

One unrelated error in test_wire_record_schema_parity.py: a bare pa.Scalar
where strict mode wants pa.Scalar[Any], matching how vgi/arguments.py spells
it.

No production code changed; no type was loosened and no ignore was added.
`mypy vgi/`, `mypy tests/`, ruff format/check, and the full suite (2438
passed) are all green.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/test_split_token.py

@@ -9,9 +9,14 @@
 
 import json
 import pathlib
+from typing import Any
 
 import pytest
+from vgi_rpc.rpc import AuthContext, CallContext
 
+from vgi.arguments import Arguments
+from vgi.invocation import FunctionType
+from vgi.protocol import BindRequest, PlanResponse, ScanSplit, TableFunctionPlanRequest
 from vgi.split_token import (
     _HEADER_STRUCT,
     FLAG_PAYLOAD_SEALED,
@@ -33,11 +38,27 @@
 PAYLOAD = MANIFEST["payload"].encode()
 
 
-class _Auth:
-    def __init__(self, principal: str) -> None:
-        self.authenticated = True
-        self.domain = "test"
-        self.principal = principal
+def _auth(principal: str) -> AuthContext:
+    """A caller identity, as the transports actually build one."""
+    return AuthContext(domain="test", authenticated=True, principal=principal)
+
+
+def _anon_ctx() -> CallContext:
+    """The unauthenticated call context a subprocess worker runs under."""
+    return CallContext(auth=AuthContext.anonymous(), emit_client_log=lambda *a, **kw: None)
+
+
+def _stamped(response: PlanResponse, index: int) -> bytes:
+    """The serialized bytes of one stamped split.
+
+    ``PlanResponse.splits`` is ``list[ScanSplit] | list[bytes]`` because a worker
+    may hand back either, but stamping always returns the serialized form — so
+    asserting it here both narrows the union for the reader and pins that half of
+    the contract.
+    """
+    entry = response.splits[index]
+    assert isinstance(entry, bytes), "stamping must serialize every split"
+    return entry
 
 
 # --------------------------------------------------------------------------- #
@@ -105,11 +126,11 @@ def test_seal_binds_the_caller_principal() -> None:
     a split token names data (files, offsets, tenant partitions).
     """
     token = build_split_token(
-        payload=PAYLOAD, fingerprint=FINGERPRINT, anchor=ANCHOR, signing_key=KEY, auth=_Auth("alice")
+        payload=PAYLOAD, fingerprint=FINGERPRINT, anchor=ANCHOR, signing_key=KEY, auth=_auth("alice")
     )
-    assert open_split_token(token, signing_key=KEY, auth=_Auth("alice")) == PAYLOAD
+    assert open_split_token(token, signing_key=KEY, auth=_auth("alice")) == PAYLOAD
     with pytest.raises(SplitTokenInvalid):
-        open_split_token(token, signing_key=KEY, auth=_Auth("bob"))
+        open_split_token(token, signing_key=KEY, auth=_auth("bob"))
 
 
 def test_header_is_covered_by_the_aad() -> None:
@@ -156,7 +177,7 @@ def test_error_kinds_are_stable_strings() -> None:
 
 
 @pytest.mark.parametrize("case", MANIFEST["cases"], ids=lambda c: c["name"])
-def test_fixture_vectors_reach_their_recorded_verdict(case: dict) -> None:
+def test_fixture_vectors_reach_their_recorded_verdict(case: dict[str, Any]) -> None:
     """Every checked-in vector must reach the verdict its manifest records."""
     raw = (FIXTURES / f"{case['name']}.bin").read_bytes()
     assert len(raw) == case["size"]
@@ -224,31 +245,22 @@ def test_anchor_defaults_to_the_live_catalog_version_not_zero() -> None:
     Invisible wherever the catalog's version is 0, which is most fixtures — so this
     pins the non-zero case specifically.
     """
-    from vgi.protocol import BindRequest as PBindRequest
-    from vgi.protocol import PlanResponse, ScanSplit, TableFunctionPlanRequest
 
-    class _VersionedWorker:
+    class _VersionedWorker(Worker):
         """Stands in for a worker whose catalog reports a non-zero version."""
 
-        _signing_key = None
-
-        def _current_split_anchor(self, request: object, ctx: object) -> bytes:
+        def _current_split_anchor(self, request: Any, ctx: CallContext | None) -> bytes:
             return (47).to_bytes(8, "little", signed=True)
 
-        _stamp_split_tokens = Worker._stamp_split_tokens
-
-    bind = PBindRequest(function_name="f", arguments=b"", function_type="TABLE")
+    bind = BindRequest(function_name="f", arguments=Arguments(positional=()), function_type=FunctionType.TABLE)
     request = TableFunctionPlanRequest(bind_call=bind)
 
-    class _Ctx:
-        auth = None
-
     # on_plan left catalog_version unset — the common case for a worker that has
     # not thought about snapshots.
     planned = PlanResponse(splits=[ScanSplit(payload=b"file=1")])
-    stamped = _VersionedWorker()._stamp_split_tokens(planned, request, _Ctx())
+    stamped = _VersionedWorker()._stamp_split_tokens(planned, request, _anon_ctx())
 
-    split = ScanSplit.deserialize_from_bytes(stamped.splits[0])
+    split = ScanSplit.deserialize_from_bytes(_stamped(stamped, 0))
     # Redemption compares against the live version; the token must name it.
     opened = open_split_token(
         split.token,
@@ -260,26 +272,17 @@ class _Ctx:
 
 def test_a_worker_that_names_its_version_is_taken_at_its_word() -> None:
     """An explicit ``catalog_version`` wins: the worker knows which snapshot it planned."""
-    from vgi.protocol import BindRequest as PBindRequest
-    from vgi.protocol import PlanResponse, ScanSplit, TableFunctionPlanRequest
-
-    class _VersionedWorker:
-        _signing_key = None
 
-        def _current_split_anchor(self, request: object, ctx: object) -> bytes:
+    class _VersionedWorker(Worker):
+        def _current_split_anchor(self, request: Any, ctx: CallContext | None) -> bytes:
             return (47).to_bytes(8, "little", signed=True)
 
-        _stamp_split_tokens = Worker._stamp_split_tokens
-
-    bind = PBindRequest(function_name="f", arguments=b"", function_type="TABLE")
+    bind = BindRequest(function_name="f", arguments=Arguments(positional=()), function_type=FunctionType.TABLE)
     request = TableFunctionPlanRequest(bind_call=bind)
 
-    class _Ctx:
-        auth = None
-
     planned = PlanResponse(splits=[ScanSplit(payload=b"file=1")], catalog_version=11)
-    stamped = _VersionedWorker()._stamp_split_tokens(planned, request, _Ctx())
-    split = ScanSplit.deserialize_from_bytes(stamped.splits[0])
+    stamped = _VersionedWorker()._stamp_split_tokens(planned, request, _anon_ctx())
+    split = ScanSplit.deserialize_from_bytes(_stamped(stamped, 0))
 
     with pytest.raises(SplitSnapshotExpired):
         open_split_token(
@@ -310,32 +313,24 @@ def test_stamping_clears_the_plaintext_payload() -> None:
     this is a leak with no consumer, which is the easiest kind to leave in place
     for years.
     """
-    from vgi.protocol import BindRequest as PBindRequest
-    from vgi.protocol import PlanResponse, ScanSplit, TableFunctionPlanRequest
-
     secret = b"file=/tenant-a/private.parquet;v=47"
 
-    class _KeyedWorker:
+    class _KeyedWorker(Worker):
         _signing_key = b"k" * 32
 
-        def _current_split_anchor(self, request: object, ctx: object) -> bytes:
+        def _current_split_anchor(self, request: Any, ctx: CallContext | None) -> bytes:
             return (0).to_bytes(8, "little", signed=True)
 
-        _stamp_split_tokens = Worker._stamp_split_tokens
-
-    class _Ctx:
-        auth = None
-
-    bind = PBindRequest(function_name="f", arguments=b"", function_type="TABLE")
+    bind = BindRequest(function_name="f", arguments=Arguments(positional=()), function_type=FunctionType.TABLE)
     request = TableFunctionPlanRequest(bind_call=bind)
-    stamped = _KeyedWorker()._stamp_split_tokens(PlanResponse(splits=[ScanSplit(payload=secret)]), request, _Ctx())
+    stamped = _KeyedWorker()._stamp_split_tokens(PlanResponse(splits=[ScanSplit(payload=secret)]), request, _anon_ctx())
 
     # Not in the field...
-    split = ScanSplit.deserialize_from_bytes(stamped.splits[0])
+    split = ScanSplit.deserialize_from_bytes(_stamped(stamped, 0))
     assert split.payload == b""
     # ...and not anywhere else in the serialized record either, which is the
     # assertion that survives someone moving the leak to a different column.
-    assert secret not in stamped.splits[0]
+    assert secret not in _stamped(stamped, 0)
 
     # Still fully recoverable from the token, so clearing the field cost nothing.
     opened = open_split_token(

tests/test_wire_record_schema_parity.py

@@ -67,7 +67,7 @@ class WireRecordCase:
     build: Callable[[], Any]
 
 
-def _scalar_args() -> tuple[list[pa.Scalar], dict[str, pa.Scalar]]:
+def _scalar_args() -> tuple[list[pa.Scalar[Any]], dict[str, pa.Scalar[Any]]]:
     return (
         [pa.scalar("s3://bucket/x.parquet", pa.string())],
         {"hive_partitioning": pa.scalar(True, pa.bool_())},