2026-10-19 21:10:48 +00:00
9 changed files with 1952 additions and 49 deletions
@@ -2,6 +2,13 @@

 ## Не выпущено

+- Добавлен read-only возобновляемый inventory корпуса с полным хешированием,
+  потоковой сверкой, индексируемыми диапазонами путей и provenance SHA-256
+  проверяющего; значения `stat` хранятся без ограничения SQLite INTEGER.
+  Сравнение source/staging сверяет путь, размер и SHA-256, удерживает shared
+  locks и отклоняет пустые корпуса без документов или фрагментов. Runbook и
+  `CURRENT_STATE.md` отражают, что полный аудит корпуса ещё выполняется; это
+  изменение не означает готовность корпуса или production.
 - CI actions теперь используют workspace, переданный Gitea, для Caddyfile и Kaniko build context; CI run 69 и smoke run 70 прошли на self-hosted runner.
 - Runner installer сохраняет `HOME` и XDG-пути в user-unit `docker.service`, перезапускает активный daemon и проверяет окружение его процесса.
 - Runner installer защищает image и mountpoint root-owned home-каталогом, оставляет регистрацию и workspace в runner-owned каталоге, безопасно форматирует образ по открытому FD и проверяет `nosuid,nodev`.
@@ -0,0 +1,131 @@
+# Corpus inventory and production handoff
+
+The repository contains a read-only verifier for the ignored local corpus at
+`data/minjust-cbd/` and `data/minjust-normalized/`. It does not download,
+normalize, copy, index, or modify source files. Keep inventory artifacts outside
+the repository and both source trees.
+
+## Run the inventory
+
+From the repository root, create a private artifact directory and run:
+
+```bash
+artifact_dir=$(mktemp -d /tmp/akyldash-corpus-audit.XXXXXXXX)
+python3 infra/deploy/verify_corpus.py \
+  --raw-root data/minjust-cbd \
+  --normalized-root data/minjust-normalized \
+  --output-dir "$artifact_dir"
+```
+
+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.
+Warnings, including documents with zero text fragments, are reported
+separately and do not turn the scan into a failure.
+
+For an interrupted run, use the same artifact directory and add `--resume`.
+Resume rehashes every source file instead of trusting cached timestamps, then
+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
+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. 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 infra/deploy/compare_corpus_inventories.py \
+  "$source_artifacts/result.json" "$staging_artifacts/result.json"
+```
+
+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. It opens both existing private `.inventory.lock` files
+read-only and holds shared nonblocking locks through comparison; a missing,
+unsafe, or busy lock fails closed. 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
+
+As measured on 2026-10-18, both local manifests contain 209,958 documents and
+pass SQLite `quick_check`. All normalized manifest entries are `success` at
+schema `1` and normalizer `1.0.0`. The raw manifest records one fetch error for
+code `6`, which is absent from its `documents` table; this does not prove that
+the source document is absent. On 2026-10-18, bounded calls through the
+repository's rate-limited MinJust client received HTTP `404` for
+`GetDocumentListByQuery`, `GetDocument(6)`, control `GetDocument(1096)`, and
+`CheckAvailable`. The listing call failed before returning a code list, and
+both document calls failed identically. The repeated `404` is consistent with
+the OpenData endpoint being unavailable or its route having changed, but does
+not identify which. It is not evidence that code `6` is absent. The [public
+document listing page](https://cbd.minjust.gov.kg/list-docs/kg) was reachable,
+but it does not confirm this numeric API code. Do not describe the corpus as
+fully verified or silently waive this error. Recheck the official listing and
+API when the endpoint responds, then record an explicit owner disposition.
+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 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
+
+Production deployment is not enabled. Before any transfer, the owner must
+review the root-only host audit, confirm the target data location is empty or
+backed up, resolve the raw code `6` finding, and approve the exact normalized
+corpus to use. Transfer only `minjust-normalized/`; keep the raw source archive
+offline. Do not preserve the source's Windows/WSL file permission bits. Copy
+to a new staging location with owner-approved permissions, verify the copied
+manifest and files there, and only then consider placing it under `DATA_ROOT`.
+Never overwrite an existing corpus or OpenSearch volume.
+
+The current bulk importer checks individual Bulk API item results, but it can
+switch an alias after a zero-fragment input and does not enforce a refresh,
+count comparison, or representative-query check before that switch. Until a
+separate change adds those gates, do not pass `--alias` during production
+loading. Load into a new versioned index only after the corpus transfer is
+verified; require a nonzero fragment count, successful item results, an
+explicit refresh, a matching OpenSearch document count, and representative
+Russian/Kyrgyz queries before an owner performs an alias change. Take and
+verify the required backup before any production index or service change.
@@ -1,25 +1,26 @@
 # Infrastructure current state

-Last host audit: 2026-10-12. Gitea Actions status checked 2026-10-18. Source: read-only SSH session as `codex`, owner-reported
-UFW state, local repository inspection, and Gitea API. Root-only Docker state,
-UFW rule details, and protected file metadata remain unverified.
+Read-only host recheck: 2026-10-18. Source: SSH as `codex`, local repository
+inspection, and Gitea API. Root-only Docker state, UFW rule details, and
+protected file metadata remain unverified.

 ## VPS facts

 - Host: `avprogs.hlab.kz`, `206.223.249.118`, SSH port `22`, user `codex`.
 - OS: Ubuntu Server `24.04.5 LTS`; kernel `6.8.0-142-generic` from the prior
  inventory audit.
- Resources: 4 vCPU, 7.7 GiB RAM, 3.8 GiB swap, 99 GiB disk with 92 GiB free
-  at this audit.
+- Resources: 4 vCPU, 7.7 GiB RAM, 3.8 GiB swap, 99 GiB root filesystem with
+  84 GiB free on 2026-10-18. Runner storage is a separate 15 GiB ext4 loop
+  mount with 12 GiB free at that check.
 - Network: interface `ens3`, IPv4 `206.223.249.118/32`.
 - `codex` is UID/GID 1000, groups `codex` and `users`; `sudo -n true` requires
  a password. Docker CLI cannot access `/var/run/docker.sock` as `codex`.
 - SSH service is running. The inspected sshd configuration has public-key and
  password authentication enabled and `PermitRootLogin yes`; no SSH changes
  were made.
- Bootstrap completed. Docker CE/CLI `29.9.0`, Compose `5.6.0`, containerd
-  `2.4.1`, and runc `1.5.2` are installed; package versions and active Docker
-  service were independently checked. The Docker daemon API remains root-only.
+- Bootstrap completed. Docker CLI `29.9.0` and Compose `5.6.0` were rechecked
+  over SSH on 2026-10-18; containerd `2.4.1` and runc `1.5.2` are from the
+  prior package audit. The Docker daemon API remains root-only.
 - `/etc/docker/daemon.json` has the expected `local` log driver and `10m` / `3`
  rotation limits. Bootstrap paths exist with root ownership: `/opt/akyldash`
  mode `0755`, `/etc/akyldash` and `/var/backups/akyldash` mode `0700`, and
@@ -27,7 +28,10 @@ UFW rule details, and protected file metadata remain unverified.
 - `/opt/akyldash/current`, verifier/rollback binaries, and the sudoers file are
  absent. `codex` has not received sudo or Docker-group access. Metadata for
  `/etc/akyldash/production.env` and `/etc/akyldash/review_secret` could not be
-  checked because `/etc/akyldash` is root-only.
+  checked because access is denied. The target
+  `/var/lib/akyldash/data/minjust-normalized` and its manifest metadata were
+  also denied; their presence, ownership, and contents are unknown. No sudo was
+  used.
 - TCP listeners observed: SSH `22` and loopback DNS only; no listener on `1500`,
  `80`, or `443`. No ISPmanager package or service unit was found through the
  unprivileged checks; root can confirm with the audit script below.
@@ -39,8 +43,10 @@ UFW rule details, and protected file metadata remain unverified.
  `https://git.vakanaut.ru/api/v1/version` returned Gitea `1.25.5`; runner
  registration and Gitea Actions protocol connectivity are not confirmed.
 - `akyldash.kg` resolves to Cloudflare IPv4/IPv6 edge addresses. From the VPS,
