Reject empty corpus staging comparisons

docs/operations/CORPUS_HANDOFF.md

@@ -56,41 +56,22 @@ semantic correctness of MinJust content.
 Run a complete inventory of the approved normalized source and the copied
 staging tree into separate private artifact directories, using
 `--normalized-root` for each. Require `status: complete` and
-`verification_status: passed` in both results. Then compare every relative
-path, size, and SHA-256 from the two `inventory.sqlite3` files. This streaming
-comparison uses constant memory and reports only a mismatch count and the first
-few differing paths:
+`verification_status: passed` in both results. The comparison command also
+requires each manifest to contain at least one document and each normalized
+tree to contain at least one fragment; an empty source or staging tree is a
+failure.
 
 ```bash
-python3 - "$source_artifacts/inventory.sqlite3" "$staging_artifacts/inventory.sqlite3" <<'PY'
-import itertools
-import sqlite3
-import sys
-
-def rows(path):
-    db = sqlite3.connect(f"file:{path}?mode=ro&immutable=1", uri=True)
-    yield from db.execute(
-        "SELECT relative_path,size,sha256 FROM files "
-        "WHERE root_id='normalized' AND status='hashed' ORDER BY relative_path"
-    )
-    db.close()
-
-differences = 0
-for source, staging in itertools.zip_longest(rows(sys.argv[1]), rows(sys.argv[2])):
-    if source != staging:
-        differences += 1
-        if differences <= 20:
-            print("mismatch:", source, staging)
-print("mismatch_rows:", differences)
-sys.exit(1 if differences else 0)
-PY
+python3 infra/deploy/compare_corpus_inventories.py \
+  "$source_artifacts/result.json" "$staging_artifacts/result.json"
 ```
 
-Use absolute artifact database paths in the command. A zero exit status confirms
-that both inventories contain the same ordered path/size/SHA-256 rows. Keep the
-source and staging trees quiescent during their scans and comparison. This is
-an equality check only; the owner still reviews the code `6` disposition and
-approves the transfer destination.
+The checker streams every relative path, size, and SHA-256 from both SQLite
+inventories, and requires the same verifier/schema version and different
+normalized roots. A zero exit status confirms a non-empty staging tree matches
+the source. Keep both trees quiescent during their scans and comparison. This
+is an equality check only; the owner still reviews the code `6` disposition
+and approves the transfer destination.
 
 ## Current data disposition
 

docs/operations/CURRENT_STATE.md

@@ -198,7 +198,8 @@ protected file metadata remain unverified.
 ## Change in this PR
 
 Adds a read-only, resumable full-file corpus inventory and normalized-data
-cross-check, synthetic regression tests, and an operator handoff runbook. CI's
+cross-check, a source/staging comparator requiring nonzero documents and
+fragments, synthetic regression tests, and an operator handoff runbook. CI's
 existing `infra/deploy` unittest discovery includes the new tests. No workflow,
 runner, VPS, production data, or Docker settings are changed. The first real
 corpus scan stopped on an integer-storage bug; full validation with the

infra/deploy/compare_corpus_inventories.py

