fix(audit): bump SCHEMA_VERSION to 2 for scorer-stamped records (#315)

agentdescent/audit/records.py

@@ -34,8 +34,15 @@
 
 #: Bumped when the on-disk shape of a record changes incompatibly. A reader that
 #: meets a record it does not understand must say so rather than guess -- see
-#: :meth:`AuditStore.load`.
-SCHEMA_VERSION = 1
+#: :meth:`AuditStore.load`. Version 2 adds optional ``scorer_version``; this
+#: build still migrates version-1 lines. Stamp the new marker whenever the
+#: persisted shape is one an older v1-only reader cannot decode, so that reader
+#: hits its explicit unsupported-version path instead of treating a valid line
+#: as corrupt and silently skipping pending work (#315).
+SCHEMA_VERSION = 2
+#: On-disk versions this build can load. ``1`` is migrated in place; anything
+#: else raises.
+SUPPORTED_SCHEMA_VERSIONS = frozenset({1, SCHEMA_VERSION})
 
 
 class Purpose(str, Enum):
@@ -144,19 +151,33 @@ def residual(self) -> Optional[float]:
     def to_dict(self) -> Dict[str, Any]:
         d = asdict(self)
         d["purpose"] = self.purpose.value
+        # Always stamp the current marker. Unstamped (empty scorer) lines still
+        # omit the field so a pure v1 file that never saw a boost stays small;
+        # the version bump alone is what makes an old reader refuse rather than
+        # TypeError-skip a stamped line (#315).
+        d["schema_version"] = SCHEMA_VERSION
         if not d["scorer_version"]:
-            del d["scorer_version"]     # keep pre-existing lines byte-identical
+            del d["scorer_version"]
         return d
 
     @classmethod
     def from_dict(cls, d: Dict[str, Any]) -> "AuditRecord":
         d = dict(d)
         version = d.get("schema_version")
-        if version != SCHEMA_VERSION:
+        if version not in SUPPORTED_SCHEMA_VERSIONS:
             raise ValueError(
                 f"audit record schema_version={version!r}, this build reads "
-                f"{SCHEMA_VERSION}. Refusing to guess at the difference.")
+                f"{SCHEMA_VERSION} (and migrates v1). Refusing to guess at "
+                f"the difference.")
+        # v1: pre-scorer shape. Also accept a mistaken early stamp that wrote
+        # ``scorer_version`` under schema_version=1 (the #312/#315 bug): keep
+        # the field and promote the marker so a re-write is v2.
+        d.setdefault("scorer_version", "")
+        d["schema_version"] = SCHEMA_VERSION
         d["purpose"] = Purpose(d["purpose"])
+        # No silent filtering: an unknown key (say a misspelt ``strtaum``) is a
+        # malformed line, not a record to load under default values -- the
+        # ``TypeError`` it raises sends it down the store's corrupt-line path.
         return cls(**d)
 
 

agentdescent/audit/store.py

@@ -28,7 +28,8 @@
 import time
 from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple
 
-from .records import SCHEMA_VERSION, AuditRecord, Purpose
+from .records import (SCHEMA_VERSION, SUPPORTED_SCHEMA_VERSIONS, AuditRecord,
+                      Purpose)
 
 #: Marks a JSONL line as a moments snapshot rather than an audit record. Absent
 #: on every record ever written, so a file from before this existed still loads.
@@ -131,7 +132,8 @@ def load(self) -> None:
         A line that parses but carries an unknown ``schema_version`` **raises**:
         that is a different build's file, the fields do not mean what this code
         thinks they mean, and continuing would produce numbers rather than an