-  HTTP, HTTPS, and `/review` still return Cloudflare `521`; origin TLS and
-  Cloudflare Full (strict) are not verified.
+  HTTPS returned Cloudflare `521` with TLS verification result `0` on
+  2026-10-18. TCP listeners were SSH `22` and local DNS only, so no origin
+  listener on `80`/`443` was observed. Origin TLS and Cloudflare Full (strict)
+  are not verified.

 ## Application and Gitea facts

@@ -49,13 +55,12 @@ UFW rule details, and protected file metadata remain unverified.
  PRs enabled. `main` is protected and requires one approval; the current API
  principal cannot push or merge directly to `main`.
 - The repository runner-list API returns `403` because this API principal is
-  not the repository owner. Gitea status checked 2026-10-18 showed smoke run 62
-  successful and CI run 61 pending on PR head `69c62466`. That workflow used
-  `ubuntu-latest`, while the smoke workflow used `akyldash-ci`; this label
-  difference is a plausible explanation, not a confirmed scheduler diagnosis.
-  The owner reports that the runner is registered and Idle and rootless Docker
-  works (report date not recorded). Runner inventory and package publishing
-  permissions remain unverified.
+  not the repository owner. The latest verified CI run 75 and smoke run 76 both
+  succeeded on `main` head `0e276ae6`. The Actions run-list endpoint returned
+  `403` for this API principal during this recheck, so no newer run state was
+  independently retrieved. The owner reports the runner is registered and
+  Idle; repository runner inventory and package publishing permissions remain
+  unverified.
 - No public application listener is observed on TCP `80`/`443`. Docker
  container/volume state still requires the root audit. Repository Compose describes
  Caddy `2.11.6` pinned by digest, Python Search API `0.11.3`, SQLite
