Fix corpus inventory indexing and provenance

docs/operations/CORPUS_HANDOFF.md

@@ -17,8 +17,10 @@ python3 infra/deploy/verify_corpus.py \
   --output-dir "$artifact_dir"
 ```
 
-The tool writes `result.json` and `inventory.sqlite3` into that directory. Keep
-the directory private: the inventory contains source paths and checksums. Exit
+The tool writes `result.json`, `inventory.sqlite3`, and a persistent lock file
+into that directory. Keep the directory private: the inventory contains source
+paths and checksums. A second run against the same artifact directory is
+rejected while a scan is active. Exit
 status `0` means the scan and structural checks completed without errors; `1`
 means findings were recorded or the run is incomplete; `2` means the arguments
 or artifact path were invalid. Read `result.json` to distinguish those states.
@@ -31,19 +33,65 @@ reports added, changed, and removed paths relative to entries already stored
 in the inventory. After resuming an interrupted first pass, those counters are
 relative to the partial saved inventory; the full rehash and final checks still
 cover every path. A successful result therefore describes the latest complete
-pass. Run against a quiescent corpus; do not
-normalize, fetch, or copy files into either source tree during the scan. Do not
-point `--output-dir` inside either
-source tree or through a symlink. The verifier rejects such paths before
+pass. Run against a quiescent corpus; do not normalize, fetch, or copy files
+into either source tree during the scan. Do not point `--output-dir` inside
+either source tree or through a symlink. The verifier rejects such paths before
 creating or changing them; it also flags symlinks, reparse points, hard links,
 unsupported file types, and files that change while being read.
 
+The inventory database schema is versioned. Resume requires the same root
+metadata and verifier SHA-256 recorded in `result.json`; start a new artifact
+directory after changing the verifier. `entries` counts files plus directories;
+`files` and `directories` are separate counts. Document-prefix checks use
+indexed path ranges, and stale-path reporting streams rows from SQLite.
+
 The audit checks SQLite integrity, manifest/document directory agreement,
 normalized schema and normalizer versions, source inventory checksums,
 fragment identities and text hashes, required output files, and text assembled
 from fragments. It does not establish the legal authority, completeness, or
 semantic correctness of MinJust content.
 
+## Compare source and staging before transfer
+
+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:
+
+```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
+```
+
+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.
+
 ## Current data disposition
 
 As measured on 2026-10-18, both local manifests contain 209,958 documents and
@@ -65,9 +113,18 @@ From the VPS, a metadata-only HTTPS check to `CheckAvailable` failed at the
 HTTP/2 transport layer (`curl` exit 92, `INTERNAL_ERROR`, HTTP status `000`,
 TLS verification `0`). An HTTP/1.1 retry was not completed.
 
-The full local file inventory and normalized cross-check are still in progress
-until a `result.json` reports `status: complete` or `status: failed` and
-`scan_status: complete`. Do not treat progress output as a completed audit.
+The first full pass (run `b4a4556e30cc464d90bc8d75ef1673e9`) stopped before
+validation with `Python int too large to convert to SQLite INTEGER` after
+2,120,420 raw entries and 397,856 normalized entries. This was an inventory
+metadata storage failure, not a corpus finding. The verifier now stores file
+identity values as text and records its own SHA-256. A complete rehash with the
+updated verifier is required; do not treat the failed pass or progress output
+as a completed audit.
+
+A fresh full rehash is now running in `/tmp/akyldash-corpus-audit-final.gtsD87vc`,
+run `f4dd5e53020f442bbbecee2d0e4986a6`, verifier SHA-256
+`83dac230099fc91972cd14e86e9f8ec72d13bca08f53d402f13a169e096d98f5`. Check
+its private `result.json` for final counters, findings, and manifest comparison.
 
 ## Production transfer and indexing gates
 

docs/operations/CURRENT_STATE.md

@@ -133,12 +133,18 @@ protected file metadata remain unverified.
   is 31,571,968 bytes with SHA-256
   `1679c20e6d0f3e5a9b50737c006c4417a28ebcaf7467d194d3bf8243538f83bf`; all
   entries are `success`, schema `1`, normalizer `1.0.0`.
-- A streaming full-file inventory and cross-check is running read-only with
-  artifacts only in `/tmp/akyldash-corpus-audit.Mjr6aa`. Latest progress was
-  run `b4a4556e30cc464d90bc8d75ef1673e9`; live counters are in its private
-  `result.json`, updated every 1,000 entries. The normalized tree has not yet
-  started. Progress is not a completed audit; record the final result and full
-  manifest comparison when the run finishes.
+- The read-only full-file inventory in `/tmp/akyldash-corpus-audit.Mjr6aa`
+  stopped before validation in run `b4a4556e30cc464d90bc8d75ef1673e9` with
+  `Python int too large to convert to SQLite INTEGER`. At stop it had processed
+  2,120,420 raw entries (1,081,428 files, 1,038,992 directories) and 397,856
+  normalized entries (246,321 files, 151,535 directories). This is a verifier
+  metadata storage failure, not a corpus finding. A full rehash with the
+  corrected, versioned verifier is required before readiness can be stated.
+- A fresh full rehash is running in `/tmp/akyldash-corpus-audit-final.gtsD87vc`,
+  run `f4dd5e53020f442bbbecee2d0e4986a6`, with verifier SHA-256
+  `83dac230099fc91972cd14e86e9f8ec72d13bca08f53d402f13a169e096d98f5`.
+  `result.json` currently reports `scan_status: running`; its live counters
+  must be recorded after completion. Sources remain read-only.
 - Through the repository client, bounded MinJust OpenData requests for
   `GetDocumentListByQuery`, `GetDocument(6)`, control `GetDocument(1096)`, and
   `CheckAvailable` each returned HTTP `404`. The list API returned no code list;
@@ -194,6 +200,7 @@ protected file metadata remain unverified.
 Adds a read-only, resumable full-file corpus inventory and normalized-data
 cross-check, 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 real corpus
-scan is still in progress and the source API's code `6` outcome remains
+runner, VPS, production data, or Docker settings are changed. The first real
+corpus scan stopped on an integer-storage bug; full validation with the
+corrected verifier is still required. The source API's code `6` outcome remains
 unconfirmed.

docs/operations/DEPLOYMENT.md

@@ -71,8 +71,9 @@ Application deployment remains owner-run. The repository-scoped `act_runner`
 separate loop-backed filesystem. CI run 75 and smoke run 76 succeeded on main
 head `0e276ae6`. Runner job containers have neither Docker socket nor production
 secrets. CI has no image publishing or production deployment step. The
-installed sudo helpers support fixed verification, read-only Compose
-`config`/`status`, and guarded rollback. They do not start or stop services.
+repository defines sudo helpers for fixed verification, read-only Compose
+`config`/`status`, and guarded rollback, but none is installed on the VPS. They
+do not start or stop services.
 
 ## CI
 

infra/deploy/test_verify_corpus.py

@@ -1,6 +1,7 @@
 import hashlib
 import json
 import os
+import fcntl
 import sqlite3
 import tempfile
 import unittest
@@ -9,6 +10,10 @@ from pathlib import Path
 from verify_corpus import (
     InventoryError,
     database,
+    insert_or_update_file,
+    prefix_upper_bound,
+    raw_file_rows,
+    combined_source_hash,
     safe_relative_path,
     validate_fragment_file,
     verify_corpus,
@@ -34,6 +39,70 @@ class CorpusInventoryTests(unittest.TestCase):
         )
         connection.close()
 
+    def _nonempty_corpus(self):
+        document = self.raw / "documents/1"
+        edition = document / "editions/2"
+        edition.mkdir(parents=True)
+        (document / "metadata.json").write_text('{"Code":"1"}\n', encoding="utf-8")
+        (edition / "metadata.json").write_text('{"Code":"2"}\n', encoding="utf-8")
+        html = edition / "ru.html"
+        html.write_text("<p>Фрагмент</p>\n", encoding="utf-8")
+        source_files = []
+        for path in sorted(document.rglob("*")):
+            if path.is_file():
+                digest = hashlib.sha256(path.read_bytes()).hexdigest()
+                source_files.append({"path": path.relative_to(self.raw).as_posix(), "sha256": digest})
+        source_sha = combined_source_hash(source_files)
+        raw_manifest = sqlite3.connect(self.raw / "manifest.sqlite3")
+        raw_manifest.execute("INSERT INTO documents VALUES (1, ?, 'now')", (source_sha,))
+        raw_manifest.commit()
+        raw_manifest.close()
+
+        normalized = self.base / "normalized"
+        language = normalized / "documents/1/editions/2/ru"
+        language.mkdir(parents=True)
+        fragment_text = "Фрагмент"
+        fragments = [{
+            "id": "document:1:edition:2:lang:ru:fragment:1",
+            "document_code": "1",
+            "edition_code": "2",
+            "language": "ru",
+            "position": 1,
+            "text": fragment_text,
+            "text_sha256": hashlib.sha256(fragment_text.encode("utf-8")).hexdigest(),
+            "source_path": "documents/1/editions/2/ru.html",
+            "source_sha256": next(item["sha256"] for item in source_files if item["path"].endswith("/ru.html")),
+        }]
+        (language / "fragments.json").write_text(json.dumps(fragments, ensure_ascii=False), encoding="utf-8")
+        (language / "content.txt").write_text(fragment_text + "\n", encoding="utf-8")
+        (language / "content.html").write_text("<p>Фрагмент</p>\n", encoding="utf-8")
+        document_json = {
+            "schema_version": "1",
+            "source_code": "1",
+            "normalizer": {"version": "1.0.0"},
+            "source": {"path": "documents/1", "files": source_files, "sha256": source_sha},
+            "editions": [{"source_code": "2"}],
+        }
+        (normalized / "documents/1/document.json").write_text(json.dumps(document_json), encoding="utf-8")
+        edition_json = {
+            "source_code": "2",
+            "available_languages": ["ru"],
+            "source": {
+                "path": "documents/1/editions/2",
+                "files": [item for item in source_files if item["path"].startswith("documents/1/editions/2/")],
+            },
+            "images": [],
+        }
+        (normalized / "documents/1/editions/2/edition.json").write_text(json.dumps(edition_json), encoding="utf-8")
+        normalized_manifest = sqlite3.connect(normalized / "manifest.sqlite3")
+        normalized_manifest.execute(
+            "CREATE TABLE documents (code TEXT PRIMARY KEY, source_sha256 TEXT, schema_version TEXT, normalizer_version TEXT, state TEXT)"
+        )
+        normalized_manifest.execute("INSERT INTO documents VALUES ('1', ?, '1', '1.0.0', 'success')", (source_sha,))
+        normalized_manifest.commit()
+        normalized_manifest.close()
+        return normalized, source_sha
+
     def test_output_inside_source_is_rejected_without_creating_it(self):
         output = self.raw / "artifacts"
         before_mode = self.raw.stat().st_mode
@@ -106,6 +175,68 @@ class CorpusInventoryTests(unittest.TestCase):
         ).fetchone()
         self.assertEqual(rows[0], hashlib.sha256(b"after").hexdigest())
 
+    def test_large_filesystem_identity_values_are_stored_without_integer_overflow(self):
+        connection = database(self.base / "identity.sqlite3")
+        insert_or_update_file(
+            connection, "run", "raw", "payload", 1, "a" * 64,
+            (2**100, 2**120, 1, 2**110, 2**115, 1), {"raw": {"added": 0, "changed": 0}},
+        )
+        device, inode, mtime_ns, ctime_ns = connection.execute(
+            "SELECT device,inode,mtime_ns,ctime_ns FROM files"
+        ).fetchone()
+        self.assertEqual((device, inode, mtime_ns, ctime_ns), tuple(map(str, (2**100, 2**120, 2**110, 2**115))))
+        connection.close()
+
+    def test_document_prefix_uses_primary_key_range_and_excludes_neighbor_codes(self):
+        connection = database(self.base / "paths.sqlite3")
+        prefix = "documents/1/"
+        upper = prefix_upper_bound(prefix)
+        plan = connection.execute(
+            "EXPLAIN QUERY PLAN SELECT relative_path FROM files WHERE root_id='raw' AND relative_path>=? AND relative_path<?",
+            (prefix, upper),
+        ).fetchone()[3]
+        self.assertIn("USING COVERING INDEX sqlite_autoindex_files_1", plan)
+        connection.executemany(
+            "INSERT INTO files(root_id,relative_path,status,last_seen_run) VALUES('raw',?,'hashed','run')",
+            [(f"documents/{code}/a",) for code in range(20000)]
+            + [("documents/1x/a",)],
+        )
+        self.assertEqual([item["path"] for item in raw_file_rows(connection, "1")], ["documents/1/a"])
+        connection.close()
+
+    def test_inventory_rejects_a_concurrent_writer_and_records_verifier_hash(self):
+        output = self.base / "artifacts"
+        output.mkdir(mode=0o700)
+        lock_path = output / ".inventory.lock"
+        descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaisesRegex(InventoryError, "another inventory run"):
+                verify_corpus(self.raw, None, output)
+        finally:
+            os.close(descriptor)
+
+        result = verify_corpus(self.raw, None, output)
+        self.assertEqual(result["status"], "complete")
+        self.assertRegex(result["verifier_sha256"], r"^[a-f0-9]{64}$")
+
+    def test_removed_file_is_reported_on_resume(self):
+        payload = self.raw / "payload.txt"
+        payload.write_text("payload", encoding="utf-8")
+        output = self.base / "artifacts"
+        self.assertEqual(verify_corpus(self.raw, None, output)["status"], "complete")
+        payload.unlink()
+        result = verify_corpus(self.raw, None, output, resume=True)
+        self.assertEqual(result["input_changes"]["raw"]["removed"], 1)
+        self.assertIn("removed_file", {item["kind"] for item in result["finding_examples"]})
+
+    def test_added_file_is_reported_on_resume(self):
+        output = self.base / "artifacts"
+        self.assertEqual(verify_corpus(self.raw, None, output)["status"], "complete")
+        (self.raw / "extra.txt").write_text("extra", encoding="utf-8")
+        result = verify_corpus(self.raw, None, output, resume=True)
+        self.assertEqual(result["input_changes"]["raw"]["added"], 1)
+
     def test_manifest_error_missing_document_and_extra_directory_are_reported(self):
         connection = sqlite3.connect(self.raw / "manifest.sqlite3")
         connection.execute("INSERT INTO documents VALUES (1, ?, 'now')", ("a" * 64,))
@@ -121,6 +252,39 @@ class CorpusInventoryTests(unittest.TestCase):
         self.assertIn("missing_raw_document_directory", kinds)
         self.assertIn("extra_raw_document_directory", kinds)
 
+    def test_empty_raw_and_normalized_manifests_complete_successfully(self):
+        normalized = self.base / "normalized"
+        normalized.mkdir()
+        connection = sqlite3.connect(normalized / "manifest.sqlite3")
+        connection.execute(
+            "CREATE TABLE documents (code TEXT PRIMARY KEY, source_sha256 TEXT, schema_version TEXT, normalizer_version TEXT, state TEXT)"
+        )
+        connection.commit()
+        connection.close()
+        result = verify_corpus(self.raw, normalized, self.base / "artifacts")
+        self.assertEqual(result["status"], "complete", result["finding_examples"])
+        self.assertEqual(result["manifest_summary"]["raw"]["documents"], 0)
+        self.assertEqual(result["manifest_summary"]["normalized"]["documents"], 0)
+
+    def test_nonempty_raw_and_normalized_manifests_cross_validate_fragment_sources(self):
+        normalized, _ = self._nonempty_corpus()
+        result = verify_corpus(self.raw, normalized, self.base / "artifacts")
+        self.assertEqual(result["status"], "complete", result["finding_examples"])
+        self.assertEqual(result["manifest_summary"]["raw"]["documents"], 1)
+        self.assertEqual(result["manifest_summary"]["normalized"]["fragments"], 1)
+
+    def test_raw_normalized_source_mismatch_is_reported(self):
+        normalized, _ = self._nonempty_corpus()
+        manifest = sqlite3.connect(normalized / "manifest.sqlite3")
+        manifest.execute("UPDATE documents SET source_sha256=? WHERE code='1'", ("0" * 64,))
+        manifest.commit()
+        manifest.close()
+        result = verify_corpus(self.raw, normalized, self.base / "artifacts")
+        kinds = {item["kind"] for item in result["finding_examples"]}
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("normalized_source_hash_mismatch", kinds)
+        self.assertIn("document_source_manifest_mismatch", kinds)
+
     def test_fragment_text_hash_mismatch_is_reported(self):
         normalized = self.base / "normalized"
         fragment_rel = "documents/1/editions/2/ru/fragments.json"

infra/deploy/verify_corpus.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import fcntl
 import hashlib
 import json
 import ntpath
@@ -12,16 +13,18 @@ import sqlite3
 import stat
 import sys
 import uuid
+from contextlib import contextmanager
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Callable, Iterable
 
-INVENTORY_SCHEMA_VERSION = 1
+INVENTORY_SCHEMA_VERSION = 2
 CHECKPOINT_INTERVAL = 1000
 CHUNK_SIZE = 1024 * 1024
 MAX_JSON_BYTES = 64 * 1024 * 1024
 REPARSE_POINT = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)
 NORMALIZER_ROOT = Path(__file__).resolve().parents[2]
+TOOL_SHA256 = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
 sys.path.insert(0, str(NORMALIZER_ROOT))
 from backend.normalization.minjust_cbd import NORMALIZER_VERSION, SCHEMA_VERSION as NORMALIZED_SCHEMA_VERSION  # noqa: E402
 
@@ -150,10 +153,10 @@ def database(path: Path) -> sqlite3.Connection:
             size INTEGER,
             sha256 TEXT,
             status TEXT NOT NULL,
-            device INTEGER,
-            inode INTEGER,
-            mtime_ns INTEGER,
-            ctime_ns INTEGER,
+            device TEXT,
+            inode TEXT,
+            mtime_ns TEXT,
+            ctime_ns TEXT,
             last_seen_run TEXT NOT NULL,
             PRIMARY KEY (root_id, relative_path)
         );
@@ -199,6 +202,35 @@ def database(path: Path) -> sqlite3.Connection:
     return connection
 
 
+def prefix_upper_bound(prefix: str) -> str:
+    for index in range(len(prefix) - 1, -1, -1):
+        codepoint = ord(prefix[index])
+        if codepoint < 0x10FFFF:
+            return prefix[:index] + chr(codepoint + 1)
+    raise InventoryError("path prefix has no upper bound")
+
+
+@contextmanager
+def inventory_lock(output_dir: Path):
+    path = output_dir / ".inventory.lock"
+    descriptor = os.open(path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600)
+    try:
+        metadata = os.fstat(descriptor)
+        if (
+            not stat.S_ISREG(metadata.st_mode)
+            or metadata.st_nlink != 1
+            or stat.S_IMODE(metadata.st_mode) & 0o077
+        ):
+            raise InventoryError("inventory lock must be a regular private file")
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as error:
+            raise InventoryError("another inventory run is using this artifact directory") from error
+        yield
+    finally:
+        os.close(descriptor)
+
+
 def finding(
     connection: sqlite3.Connection,
     run_id: str,
@@ -234,6 +266,7 @@ def result_summary(
     ).fetchall()
     return {
         "schema_version": INVENTORY_SCHEMA_VERSION,
+        "verifier_sha256": TOOL_SHA256,
         "run_id": run_id,
         "updated_at": utc_now(),
         "status": status,
@@ -330,7 +363,7 @@ def insert_or_update_file(
              ctime_ns=excluded.ctime_ns,last_seen_run=excluded.last_seen_run""",
         (
             root_id, relative_path, size, sha256, "hashed",
-            signature[0], signature[1], signature[3], signature[4], run_id,
+            *(str(value) for value in (signature[0], signature[1], signature[3], signature[4])), run_id,
         ),
     )
 
@@ -435,7 +468,7 @@ def scan_root(
     stale_files = connection.execute(
         "SELECT relative_path FROM files WHERE root_id=? AND last_seen_run<>?",
         (root_id, run_id),
-    ).fetchall()
+    )
     for row in stale_files:
         finding(connection, run_id, root_id, "removed_file", row["relative_path"], "file absent from resumed input")
         changed[root_id]["removed"] += 1
@@ -443,7 +476,7 @@ def scan_root(
     stale_dirs = connection.execute(
         "SELECT relative_path FROM directories WHERE root_id=? AND last_seen_run<>?",
         (root_id, run_id),
-    ).fetchall()
+    )
     for row in stale_dirs:
         changed[root_id]["removed_directories"] += 1
     connection.execute("DELETE FROM directories WHERE root_id=? AND last_seen_run<>?", (root_id, run_id))
@@ -557,8 +590,8 @@ def raw_file_rows(connection: sqlite3.Connection, code: str) -> list[dict[str, s
     return [
         {"path": row[0], "sha256": row[1]}
         for row in connection.execute(
-            "SELECT relative_path,sha256 FROM files WHERE root_id='raw' AND relative_path LIKE ? ORDER BY relative_path",
-            (prefix + "%",),
+            "SELECT relative_path,sha256 FROM files WHERE root_id='raw' AND relative_path>=? AND relative_path<? ORDER BY relative_path",
+            (prefix, prefix_upper_bound(prefix)),
         )
     ]
 
@@ -602,9 +635,9 @@ def validate_raw(
         except (OSError, ValueError, InventoryError) as error:
             finding(connection, run_id, "raw", "raw_document_metadata_invalid", f"documents/{code}/metadata.json", str(error))
         edition_dirs = connection.execute(
-            "SELECT relative_path FROM directories WHERE root_id='raw' AND relative_path LIKE ?",
-            (f"documents/{code}/editions/%",),
-        ).fetchall()
+            "SELECT relative_path FROM directories WHERE root_id='raw' AND relative_path>=? AND relative_path<?",
+            (f"documents/{code}/editions/", prefix_upper_bound(f"documents/{code}/editions/")),
+        )
         edition_codes = [row[0].rsplit("/", 1)[-1] for row in edition_dirs if row[0].count("/") == 3]
         if not edition_codes:
             zero_editions += 1
@@ -769,8 +802,8 @@ def validate_normalized(
             edition_dirs = {
                 row[0][len(edition_dir_prefix):]
                 for row in connection.execute(
-                    "SELECT relative_path FROM directories WHERE root_id='normalized' AND relative_path LIKE ?",
-                    (edition_dir_prefix + "%",),
+                    "SELECT relative_path FROM directories WHERE root_id='normalized' AND relative_path>=? AND relative_path<?",
+                    (edition_dir_prefix, prefix_upper_bound(edition_dir_prefix)),
                 )
                 if row[0].count("/") == 3 and row[0].startswith(edition_dir_prefix)
             }
@@ -890,7 +923,7 @@ def setup_run(
         raise InventoryError("inventory already exists; use --resume or choose a new artifact directory")
     if resume and not database_exists:
         raise InventoryError("--resume requested but inventory.sqlite3 is missing")
-    if not resume and next(output_dir.iterdir(), None) is not None:
+    if not resume and any(path.name != ".inventory.lock" for path in output_dir.iterdir()):
         raise InventoryError("artifact output must be empty for a new inventory")
     connection = database(database_path)
     root_metadata = {key: path_metadata(value) for key, value in roots.items()}
@@ -898,12 +931,18 @@ def setup_run(
         stored = {row["key"]: json.loads(row["value"]) for row in connection.execute("SELECT key,value FROM metadata")}
         if stored.get("schema_version") != INVENTORY_SCHEMA_VERSION:
             raise InventoryError("inventory schema differs; use a new artifact directory")
+        if stored.get("verifier_sha256") != TOOL_SHA256:
+            raise InventoryError("verifier code differs; use a new artifact directory")
         if stored.get("roots") != root_metadata:
             raise InventoryError("--resume roots differ from the saved inventory")
     else:
         connection.executemany(
             "INSERT INTO metadata VALUES (?,?)",
-            (("schema_version", json.dumps(INVENTORY_SCHEMA_VERSION)), ("roots", json.dumps(root_metadata, sort_keys=True))),
+            (
+                ("schema_version", json.dumps(INVENTORY_SCHEMA_VERSION)),
+                ("verifier_sha256", json.dumps(TOOL_SHA256)),
+                ("roots", json.dumps(root_metadata, sort_keys=True)),
+            ),
         )
     run_id = uuid.uuid4().hex
     connection.execute(
@@ -938,6 +977,21 @@ def verify_corpus(
         roots[key] = path
     if checkpoint_every <= 0:
         raise InventoryError("checkpoint interval must be greater than zero")
+    output_dir = validate_output_dir(output_dir, roots)
+    with inventory_lock(output_dir):
+        return _run_inventory(
+            roots, output_dir, resume, checkpoint_every, stop_after_entries, progress_callback
+        )
+
+
+def _run_inventory(
+    roots: dict[str, Path],
+    output_dir: Path,
+    resume: bool,
+    checkpoint_every: int,
+    stop_after_entries: int | None,
+    progress_callback: Callable[[dict[str, Any]], None] | None,
+) -> dict[str, Any]:
     connection, run_id, result_path = setup_run(output_dir, roots, resume)
     progress = {key: {"entries": 0, "files": 0, "directories": 0, "bytes": 0, "errors": 0} for key in roots}
     changed = {key: {"added": 0, "changed": 0, "removed": 0, "removed_directories": 0} for key in roots}