From 209c01205804da1702fcf4ebe9bfeaf680ffcde3 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 10 Jul 2026 11:35:21 -0400 Subject: [PATCH] Harden thesis-facts append gate: base-anchor prefix, av2 addressing, byte-append Address ledger review findings 3, 6, and 13 in the thesis-facts append gate. Finding 3 (HIGH): the immutable-prefix manifest is candidate-controlled, so a PR could grow prefixLineCount over its own append and have every post-cutover binding skipped. Under --base-ref the gate now loads the base manifest and requires prefixLineCount/prefixSha256/lineSha256s to be unchanged, and it uses the BASE prefix count as the post-cutover binding boundary so a candidate-controlled count can never move it. On push (no base ref) the candidate manifest is trusted for the full-file invariants only, which base-anchoring cannot cover without a witness. Finding 6 (HIGH): replace the incomplete av1 content address with the av2 projection (byte-identical to the Brier writer's assertion_version): it binds the complete measure concept mapping, exact source file/digest, row/cell lineage, and the archived response digest. check_rows now reserves the effective id (explicit or recomputed) of EVERY row, closing legacy synthetic-id reuse and A->B->A restoration. Add effective_current_rows() and validate the aggregate-fact journal as its supersede-aware current view so a legitimate correction no longer fails as a duplicate semantic key. Finding 13 (LOW): _lines dropped blank lines, so a blank line inserted into the frozen JSONL normalized away. reject_non_append_bytes rejects any blank/whitespace-only line in the covered region and a non-single trailing newline. Ports Sol's adversarial counterexamples into tests/test_thesis_append_adversarial.py: the finding 3/6/13 cases are now real asserts; the remaining strict-xfails document out-of-scope boundaries (provenance-format validation, no-base-ref full-file mode, per-commit history walking). Co-Authored-By: Claude Fable 5 --- scripts/check_thesis_facts_append.py | 163 +++++++- tests/test_policyengine_ledger.py | 9 +- tests/test_thesis_append_adversarial.py | 520 ++++++++++++++++++++++++ 3 files changed, 668 insertions(+), 24 deletions(-) create mode 100644 tests/test_thesis_append_adversarial.py diff --git a/scripts/check_thesis_facts_append.py b/scripts/check_thesis_facts_append.py index 77e18bc..38b2456 100644 --- a/scripts/check_thesis_facts_append.py +++ b/scripts/check_thesis_facts_append.py @@ -63,12 +63,37 @@ def _lines(text: str) -> list[str]: return [line for line in text.split("\n") if line.strip()] +def reject_non_append_bytes(text: str) -> None: + """Reject blank/whitespace-only lines and any non-single trailing newline. + + ``_lines`` drops blank lines so row parsing is convenient, but that means a + blank line inserted into the frozen JSONL would normalize away and pass both + the prefix hash and the append-only diff. A JSONL row is exactly one + non-empty line: a blank/whitespace-only line inside the covered region is a + byte tamper, and the file must end with exactly one trailing newline. + """ + parts = text.split("\n") + if parts[-1] != "": + raise AppendError("ledger must end with exactly one trailing newline") + for index, part in enumerate(parts[:-1], start=1): + if not part.strip(): + raise AppendError( + f"line {index} is blank or whitespace-only; a JSONL row is one " + "non-empty line and a stray blank line is a tamper" + ) + + def expected_assertion_version_id(row: dict[str, Any]) -> str: """Recompute the content address the resolver must have written. - Mirrors ``assertion_version`` in the Thesis resolver: the ID commits to - everything that changes what the assertion means, so an in-place edit - is detectable and a correction must supersede explicitly. + Mirrors ``assertion_version`` in the Thesis resolver (av1 v2 spec): the ID + commits to everything that changes what the assertion MEANS — identity, + value, timing, population, the complete measure concept mapping, exact + source lineage/digest, row/cell lineage, and the archived response digest — + so an in-place edit is detectable and a correction must supersede + explicitly. This projection must stay byte-identical to the Brier writer's + ``assertion_version`` (both fed to the shared ``canonical_sha256``), so any + change here is a coordinated schema migration on both sides. """ measure = row.get("measure") or {} source = row.get("source") or {} @@ -76,14 +101,59 @@ def expected_assertion_version_id(row: dict[str, Any]) -> str: projection["measure"] = { "concept": measure.get("concept"), "unit": measure.get("unit"), + "source_concept": measure.get("source_concept"), + "concept_relation": measure.get("concept_relation"), + "concept_authority": measure.get("concept_authority"), + "legal_vintage": measure.get("legal_vintage"), } projection["source"] = { "source_name": source.get("source_name"), "source_table": source.get("source_table"), + "source_file": source.get("source_file"), "url": source.get("url"), "vintage": source.get("vintage"), + "source_sha256": source.get("source_sha256"), + } + projection["lineage"] = { + "source_row_keys": row.get("source_row_keys"), + "source_cell_keys": row.get("source_cell_keys"), } - return f"av1:{canonical_sha256(projection)}" + projection["responseArchiveSha256"] = (row.get("responseArchive") or {}).get( + "sha256" + ) + return f"av2:{canonical_sha256(projection)}" + + +def _effective_assertion_id(row: dict[str, Any]) -> str: + """Return the row's effective assertion version ID. + + Post-cutover rows carry an explicit ``assertionVersion.id`` (validated + against the recomputed content address in :func:`check_rows`); legacy + pre-versioning rows are addressable by their recomputed content address. + Either way every row has exactly one effective ID that a correction must + name and that no later row may reissue. + """ + version = row.get("assertionVersion") + if isinstance(version, dict) and version.get("id"): + return str(version["id"]) + return expected_assertion_version_id(row) + + +def effective_current_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return the latest non-superseded row per assertion identity. + + A correction names the version it replaces via + ``assertionVersion.supersedes``; the replaced row drops out of the current + view. Aggregate-fact validation runs on this supersede-aware view so a + legitimate correction (same semantic key, new value) is not mistaken for a + duplicate key. + """ + superseded: set[str] = set() + for row in rows: + version = row.get("assertionVersion") + if isinstance(version, dict) and version.get("supersedes"): + superseded.add(str(version["supersedes"])) + return [row for row in rows if _effective_assertion_id(row) not in superseded] def check_prefix(lines: list[str]) -> dict[str, Any]: @@ -137,26 +207,35 @@ def check_rows(lines: list[str], prefix_count: int) -> None: if not unit: raise AppendError(f"line {number} ({record_id}) has no measure unit") + recomputed = expected_assertion_version_id(row) version = row.get("assertionVersion") - version_id = None supersedes = None if version is not None: if not isinstance(version, dict): raise AppendError(f"line {number} assertionVersion is not an object") version_id = str(version.get("id", "")) supersedes = version.get("supersedes") - expected = expected_assertion_version_id(row) - if version_id != expected: + if version_id != recomputed: raise AppendError( f"line {number} ({record_id}) assertionVersion.id does not " - f"match its content ({version_id} != {expected})" - ) - if version_id in versions: - raise AppendError( - f"line {number} restates assertion version {version_id} " - f"from line {versions[version_id]}" + f"match its content ({version_id} != {recomputed})" ) - versions[version_id] = number + effective_id = version_id + else: + # Pre-versioning rows are addressable by their recomputed content + # address; that ID is reserved just like an explicit one so a legacy + # synthetic ID cannot be silently reissued. + effective_id = recomputed + + # Reserve the effective ID of EVERY row. A collision means two rows + # claim the same assertion version — a duplicate legacy ID or an + # A->B->A chain trying to restore a superseded value. + if effective_id in versions: + raise AppendError( + f"line {number} restates assertion version {effective_id} " + f"from line {versions[effective_id]}" + ) + versions[effective_id] = number if number > prefix_count: for field in ( @@ -214,12 +293,7 @@ def check_rows(lines: list[str], prefix_count: int) -> None: f"line {number} supersedes {supersedes} but {record_id} has " "no earlier row" ) - # Rows that predate explicit versioning are still addressable: their - # version is the content address a correction must recompute. - active_by_record_id[str(record_id)] = ( - number, - version_id or expected_assertion_version_id(row), - ) + active_by_record_id[str(record_id)] = (number, effective_id) def check_append_only(base_ref: str, lines: list[str]) -> int: @@ -245,6 +319,42 @@ def check_append_only(base_ref: str, lines: list[str]) -> int: return len(lines) - len(base_lines) +def _manifest_at_ref(base_ref: str) -> dict[str, Any]: + relative = PREFIX_PATH.relative_to(ROOT).as_posix() + try: + text = subprocess.check_output( + ["git", "show", f"{base_ref}:{relative}"], cwd=ROOT, text=True + ) + except subprocess.CalledProcessError as exc: + raise AppendError( + f"cannot read {relative} at base {base_ref}" + ) from exc + return json.loads(text) + + +def check_prefix_anchored_to_base(base_ref: str, candidate_prefix: dict[str, Any]) -> int: + """Require the frozen prefix manifest to be unchanged from the base. + + The immutable-prefix manifest lives beside the ledger and is candidate- + controlled, so a PR could grow ``prefixLineCount`` over its own append and + have every post-cutover binding skipped (the appended row would count as + "prefix"). Growing the frozen prefix is an explicit, separately reviewed + migration — never part of the automated append path — so under a base ref + the count, cumulative hash, and per-line hashes must match the base exactly. + Returns the BASE prefix line count, which callers use as the post-cutover + binding boundary so a candidate-controlled count can never move it. + """ + base_prefix = _manifest_at_ref(base_ref) + for field in ("prefixLineCount", "prefixSha256", "lineSha256s"): + if candidate_prefix.get(field) != base_prefix.get(field): + raise AppendError( + f"immutable prefix manifest {field} changed vs base {base_ref}; " + "the frozen prefix cannot grow through the automated append path " + "— growing it is an explicit reviewed migration" + ) + return int(base_prefix["prefixLineCount"]) + + def main() -> int: parser = argparse.ArgumentParser() parser.add_argument( @@ -252,13 +362,22 @@ def main() -> int: help="enforce an append-only diff against this git ref", ) args = parser.parse_args() - lines = _lines(LEDGER_PATH.read_text(encoding="utf-8")) + text = LEDGER_PATH.read_text(encoding="utf-8") try: + reject_non_append_bytes(text) + lines = _lines(text) prefix = check_prefix(lines) - check_rows(lines, int(prefix["prefixLineCount"])) + # The post-cutover binding boundary is the BASE prefix count under a + # base ref, so a PR cannot grandfather an unbound append by growing the + # candidate manifest over it. Without a base ref (push) there is nothing + # to anchor against, so the candidate manifest is trusted for the + # full-file invariants only — base-anchoring requires the PR path. + binding_boundary = int(prefix["prefixLineCount"]) appended = None if args.base_ref: + binding_boundary = check_prefix_anchored_to_base(args.base_ref, prefix) appended = check_append_only(args.base_ref, lines) + check_rows(lines, binding_boundary) except AppendError as exc: print(f"thesis-facts append check failed: {exc}", file=sys.stderr) return 1 diff --git a/tests/test_policyengine_ledger.py b/tests/test_policyengine_ledger.py index ec6b1a0..970e429 100644 --- a/tests/test_policyengine_ledger.py +++ b/tests/test_policyengine_ledger.py @@ -34,6 +34,7 @@ from check_thesis_facts_append import ( # noqa: E402 check_prefix, check_rows, + effective_current_rows, expected_assertion_version_id, ) @@ -113,7 +114,11 @@ def test_official_observation_ledger_contains_facts_not_predictions(): def test_official_observations_validate_as_aggregate_facts(): - facts = [_to_aggregate_fact(row) for row in _read_ledger_facts()] + # ``validate_facts`` rejects two rows sharing a semantic aggregate key, so + # the journal is validated as its supersede-aware effective current view + # (latest non-superseded row per identity) rather than as raw duplicates. + current = effective_current_rows(_read_ledger_facts()) + facts = [_to_aggregate_fact(row) for row in current] report = validate_facts(facts) @@ -207,7 +212,7 @@ def test_a_correction_naming_a_stale_version_is_rejected(): correction = _appended_row(original, value_delta=1) correction["assertionVersion"] = { "id": expected_assertion_version_id(correction), - "supersedes": "av1:" + "0" * 64, + "supersedes": "av2:" + "0" * 64, } try: check_rows( diff --git a/tests/test_thesis_append_adversarial.py b/tests/test_thesis_append_adversarial.py new file mode 100644 index 0000000..f1a3868 --- /dev/null +++ b/tests/test_thesis_append_adversarial.py @@ -0,0 +1,520 @@ +"""Adversarial verification of the Thesis observation append gate. + +These tests exercise counterexamples from the 2026-07-10 ledger review. The +cases that pinned open findings 3, 6, and 13 are now real assertions that the +hardened gate rejects the attack. The remaining strict-xfail tests document +boundaries that are deliberately out of scope for this pass (provenance-format +validation, the no-base-ref full-file mode, and per-commit history walking); +each fails loudly if it ever starts passing. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from arch.core import ( + AggregateFact, + Aggregation, + EntityDimension, + GeographyDimension, + Measure, + PeriodDimension, + SourceProvenance, + validate_facts, +) + +ROOT = Path(__file__).resolve().parents[1] +LEDGER_PATH = ROOT / "ledger" / "official_observations.jsonl" +PREFIX_PATH = ROOT / "ledger" / "immutable_prefix.json" + +sys.path.insert(0, str(ROOT / "scripts")) + +from check_thesis_facts_append import ( # noqa: E402 + AppendError, + check_append_only, + check_rows, + effective_current_rows, + expected_assertion_version_id, +) + + +def _read_lines() -> list[str]: + return [ + line + for line in LEDGER_PATH.read_text(encoding="utf-8").split("\n") + if line.strip() + ] + + +def _json_line(row: dict) -> str: + return json.dumps(row, separators=(",", ":")) + + +def _to_aggregate_fact(row: dict) -> AggregateFact: + return AggregateFact( + value=row["value"], + period=PeriodDimension(**row["period"]), + geography=GeographyDimension(**row["geography"]), + entity=EntityDimension(**row["entity"]), + measure=Measure(**row["measure"]), + aggregation=Aggregation(**row["aggregation"]), + source=SourceProvenance(**row["source"]), + filters=row.get("filters", {}), + domain=row.get("domain", "all"), + label=row.get("label"), + source_record_id=row.get("source_record_id"), + source_cell_keys=tuple(row.get("source_cell_keys", ())), + source_row_keys=tuple(row.get("source_row_keys", ())), + ) + + +def _appended_row( + original: dict, + *, + value_delta: float = 0, + source_record_id: str | None = None, +) -> dict: + row = copy.deepcopy(original) + row["value"] += value_delta + if source_record_id is not None: + row["source_record_id"] = source_record_id + row.update( + { + "retrievedAt": "2026-07-11T00:00:00Z", + "sourceVintage": "2026-07-11", + "ledgerRepoSha": "a" * 40, + "responseArchive": { + "sha256": "b" * 64, + "contentEncoding": "gzip", + }, + } + ) + row.pop("targetContentHash", None) + row.pop("sourceBindingProjection", None) + row["assertionVersion"] = { + "id": expected_assertion_version_id(row), + "supersedes": None, + } + return row + + +def _write_checker_fixture(path: Path, ledger_text: str, manifest: dict) -> None: + ledger_dir = path / "ledger" + ledger_dir.mkdir(parents=True, exist_ok=True) + (ledger_dir / "official_observations.jsonl").write_text( + ledger_text, encoding="utf-8" + ) + (ledger_dir / "immutable_prefix.json").write_text( + json.dumps(manifest, indent=1) + "\n", encoding="utf-8" + ) + scripts_dir = path / "scripts" + scripts_dir.mkdir(exist_ok=True) + for name in ("check_thesis_facts_append.py", "canonical_json.py"): + (scripts_dir / name).write_text( + (ROOT / "scripts" / name).read_text(encoding="utf-8"), + encoding="utf-8", + ) + + +def _run_checker(path: Path, base_ref: str | None = None) -> subprocess.CompletedProcess: + command = [sys.executable, str(path / "scripts/check_thesis_facts_append.py")] + if base_ref is not None: + command.extend(["--base-ref", base_ref]) + return subprocess.run(command, cwd=path, capture_output=True, text=True) + + +def _git(path: Path, *args: str) -> str: + completed = subprocess.run( + ["git", *args], cwd=path, capture_output=True, text=True, check=True + ) + return completed.stdout.strip() + + +def _init_fixture_repo(path: Path) -> str: + _git(path, "init", "-q") + _git(path, "config", "user.name", "Append Gate Test") + _git(path, "config", "user.email", "append-gate@example.invalid") + _git(path, "add", ".") + _git(path, "commit", "-qm", "base") + return _git(path, "rev-parse", "HEAD") + + +def _rehash_manifest(lines: list[str]) -> dict: + manifest = json.loads(PREFIX_PATH.read_text(encoding="utf-8")) + count = int(manifest["prefixLineCount"]) + manifest["lineSha256s"] = [ + hashlib.sha256(line.encode("utf-8")).hexdigest() + for line in lines[:count] + ] + manifest["prefixSha256"] = hashlib.sha256( + ("\n".join(lines[:count]) + "\n").encode("utf-8") + ).hexdigest() + return manifest + + +def _extend_manifest_to_all_lines(lines: list[str]) -> dict: + manifest = json.loads(PREFIX_PATH.read_text(encoding="utf-8")) + manifest["prefixLineCount"] = len(lines) + manifest["lineSha256s"] = [ + hashlib.sha256(line.encode("utf-8")).hexdigest() for line in lines + ] + manifest["prefixSha256"] = hashlib.sha256( + ("\n".join(lines) + "\n").encode("utf-8") + ).hexdigest() + return manifest + + +def test_base_check_rejects_an_existing_line_rewrite(): + lines = _read_lines() + rewritten = json.loads(lines[-1]) + rewritten["value"] += 1 + candidate = [*lines[:-1], _json_line(rewritten)] + + with pytest.raises(AppendError, match="rewrites existing line 128"): + check_append_only("HEAD", candidate) + + +def test_base_check_rejects_truncation(): + lines = _read_lines() + + with pytest.raises(AppendError, match="truncates the ledger: 128 -> 127"): + check_append_only("HEAD", lines[:-1]) + + +def test_base_check_accepts_a_true_append(): + lines = _read_lines() + + assert check_append_only("HEAD", [*lines, "{}"]) == 1 + + +def test_duplicate_identity_without_supersedes_is_rejected(): + lines = _read_lines() + duplicate = _appended_row(json.loads(lines[-1]), value_delta=1) + + with pytest.raises(AppendError, match="without superseding"): + check_rows([*lines, _json_line(duplicate)], len(lines)) + + +def test_mismatched_av2_id_is_rejected(): + lines = _read_lines() + row = _appended_row( + json.loads(lines[-1]), + source_record_id="verification.unique.mismatched_av2", + ) + row["assertionVersion"]["id"] = "av2:" + "0" * 64 + + with pytest.raises(AppendError, match="does not match its content"): + check_rows([*lines, _json_line(row)], len(lines)) + + +@pytest.mark.parametrize( + ("path", "replacement"), + [ + ("measure.source_concept", "DIFFERENT_PUBLISHER_SERIES"), + ("measure.concept_relation", "approximate"), + ("measure.concept_authority", "different_authority"), + ("measure.legal_vintage", "different_legal_vintage"), + ("source.source_file", "different_release.csv"), + ("source.source_sha256", "d" * 64), + ("source_cell_keys", ["different:publisher:cell"]), + ("source_row_keys", ["different:publisher:row"]), + ], +) +def test_av2_binds_material_concept_and_source_lineage( + path: str, replacement: object +): + # Review finding 6: av1 projected only measure concept/unit and four source + # fields, so these material changes collided. The av2 projection binds the + # complete concept mapping, exact source file/digest, and row/cell lineage. + original = json.loads(_read_lines()[-1]) + changed = copy.deepcopy(original) + if "." in path: + parent, key = path.split(".", maxsplit=1) + changed[parent][key] = replacement + else: + changed[path] = replacement + + assert expected_assertion_version_id(changed) != ( + expected_assertion_version_id(original) + ) + + +def test_av2_binds_the_archived_response_digest(): + # av1 could reuse an id across different archived response bytes; av2 binds + # responseArchive.sha256 so re-fetched bytes are a distinct assertion. + original = json.loads(_read_lines()[-1]) + changed = copy.deepcopy(original) + changed["responseArchive"] = {**changed["responseArchive"], "sha256": "e" * 64} + + assert expected_assertion_version_id(changed) != ( + expected_assertion_version_id(original) + ) + + +def test_inconsistent_supersedes_chain_is_rejected(): + lines = _read_lines() + original = json.loads(lines[-1]) + original_id = expected_assertion_version_id(original) + first_correction = _appended_row(original, value_delta=1) + first_correction["assertionVersion"]["supersedes"] = original_id + second_correction = _appended_row(original, value_delta=2) + second_correction["assertionVersion"]["supersedes"] = original_id + + with pytest.raises(AppendError, match="but the active version"): + check_rows( + [*lines, _json_line(first_correction), _json_line(second_correction)], + len(lines), + ) + + +def test_pre_versioning_row_is_addressable_by_recomputed_av2_id(): + lines = _read_lines() + original = json.loads(lines[-1]) + correction = _appended_row(original, value_delta=1) + correction["assertionVersion"]["supersedes"] = ( + expected_assertion_version_id(original) + ) + + check_rows([*lines, _json_line(correction)], len(lines)) + + +def test_full_ci_fact_validation_accepts_an_explicit_correction(): + # Review finding 6: the append gate accepts an explicit correction, but the + # required aggregate-fact validation used to reject it as a duplicate key. + # Validating the supersede-aware effective current view resolves the + # documented correction path. + rows = [json.loads(line) for line in _read_lines()] + original = rows[-1] + correction = _appended_row(original, value_delta=1) + correction["assertionVersion"]["supersedes"] = ( + expected_assertion_version_id(original) + ) + + # The append gate itself accepts exactly the advertised correction shape. + lines = [_json_line(row) for row in rows] + check_rows([*lines, _json_line(correction)], len(lines)) + + # The subsequently required aggregate-fact validation now runs on the + # supersede-aware current view, which drops the superseded original. + current = effective_current_rows([*rows, correction]) + report = validate_facts([_to_aggregate_fact(row) for row in current]) + assert report.valid, report.to_dict() + assert len(current) == len(rows) + + +def test_correction_chain_cannot_restore_a_superseded_legacy_value(): + # Review finding 6: an A->B->A chain must not resurrect a superseded value. + # Restoring A means re-asserting A's exact content, which recomputes A's + # effective id — reserved when A was first seen — so the restore is rejected. + lines = _read_lines() + original = json.loads(lines[-1]) + original_id = expected_assertion_version_id(original) + first_correction = _appended_row(original, value_delta=1) + first_correction["assertionVersion"]["supersedes"] = original_id + restore = copy.deepcopy(original) + restore["assertionVersion"] = { + "id": expected_assertion_version_id(restore), + "supersedes": first_correction["assertionVersion"]["id"], + } + assert restore["assertionVersion"]["id"] == original_id + + with pytest.raises(AppendError, match="restates assertion version"): + check_rows( + [*lines, _json_line(first_correction), _json_line(restore)], + len(lines), + ) + + +def test_provenance_correction_gets_a_distinct_av2_id_and_supersedes(): + # Review finding 6: under av1 a provenance-only correction (changed lineage, + # re-archived bytes) collided with the original id and could be reissued. + # av2 binds lineage, so the correction is a DISTINCT assertion that merges + # cleanly as an explicit supersede rather than reusing the legacy id. + lines = _read_lines() + original = json.loads(lines[-1]) + original_id = expected_assertion_version_id(original) + correction = copy.deepcopy(original) + correction["source_cell_keys"] = ["different:publisher:cell"] + correction["assertionVersion"] = { + "id": expected_assertion_version_id(correction), + "supersedes": original_id, + } + + assert correction["assertionVersion"]["id"] != original_id + check_rows([*lines, _json_line(correction)], len(lines)) + + +@pytest.mark.xfail( + strict=True, + reason=( + "out of scope for this pass: provenance fields are checked for presence " + "and binding, not for retrievedAt/sha/git-sha string formats" + ), +) +@pytest.mark.parametrize( + ("field", "invalid_value"), + [ + ("retrievedAt", "not-a-timestamp"), + ("ledgerRepoSha", "not-a-git-sha"), + ("responseArchive.sha256", "not-a-sha256"), + ], +) +def test_appended_provenance_requires_valid_integrity_metadata( + field: str, invalid_value: str +): + lines = _read_lines() + row = _appended_row( + json.loads(lines[-1]), + source_record_id=f"verification.unique.invalid.{field}", + ) + if field == "responseArchive.sha256": + row["responseArchive"]["sha256"] = invalid_value + else: + row[field] = invalid_value + row["assertionVersion"]["id"] = expected_assertion_version_id(row) + + with pytest.raises(AppendError): + check_rows([*lines, _json_line(row)], len(lines)) + + +def test_joint_manifest_and_file_rewrite_is_rejected_against_git_base(tmp_path): + # A coordinated rewrite of a frozen row plus a rehashed manifest is rejected + # end-to-end: the base-anchored manifest check catches the co-edited + # manifest before the append-only line diff even runs. + lines = _read_lines() + original_text = "\n".join(lines) + "\n" + original_manifest = json.loads(PREFIX_PATH.read_text(encoding="utf-8")) + _write_checker_fixture(tmp_path, original_text, original_manifest) + base = _init_fixture_repo(tmp_path) + + rewritten = json.loads(lines[0]) + rewritten["value"] += 1 + lines[0] = _json_line(rewritten) + manifest = _rehash_manifest(lines) + _write_checker_fixture(tmp_path, "\n".join(lines) + "\n", manifest) + + completed = _run_checker(tmp_path, base) + + assert completed.returncode == 1 + assert "immutable prefix manifest" in completed.stderr + + +def test_manifest_extension_cannot_grandfather_an_unbound_append(tmp_path): + # Review finding 3: a PR extends the co-editable prefix manifest over its own + # append so the unbound row counts as "prefix" and every post-cutover binding + # is skipped. Base-anchoring the manifest and using the BASE prefix count for + # the binding boundary both reject it. + lines = _read_lines() + original_text = "\n".join(lines) + "\n" + original_manifest = json.loads(PREFIX_PATH.read_text(encoding="utf-8")) + _write_checker_fixture(tmp_path, original_text, original_manifest) + base = _init_fixture_repo(tmp_path) + + row = copy.deepcopy(json.loads(lines[-1])) + row["source_record_id"] = "verification.unbound.grandfathered_append" + row["observed_at"] = "2099-02-01" + row["period"] = {"type": "month", "value": "2099-01"} + for field in ( + "retrievedAt", + "sourceVintage", + "ledgerRepoSha", + "responseArchive", + "targetContentHash", + "sourceBindingProjection", + "assertionVersion", + ): + row.pop(field, None) + lines.append(_json_line(row)) + manifest = _extend_manifest_to_all_lines(lines) + _write_checker_fixture(tmp_path, "\n".join(lines) + "\n", manifest) + + # This counterexample also survives the separately required aggregate-fact + # validation: the new period gives it a unique semantic fact key. + facts = [_to_aggregate_fact(json.loads(line)) for line in lines] + assert validate_facts(facts).valid + + completed = _run_checker(tmp_path, base) + + assert completed.returncode == 1, completed.stdout + completed.stderr + assert "immutable prefix manifest" in completed.stderr + + +@pytest.mark.xfail( + strict=True, + reason=( + "by design: the full-file (no base-ref) mode trusts the co-editable " + "manifest, so a joint manifest/file rewrite is only anchored on the PR " + "path (--base-ref) or by an external witness" + ), +) +def test_full_file_check_alone_rejects_joint_manifest_and_file_rewrite(tmp_path): + lines = _read_lines() + rewritten = json.loads(lines[0]) + rewritten["value"] += 1 + lines[0] = _json_line(rewritten) + manifest = _rehash_manifest(lines) + _write_checker_fixture(tmp_path, "\n".join(lines) + "\n", manifest) + + completed = _run_checker(tmp_path) + + assert completed.returncode == 1, completed.stdout + completed.stderr + + +def test_prefix_byte_check_rejects_an_inserted_blank_line(tmp_path): + # Review finding 13: _lines dropped blank lines, so a blank line inserted + # into the frozen JSONL normalized away and passed even with --base-ref. The + # byte guard rejects any blank/whitespace-only line in the covered region. + lines = _read_lines() + manifest = json.loads(PREFIX_PATH.read_text(encoding="utf-8")) + original_text = "\n".join(lines) + "\n" + _write_checker_fixture(tmp_path, original_text, manifest) + base = _init_fixture_repo(tmp_path) + text_with_blank_line = lines[0] + "\n\n" + "\n".join(lines[1:]) + "\n" + _write_checker_fixture(tmp_path, text_with_blank_line, manifest) + + completed = _run_checker(tmp_path, base) + + assert completed.returncode == 1, completed.stdout + completed.stderr + assert "blank or whitespace-only" in completed.stderr + + +@pytest.mark.xfail( + strict=True, + reason=( + "out of scope for this pass: the checker compares only base and final " + "trees, not every intermediate commit, so a rewrite-then-restore whose " + "final tree equals the base passes" + ), +) +def test_base_check_rejects_rewrite_then_restore_across_commits(tmp_path): + lines = _read_lines() + original_text = "\n".join(lines) + "\n" + manifest = json.loads(PREFIX_PATH.read_text(encoding="utf-8")) + _write_checker_fixture(tmp_path, original_text, manifest) + base = _init_fixture_repo(tmp_path) + + rewritten = json.loads(lines[0]) + rewritten["value"] += 1 + rewritten_lines = [_json_line(rewritten), *lines[1:]] + (tmp_path / "ledger" / "official_observations.jsonl").write_text( + "\n".join(rewritten_lines) + "\n", encoding="utf-8" + ) + _git(tmp_path, "add", "ledger/official_observations.jsonl") + _git(tmp_path, "commit", "-qm", "rewrite frozen row") + + (tmp_path / "ledger" / "official_observations.jsonl").write_text( + original_text, encoding="utf-8" + ) + _git(tmp_path, "add", "ledger/official_observations.jsonl") + _git(tmp_path, "commit", "-qm", "restore frozen row") + + completed = _run_checker(tmp_path, base) + + assert completed.returncode == 1, completed.stdout + completed.stderr