@@ -89,11 +94,15 @@ UFW rule details, and protected file metadata remain unverified.
  ext4 loop-backed image/workspace storage, capacity 1, per-job limits (2 CPU,
  3 GiB RAM, 512 processes), account slice limits (2.5 CPU, 4 GiB RAM, 2048
  tasks), rotated file logs, and a systemd health timer. Jobs target the
-  `akyldash-ci` label and do not receive either Docker socket. The owner reports
-  that runner registration and rootless Docker work (report date not recorded).
-  Gitea status checked 2026-10-18 showed smoke run 62 successful and CI run 61
-  pending on PR head `69c62466`. The workflow/label difference may explain that
-  pending status, but the scheduler reason was not independently verified. On
+  `akyldash-ci` label and do not receive either Docker socket. On 2026-10-18,
+  SSH showed `akyldash-act-runner.service` active under `akyldash-runner`, with
+  zero restarts and about 115 MiB current memory. Its rootless Docker store is
+  mounted from `/dev/loop0` as ext4 with `rw,nosuid,nodev`; the 15 GiB mount had
+  12 GiB free. The separate `akyldash-runner-restart.service` remains failed
+  with exit status 1 since 2026-10-14; its journal had no entries, so the cause
+  is unknown. The latest verified main CI run 75 and smoke run 76 succeeded on
+  head `0e276ae6`. The Actions API returned 403 during this recheck, so no newer
+  run state was independently retrieved. On
  head `3a89dc21`, CI run 63 failed at the host audit root guard and smoke run 64
  succeeded. On head `436806a8`, CI run 65 failed in the Caddy container action
  with `exec: "validate": executable file not found`; Kaniko was skipped and
@@ -113,6 +122,43 @@ UFW rule details, and protected file metadata remain unverified.
  backend `0.11.3`, frontend `0.1.2`, Telegram bot `0.2.2` (the bot is not
  deployed on this VPS).

+## Corpus readiness (2026-10-18)
+
+- The local raw archive and normalized corpus are ignored under `data/`; no
+  corpus files were changed. Their read-only SQLite manifests both pass
+  `quick_check` and each report 209,958 documents. The raw manifest is
+  28,295,168 bytes with SHA-256
+  `f8e3dd2fdec607583ca09cfdb5b76171bb7bfa1e41a0e07ee5aca8481e18e17e`; it has
+  one fetch error, code `6`, absent from `documents`. The normalized manifest
+  is 31,571,968 bytes with SHA-256
+  `1679c20e6d0f3e5a9b50737c006c4417a28ebcaf7467d194d3bf8243538f83bf`; all
+  entries are `success`, schema `1`, normalizer `1.0.0`.
+- 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;
+  the direct requests do not establish whether code `6` exists. The public
+  listing page is reachable, but it does not map the numeric API code. A
+  metadata-only `CheckAvailable` request from the VPS failed in HTTP/2 with
+  `curl` exit 92 (`INTERNAL_ERROR`), HTTP status `000`, and TLS verification 0;
+  the HTTP/1.1 retry was not completed. Treat the code `6` disposition and
+  full corpus readiness as unresolved.
+- On the VPS, read-only `stat` of `/var/lib/akyldash/data/minjust-normalized`
+  and its manifest was denied. Production corpus presence and metadata remain
+  unknown. No corpus was transferred and no production service, Docker socket,
+  runner policy, or VPS configuration was changed.
+
 ## Checks not completed

 - Docker Engine version from the server API, container and volume inventory,