@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""Require non-empty completed normalized audits and compare every file row."""
+
+from __future__ import annotations
+
+import argparse
+import itertools
+import json
+import sqlite3
+import stat
+import sys
+from pathlib import Path
+from typing import Any
+
+
+class ComparisonError(ValueError):
+    pass
+
+
+def read_audit(result_path: Path, label: str) -> tuple[sqlite3.Connection, dict[str, Any]]:
+    result_path = result_path.absolute()
+    try:
+        metadata = result_path.lstat()
+        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
+            raise ComparisonError(f"{label}: result.json must be a regular file")
+        result = json.loads(result_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError) as error:
+        raise ComparisonError(f"{label}: cannot read result.json: {error}") from error
+
+    if not isinstance(result, dict):
+        raise ComparisonError(f"{label}: result.json must contain an object")
+    if result.get("status") != "complete" or result.get("verification_status") != "passed":
+        raise ComparisonError(f"{label}: inventory must be complete and passed")
+    verifier_sha256 = result.get("verifier_sha256")
+    if not isinstance(verifier_sha256, str) or len(verifier_sha256) != 64:
+        raise ComparisonError(f"{label}: verifier SHA-256 is missing")
+    normalized_root = result.get("roots", {}).get("normalized")
+    if not isinstance(normalized_root, str):
+        raise ComparisonError(f"{label}: normalized root is missing")
+    summary = result.get("manifest_summary", {}).get("normalized")
+    if not isinstance(summary, dict):
+        raise ComparisonError(f"{label}: normalized manifest summary is missing")
+    for key in ("documents", "fragments"):
+        if not isinstance(summary.get(key), int) or summary[key] <= 0:
+            raise ComparisonError(f"{label}: normalized {key} must be greater than zero")
+
+    database_path = result_path.with_name("inventory.sqlite3")
+    try:
+        db_metadata = database_path.lstat()
+        if (
+            stat.S_ISLNK(db_metadata.st_mode)
+            or not stat.S_ISREG(db_metadata.st_mode)
+            or db_metadata.st_nlink != 1
+        ):
+            raise ComparisonError(f"{label}: inventory.sqlite3 must be a regular unlinked file")
+        for suffix in ("-wal", "-shm", "-journal"):
+            if Path(f"{database_path}{suffix}").exists():
+                raise ComparisonError(f"{label}: SQLite sidecar files are present")
+        connection = sqlite3.connect(database_path.resolve().as_uri() + "?mode=ro&immutable=1", uri=True)
+        connection.row_factory = sqlite3.Row
+        if connection.execute("PRAGMA quick_check").fetchone()[0] != "ok":
+            raise ComparisonError(f"{label}: inventory database failed quick_check")
+        manifest_documents = connection.execute(
+            "SELECT count(*) FROM manifest_docs WHERE root_id='normalized'"
+        ).fetchone()[0]
+        if manifest_documents != summary["documents"]:
+            raise ComparisonError(f"{label}: normalized document count does not match inventory")
+        files, byte_count, unfinished = connection.execute(
+            "SELECT count(*),coalesce(sum(size),0),"
+            "sum(CASE WHEN status!='hashed' OR size IS NULL OR sha256 IS NULL THEN 1 ELSE 0 END) "
+            "FROM files WHERE root_id='normalized'"
+        ).fetchone()
+        progress = result.get("progress", {}).get("normalized", {})
+        if unfinished or files != progress.get("files") or byte_count != progress.get("bytes"):
+            raise ComparisonError(f"{label}: inventory rows do not match completed scan counters")
+        return connection, {
+            "root": normalized_root,
+            "documents": summary["documents"],
+            "fragments": summary["fragments"],
+            "files": files,
+            "bytes": byte_count,
+            "verifier_sha256": verifier_sha256,
+            "schema_version": result.get("schema_version"),
+        }
+    except sqlite3.Error as error:
+        raise ComparisonError(f"{label}: cannot validate inventory database: {error}") from error
+
+
+def compare_inventories(source_result: Path, staging_result: Path) -> dict[str, int]:
+    source_db, source = read_audit(source_result, "source")
+    try:
+        staging_db, staging = read_audit(staging_result, "staging")
+    except Exception:
+        source_db.close()
+        raise
+    try:
+        if source["root"] == staging["root"]:
+            raise ComparisonError("source and staging must be different normalized roots")
+        if source["verifier_sha256"] != staging["verifier_sha256"] or source["schema_version"] != staging["schema_version"]:
+            raise ComparisonError("source and staging audits must use the same verifier and schema")
+        if (source["documents"], source["fragments"]) != (staging["documents"], staging["fragments"]):
+            raise ComparisonError("source and staging document or fragment counts differ")
+
+        query = (
+            "SELECT relative_path,size,sha256 FROM files "
+            "WHERE root_id='normalized' ORDER BY relative_path"
+        )
+        left = iter(source_db.execute(query))
+        right = iter(staging_db.execute(query))
+        mismatch_count = 0
+        for source_row, staging_row in itertools.zip_longest(left, right):
+            if source_row != staging_row:
+                mismatch_count += 1
+        if mismatch_count:
+            raise ComparisonError(f"source and staging differ in {mismatch_count} inventory rows")
+        return {"documents": source["documents"], "fragments": source["fragments"], "files": source["files"], "bytes": source["bytes"]}
+    finally:
+        source_db.close()
+        staging_db.close()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("source_result", type=Path)
+    parser.add_argument("staging_result", type=Path)
+    arguments = parser.parse_args()
+    try:
+        summary = compare_inventories(arguments.source_result, arguments.staging_result)
+    except (ComparisonError, OSError) as error:
+        print(f"FAIL: {error}", file=sys.stderr)
+        return 1
+    print(
+        "PASS: matched "
+        f"{summary['documents']} documents, {summary['fragments']} fragments, "
+        f"{summary['files']} files, {summary['bytes']} bytes"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

infra/deploy/test_compare_corpus_inventories.py

@@ -0,0 +1,94 @@
+import json
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from compare_corpus_inventories import ComparisonError, compare_inventories
+
+
+class CompareCorpusInventoryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.base = Path(self.temp.name)
+        self.source = self._artifact("source", "/data/source")
+        self.staging = self._artifact("staging", "/data/staging")
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def _artifact(self, name, root, documents=1, fragments=2, rows=None):
+        directory = self.base / name
+        directory.mkdir(mode=0o700)
+        database = directory / "inventory.sqlite3"
+        connection = sqlite3.connect(database)
+        connection.execute(
+            "CREATE TABLE files (root_id TEXT, relative_path TEXT, size INTEGER, sha256 TEXT, status TEXT)"
+        )
+        connection.execute("CREATE TABLE manifest_docs (root_id TEXT, code TEXT)")
+        connection.executemany(
+            "INSERT INTO manifest_docs VALUES ('normalized',?)",
+            [(str(code),) for code in range(documents)],
+        )
+        if rows is None:
+            rows = [
+                ("manifest.sqlite3", 16, "a" * 64),
+                ("documents/1/document.json", 24, "b" * 64),
+            ]
+        connection.executemany(
+            "INSERT INTO files VALUES ('normalized',?,?,?,'hashed')", rows
+        )
+        connection.commit()
+        connection.close()
+        result = {
+            "status": "complete",
+            "verification_status": "passed",
+            "schema_version": 2,
+            "verifier_sha256": "c" * 64,
+            "roots": {"normalized": root},
+            "manifest_summary": {
+                "normalized": {"documents": documents, "fragments": fragments}
+            },
+            "progress": {
+                "normalized": {
+                    "files": len(rows),
+                    "bytes": sum(row[1] for row in rows),
+                }
+            },
+        }
+        result_path = directory / "result.json"
+        result_path.write_text(json.dumps(result), encoding="utf-8")
+        return result_path
+
+    def test_nonempty_source_and_staging_inventories_match(self):
+        self.assertEqual(
+            compare_inventories(self.source, self.staging),
+            {"documents": 1, "fragments": 2, "files": 2, "bytes": 40},
+        )
+
+    def test_empty_source_is_rejected(self):
+        empty = self._artifact("empty-source", "/data/empty-source", documents=0, fragments=0)
+        with self.assertRaisesRegex(ComparisonError, "source: normalized documents"):
+            compare_inventories(empty, self.staging)
+
+    def test_empty_staging_is_rejected(self):
+        empty = self._artifact("empty-staging", "/data/empty-staging", documents=0, fragments=0)
+        with self.assertRaisesRegex(ComparisonError, "staging: normalized documents"):
+            compare_inventories(self.source, empty)
+
+    def test_zero_fragments_are_rejected_even_with_documents(self):
+        empty = self._artifact("no-fragments", "/data/no-fragments", fragments=0)
+        with self.assertRaisesRegex(ComparisonError, "normalized fragments"):
+            compare_inventories(self.source, empty)
+
+    def test_different_file_sha_is_rejected(self):
+        changed = self._artifact(
+            "changed", "/data/changed",
+            rows=[("manifest.sqlite3", 16, "a" * 64), ("documents/1/document.json", 24, "d" * 64)],
+        )
+        with self.assertRaisesRegex(ComparisonError, "differ in 1 inventory rows"):
+            compare_inventories(self.source, changed)
+
+
+if __name__ == "__main__":
+    unittest.main()