-        error.
+        error. Version 1 lines (pre-``scorer_version``) are migrated on read;
+        version 2 is current (#315).
         """
         if not self.path:
             return
@@ -188,8 +190,11 @@ def _load_line(self, payload: Dict[str, Any]) -> bool:
                 return True
         except (KeyError, TypeError, ValueError, AttributeError):
             return False
-        if payload.get("schema_version") != SCHEMA_VERSION:
+        if payload.get("schema_version") not in SUPPORTED_SCHEMA_VERSIONS:
             AuditRecord.from_dict(payload)          # raises on schema drift
+        # Every supported version (v1 is migrated) goes through the guarded
+        # construction below, so a malformed v1 line is one corrupt line rather
+        # than an unopenable store.
         try:
             rec = AuditRecord.from_dict(payload)
         except (KeyError, TypeError, ValueError):

changelog.d/315-audit-schema-version.md

@@ -0,0 +1,3 @@
+Fixed
+Audit records that carry a nonempty `scorer_version` (VerifierBoost identity) now persist `schema_version=2`. The previous release kept `SCHEMA_VERSION=1` while adding the field, so an older v0.6.0 reader hit `TypeError` on the unknown keyword, counted the line corrupt, and silently skipped pending work. The new reader still migrates v1 lines (unstamped and the mistaken early stamp); unsupported versions keep raising explicitly (#315). Scorer-identity behaviour from #304/#312 is unchanged.
+A v1 line that parses but cannot construct a record (missing key, invalid inclusion probability) is now one corrupt line rather than an unopenable store, and a record with an unknown key is counted corrupt instead of loading under default values.

docs/api.md

@@ -2502,7 +2502,7 @@ AuditRecord(
     dispatched_at: float = <factory>,
     resolved_at: Optional[float] = None,
     sampler_seed: int = 0,
-    schema_version: int = 1,
+    schema_version: int = 2,
     scorer_version: str = ''
 ) -> None
 ```
@@ -3889,7 +3889,7 @@ One rollout, described completely enough to run somewhere else.
 
 ### `SCHEMA_VERSION`
 
-Bumped when the on-disk shape of a record changes incompatibly. A reader that meets a record it does not understand must say so rather than guess -- see `load`.
+Bumped when the on-disk shape of a record changes incompatibly. A reader that meets a record it does not understand must say so rather than guess -- see `load`. Version 2 adds optional `scorer_version`; this build still migrates version-1 lines. Stamp the new marker whenever the persisted shape is one an older v1-only reader cannot decode, so that reader hits its explicit unsupported-version path instead of treating a valid line as corrupt and silently skipping pending work (#315).
 
 ### `SCORERS`
 

tests/test_audit_store_durability.py

@@ -385,3 +385,208 @@ def test_a_replayed_batch_still_counts_duplicates(tmp_path):
     assert again == 1 and again.already_resolved == 1
     assert store.get("r1").oracle_score == 1.0
     assert store.get("r2").oracle_score == 1.0
+
+
+# -- schema v1 → v2 / scorer_version (#315) ------------------------------------
+
+
+def test_unstamped_new_records_write_schema_version_2():
+    """Current writers stamp SCHEMA_VERSION even when scorer_version is omitted."""
+    from agentdescent.audit.records import SCHEMA_VERSION
+    payload = _record().to_dict()
+    assert SCHEMA_VERSION == 2
+    assert payload["schema_version"] == 2
+    assert "scorer_version" not in payload
+
+
+def test_stamped_records_write_schema_version_2_with_scorer():
+    from agentdescent.audit.records import SCHEMA_VERSION, AuditRecord, Purpose
+    rec = AuditRecord(
+        record_id="s1", task_id="t0", artifact_signature="sig",
+        output="out", verifier_version="v1+boost", verifier_score=0.5,
+        inclusion_prob=0.5, purpose=Purpose.CALIBRATION, dispatched_at=1.0,
+        scorer_version="scorer-alone")
+    payload = rec.to_dict()
+    assert payload["schema_version"] == SCHEMA_VERSION == 2
+    assert payload["scorer_version"] == "scorer-alone"
+    assert AuditRecord.from_dict(payload) == rec
+
+
+def test_v1_records_migrate_and_preserve_pending_state(tmp_path):
+    """A new reader loads pre-scorer v1 lines (pending and resolved) as v2."""
+    from agentdescent.audit.records import SCHEMA_VERSION, AuditRecord, Purpose
+    path = _path(tmp_path)
+    pending = {
+        "record_id": "p1", "task_id": "t0", "artifact_signature": "sig",
+        "output": "ans", "verifier_version": "v1", "verifier_score": 0.7,
+        "inclusion_prob": 0.5, "purpose": "calibration", "stratum": "all",
+        "oracle_score": None, "dispatched_at": 1.0, "resolved_at": None,
+        "sampler_seed": 0, "schema_version": 1,
+    }
+    resolved = dict(pending, record_id="r1", oracle_score=0.4, resolved_at=2.0)
+    # Mistaken early stamp: scorer_version under schema_version=1 (the #315 bug).
+    stamped_as_v1 = dict(pending, record_id="buggy", scorer_version="scorer")
+    with open(path, "w", encoding="utf-8") as fh:
+        for payload in (pending, resolved, stamped_as_v1):
+            fh.write(json.dumps(payload) + "\n")
+
+    store = AuditStore(path)
+    assert store.corrupt == 0
+    assert [r.record_id for r in store] == ["p1", "r1", "buggy"]
+    p1 = store.get("p1")
+    assert p1.oracle_score is None and p1.scorer_version == ""
+    assert p1.schema_version == SCHEMA_VERSION
+    r1 = store.get("r1")
+    assert r1.oracle_score == 0.4 and r1.resolved_at == 2.0
+    buggy = store.get("buggy")
+    assert buggy.scorer_version == "scorer"
+    assert buggy.schema_version == SCHEMA_VERSION
+    # Re-persisting a migrated pending record must write v2, so an old reader
+    # would hit its unsupported-version raise rather than TypeError-skip.
+    rewritten = buggy.to_dict()
+    assert rewritten["schema_version"] == 2
+    assert rewritten["scorer_version"] == "scorer"
+
+
+def test_mixed_v1_and_v2_lines_load_without_rewriting_bytes(tmp_path):
+    path = _path(tmp_path)
+    v1 = dict(_record("old").to_dict(), schema_version=1)
+    v1.pop("scorer_version", None)
+    v2 = _record("new", score=0.9)
+    # Force a stamped v2 line onto disk.
+    from agentdescent.audit.records import AuditRecord, Purpose
+    stamped = AuditRecord(
+        record_id="boosted", task_id="t0", artifact_signature="sig",
+        output="answer", verifier_version="v1+b", verifier_score=0.5,
+        inclusion_prob=0.5, purpose=Purpose.IMPROVEMENT, dispatched_at=1.0,
+        scorer_version="scorer")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(json.dumps(v1) + "\n")
+        fh.write(json.dumps(v2.to_dict()) + "\n")
+        fh.write(json.dumps(stamped.to_dict()) + "\n")
+    raw_before = open(path, "rb").read()
+    store = AuditStore(path)
+    assert store.corrupt == 0
+    assert [r.record_id for r in store] == ["old", "new", "boosted"]
+    assert store.get("boosted").scorer_version == "scorer"
+    # load is read-only: original bytes unchanged
+    assert open(path, "rb").read() == raw_before
+
+
+def test_old_reader_contract_rejects_v2_explicitly_not_as_corrupt():
+    """Simulate a v1-only from_dict: schema_version=2 must raise ValueError.
+
+    The #315 failure mode was schema_version=1 + unknown ``scorer_version``
+    keyword → TypeError → store counted the line corrupt and skipped pending
+    work. Bumping the marker makes an unchanged old reader take its
+    unsupported-version path instead.
+    """
+    from agentdescent.audit.records import AuditRecord, Purpose
+
+    def old_v1_from_dict(d):
+        d = dict(d)
+        version = d.get("schema_version")
+        if version != 1:
+            raise ValueError(
+                f"audit record schema_version={version!r}, this build reads 1. "
+                f"Refusing to guess at the difference.")
+        d["purpose"] = Purpose(d["purpose"])
+        return AuditRecord(**{k: v for k, v in d.items()
+                              if k in AuditRecord.__dataclass_fields__
+                              and k != "scorer_version"})
+
+    stamped = AuditRecord(
+        record_id="s", task_id="t0", artifact_signature="sig",
+        output="o", verifier_version="v", verifier_score=0.5,
+        inclusion_prob=0.5, purpose=Purpose.CALIBRATION, dispatched_at=1.0,
+        scorer_version="scorer")
+    payload = stamped.to_dict()
+    assert payload["schema_version"] == 2
+    with pytest.raises(ValueError, match="schema_version"):
+        old_v1_from_dict(payload)
+
+    # Contrast: the pre-fix shape (v1 + scorer_version) is what caused the
+    # silent TypeError skip on a real v0.6.0 constructor that lacked the field.
+    broken = dict(payload, schema_version=1)
+    try:
+        AuditRecord(
+            **{k: (Purpose(v) if k == "purpose" else v)
+               for k, v in broken.items()
+               if k in AuditRecord.__dataclass_fields__})
+        # Current constructor accepts scorer_version; emulate rejection:
+        raise AssertionError("current build accepts scorer_version")
+    except AssertionError:
+        pass
+    # Document the TypeError an old constructor would raise:
+    import dataclasses
+    fields_without = tuple(
+        f for f in dataclasses.fields(AuditRecord) if f.name != "scorer_version")
+
+    @dataclasses.dataclass(frozen=True)
+    class OldAuditRecord:
+        pass
+    # Cheaper pin: constructing with an unexpected kwarg raises TypeError.
+    def old_ctor(**kwargs):
+        known = {"record_id", "task_id", "artifact_signature", "output",
+                 "verifier_version", "verifier_score", "inclusion_prob",
+                 "purpose", "stratum", "oracle_score", "dispatched_at",
+                 "resolved_at", "sampler_seed", "schema_version"}
+        unknown = set(kwargs) - known
+        if unknown:
+            raise TypeError(
+                f"__init__() got an unexpected keyword argument "
+                f"{unknown.pop()!r}")
+        return True
+    with pytest.raises(TypeError, match="scorer_version"):
+        old_ctor(**{**broken, "purpose": "calibration"})
+
+
+@pytest.mark.parametrize("drop", ["purpose", "task_id", "verifier_score"])
+def test_a_malformed_v1_line_is_one_corrupt_line_not_an_unopenable_store(
+        tmp_path, drop):
+    """A v1 line that parses but cannot construct a record is skipped and
+    counted, exactly like a malformed current-schema line (Codex P1 on #324)."""
+    path = _path(tmp_path)
+    bad = dict(_record("bad").to_dict(), schema_version=1)
+    bad.pop(drop)
+    good = _record("good")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(json.dumps(bad) + "\n")
+        fh.write(json.dumps(good.to_dict()) + "\n")
+    store = AuditStore(path)
+    assert store.corrupt == 1
+    assert [r.record_id for r in store] == ["good"]
+
+
+def test_a_v1_line_with_an_invalid_inclusion_prob_is_counted_corrupt(tmp_path):
+    path = _path(tmp_path)
+    bad = dict(_record("bad").to_dict(), schema_version=1, inclusion_prob=0.0)
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(json.dumps(bad) + "\n")
+    store = AuditStore(path)
+    assert store.corrupt == 1
+    assert list(store) == []
+
+
+@pytest.mark.parametrize("version", [1, 2])
+def test_an_unknown_field_is_corrupt_not_silently_defaulted(tmp_path, version):
+    """A misspelt optional key must not load under the default value (Codex P2
+    on #324): ``strtaum`` would otherwise land the record in stratum "all"."""
+    path = _path(tmp_path)
+    bad = _record("typo").to_dict()
+    bad["schema_version"] = version
+    bad["strtaum"] = bad.pop("stratum")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(json.dumps(bad) + "\n")
+    store = AuditStore(path)
+    assert store.corrupt == 1
+    assert list(store) == []
+
+
+def test_an_unknown_schema_version_still_raises(tmp_path):
+    path = _path(tmp_path)
+    future = dict(_record("future").to_dict(), schema_version=3)
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(json.dumps(future) + "\n")
+    with pytest.raises(ValueError, match="schema_version=3"):
+        AuditStore(path)