@@ -149,17 +195,14 @@ UFW rule details, and protected file metadata remain unverified.
 5. Test backup/restore and rollback only with synthetic data in an isolated
   environment before any production deployment.

-## CI workflow change in this PR
+## Change in this PR

-The branch targets `akyldash-ci` and uses digest-pinned isolated actions for
-Caddy validation and the two production Dockerfile builds. Kaniko image
-publishing and cache publishing are disabled. This repository change does not
-modify the VPS or grant workflow containers a Docker socket. CI run 65 failed
-in the Caddy action because the runner tried to execute `validate` directly;
-Kaniko was skipped. CI run 67 then found that the action's fixed
-`/github/workspace` path did not contain the repository; Kaniko was skipped.
-The branch now passes Gitea's `${{ github.workspace }}` value into all three
-container action invocations. CI run 69 validated Caddy and completed both
-image builds; smoke run 70 passed. Actual per-action resource-limit enforcement
-and peak usage were not independently measured or verified. Host-wide runner
-metrics, image publishing, and production deployment remain unverified.
+Adds a read-only, resumable full-file corpus inventory and normalized-data
+cross-check, a source/staging comparator requiring nonzero documents and
+fragments plus held shared artifact locks, 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
+corrected verifier is still required. The source API's code `6` outcome remains
+unconfirmed.
@@ -34,8 +34,9 @@ backup/restore have been tested.
   It creates `/etc/akyldash/production.env` from the tracked example and
   generates `/etc/akyldash/review_secret`; both are root-owned mode `0600`.
   It refuses to overwrite either file and does not start services or create
-   volumes. Review `DATA_ROOT` and place the approved normalized corpus there
-   before deployment.
+   volumes. Review `DATA_ROOT` and follow the
+   [corpus inventory and handoff gates](CORPUS_HANDOFF.md) before placing the
+   approved normalized corpus there.
 3. After reviewing the root-owned checkout and audit output, run
   `sudo bash /opt/akyldash/current/infra/deploy/install-helper.sh`. It installs
   the backup reader, read-only verifier, `akyldash-compose config|status`, and
