test: add portable renderer conformance checks

.github/workflows/test.yml

@@ -9,7 +9,7 @@ on:
       - "uv.lock"
       - ".github/workflows/test.yml"
   pull_request:
-    branches: [main]
+    branches: [main, "codex/**"]
     paths:
       - "**.py"
       - "pyproject.toml"

tests/conformance/__init__.py

@@ -0,0 +1 @@
+"""Independent reference tests through the public renderer interface."""

tests/conformance/conftest.py

@@ -0,0 +1,14 @@
+"""Release real tokenizer assets between conformance modules."""
+
+import gc
+
+import pytest
+
+from tests.conformance.harness import tokenizer
+
+
+@pytest.fixture(scope="module", autouse=True)
+def release_reference_assets():
+    yield
+    tokenizer.cache_clear()
+    gc.collect()

tests/conformance/encoder.py

@@ -0,0 +1,87 @@
+"""Execute the reviewed, pinned model-supplied Python encoder directly."""
+
+import copy
+import hashlib
+import importlib.util
+import json
+from functools import cache
+from pathlib import Path
+
+from huggingface_hub import hf_hub_download
+
+from tests.model_assets import model_revision
+
+DEEPSEEK_ENCODER = {
+    "repo": "deepseek-ai/DeepSeek-V4-Flash-0731",
+    "file": "encoding/encoding_dsv4.py",
+    "sha256": "abc0d26120250dda0ae077dc64aa28836026e61e970854aaeb792445e6a0dde6",
+}
+
+
+@cache
+def encoder():
+    source = DEEPSEEK_ENCODER
+    path = hf_hub_download(
+        source["repo"],
+        source["file"],
+        revision=model_revision(source["repo"]),
+    )
+    assert hashlib.sha256(Path(path).read_bytes()).hexdigest() == source["sha256"]
+    spec = importlib.util.spec_from_file_location("pinned_deepseek_encoder", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def deepseek_reference(
+    tok,
+    messages,
+    *,
+    tools=None,
+    add_generation_prompt=False,
+    enable_thinking=False,
+    drop_thinking=True,
+    reasoning_effort="low",
+):
+    upstream = encoder()
+    messages = copy.deepcopy(messages)
+    if tools:
+        target = next((m for m in messages if m["role"] == "developer"), None)
+        if target is None:
+            target = next((m for m in messages if m["role"] == "system"), None)
+        if target is None:
+            target = {"role": "system", "content": ""}
+            messages.insert(0, target)
+        target["tools"] = tools
+    for message in messages:
+        for call in message.get("tool_calls") or []:
+            function = call["function"]
+            if not isinstance(function["arguments"], str):
+                function["arguments"] = json.dumps(
+                    function["arguments"], ensure_ascii=False
+                )
+    text = upstream.encode_messages(
+        messages,
+        thinking_mode="thinking" if enable_thinking else "chat",
+        drop_thinking=drop_thinking,
+        reasoning_effort=reasoning_effort,
+    )
+    # Upstream's API always opens generation after a terminal query. The
+    # renderer API also exposes a completed-transcript view. Remove only that
+    # terminal generation prefill; all internal transitions remain upstream's.
+    if not add_generation_prompt:
+        merged = upstream.merge_tool_messages(messages)
+        if (
+            merged
+            and merged[-1]["role"] in {"user", "developer"}
+            and not merged[-1].get("task")
+        ):
+            suffix = upstream.ASSISTANT_SP_TOKEN + (
+                upstream.thinking_start_token
+                if enable_thinking
+                else upstream.thinking_end_token
+            )
+            assert text.endswith(suffix)
+            text = text[: -len(suffix)]
+    return tok.encode(text, add_special_tokens=False)

tests/conformance/harness.py

@@ -0,0 +1,102 @@
+"""No renderer internals: official encoder → token backend → public client."""
+
+from __future__ import annotations
+
+import json
+from functools import lru_cache
+from typing import Any
+
+import httpx
+
+from renderers import config_from_name, create_renderer
+from tests.conformance.encoder import deepseek_reference
+from tests.model_assets import load_test_tokenizer
+from tests.parity import ModelCase
+from tests.reference_rendering import render_reference
+
+
+@lru_cache(maxsize=1)
+def tokenizer(model: str):
+    return load_test_tokenizer(model)
+
+
+def renderer(case: ModelCase, kwargs: dict[str, Any]):
+    return create_renderer(
+        tokenizer(case.model),
+        config_from_name(case.resolved_renderer),
+        chat_template_kwargs=kwargs or None,
+    )
+
+
+def reference(case: ModelCase, messages: list, kwargs: dict, **render_kwargs):
+    options = dict(case.oracle_defaults) | kwargs
+    options = {key: value for key, value in options.items() if value is not None}
+    if case.oracle == "deepseek-v4":
+        return deepseek_reference(
+            tokenizer(case.model), messages, **options, **render_kwargs
+        )
+    return render_reference(
+        tokenizer(case.model), messages, oracle=case.oracle, **options, **render_kwargs
+    )
+
+
+def assert_tokens(actual: list[int], expected: list[int], case: ModelCase):
+    if actual == expected:
+        return
+    first = next(
+        (i for i, (a, b) in enumerate(zip(actual, expected)) if a != b),
+        min(len(actual), len(expected)),
+    )
+    tok = tokenizer(case.model)
+    window = slice(max(0, first - 12), first + 24)
+    raise AssertionError(
+        f"{case.model}: first differing token {first}; "
+        f"lengths native={len(actual)}, reference={len(expected)}\n"
+        f"native: {tok.decode(actual[window], skip_special_tokens=False)!r}\n"
+        f"reference: {tok.decode(expected[window], skip_special_tokens=False)!r}"
+    )
+
+
+def assert_attribution(rendered):
+    size = len(rendered.token_ids)
+    for field in ("message_indices",):
+        assert len(getattr(rendered, field)) == size, field
+    assert len(rendered.sampled_mask) in {0, size}
+    assert len(rendered.is_content) in {0, size}
+
+
+class TokenBackend:
+    """Script sampling only. Rendering and parsing remain the real implementation."""
+
+    base_url = "http://conformance.invalid/v1"
+
+    def __init__(self, prompt: list[int], completion: list[int], finish="stop"):
+        self.prompt = prompt
+        self.completion = completion
+        self.finish = finish
+        self.requests: list[dict] = []
+
+    async def post(self, endpoint, *, body, cast_to, **kwargs):
+        assert endpoint == "http://conformance.invalid/inference/v1/generate"
+        assert body["token_ids"] == self.prompt
+        assert body["sampling_params"]["skip_special_tokens"] is False
+        self.requests.append(body)
+        return httpx.Response(
+            200,
+            content=json.dumps(
+                {
+                    "choices": [
+                        {
+                            "token_ids": self.completion,
+                            "finish_reason": self.finish,
+                            "logprobs": {
+                                "content": [
+                                    {"token": f"token_id:{t}", "logprob": -0.5}
+                                    for t in self.completion
+                                ]
+                            },
+                        }
+                    ]
+                }
+            ).encode(),
+        )