@@ -65,13 +66,14 @@ backup/restore have been tested.
   sudo chmod 0600 /opt/akyldash/deployed-image.txt
   ```

-Application deployment remains owner-run. A repository-level Actions runner
-installation is prepared in [CI_RUNNER.md](CI_RUNNER.md), but it is not yet
-installed or registered; its Online state and smoke workflow are unverified.
-The runner's job containers have no Docker socket or 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.
+Application deployment remains owner-run. The repository-scoped `act_runner`
+`0.2.13` service is active under `akyldash-runner` and uses rootless Docker on a
+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
+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

@@ -79,5 +81,7 @@ rollback. They do not start or stop services.
 production images on pull requests and pushes to `main`. It does not publish an
 image or deploy production. `.gitea/workflows/runner-smoke.yaml` exercises the
 `akyldash-ci` runner on pull requests, pushes to `main`, or manual dispatch.
-Runner registration is incomplete, so neither workflow execution on that
-runner nor current `ubuntu-latest` runner availability has been confirmed.
+Both workflows passed on main head `0e276ae6` in runs 75 and 76 respectively.
+The repository runner-list endpoint is not available to the current API
+principal, so its registration details cannot be independently read through
+Gitea API.
@@ -11,8 +11,13 @@

 - загрузчик находится в `backend/ingestion/minjust_cbd.py`;
 - сырой архив хранится в `data/minjust-cbd` и исключён из Git;
- в манифесте 209 811 успешно загруженных документов;
- 13 кодов остаются в таблице `errors` и отсутствуют в таблице `documents`;
+- в read-only проверенном манифесте 209 958 документов;
+- один код, `6`, остаётся в таблице `errors` и отсутствует в `documents`;
+  это не подтверждает отсутствие документа в источнике: при последней
+  проверке OpenData API возвращал HTTP `404` для списка, `GetDocument(6)`,
+  контрольного `GetDocument(1096)` и `CheckAvailable`;
+- в нормализованном манифесте 209 958 успешных записей, schema `1`,
+  normalizer `1.0.0`; полная потоковая проверка всех файлов ещё выполняется.
 - документ содержит `metadata.json` и каталог `editions`;
 - редакция содержит `metadata.json`, `ru.html` и/или `ky.html`, иногда
  изображения;
@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""Require non-empty completed normalized audits and compare every file row."""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import itertools
+import json
+import os
+import sqlite3
+import stat
+import sys
+from contextlib import ExitStack, contextmanager
+from pathlib import Path
+from typing import Any
+
+
+class ComparisonError(ValueError):
+    pass
+
+
+@contextmanager
+def shared_inventory_lock(result_path: Path, label: str):
+    lock_path = result_path.absolute().parent / ".inventory.lock"
+    if not hasattr(os, "O_NOFOLLOW"):
+        raise ComparisonError(f"{label}: safe inventory lock opens are unsupported")
+    try:
+        descriptor = os.open(lock_path, os.O_RDONLY | os.O_NOFOLLOW)
+    except OSError as error:
+        raise ComparisonError(f"{label}: cannot open inventory lock: {error}") from error
+    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 ComparisonError(f"{label}: inventory lock must be a private regular file")
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        except BlockingIOError as error:
+            raise ComparisonError(f"{label}: inventory is active (lock busy)") from error
+        yield
+    finally:
+        os.close(descriptor)
+
+
+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]:
+    with ExitStack() as locks:
+        locks.enter_context(shared_inventory_lock(source_result, "source"))
+        locks.enter_context(shared_inventory_lock(staging_result, "staging"))
+        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())
@@ -0,0 +1,132 @@
+import fcntl
+import json
+import os
+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)
+        (directory / ".inventory.lock").touch(mode=0o600)
+        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)
+
+    def test_busy_source_lock_is_rejected(self):
+        self._assert_busy_lock(self.source, "source")
+
+    def test_busy_staging_lock_is_rejected(self):
+        self._assert_busy_lock(self.staging, "staging")
+
+    def _assert_busy_lock(self, result_path, label):
+        descriptor = os.open(result_path.parent / ".inventory.lock", os.O_RDWR)
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaisesRegex(ComparisonError, rf"{label}: inventory is active"):
+                compare_inventories(self.source, self.staging)
+        finally:
+            os.close(descriptor)
+
+    def test_missing_lock_is_rejected_without_creating_it(self):
+        lock = self.source.parent / ".inventory.lock"
+        lock.unlink()
+        with self.assertRaisesRegex(ComparisonError, "source: cannot open inventory lock"):
+            compare_inventories(self.source, self.staging)
+        self.assertFalse(lock.exists())
+
+    def test_symlink_lock_is_rejected(self):
+        lock = self.source.parent / ".inventory.lock"
+        lock.unlink()
+        target = self.base / "external-lock"
+        target.touch(mode=0o600)
+        lock.symlink_to(target)
+        with self.assertRaisesRegex(ComparisonError, "source: cannot open inventory lock"):
+            compare_inventories(self.source, self.staging)
+
+    def test_read_only_lock_can_be_shared(self):
+        (self.source.parent / ".inventory.lock").chmod(0o400)
+        self.assertEqual(compare_inventories(self.source, self.staging)["documents"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,328 @@
+import hashlib
+import json
+import os
+import fcntl
+import sqlite3
+import tempfile
+import unittest
+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,
+)
+
+
+class CorpusInventoryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.base = Path(self.temp.name)
+        self.raw = self.base / "raw"
+        self.raw.mkdir()
+        self._manifest()
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def _manifest(self):
+        connection = sqlite3.connect(self.raw / "manifest.sqlite3")
+        connection.executescript(
+            "CREATE TABLE documents (code INTEGER PRIMARY KEY, source_sha256 TEXT, fetched_at TEXT);"
+             "CREATE TABLE errors (code INTEGER PRIMARY KEY, message TEXT, failed_at TEXT);"
+        )
+        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
+        with self.assertRaisesRegex(InventoryError, "separate"):
+            verify_corpus(self.raw, None, output)
+        self.assertFalse(output.exists())
+        self.assertEqual(self.raw.stat().st_mode, before_mode)
+        self.assertEqual({path.name for path in self.raw.iterdir()}, {"manifest.sqlite3"})
+
+    def test_symlink_output_is_rejected_without_changing_target(self):
+        target = self.base / "target"
+        target.mkdir(mode=0o755)
+        link = self.base / "output-link"
+        try:
+            link.symlink_to(target, target_is_directory=True)
+        except (NotImplementedError, OSError):
+            self.skipTest("directory symlinks are unavailable")
+        before_mode = target.stat().st_mode
+        with self.assertRaisesRegex(InventoryError, "link or file"):
+            verify_corpus(self.raw, None, link)
+        self.assertEqual(target.stat().st_mode, before_mode)
+        self.assertEqual(list(target.iterdir()), [])
+
+    def test_public_existing_output_is_rejected_without_chmod(self):
+        output = self.base / "public-output"
+        output.mkdir(mode=0o755)
+        output.chmod(0o755)
+        before_mode = output.stat().st_mode
+        with self.assertRaisesRegex(InventoryError, "must be private"):
+            verify_corpus(self.raw, None, output)
+        self.assertEqual(output.stat().st_mode, before_mode)
+        self.assertEqual(list(output.iterdir()), [])
+
+    def test_resume_rejects_symlink_inventory_without_touching_target(self):
+        output = self.base / "artifacts"
+        output.mkdir(mode=0o700)
+        target = self.base / "do-not-touch.sqlite3"
+        target.write_bytes(b"sentinel")
+        link = output / "inventory.sqlite3"
+        try:
+            link.symlink_to(target)
+        except (NotImplementedError, OSError):
+            self.skipTest("file symlinks are unavailable")
+        with self.assertRaisesRegex(InventoryError, "private regular file"):
+            verify_corpus(self.raw, None, output, resume=True)
+        self.assertEqual(target.read_bytes(), b"sentinel")
+
+    def test_rejects_unsafe_cross_platform_relative_paths(self):
+        for value in ("../outside", "C:\\outside", "folder\\file", "//server/share"):
+            with self.subTest(value=value), self.assertRaises(InventoryError):
+                safe_relative_path(value)
+
+    def test_interrupted_inventory_resumes_and_detects_changed_input(self):
+        payload = self.raw / "payload.txt"
+        payload.write_text("before", encoding="utf-8")
+        output = self.base / "artifacts"
+        stopped = verify_corpus(self.raw, None, output, stop_after_entries=1)
+        self.assertEqual(stopped["status"], "incomplete")
+
+        resumed = verify_corpus(self.raw, None, output, resume=True)
+        self.assertEqual(resumed["status"], "complete")
+        self.assertEqual(resumed["input_changes"]["raw"]["added"], 1)
+
+        payload.write_text("after", encoding="utf-8")
+        changed = verify_corpus(self.raw, None, output, resume=True)
+        self.assertEqual(changed["status"], "complete")
+        self.assertEqual(changed["input_changes"]["raw"]["changed"], 1)
+        rows = sqlite3.connect(output / "inventory.sqlite3").execute(
+            "SELECT sha256 FROM files WHERE relative_path='payload.txt'"
+        ).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,))
+        connection.execute("INSERT INTO errors VALUES (6, 'fetch failed', 'now')")
+        connection.commit()
+        connection.close()
+        (self.raw / "documents" / "999").mkdir(parents=True)
+
+        result = verify_corpus(self.raw, None, self.base / "artifacts")
+        kinds = {finding["kind"] for finding in result["finding_examples"]}
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("raw_manifest_error", kinds)
+        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"
+        fragment_path = normalized / fragment_rel
+        fragment_path.parent.mkdir(parents=True)
+        fragment = {
+            "id": "document:1:edition:2:lang:ru:fragment:1",
+            "document_code": "1",
+            "edition_code": "2",
+            "language": "ru",
+            "position": 1,
+            "text": "text",
+            "text_sha256": "0" * 64,
+            "source_path": "documents/1/source.txt",
+            "source_sha256": "1" * 64,
+        }
+        fragment_path.write_text(json.dumps([fragment]), encoding="utf-8")
+        content_path = fragment_path.with_name("content.txt")
+        content_path.write_text("text\n", encoding="utf-8")
+        connection = database(self.base / "test.sqlite3")
+        content_digest = hashlib.sha256(b"text\n").hexdigest()
+        connection.execute(
+            "INSERT INTO files(root_id,relative_path,sha256,status,last_seen_run) VALUES('normalized',?,?, 'hashed','run')",
+            (content_path.relative_to(normalized).as_posix(), content_digest),
+        )
+        validate_fragment_file(
+            connection,
+            "run",
+            normalized,
+            fragment_path,
+            fragment_rel,
+            hashlib.sha256(fragment_path.read_bytes()).hexdigest(),
+            {"documents/1/source.txt": "1" * 64},
+        )
+        kinds = {row[0] for row in connection.execute("SELECT kind FROM findings WHERE run_id='run'")}
+        connection.close()
+        self.assertIn("fragment_text_checksum_mismatch", kinds)
+
+
+if __name__ == "__main__":
+    unittest.main()