Document runner ProtectHome fix

# Conflicts:
#	CHANGELOG.md
#	docs/operations/CURRENT_STATE.md

.gitea/workflows/ci.yaml

@@ -51,7 +51,8 @@ jobs:
             infra/runner/test-install-runner-idempotency.sh infra/runner/test-install-runner-guard.sh \
             infra/runner/test-ci-container-builds.sh
           python3 -c "import ast,pathlib; ast.parse(pathlib.Path('infra/runner/secure-runner-image.py').read_text())"
-      - name: Run rollback helper tests
+          python3 -c "import ast,pathlib; ast.parse(pathlib.Path('infra/deploy/validate_release.py').read_text())"
+      - name: Run deployment contract and rollback tests
         run: python3 -m unittest discover -s infra/deploy -p 'test_*.py' -v
       - name: Validate production Compose and Caddyfile
         run: |

CHANGELOG.md

@@ -2,6 +2,28 @@
 
 ## Не выпущено
 
+- Infrastructure: systemd unit Gitea Actions runner теперь использует
+  `ProtectHome=read-only`, сохраняя доступность `/run/user` и существующий
+  writable carve-out для rootless runtime; VPS не изменялся.
+- Backend `0.12.1`: добавлен ограниченный возобновляемый staging редакций из
+  текущего API Минюста с проверкой точных ID и метаданных, изоляцией от
+  legacy-корпуса и сохранением исходного порядка редакций при нормализации и
+  экспорте. Тест overlap guard задаёт безопасные права fixture независимо от
+  `umask` CI; сам guard не ослаблен.
+- Rollback preflight проверяет digest-pinned image references по сохранённому
+  image ID и останавливается до изменений deployment; переключение production
+  на digest и полноценный digest rollback пока не реализованы.
+- Infrastructure: добавлен офлайн-контракт release manifest с точной схемой,
+  digest-ссылками и repository allowlist; проверен Compose override с
+  `--no-build` как требованием будущего deploy. Публикация, provenance и
+  production-деплой не включены.
+- Добавлен 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`.

README.md

@@ -10,7 +10,7 @@ Telegram-бот — только часть рабочего окружения
 ## Текущее состояние
 
 Сейчас реализованы Telegram-бот-секретарь версии `0.2.2`, backend версии
-`0.11.3` и внутренний frontend лаборатории `0.1.2`. Авторизация хранится в
+`0.12.2` и внутренний frontend лаборатории `0.1.2`. Авторизация хранится в
 SQLite; первый пользователь `admin` создаётся интерактивной CLI-командой.
 Поиск расширяет аббревиатуры организационно-правовых форм и поднимает
 действующий Налоговый кодекс для запросов о налоге на прибыль.
@@ -18,7 +18,7 @@ SQLite; первый пользователь `admin` создаётся инт
 | Компонент | Версия | Состояние |
 |---|---:|---|
 | Telegram-бот | `0.2.2` | на Synology работает `0.2.1`; обновление после слияния |
-| Backend | `0.11.3` | авторизация и поиск с фильтрами; тематический поиск ОсОО/ЖЧК |
+| Backend | `0.12.2` | авторизация, поиск с фильтрами и ограниченный staging текущего API Минюста |
 | Frontend | `0.1.2` | лаборатория поиска, личный кабинет и управление учетными записями |
 | Сбор и обработка правовых данных | `0.8.0` | добавлены relevance set и оценка выдачи |
 | RAG и база знаний | — | ещё не созданы |
@@ -62,4 +62,4 @@ python3 -m unittest discover -s tools/telegram-bot -v
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.2 · Frontend v0.1.2

backend/README.md

@@ -1,6 +1,6 @@
 # Backend Акылдаш
 
-Версия: `0.11.3`
+Версия: `0.12.2`
 
 Первая backend-область проекта — загрузка правовых документов из официального
 Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в
@@ -54,6 +54,49 @@ result = sync_archive(output_path)
 Планировщик, очередь задач и PostgreSQL пока не добавлены: модуль не зависит от
 выбора будущего backend-фреймворка.
 
+## Текущий API Минюста и отдельный staging
+
+Старый OpenData endpoint может возвращать `404`; для отдельной ограниченной
+выгрузки официальный сайт предоставляет `/api/v1/GetDocument` и
+`/api/v1/GetEdition`. Адаптер `ingestion.minjust_current` получает один
+указанный документ и сохраняет его в отдельное поколение staging:
+
+```bash
+PYTHONPATH=backend python3 -m ingestion.minjust_current \
+  --output /tmp/minjust-current-code6 \
+  --document-code 6
+```
+
+Продолжайте прерванную выгрузку только с тем же каталогом и `--resume`. По
+умолчанию лимит скачанного текста одного документа — 4 GiB, лимит запросов —
+один в секунду; каталог не должен совпадать или пересекаться с
+`data/minjust-cbd` и
+`data/minjust-normalized`. Адаптер сохраняет исходные ответы и checkpoints,
+публикует папку документа атомарно и не заменяет существующий код в staging.
+Каталог staging должен быть отдельной директорией режима 0700, принадлежащей
+текущему пользователю; его родитель должен быть существующим доверенным
+каталогом. Символические ссылки в generation, checkpoints и каталогах
+документов отклоняются. Legacy-корни вычисляются от корня репозитория, а не
+от текущего рабочего каталога.
+Для `/GetEdition?lang=kg` официальный ответ помещает кыргызский HTML в
+`contentRu` при `contentKg=null`; архивный язык определяется запрошенным
+`lang`, а не именем поля ответа.
+
+Исторические редакции запрашиваются только с `exact=true`; без этого параметра
+официальный endpoint возвращал ID последней редакции даже для другого
+запрошенного ID. Адаптер требует совпадения response ID и запрошенного ID,
+сверяет названия RU/KY с metadata `GetDocument` и сохраняет request contract в
+checkpoint. Старые staging generations без этого contract нельзя возобновить.
+`exact=true` и совпадение ID проверены на двух контрольных редакциях; это не
+подтверждает полноту всего источника.
+
+Порядок редакций сохраняется по ответу `/GetDocument` и используется как
+официальный порядок при нормализации и выборе текущей редакции в поисковом
+индексе. Вложения и встроенные изображения этим адаптером не подтверждаются:
+staging остаётся неполным до отдельной проверки медиа и не заменяет корпус.
+Получение полного списка документов не является частью этой ограниченной
+процедуры.
+
 ## Нормализация архива
 
 Нормализатор читает исходный архив без изменений и создаёт отдельный набор
@@ -253,4 +296,4 @@ PYTHONPATH=backend python3 -m search.query "ЖЧК ачуу тартиби" --la
 
 ---
 
-Акылдаш · Backend v0.11.3 · Frontend v0.1.2
+Акылдаш · Backend v0.12.2 · Frontend v0.1.2

backend/ingestion/minjust_current.py

@@ -0,0 +1,720 @@
+"""Read-only staging client for the current MinJust CBD API."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import http.client
+import json
+import logging
+import os
+import sqlite3
+import stat
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
+from pathlib import Path
+from typing import Callable
+
+from ingestion.minjust_cbd import atomic_write, connect_manifest, json_bytes
+
+API_ROOT = "https://cbd.minjust.gov.kg/api/v1"
+APP_VERSION = "0.12.2"
+REQUEST_CONTRACT = {
+    "version": 2,
+    "endpoint": "/GetEdition",
+    "edition_id_parameter": "editionId",
+    "exact_parameter": "true",
+    "language_parameter": "lang",
+    "response_id_must_match": True,
+}
+MAX_RESPONSE_BYTES = 16 * 1024 * 1024
+DEFAULT_DOCUMENT_BUDGET = 4 * 1024 * 1024 * 1024
+LANGUAGES = (("ru", "ru", "textRusType"), ("kg", "ky", "textKyrType"))
+LOGGER = logging.getLogger(__name__)
+REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
+
+
+class SourceError(RuntimeError):
+    pass
+
+
+class SourceChanged(SourceError):
+    pass
+
+
+@dataclass(frozen=True)
+class ApiResponse:
+    value: dict
+    body: bytes
+    sha256: str
+
+
+class CurrentCbdClient:
+    def __init__(
+        self,
+        requests_per_second: float = 1,
+        timeout: int = 60,
+        retries: int = 5,
+        max_response_bytes: int = MAX_RESPONSE_BYTES,
+        opener: Callable = urllib.request.urlopen,
+        sleep: Callable[[float], None] = time.sleep,
+        clock: Callable[[], float] = time.monotonic,
+    ) -> None:
+        if requests_per_second <= 0 or timeout <= 0 or retries <= 0 or max_response_bytes <= 0:
+            raise ValueError("rate, timeout, retries, and response limit must be positive")
+        self.interval = 1 / requests_per_second
+        self.timeout = timeout
+        self.retries = retries
+        self.max_response_bytes = max_response_bytes
+        self.opener = opener
+        self.sleep = sleep
+        self.clock = clock
+        self.last_request = 0.0
+
+    def request_json(
+        self,
+        path: str,
+        *,
+        method: str = "GET",
+        parameters: tuple[tuple[str, object], ...] = (),
+        payload: dict | None = None,
+    ) -> ApiResponse:
+        if not path.startswith("/") or ".." in path.split("/"):
+            raise ValueError("API path must be an absolute endpoint path")
+        query = urllib.parse.urlencode(parameters, doseq=True)
+        url = API_ROOT + path + (f"?{query}" if query else "")
+        body = json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":")).encode() if method == "POST" else None
+        for attempt in range(self.retries):
+            wait = self.interval - (self.clock() - self.last_request)
+            if wait > 0:
+                self.sleep(wait)
+            request = urllib.request.Request(
+                url,
+                data=body,
+                method=method,
+                headers={
+                    "Accept": "application/json",
+                    **({"Content-Type": "application/json"} if body is not None else {}),
+                    "User-Agent": f"Akyldash/{APP_VERSION} (+https://cbd.minjust.gov.kg)",
+                },
+            )
+            try:
+                self.last_request = self.clock()
+                with self.opener(request, timeout=self.timeout) as response:
+                    raw = response.read(self.max_response_bytes + 1)
+                if len(raw) > self.max_response_bytes:
+                    raise SourceError(f"MinJust response exceeds {self.max_response_bytes} bytes")
+                value = json.loads(raw)
+                if not isinstance(value, dict):
+                    raise SourceError("MinJust response must be a JSON object")
+                return ApiResponse(value, raw, hashlib.sha256(raw).hexdigest())
+            except urllib.error.HTTPError as error:
+                if error.code != 429 and error.code < 500:
+                    raise SourceError(f"MinJust returned HTTP {error.code} for {path}") from error
+                delay = 2**attempt
+                if error.code == 429:
+                    delay = max(delay, retry_after_seconds(error.headers.get("Retry-After")))
+            except (TimeoutError, urllib.error.URLError, http.client.IncompleteRead, json.JSONDecodeError) as error:
+                delay = 2**attempt
+                LOGGER.warning("MinJust request failed for %s (attempt %s/%s)", path, attempt + 1, self.retries)
+                last_error = error
+            if attempt + 1 < self.retries:
+                self.sleep(delay)
+        raise SourceError(f"MinJust request failed after {self.retries} attempts: {path}") from locals().get("last_error")
+
+    def document(self, alias: str) -> ApiResponse:
+        response = self.request_json("/GetDocument", parameters=(("DocumentCode", alias),))
+        value = response.value
+        canonical = value.get("documentCode")
+        editions = value.get("editions")
+        if not isinstance(canonical, int) or isinstance(canonical, bool) or canonical < 1:
+            raise SourceError("MinJust document has an invalid canonical documentCode")
+        if not isinstance(editions, list):
+            raise SourceError("MinJust document is missing editions")
+        ids = [edition_id(item) for item in editions]
+        if len(ids) != len(set(ids)):
+            raise SourceError(f"MinJust document {canonical} contains duplicate edition IDs")
+        return response
+
+    def edition(self, identifier: int, language: str) -> ApiResponse:
+        if identifier < 1 or language not in {"ru", "kg"}:
+            raise ValueError("edition ID or requested language is invalid")
+        response = self.request_json(
+            "/GetEdition",
+            parameters=(("editionId", identifier), ("lang", language), ("exact", "true")),
+        )
+        returned_id = response.value.get("id")
+        if not isinstance(returned_id, int) or isinstance(returned_id, bool) or returned_id < 1:
+            raise SourceError(f"MinJust returned an invalid edition response ID for {identifier}")
+        if returned_id != identifier:
+            raise SourceChanged(f"MinJust exact edition ID {returned_id} does not match requested ID {identifier}")
+        return response
+
+
+def retry_after_seconds(value: str | None) -> float:
+    if not value:
+        return 0
+    try:
+        return max(0, float(value))
+    except ValueError:
+        try:
+            return max(0, (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds())
+        except (TypeError, ValueError, OverflowError):
+            return 0
+
+
+def edition_id(value: object) -> int:
+    if not isinstance(value, dict):
+        raise SourceError("MinJust document contains an invalid edition")
+    identifier = value.get("id")
+    if not isinstance(identifier, int) or isinstance(identifier, bool) or identifier < 1:
+        raise SourceError("MinJust edition has an invalid id")
+    return identifier
+
+
+def open_manifest(path: Path) -> sqlite3.Connection:
+    connection = sqlite3.connect(path)
+    connection.execute("PRAGMA journal_mode=DELETE")
+    connection.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS documents (
+            alias TEXT PRIMARY KEY,
+            canonical_id INTEGER,
+            response_sha256 TEXT,
+            expected_editions INTEGER,
+            state TEXT NOT NULL,
+            error TEXT,
+            updated_at TEXT NOT NULL
+        );
+        CREATE TABLE IF NOT EXISTS editions (
+            canonical_id INTEGER NOT NULL,
+            edition_id INTEGER NOT NULL,
+            source_order INTEGER NOT NULL,
+            response_sha256 TEXT NOT NULL,
+            PRIMARY KEY (canonical_id, edition_id)
+        );
+        CREATE TABLE IF NOT EXISTS languages (
+            canonical_id INTEGER NOT NULL,
+            edition_id INTEGER NOT NULL,
+            request_language TEXT NOT NULL,
+            archive_language TEXT NOT NULL,
+            response_sha256 TEXT,
+            content_sha256 TEXT,
+            byte_count INTEGER NOT NULL DEFAULT 0,
+            state TEXT NOT NULL,
+            error TEXT,
+            updated_at TEXT NOT NULL,
+            PRIMARY KEY (canonical_id, edition_id, request_language)
+        );
+        CREATE TABLE IF NOT EXISTS generation (
+            contract_sha256 TEXT NOT NULL
+        );
+        """
+    )
+    return connection
+
+
+def reject_symlink_components(path: Path) -> None:
+    current = Path(path.anchor)
+    for part in path.parts[1:]:
+        current /= part
+        try:
+            mode = current.lstat().st_mode
+        except FileNotFoundError:
+            continue
+        if stat.S_ISLNK(mode):
+            raise ValueError(f"staging path must not contain a symbolic link: {current}")
+        if current != path and not stat.S_ISDIR(mode):
+            raise ValueError(f"staging parent must be a directory: {current}")
+
+
+def ensure_stage_path(root: Path, path: Path) -> Path:
+    root = root.absolute()
+    path = path.absolute()
+    try:
+        relative = path.relative_to(root)
+    except ValueError as error:
+        raise ValueError("staging file escaped its private generation") from error
+    current = root
+    if current.is_symlink():
+        raise ValueError("staging root must not be a symbolic link")
+    for part in relative.parts:
+        current /= part
+        try:
+            mode = current.lstat().st_mode
+        except FileNotFoundError:
+            continue
+        if stat.S_ISLNK(mode):
+            raise ValueError(f"staging path must not contain a symbolic link: {relative}")
+        if current != path and not stat.S_ISDIR(mode):
+            raise ValueError(f"staging parent must be a directory: {relative}")
+    return path
+
+
+def stage_write(root: Path, path: Path, content: bytes) -> None:
+    path = ensure_stage_path(root, path)
+    atomic_write(path, content)
+    ensure_stage_path(root, path)
+
+
+def stage_mkdir(root: Path, path: Path) -> None:
+    path = ensure_stage_path(root, path)
+    path.mkdir(parents=True, exist_ok=True)
+    ensure_stage_path(root, path)
+    if not path.is_dir():
+        raise ValueError("staging path must be a directory")
+
+
+def validate_stage_tree(output: Path) -> None:
+    for relative in (
+        Path("current-api.sqlite3"), Path("current-api.sqlite3-journal"),
+        Path("current-api.sqlite3-wal"), Path("current-api.sqlite3-shm"),
+        Path("manifest.sqlite3"), Path("manifest.sqlite3-journal"),
+        Path("manifest.sqlite3-wal"), Path("manifest.sqlite3-shm"),
+        Path("request-contract.json"),
+    ):
+        path = ensure_stage_path(output, output / relative)
+        if path.exists() and not path.is_file():
+            raise ValueError(f"staging checkpoint must be a regular file: {relative}")
+    for relative in (Path(".work"), Path("documents")):
+        root = ensure_stage_path(output, output / relative)
+        if root.exists() and not root.is_dir():
+            raise ValueError(f"staging path must be a directory: {relative}")
+        if root.is_dir():
+            for directory, directories, files in os.walk(root, followlinks=False):
+                for name in directories + files:
+                    ensure_stage_path(output, Path(directory) / name)
+
+
+def validate_stage_root(output: Path, resume: bool) -> Path:
+    requested = Path(output)
+    if ".." in requested.parts:
+        raise ValueError("staging output path must not contain '..'")
+    destination = Path(os.path.abspath(requested))
+    reject_symlink_components(destination)
+    parent = destination.parent
+    if not parent.is_dir():
+        raise ValueError("staging output parent must already exist")
+    for ancestor in (parent, *parent.parents):
+        info = ancestor.lstat()
+        mode = stat.S_IMODE(info.st_mode)
+        if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
+            raise ValueError("staging output ancestors must be real directories")
+        sticky_parent = bool(mode & stat.S_ISVTX)
+        if mode & 0o022 and not sticky_parent:
+            raise ValueError("staging output parent is not trusted")
+    for relative in (Path("data/minjust-cbd"), Path("data/minjust-normalized")):
+        legacy = (REPOSITORY_ROOT / relative).resolve()
+        if destination == legacy or destination.is_relative_to(legacy) or legacy.is_relative_to(destination):
+            raise ValueError("current API staging must not overlap the existing corpus")
+    if destination.exists():
+        if destination.is_symlink() or not destination.is_dir():
+            raise ValueError("staging output must be a real directory")
+    else:
+        destination.mkdir(mode=0o700)
+    root_info = destination.lstat()
+    root_mode = stat.S_IMODE(root_info.st_mode)
+    if (
+        stat.S_ISLNK(root_info.st_mode)
+        or not stat.S_ISDIR(root_info.st_mode)
+        or root_info.st_uid != os.geteuid()
+        or root_mode & 0o077
+        or (root_mode & 0o700) != 0o700
+    ):
+        raise ValueError("staging output must be a private directory owned by the current user")
+    validate_stage_tree(destination)
+    if any(destination.iterdir()) and not resume:
+        raise ValueError("staging output is not empty; use --resume for this generation")
+    contract_path = destination / "request-contract.json"
+    if contract_path.exists():
+        try:
+            contract = json.loads(contract_path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeError, json.JSONDecodeError) as error:
+            raise SourceChanged("staging generation request contract is invalid") from error
+        if contract != REQUEST_CONTRACT:
+            raise SourceChanged("staging generation uses a different request contract")
+    else:
+        if any(destination.iterdir()):
+            raise SourceChanged("staging generation has no exact-edition request contract; start a new generation")
+        stage_write(destination, contract_path, json_bytes(REQUEST_CONTRACT))
+    return destination
+
+
+def stage_document(
+    output: Path,
+    alias: str,
+    client: CurrentCbdClient,
+    *,
+    resume: bool = False,
+    max_document_bytes: int = DEFAULT_DOCUMENT_BUDGET,
+) -> dict:
+    """Stage one document and its official editions without touching legacy roots."""
+    if not alias or len(alias) > 64 or "/" in alias or "\\" in alias:
+        raise ValueError("document code is invalid")
+    if max_document_bytes <= 0:
+        raise ValueError("document byte budget must be positive")
+    output = validate_stage_root(output, resume)
+    work_root = output / ".work" / "documents"
+    target_root = output / "documents"
+    connection = open_manifest(output / "current-api.sqlite3")
+    raw_manifest = connect_manifest(output / "manifest.sqlite3")
+    try:
+        contract_sha = hashlib.sha256(json.dumps(REQUEST_CONTRACT, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+        saved_contract = connection.execute("SELECT contract_sha256 FROM generation").fetchone()
+        if saved_contract and saved_contract[0] != contract_sha:
+            raise SourceChanged("checkpoint request contract changed; start a new staging generation")
+        if not saved_contract:
+            if connection.execute("SELECT COUNT(*) FROM documents").fetchone()[0]:
+                raise SourceChanged("existing checkpoints have no request contract; start a new staging generation")
+            with connection:
+                connection.execute("INSERT INTO generation (contract_sha256) VALUES (?)", (contract_sha,))
+        response = client.document(alias)
+        source = response.value
+        canonical = source["documentCode"]
+        record_sha = response.sha256
+        editions = source["editions"]
+        previous = connection.execute(
+            "SELECT canonical_id, response_sha256, expected_editions FROM documents WHERE alias = ?",
+            (alias,),
+        ).fetchone()
+        if previous and previous[1] is not None and (
+            previous[0] != canonical or previous[1] != record_sha or previous[2] != len(editions)
+        ):
+            raise SourceChanged("document metadata changed; start a new staging generation")
+        working_document = work_root / str(canonical)
+        final_document = target_root / str(canonical)
+        ensure_stage_path(output, working_document)
+        ensure_stage_path(output, final_document)
+        if final_document.is_symlink():
+            raise ValueError("staging target must not be a symbolic link")
+        if final_document.exists() and not previous:
+            raise ValueError("staging target already exists without a matching checkpoint")
+        if final_document.exists():
+            final_metadata = ensure_stage_path(output, final_document / "metadata.json")
+            source_document = ensure_stage_path(output, final_document / "source-document.json")
+            metadata = json.loads(final_metadata.read_text(encoding="utf-8"))
+            if (
+                metadata.get("SourceApi", {}).get("response_sha256") != record_sha
+                or hashlib.sha256(source_document.read_bytes()).hexdigest() != record_sha
+            ):
+                raise SourceChanged("published staging document does not match the current source record")
+            for edition in editions:
+                identifier = edition_id(edition)
+                for request_language, archive_language, type_field in LANGUAGES:
+                    html_path = ensure_stage_path(
+                        output,
+                        final_document / "editions" / str(identifier) / f"{archive_language}.html",
+                    )
+                    checkpoint = connection.execute(
+                        "SELECT state, content_sha256 FROM languages WHERE canonical_id = ? AND edition_id = ? AND request_language = ?",
+                        (canonical, identifier, request_language),
+                    ).fetchone()
+                    if not edition.get(type_field):
+                        if not checkpoint or checkpoint[0] != "unavailable":
+                            raise SourceError(f"published staging document has no checkpoint for edition {identifier} {request_language}")
+                    elif (
+                        not checkpoint
+                        or checkpoint[0] != "complete"
+                        or not html_path.is_file()
+                        or sha256_file(html_path) != checkpoint[1]
+                    ):
+                        raise SourceError(f"published staging document failed verification for edition {identifier} {request_language}")
+            document_sha = hash_tree(final_document)
+            fetched_at = utc_now()
+            manifest_row = raw_manifest.execute(
+                "SELECT source_sha256 FROM documents WHERE code = ?", (canonical,)
+            ).fetchone()
+            if manifest_row and manifest_row[0] != document_sha:
+                raise SourceError("published staging document does not match its raw manifest")
+            with raw_manifest:
+                raw_manifest.execute(
+                    "INSERT OR IGNORE INTO documents (code, source_sha256, fetched_at) VALUES (?, ?, ?)",
+                    (canonical, document_sha, fetched_at),
+                )
+                raw_manifest.execute(
+                    "INSERT INTO errors (code, message, failed_at) VALUES (?, ?, ?) "
+                    "ON CONFLICT(code) DO UPDATE SET message=excluded.message, failed_at=excluded.failed_at",
+                    (canonical, "attachments and embedded media are not independently verified", fetched_at),
+                )
+            with connection:
+                connection.execute(
+                    "UPDATE documents SET state='text_complete_incomplete_attachments', error=NULL, updated_at=? WHERE alias=?",
+                    (fetched_at, alias),
+                )
+            return {
+                "alias": alias,
+                "canonical_id": canonical,
+                "editions": len(editions),
+                "source_sha256": document_sha,
+                "state": "text_complete_incomplete_attachments",
+                "resumed": True,
+            }
+        stage_mkdir(output, working_document)
+        document_metadata = map_document_metadata(source, alias, response.sha256)
+        stage_write(output, working_document / "metadata.json", json_bytes(document_metadata))
+        stage_write(output, working_document / "source-document.json", response.body)
+        with connection:
+            connection.execute(
+                "INSERT INTO documents (alias, canonical_id, response_sha256, expected_editions, state, updated_at) VALUES (?, ?, ?, ?, 'fetching', ?) "
+                "ON CONFLICT(alias) DO UPDATE SET canonical_id=excluded.canonical_id, response_sha256=excluded.response_sha256, "
+                "expected_editions=excluded.expected_editions, state='fetching', error=NULL, updated_at=excluded.updated_at",
+                (alias, canonical, record_sha, len(editions), utc_now()),
+            )
+            for order, edition in enumerate(editions):
+                identifier = edition_id(edition)
+                existing = connection.execute(
+                    "SELECT source_order, response_sha256 FROM editions WHERE canonical_id = ? AND edition_id = ?",
+                    (canonical, identifier),
+                ).fetchone()
+                if existing and existing != (order, hashlib.sha256(json_bytes(edition)).hexdigest()):
+                    raise SourceChanged(f"edition metadata changed for {identifier}")
+                connection.execute(
+                    "INSERT OR IGNORE INTO editions (canonical_id, edition_id, source_order, response_sha256) VALUES (?, ?, ?, ?)",
+                    (canonical, identifier, order, hashlib.sha256(json_bytes(edition)).hexdigest()),
+                )
+
+        incomplete = []
+        for order, edition in enumerate(editions):
+            identifier = edition_id(edition)
+            directory = working_document / "editions" / str(identifier)
+            stage_mkdir(output, directory)
+            stage_write(output, directory / "metadata.json", json_bytes(map_edition_metadata(edition, order)))
+            for request_language, archive_language, type_field in LANGUAGES:
+                html_path = ensure_stage_path(output, directory / f"{archive_language}.html")
+                checkpoint = connection.execute(
+                    "SELECT state, content_sha256, byte_count FROM languages WHERE canonical_id = ? AND edition_id = ? AND request_language = ?",
+                    (canonical, identifier, request_language),
+                ).fetchone()
+                if checkpoint and checkpoint[0] == "complete":
+                    if not html_path.is_file() or sha256_file(html_path) != checkpoint[1]:
+                        raise SourceError(f"staged content hash mismatch for edition {identifier} {request_language}")
+                    continue
+                if not edition.get(type_field):
+                    save_language_checkpoint(
+                        connection, canonical, identifier, request_language, archive_language,
+                        None, None, 0, "unavailable", "source metadata reports no language file",
+                    )
+                    continue
+                try:
+                    language_response = client.edition(identifier, request_language)
+                    response_id = language_response.value.get("id")
+                    if isinstance(response_id, bool) or not isinstance(response_id, int) or response_id != identifier:
+                        raise SourceChanged(f"MinJust exact edition ID mismatch for {identifier}")
+                    for field in ("nameRus", "nameKyr"):
+                        if edition.get(field) is not None and language_response.value.get(field) != edition[field]:
+                            raise SourceChanged(f"MinJust exact edition metadata mismatch for {identifier}: {field}")
+                    content = language_response.value.get("contentRu")
+                    if not isinstance(content, str) or not content:
+                        raise SourceError(f"MinJust returned no content for edition {identifier} language {request_language}")
+                    encoded = content.encode("utf-8")
+                    completed_bytes = connection.execute(
+                        "SELECT COALESCE(SUM(byte_count), 0) FROM languages WHERE canonical_id = ? AND state = 'complete'",
+                        (canonical,),
+                    ).fetchone()[0]
+                    if completed_bytes + len(encoded) > max_document_bytes:
+                        raise SourceError(f"document {canonical} exceeds its {max_document_bytes}-byte staging budget")
+                    response_metadata = {
+                        key: value
+                        for key, value in language_response.value.items()
+                        if key not in {"contentRu", "contentKg"}
+                    }
+                    response_metadata["requested_language"] = request_language
+                    response_metadata["archive_language"] = archive_language
+                    response_metadata["requested_edition_id"] = identifier
+                    response_metadata["exact"] = True
+                    response_metadata["response_sha256"] = language_response.sha256
+                    stage_write(
+                        output,
+                        directory / "source" / f"{request_language}.json",
+                        json_bytes(response_metadata),
+                    )
+                    stage_write(output, html_path, encoded)
+                    save_language_checkpoint(
+                        connection, canonical, identifier, request_language, archive_language,
+                        language_response.sha256, hashlib.sha256(encoded).hexdigest(), len(encoded), "complete", None,
+                    )
+                except Exception as error:
+                    message = str(error)[:500]
+                    save_language_checkpoint(
+                        connection, canonical, identifier, request_language, archive_language,
+                        None, None, 0, "error", message,
+                    )
+                    incomplete.append({"edition_id": identifier, "language": request_language, "error": message})
+                    if isinstance(error, SourceChanged) or (isinstance(error, SourceError) and "budget" in message):
+                        raise
+
+        if incomplete:
+            with connection:
+                connection.execute(
+                    "UPDATE documents SET state='incomplete', error=?, updated_at=? WHERE alias=?",
+                    (json.dumps(incomplete, ensure_ascii=False), utc_now(), alias),
+                )
+            return {"alias": alias, "canonical_id": canonical, "editions": len(editions), "incomplete": incomplete}
+
+        working_metadata = ensure_stage_path(output, working_document / "metadata.json")
+        metadata = json.loads(working_metadata.read_text(encoding="utf-8"))
+        metadata["SourceCompleteness"] = {
+            "text": "complete",
+            "attachments": "not independently verified",
+        }
+        stage_write(output, working_metadata, json_bytes(metadata))
+        stage_mkdir(output, target_root)
+        ensure_stage_path(output, working_document)
+        ensure_stage_path(output, final_document)
+        if final_document.exists():
+            raise ValueError("refusing to replace a document in this staging generation")
+        os.replace(working_document, final_document)
+        validate_stage_tree(output)
+        document_sha = hash_tree(final_document)
+        fetched_at = utc_now()
+        with raw_manifest:
+            raw_manifest.execute(
+                "INSERT INTO documents (code, source_sha256, fetched_at) VALUES (?, ?, ?)",
+                (canonical, document_sha, fetched_at),
+            )
+            raw_manifest.execute(
+                "INSERT INTO errors (code, message, failed_at) VALUES (?, ?, ?)",
+                (canonical, "attachments and embedded media are not independently verified", fetched_at),
+            )
+        with connection:
+            connection.execute(
+                "UPDATE documents SET state='text_complete_incomplete_attachments', updated_at=? WHERE alias=?",
+                (fetched_at, alias),
+            )
+        return {
+            "alias": alias,
+            "canonical_id": canonical,
+            "editions": len(editions),
+            "source_sha256": document_sha,
+            "state": "text_complete_incomplete_attachments",
+        }
+    except Exception as error:
+        with connection:
+            connection.execute(
+                "INSERT INTO documents (alias, state, error, updated_at) VALUES (?, 'error', ?, ?) "
+                "ON CONFLICT(alias) DO UPDATE SET state='error', error=excluded.error, updated_at=excluded.updated_at",
+                (alias, str(error)[:500], utc_now()),
+            )
+        raise
+    finally:
+        raw_manifest.close()
+        connection.close()
+
+
+def map_document_metadata(source: dict, alias: str, response_sha256: str) -> dict:
+    canonical = source["documentCode"]
+    return {
+        "Code": canonical,
+        "Name": {"Rus": source.get("nameRus"), "Kyr": source.get("nameKyr")},
+        "Title": {"Rus": source.get("headerRu"), "Kyr": source.get("headerKg")},
+        "Number": source.get("number"),
+        "DateAdopted": source.get("dateAdopted"),
+        "DateInclusion": source.get("dateInclusion"),
+        "DateOfEntry": source.get("dateOfEntry"),
+        "DatePublication": source.get("datePublication"),
+        "DateRegistration": source.get("dateRegistration"),
+        "IsPublicInRegister": source.get("isPublicInRegister"),
+        "SourceStatus": source.get("status"),
+        "SourceClassId": source.get("classId"),
+        "SourceRefTypeId": source.get("refTypeId"),
+        "SourceAuthorityIds": source.get("authoritiesId"),
+        "SourcePublicationIds": source.get("sourcePublicationsId"),
+        "SourceMetadata": {key: value for key, value in source.items() if key != "editions"},
+        "SourceDocumentCode": alias,
+        "SourceMediaStatus": "not independently verified",
+        "SourceApi": {"endpoint": "/GetDocument", "response_sha256": response_sha256},
+    }
+
+
+def map_edition_metadata(source: dict, source_order: int) -> dict:
+    return {
+        "Code": edition_id(source),
+        "Name": {"Rus": source.get("nameRus"), "Kyr": source.get("nameKyr")},
+        "SourceOrder": source_order,
+        "SourceTextTypes": {
+            "ru": source.get("textRusType"),
+            "ky": source.get("textKyrType"),
+        },
+        "SourceMediaStatus": "not independently verified",
+        "SourceMetadata": source,
+    }
+
+
+def save_language_checkpoint(
+    connection: sqlite3.Connection,
+    canonical: int,
+    edition: int,
+    request_language: str,
+    archive_language: str,
+    response_sha256: str | None,
+    content_sha256: str | None,
+    byte_count: int,
+    state: str,
+    error: str | None,
+) -> None:
+    with connection:
+        connection.execute(
+            "INSERT INTO languages (canonical_id, edition_id, request_language, archive_language, response_sha256, content_sha256, byte_count, state, error, updated_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(canonical_id, edition_id, request_language) DO UPDATE SET "
+            "archive_language=excluded.archive_language, response_sha256=excluded.response_sha256, content_sha256=excluded.content_sha256, "
+            "byte_count=excluded.byte_count, state=excluded.state, error=excluded.error, updated_at=excluded.updated_at",
+            (canonical, edition, request_language, archive_language, response_sha256, content_sha256, byte_count, state, error, utc_now()),
+        )
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for block in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def hash_tree(root: Path) -> str:
+    digest = hashlib.sha256()
+    for path in sorted(item for item in root.rglob("*") if item.is_file()):
+        digest.update(path.relative_to(root).as_posix().encode())
+        digest.update(b"\0")
+        digest.update(sha256_file(path).encode("ascii"))
+        digest.update(b"\0")
+    return digest.hexdigest()
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True, help="new, separate staging generation")
+    parser.add_argument("--document-code", required=True, help="one official document alias or canonical code")
+    parser.add_argument("--resume", action="store_true", help="resume only this staging generation")
+    parser.add_argument("--requests-per-second", type=float, default=1)
+    parser.add_argument("--timeout", type=int, default=60)
+    parser.add_argument("--retries", type=int, default=5)
+    parser.add_argument("--max-document-bytes", type=int, default=DEFAULT_DOCUMENT_BUDGET)
+    parser.add_argument("--version", action="version", version=APP_VERSION)
+    return parser.parse_args()
+
+
+def main() -> int:
+    arguments = parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    result = stage_document(
+        arguments.output,
+        arguments.document_code,
+        CurrentCbdClient(arguments.requests_per_second, arguments.timeout, arguments.retries),
+        resume=arguments.resume,
+        max_document_bytes=arguments.max_document_bytes,
+    )
+    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
+    return int(result.get("state") != "text_complete_incomplete_attachments")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/normalization/minjust_cbd.py

@@ -23,9 +23,9 @@ from pathlib import Path
 from typing import Callable
 from urllib.parse import urlsplit
 
-APP_VERSION = "0.9.3"
+APP_VERSION = "0.10.0"
 SCHEMA_VERSION = "1"
-NORMALIZER_VERSION = "1.0.0"
+NORMALIZER_VERSION = "1.1.0"
 LANGUAGES = ("ru", "ky")
 LOGGER = logging.getLogger(__name__)
 
@@ -376,13 +376,16 @@ def load_json(path: Path) -> dict:
 def edition_summary(edition_directory: Path, input_root: Path) -> dict:
     metadata = load_json(edition_directory / "metadata.json")
     languages = [language for language in LANGUAGES if (edition_directory / f"{language}.html").is_file()]
-    return {
+    summary = {
         "source_code": str(metadata.get("Code", edition_directory.name)),
         "name": bilingual(metadata.get("Name")),
         "source_type": clean_value(metadata.get("Type")),
         "available_languages": languages,
         "source_path": edition_directory.relative_to(input_root).as_posix(),
     }
+    if "SourceOrder" in metadata:
+        summary["source_order"] = metadata["SourceOrder"]
+    return summary
 
 
 def normalize_document(
@@ -406,6 +409,15 @@ def normalize_document(
         key=lambda path: (not path.name.isdigit(), int(path.name) if path.name.isdigit() else path.name),
     ) if edition_root.is_dir() else []
     summaries = [edition_summary(path, input_root) for path in edition_directories]
+    source_orders = [item.get("source_order") for item in summaries]
+    if any(order is not None for order in source_orders):
+        if any(not isinstance(order, int) or isinstance(order, bool) or order < 0 for order in source_orders):
+            raise ValueError(f"Invalid source edition order in {document_directory}")
+        if len(set(source_orders)) != len(source_orders):
+            raise ValueError(f"Duplicate source edition order in {document_directory}")
+        ordered = sorted(zip(source_orders, edition_directories, summaries), key=lambda item: item[0])
+        edition_directories = [item[1] for item in ordered]
+        summaries = [item[2] for item in ordered]
     available_languages = [language for language in LANGUAGES if any(language in item["available_languages"] for item in summaries)]
     normalized_metadata = clean_value(metadata)
     document = {

backend/search/minjust_opensearch.py

@@ -17,7 +17,7 @@ from typing import Iterator
 
 from search.catalog import authority_codes, legal_force_code, source_code
 
-APP_VERSION = "0.11.3"
+APP_VERSION = "0.12.2"
 LANGUAGES = {"ru", "ky"}
 DEFAULT_MAPPING = Path(__file__).with_name("minjust-fragments-index.json")
 
@@ -129,7 +129,12 @@ def bulk_pairs(
         editions = [path.name for path in edition_root.iterdir() if path.is_dir()] if edition_root.is_dir() else []
         if not editions:
             continue
-        current_edition = max(editions, key=lambda value: (not value.isdigit(), int(value) if value.isdigit() else value))
+        official_editions = document.get("editions") or []
+        current_edition = (
+            str(official_editions[-1]["source_code"])
+            if official_editions
+            else max(editions, key=lambda value: (not value.isdigit(), int(value) if value.isdigit() else value))
+        )
         for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")):
             edition_code, language = fragment_path.parts[-3:-1]
             values = read_json(fragment_path)

backend/search/review.html

@@ -176,7 +176,7 @@
   <dialog id="users-dialog" aria-labelledby="users-heading"><section class="review__page"><h2 id="users-heading">Управление пользователями</h2><form id="create-user-form"><label>Логин<input id="new-username" autocomplete="off" minlength="3" maxlength="64" required></label><label>Временный пароль<input id="temporary-password" type="password" autocomplete="new-password" minlength="12" maxlength="256" required></label><p>Пароль — от 12 символов. Новый пользователь сменит его при первом входе. Передайте временный пароль отдельно.</p><p id="users-error" role="alert" class="review__status review__status--error"></p><button class="review__button--primary" type="submit">Создать пользователя</button></form><ul id="user-list" class="review__user-list"></ul><button type="button" data-close="users-dialog">Закрыть</button></section></dialog>
   <dialog id="document-dialog"><button class="review__dialog-close" id="dialog-close" type="button">Закрыть</button><h2 id="dialog-heading">Документ</h2><div class="review__document-body" id="dialog-body"></div></dialog>
   <div class="review__tour-spotlight" id="tutorial-spotlight" hidden aria-hidden="true"></div><dialog class="review__tour" id="tutorial-dialog" aria-labelledby="tutorial-heading" aria-describedby="tutorial-text"><div class="review__tour-content"><p class="review__tour-kicker">Короткое знакомство</p><h2 id="tutorial-heading"></h2><p id="tutorial-text"></p><div class="review__tour-footer"><span class="review__tour-progress" id="tutorial-progress"></span><div class="review__tour-actions"><button id="tutorial-skip" type="button">Пропустить</button><button class="review__button--primary" id="tutorial-next" type="button">Далее</button></div></div></div></dialog>
-  <footer class="review__header"><div class="review__header-content">Акылдаш · Frontend v0.1.2 · Backend v0.11.3 · Внутренняя лаборатория релевантности</div></footer>
+  <footer class="review__header"><div class="review__header-content">Акылдаш · Frontend v0.1.2 · Backend v0.12.2 · Внутренняя лаборатория релевантности</div></footer>
   <script>
     document.body.classList.add('review__locked');
     const DRAFT_KEY = 'akyldash-search-review-draft';
@@ -193,7 +193,7 @@
     let account = null;
     const reviewRoute = location.pathname === '/review';
     $('review').hidden = !reviewRoute; $('search-page').hidden = reviewRoute; $('review-nav').classList.toggle('is-active', reviewRoute); $('search-nav').classList.toggle('is-active', !reviewRoute);
-    if (!reviewRoute) { document.title = 'Поиск по правовой базе · Акылдаш'; document.querySelector('.review__header-content').hidden = true; $('tutorial-open').hidden = true; document.querySelector('.review__skip').href = '#public-results'; document.querySelector('footer .review__header-content').textContent = 'Акылдаш · Frontend v0.1.2 · Backend v0.11.3'; }
+    if (!reviewRoute) { document.title = 'Поиск по правовой базе · Акылдаш'; document.querySelector('.review__header-content').hidden = true; $('tutorial-open').hidden = true; document.querySelector('.review__skip').href = '#public-results'; document.querySelector('footer .review__header-content').textContent = 'Акылдаш · Frontend v0.1.2 · Backend v0.12.2'; }
     async function apiFetch(path, options = {}) { const response = await fetch(path, {credentials: 'same-origin', ...options}); if (response.status === 401 && path !== '/auth/login') { document.body.classList.add('review__locked'); if (!$('login-dialog').open) $('login-dialog').showModal(); $('login-username').focus(); } return response; }
     const showLogin = () => { document.body.classList.add('review__locked'); if (!$('login-dialog').open) $('login-dialog').showModal(); $('login-username').focus(); };
     const enterWorkspace = (user) => { account = user; $('signed-in-user').textContent = `${user.username} · ${user.role}`; $('reviewer').value = user.username; $('profile-username').textContent = user.username; $('profile-role').textContent = user.role; $('profile-created').textContent = new Date(user.created_at * 1000).toLocaleDateString('ru-RU'); $('users-open').hidden = user.username.toLowerCase() !== 'admin'; document.body.classList.remove('review__locked'); if (!reviewRoute) initSearchPage(); else if (!hasSeenTour()) setTimeout(openTutorial, 350); };

backend/test_minjust_current.py

@@ -0,0 +1,405 @@
+import json
+import os
+import sqlite3
+import tempfile
+import unittest
+import urllib.error
+from email.message import Message
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from ingestion import minjust_current
+from ingestion.minjust_current import (
+    ApiResponse,
+    CurrentCbdClient,
+    SourceChanged,
+    SourceError,
+    stage_document,
+)
+
+
+def api_response(value):
+    body = json.dumps(value, ensure_ascii=False).encode()
+    import hashlib
+
+    return ApiResponse(value, body, hashlib.sha256(body).hexdigest())
+
+
+def document_record(alias="6", canonical=6, editions=None):
+    return api_response(
+        {
+            "documentCode": canonical,
+            "nameRus": "Закон",
+            "nameKyr": "Мыйзам",
+            "headerRu": "Заголовок",
+            "headerKg": "Баш сөз",
+            "number": 114,
+            "dateAdopted": "1998-08-04",
+            "classId": 1,
+            "refTypeId": 10,
+            "status": 1,
+            "editions": editions if editions is not None else [
+                {"id": 920492, "nameRus": "Редакция A", "nameKyr": "Редакция A", "textRusType": "docx", "textKyrType": "docx"},
+                {"id": 920292, "nameRus": "Редакция B", "nameKyr": "Редакция B", "textRusType": "docx", "textKyrType": "docx"},
+            ],
+        }
+    )
+
+
+class CurrentCbdClientTests(unittest.TestCase):
+    def test_rejects_legacy_roots_when_called_from_an_unrelated_working_directory(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            repository = base / "repository"
+            elsewhere = base / "elsewhere"
+            elsewhere.mkdir()
+            legacy_roots = [
+                repository / "data/minjust-cbd",
+                repository / "data/minjust-normalized",
+            ]
+            for legacy in legacy_roots:
+                legacy.mkdir(parents=True)
+            # Root CI uses umask 000, so make the trusted fixture's parent permissions explicit.
+            for directory in (repository, repository / "data", *legacy_roots):
+                directory.chmod(0o755)
+            original_cwd = Path.cwd()
+            client = Mock()
+            try:
+                os.chdir(elsewhere)
+                with patch.object(minjust_current, "REPOSITORY_ROOT", repository):
+                    for legacy in legacy_roots:
+                        with self.subTest(legacy=legacy), self.assertRaisesRegex(ValueError, "must not overlap"):
+                            stage_document(legacy / "staging", "6", client)
+                        self.assertFalse((legacy / "staging").exists())
+            finally:
+                os.chdir(original_cwd)
+            client.document.assert_not_called()
+
+    def test_requires_private_owned_staging_root(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "shared"
+            output.mkdir(mode=0o755)
+            output.chmod(0o755)
+
+            with self.assertRaisesRegex(ValueError, "private directory"):
+                stage_document(output, "6", Mock())
+
+    def test_resume_rejects_symlinked_checkpoints_and_document_paths(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            targets = [
+                ("request-contract.json", False),
+                ("current-api.sqlite3", False),
+                ("manifest.sqlite3", False),
+                (".work/documents/6", True),
+                (".work/documents/6/editions/920492/ru.html", False),
+                ("documents/6", True),
+            ]
+            client = Mock()
+            for index, (relative, directory) in enumerate(targets):
+                with self.subTest(relative=relative):
+                    output = base / f"generation-{index}"
+                    output.mkdir(mode=0o700)
+                    contract = output / "request-contract.json"
+                    if relative != "request-contract.json":
+                        contract.write_text(json.dumps(minjust_current.REQUEST_CONTRACT), encoding="utf-8")
+                    outside = base / f"outside-{index}"
+                    if directory:
+                        outside.mkdir()
+                    else:
+                        outside.write_text("preserve", encoding="utf-8")
+                    link = output / relative
+                    link.parent.mkdir(parents=True, exist_ok=True)
+                    try:
+                        link.symlink_to(outside, target_is_directory=directory)
+                    except (NotImplementedError, OSError):
+                        self.skipTest("symlinks are unavailable")
+
+                    with self.assertRaisesRegex(ValueError, "symbolic link"):
+                        stage_document(output, "6", client, resume=True)
+
+                    client.document.assert_not_called()
+                    if not directory:
+                        self.assertEqual(outside.read_text(encoding="utf-8"), "preserve")
+                    else:
+                        self.assertEqual(list(outside.iterdir()), [])
+
+    def test_respects_retry_after_and_retries_server_errors(self):
+        headers = Message()
+        headers["Retry-After"] = "7"
+        response = Mock()
+        response.read.return_value = b"{}"
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+        opener = Mock(side_effect=[
+            urllib.error.HTTPError("https://example.test", 429, "", headers, None),
+            urllib.error.HTTPError("https://example.test", 503, "", Message(), None),
+            response,
+        ])
+        delays = []
+
+        CurrentCbdClient(opener=opener, sleep=delays.append, clock=lambda: 0, retries=3).request_json("/GetEdition")
+
+        self.assertIn(7.0, delays)
+        self.assertIn(2, delays)
+        self.assertEqual(opener.call_count, 3)
+
+    def test_enforces_request_interval(self):
+        now = [0.0]
+        starts = []
+        response = Mock()
+        response.read.return_value = b"{}"
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+
+        def opener(_request, timeout):
+            starts.append(now[0])
+            return response
+
+        client = CurrentCbdClient(
+            opener=opener,
+            clock=lambda: now[0],
+            sleep=lambda delay: now.__setitem__(0, now[0] + delay),
+        )
+        client.request_json("/GetEdition")
+        client.request_json("/GetEdition")
+
+        self.assertGreaterEqual(starts[1] - starts[0], 1.0)
+
+    def test_rejects_oversized_response(self):
+        response = Mock()
+        response.read.return_value = b"x" * 11
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+        with self.assertRaisesRegex(SourceError, "exceeds 10 bytes"):
+            CurrentCbdClient(opener=lambda *_args, **_kwargs: response, max_response_bytes=10).request_json("/GetEdition")
+
+    def test_get_edition_uses_exact_query_and_checks_requested_id(self):
+        response = Mock()
+        response.read.return_value = json.dumps({"id": 920292, "contentRu": "text", "contentKg": None}).encode()
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+        requests = []
+
+        def opener(request, timeout):
+            requests.append(request)
+            return response
+
+        result = CurrentCbdClient(opener=opener, sleep=lambda _delay: None).edition(920292, "kg")
+
+        self.assertEqual(requests[0].full_url, "https://cbd.minjust.gov.kg/api/v1/GetEdition?editionId=920292&lang=kg&exact=true")
+        self.assertEqual(requests[0].method, "GET")
+        self.assertEqual(result.value["id"], 920292)
+
+    def test_rejects_response_id_that_does_not_match_exact_request(self):
+        response = Mock()
+        response.read.return_value = json.dumps({"id": 920492, "contentRu": "text", "contentKg": None}).encode()
+        response.__enter__ = Mock(return_value=response)
+        response.__exit__ = Mock(return_value=False)
+
+        with self.assertRaisesRegex(SourceChanged, "does not match requested ID"):
+            CurrentCbdClient(opener=lambda *_args, **_kwargs: response).edition(920292, "ru")
+
+    def test_stages_alias_mapping_language_quirk_and_official_edition_order(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            output = base / "generation"
+            old_raw = base / "legacy" / "documents" / "6" / "sentinel"
+            old_raw.parent.mkdir(parents=True)
+            old_raw.write_text("preserve", encoding="utf-8")
+
+            class Client:
+                def document(self, alias):
+                    self.alias = alias
+                    return document_record(alias=alias, canonical=6)
+
+                def edition(self, identifier, language):
+                    text = "<p>Russian</p>" if language == "ru" else "<p>Кыргызча</p>"
+                    name = "Редакция A" if identifier == 920492 else "Редакция B"
+                    return api_response({
+                        "id": identifier, "nameRus": name, "nameKyr": name,
+                        "contentRu": text, "contentKg": None,
+                    })
+
+            client = Client()
+            result = stage_document(output, "6", client)
+            raw = output / "documents" / "6"
+
+            self.assertEqual(result["state"], "text_complete_incomplete_attachments")
+            self.assertEqual((raw / "editions/920492/ru.html").read_text(), "<p>Russian</p>")
+            self.assertEqual((raw / "editions/920492/ky.html").read_text(), "<p>Кыргызча</p>")
+            first = json.loads((raw / "editions/920492/metadata.json").read_text())
+            second = json.loads((raw / "editions/920292/metadata.json").read_text())
+            self.assertEqual((first["SourceOrder"], second["SourceOrder"]), (0, 1))
+            language_meta = json.loads((raw / "editions/920292/source/ru.json").read_text())
+            self.assertEqual((language_meta["requested_edition_id"], language_meta["exact"]), (920292, True))
+            with sqlite3.connect(output / "current-api.sqlite3") as connection:
+                self.assertTrue(connection.execute("SELECT contract_sha256 FROM generation").fetchone()[0])
+            self.assertEqual(old_raw.read_text(), "preserve")
+            with sqlite3.connect(output / "current-api.sqlite3") as connection:
+                rows = connection.execute(
+                    "SELECT request_language, archive_language, state FROM languages ORDER BY request_language, edition_id"
+                ).fetchall()
+                self.assertEqual(rows, [("kg", "ky", "complete")] * 2 + [("ru", "ru", "complete")] * 2)
+
+    def test_resolves_official_alias_to_canonical_staging_directory(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+
+            class Client:
+                def document(self, alias):
+                    return document_record(alias=alias, canonical=205064, editions=[])
+
+            result = stage_document(output, "1-4", Client())
+
+            self.assertEqual(result["canonical_id"], 205064)
+            self.assertTrue((output / "documents/205064/metadata.json").is_file())
+            self.assertFalse((output / "documents/1-4").exists())
+
+    def test_rejects_non_exact_edition_response_id_without_publishing(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+
+            class Client:
+                def document(self, _alias):
+                    return document_record(editions=[{"id": 920292, "textRusType": "docx", "textKyrType": "docx"}])
+
+                def edition(self, _identifier, language):
+                    return api_response({"id": 920492, "contentRu": f"<{language} edition>", "contentKg": None})
+
+            with self.assertRaisesRegex(SourceChanged, "exact edition ID mismatch"):
+                stage_document(output, "6", Client())
+
+            self.assertFalse((output / "documents/6").exists())
+            with sqlite3.connect(output / "current-api.sqlite3") as connection:
+                self.assertEqual(connection.execute("SELECT state FROM languages").fetchone()[0], "error")
+
+    def test_rejects_exact_edition_metadata_contradiction(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+            client = CurrentCbdClient()
+            client.document = lambda _alias: document_record(editions=[{
+                "id": 920292, "nameRus": "04.08.1998", "nameKyr": "04.08.1998",
+                "textRusType": "docx", "textKyrType": "docx",
+            }])
+            client.request_json = lambda *_args, **_kwargs: api_response({
+                "id": 920292, "nameRus": "13.04.2017", "nameKyr": "13.04.2017",
+                "contentRu": "<wrong edition>", "contentKg": None,
+            })
+
+            with self.assertRaisesRegex(SourceChanged, "metadata mismatch"):
+                stage_document(output, "6", client)
+
+            self.assertFalse((output / "documents/6").exists())
+
+    def test_recovers_atomic_publish_before_manifest_checkpoint(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+
+            class Client:
+                def document(self, _alias):
+                    return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
+
+                def edition(self, identifier, language):
+                    return api_response({"id": identifier, "contentRu": f"<{language}>", "contentKg": None})
+
+            client = Client()
+            first = stage_document(output, "6", client)
+            with sqlite3.connect(output / "current-api.sqlite3") as connection:
+                connection.execute("UPDATE documents SET state='fetching' WHERE alias='6'")
+            with sqlite3.connect(output / "manifest.sqlite3") as connection:
+                connection.execute("DELETE FROM documents WHERE code=6")
+                connection.execute("DELETE FROM errors WHERE code=6")
+
+            resumed = stage_document(output, "6", client, resume=True)
+
+            self.assertEqual(resumed["source_sha256"], first["source_sha256"])
+            self.assertTrue(resumed["resumed"])
+            with sqlite3.connect(output / "manifest.sqlite3") as connection:
+                self.assertEqual(connection.execute("SELECT COUNT(*) FROM documents WHERE code=6").fetchone()[0], 1)
+                self.assertEqual(connection.execute("SELECT COUNT(*) FROM errors WHERE code=6").fetchone()[0], 1)
+
+    def test_retries_after_document_metadata_request_failed_before_checkpoint(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+
+            class Client:
+                failed = False
+
+                def document(self, _alias):
+                    if not self.failed:
+                        self.failed = True
+                        raise SourceError("temporary DNS failure")
+                    return document_record(editions=[])
+
+            client = Client()
+            with self.assertRaisesRegex(SourceError, "DNS"):
+                stage_document(output, "6", client)
+
+            result = stage_document(output, "6", client, resume=True)
+
+            self.assertEqual(result["canonical_id"], 6)
+            with sqlite3.connect(output / "current-api.sqlite3") as connection:
+                row = connection.execute(
+                    "SELECT canonical_id, response_sha256, expected_editions FROM documents WHERE alias='6'"
+                ).fetchone()
+                self.assertEqual(row, (6, document_record(editions=[]).sha256, 0))
+
+    def test_does_not_resume_generation_without_exact_request_contract(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "legacy-generation"
+            output.mkdir(mode=0o700)
+            sentinel = output / "old-response.json"
+            sentinel.write_text("preserve", encoding="utf-8")
+            client = Mock()
+
+            with self.assertRaisesRegex(SourceChanged, "no exact-edition request contract"):
+                stage_document(output, "6", client, resume=True)
+
+            client.document.assert_not_called()
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "preserve")
+
+    def test_incomplete_translation_resumes_without_rewriting_successful_language(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+
+            class Client:
+                failed = False
+
+                def document(self, _alias):
+                    return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
+
+                def edition(self, identifier, language):
+                    if language == "kg" and not self.failed:
+                        self.failed = True
+                        raise SourceError("simulated interrupted response")
+                    return api_response({"id": identifier, "contentRu": f"<{language}>", "contentKg": None})
+
+            client = Client()
+            first = stage_document(output, "6", client)
+            self.assertIn("incomplete", first)
+            self.assertFalse((output / "documents/6").exists())
+            second = stage_document(output, "6", client, resume=True)
+            self.assertEqual(second["state"], "text_complete_incomplete_attachments")
+            self.assertEqual((output / "documents/6/editions/920492/ru.html").read_text(), "<ru>")
+            self.assertEqual((output / "documents/6/editions/920492/ky.html").read_text(), "<kg>")
+            self.assertEqual(client.failed, True)
+
+    def test_refuses_budget_overrun_without_publishing_partial_document(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            output = Path(temporary) / "generation"
+
+            class Client:
+                def document(self, _alias):
+                    return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
+
+                def edition(self, identifier, _language):
+                    return api_response({"id": identifier, "contentRu": "123456", "contentKg": None})
+
+            with self.assertRaisesRegex(SourceError, "budget"):
+                stage_document(output, "6", Client(), max_document_bytes=5)
+            self.assertFalse((output / "documents/6").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/test_minjust_normalization.py

@@ -86,6 +86,47 @@ class MinjustNormalizationTest(unittest.TestCase):
             empty = json.loads((output / "documents/1/editions/40/edition.json").read_text(encoding="utf-8"))
             self.assertFalse(empty["quality"]["has_html"])
 
+    def test_preserves_official_edition_order_instead_of_sorting_ids(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            source = base / "source"
+            output = base / "normalized"
+            document = self.document(source, 6)
+            self.edition(document, 920492, {"ru": "<p>Первая редакция</p>"})
+            self.edition(document, 920292, {"ru": "<p>Последняя редакция</p>"})
+            for code, order in ((920492, 0), (920292, 1)):
+                metadata_path = document / "editions" / str(code) / "metadata.json"
+                metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+                metadata["SourceOrder"] = order
+                metadata_path.write_text(json.dumps(metadata, ensure_ascii=False), encoding="utf-8")
+
+            normalize_archive(source, output)
+            normalized = json.loads((output / "documents/6/document.json").read_text(encoding="utf-8"))
+
+            self.assertEqual([item["source_code"] for item in normalized["editions"]], ["920492", "920292"])
+            self.assertEqual([item["source_order"] for item in normalized["editions"]], [0, 1])
+
+    def test_rejects_partial_source_edition_order(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            source = base / "source"
+            output = base / "normalized"
+            document = self.document(source, 6)
+            self.edition(document, 10, {"ru": "<p>Один</p>"})
+            self.edition(document, 20, {"ru": "<p>Два</p>"})
+            metadata_path = document / "editions/10/metadata.json"
+            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+            metadata["SourceOrder"] = 0
+            metadata_path.write_text(json.dumps(metadata), encoding="utf-8")
+
+            with self.assertLogs("normalization.minjust_cbd", level="ERROR"):
+                result = normalize_archive(source, output)
+
+            self.assertEqual((result.failed, result.normalized), (1, 0))
+            with sqlite3.connect(output / "manifest.sqlite3") as connection:
+                error = connection.execute("SELECT error FROM documents WHERE code='6'").fetchone()[0]
+            self.assertIn("Invalid source edition order", error)
+
     def test_continues_after_bad_document_and_clears_repaired_error(self):
         with tempfile.TemporaryDirectory() as temporary:
             base = Path(temporary)

backend/test_minjust_opensearch.py

@@ -25,6 +25,50 @@ class MinjustOpenSearchTest(unittest.TestCase):
         self.assertEqual(legal_force_code({"ru": "Указ", "ky": "Жарлыгы"}), "subordinate")
         self.assertEqual(source_code("document_type", {"ru": "устав"}), "charter")
 
+    def test_export_marks_last_official_edition_current_even_when_ids_are_out_of_order(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            input_root = root / "normalized"
+            document_root = input_root / "documents/6"
+            document_root.mkdir(parents=True)
+            with closing(sqlite3.connect(input_root / "manifest.sqlite3")) as connection:
+                with connection:
+                    connection.execute("CREATE TABLE documents (code TEXT, state TEXT)")
+                    connection.execute("INSERT INTO documents VALUES ('6', 'success')")
+            (document_root / "document.json").write_text(
+                json.dumps({"schema_version": "1", "source_code": "6", "editions": [
+                    {"source_code": "920492", "source_order": 0},
+                    {"source_code": "920292", "source_order": 1},
+                ]}),
+                encoding="utf-8",
+            )
+            for edition in ("920292", "920492"):
+                text = f"Текст {edition}"
+                fragment_path = document_root / f"editions/{edition}/ru/fragments.json"
+                fragment_path.parent.mkdir(parents=True)
+                fragment_path.write_text(json.dumps([{
+                    "id": f"document:6:edition:{edition}:lang:ru:fragment:1",
+                    "document_code": "6",
+                    "edition_code": edition,
+                    "language": "ru",
+                    "position": 1,
+                    "type": "paragraph",
+                    "text": text,
+                    "text_sha256": hashlib.sha256(text.encode()).hexdigest(),
+                    "source_path": f"documents/6/editions/{edition}/ru.html",
+                    "source_sha256": "b" * 64,
+                }]), encoding="utf-8")
+
+            output = root / "bulk.ndjson"
+            self.assertEqual(export_bulk(input_root, output, "test-index"), (1, 2))
+            lines = [json.loads(line) for line in output.read_text(encoding="utf-8").splitlines()]
+            flags = {
+                lines[index]["index"]["_id"].split(":")[3]: lines[index + 1]["is_current_edition"]
+                for index in range(0, len(lines), 2)
+            }
+
+            self.assertEqual(flags, {"920292": True, "920492": False})
+
     def test_switches_alias_atomically_after_successful_load(self):
         with patch("search.minjust_opensearch.request_json", return_value={"acknowledged": True}) as request:
             switch_alias("http://127.0.0.1:9200/", "akyldash-fragments-v2", "akyldash-fragments-current")

docs/README.md

@@ -45,4 +45,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.2 · Frontend v0.1.2

docs/operations/CI_RUNNER.md

@@ -25,8 +25,13 @@ rootless Docker setup and user service use `.config/act-runner` as writable
 `.config/act-runner/.local/share` as `XDG_DATA_HOME`. This keeps Docker config
 and its systemd user unit in the existing `.config` tree without granting
 write access to the protected home.
-rootless Docker daemon
-and runner user are separate from production Docker; `codex` is not added to
+The runner systemd unit uses `ProtectHome=read-only` so `/run/user` remains
+visible; `ReadWritePaths` retains the existing `/run/user/$uid` writable
+carve-out for the runner's rootless runtime. `ProtectHome=yes` would hide
+`/run/user` and break the configured runtime and Docker socket paths. See the
+[systemd.exec documentation](https://www.freedesktop.org/software/systemd/man/255/systemd.exec.html#ProtectHome=).
+The rootless Docker daemon and runner user are separate from production Docker;
+`codex` is not added to
 sudo or Docker groups. Jobs do not receive the rootless Docker socket, the
 production Docker socket, production environment/secrets, or arbitrary host
 bind mounts. The runner account has no access to `/etc/akyldash` or the

docs/operations/CORPUS_HANDOFF.md

@@ -0,0 +1,182 @@
+# 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.
+
+## Current API and bounded code 6 recovery (2026-10-20)
+
+The current official site uses `POST /api/v1/GetDocuments` (page size 12),
+`GET /api/v1/GetDocument?DocumentCode=...`, and
+`GET /api/v1/GetEdition?editionId=...&lang=ru|kg`. A bounded first-page probe
+reported 210,639 records; this recovery does not enumerate the full list. The
+official alias `1-4` resolves to canonical code `205064`. Code `6` is present
+and has 201 editions. The edition order in `GetDocument` is authoritative;
+numeric IDs do not define currentness.
+
+For `lang=kg`, the API returns Kyrgyz HTML in `contentRu` and `contentKg=null`.
+The bounded adapter maps that response by the requested language to archive
+`ky`, preserves the response and checkpoints, limits requests to one per
+second, caps JSON responses at 16 MiB, and caps one document's text at 4 GiB.
+It writes only to a separate generation under `/tmp`; it neither resumes nor
+overwrites the existing ignored corpus.
+
+Without `exact=true`, a request for `920292` returned response field
+`id=920492` despite different text length/hash. Two exact probes for `920292`
+and `920492` returned matching IDs and names/dates (`04.08.1998` and
+`13.04.2017`) matching `GetDocument`; their RU text lengths/hashes were
+485,487 / `2b51c2ff4e9b91cbe2a7ed052a5a9ed9150bda993fc15c120ac948580926b718`
+and 1,139,908 / `e5f0fdbb47766e5a5dbb971183d765e4c1dcaf9c969a218a84206244ce9132e8`.
+The adapter now requires `exact=true`, returned-ID equality, and matching RU/KY
+names before a language checkpoint can complete. Its versioned request
+contract is part of both staging metadata and the SQLite resume identity;
+pre-contract generations cannot be resumed.
+
+The adapter does not independently validate attachments or embedded media.
+Even after the text stage completes, the raw manifest records this as an
+unresolved error. Do not merge this single document into the existing corpus,
+claim corpus completeness, or transfer it to production without a separate
+review and comparison. The prior no-exact pass at
+`/tmp/akyldash-current-code6.ZAMC37u4ex` is marked quarantined in
+`qualification.json`; its 240 text checkpoints and 304,493,520 bytes are
+retained as unverified, with no document directory published. Its preserved
+work-tree SHA-256 is
+`1e874c1d7fab1e00e7e45ff1b16422ea474e22ee13c901ac9a0612f47aa5d24a`. It must
+not be reused or normalized.
+
+The exact-only generation at `/tmp/akyldash-current-code6-exact.L4k3AvnOlP`
+contains 201 editions and 402 complete RU/KY text checkpoints (604,377,885
+raw bytes). The raw manifest passes `quick_check`; its stored source SHA-256
+matches the staged document tree at
+`81d8e1a53b2ebef61f08b56d04b3f0a80077d587a4fadae0152bb4b7f13055ea`. Separate
+normalization at `/tmp/akyldash-current-code6-exact.L4k3AvnOlP-normalized`
+passes `quick_check`, records normalizer `1.1.0`, and preserves first/last
+edition order `920292` (`04.08.1998`) to `920492` (`13.04.2017`). The normalized
+generation is about 2.05 GB. Neither temporary generation has been merged into
+the existing corpus. Attachments and embedded media remain unverified.
+
+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.

docs/operations/CURRENT_STATE.md

@@ -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,17 +55,16 @@ 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
-  account/review data, and OpenSearch `3.7.0` with `analysis-icu`.
+  Caddy `2.11.6` pinned by digest, Python Search API `0.12.2` in this branch,
+  SQLite account/review data, and OpenSearch `3.7.0` with `analysis-icu`.
 - The target publishes only Caddy on TCP `80`/`443`. Backend and Caddy share a
   proxy network; OpenSearch and backend share a separate internal network, and
   backend has no host port. Caddy certificate storage, OpenSearch, and
@@ -84,16 +89,37 @@ UFW rule details, and protected file metadata remain unverified.
   `status` helper, guarded rollback helper, and installer with exact sudoers
   command rules. None is installed on the VPS. CI contains no production
   deployment job; starting/stopping services remains owner-run.
+- Rollback preflight validates digest-pinned image references against archived
+  image IDs and local availability, then refuses before stopping services or
+  creating a backup because the current flow cannot safely switch immutable
+  digest references. This behavior is covered by synthetic local tests; it has
+  not been exercised on the VPS.
+- This branch adds an offline release-manifest validator and tests for the
+  repository allowlist intended for a future root-owned host integration. The
+  manifest format does not prove source provenance. No registry, host policy,
+  publish credential, or deployment integration is configured or installed.
 - A repository-scoped official `act_runner` `0.2.13` installation is
   configured in `infra/runner/`: rootless Docker under `akyldash-runner`, 15 GiB
   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. A read-only SSH recheck on 2026-10-21
+  confirmed the active `akyldash-act-runner.service` runs as
+  `akyldash-runner` with `ProtectHome=read-only`; `ReadWritePaths` includes the
+  runner runtime/config/log paths and `/run/user/999`. Its `ExecStart` uses
+  `/etc/akyldash-runner/config.yaml`. The live unit retains a writable carveout
+  for `/run/user/999`; the installer keeps that existing carveout and changes
+  `ProtectHome` from `yes` to `read-only` so the runtime path remains visible.
+  No VPS unit was changed. Main commit
+  `d2fb4b81` passed CI run 115 and smoke run 116. 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
@@ -109,10 +135,95 @@ UFW rule details, and protected file metadata remain unverified.
   reached healthy status for OpenSearch, backend, and Caddy, and returned HTTP
   200 for `/review` through Caddy. Its containers, networks, and volumes were
   removed afterward. This does not verify VPS startup, DNS, or public TLS.
-- The app has no migration framework or SMTP setup. Current app versions:
-  backend `0.11.3`, frontend `0.1.2`, Telegram bot `0.2.2` (the bot is not
+- The app has no migration framework or SMTP setup. Repository versions on this
+  branch are:
+  backend `0.12.2`, 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.
+
+## Current MinJust API staging (2026-10-20)
+
+- The public site's current API uses `POST /api/v1/GetDocuments` with page size
+  12, `GET /api/v1/GetDocument?DocumentCode=...`, and
+  `GET /api/v1/GetEdition?editionId=...&lang=ru|kg`. A bounded page-1 probe
+  reported 210,639 filtered records; no full listing was enumerated for this
+  recovery. Official alias `1-4` resolves to canonical document code `205064`.
+- Document code `6` has 201 official editions. The API's `lang=kg` response
+  places Kyrgyz HTML in `contentRu` while returning `contentKg=null`; staging
+  maps by requested language (`kg` to archive `ky`) and preserves the edition
+  order from the document response. Normalization and search export now retain
+  and use that order instead of sorting IDs.
+- Without `exact=true`, one probe for edition `920292` returned response field
+  `id=920492` despite a different text length/hash. Two bounded `exact=true`
+  probes returned matching IDs and names/dates from `GetDocument`: edition
+  `920292` / `04.08.1998` (485,487 RU characters, SHA-256
+  `2b51c2ff4e9b91cbe2a7ed052a5a9ed9150bda993fc15c120ac948580926b718`) and
+  edition `920492` / `13.04.2017` (1,139,908 RU characters, SHA-256
+  `e5f0fdbb47766e5a5dbb971183d765e4c1dcaf9c969a218a84206244ce9132e8`).
+  Every edition request now uses `exact=true`, requires returned ID equality,
+  and checks RU/KY names against `GetDocument` metadata.
+- `backend/ingestion/minjust_current.py` stages one document into a separate
+  generation, at no more than one request per second, with a 16 MiB response
+  limit, a 4 GiB text budget per document, contract-bound checkpoints, and
+  atomic publication. Staging requires a private 0700 directory owned by the
+  current user and rejects symlinks under checkpoints/document paths; legacy
+  corpus exclusions are anchored to the repository root. The corpus verifier
+  continues to validate unchanged normalized manifests from versions 1.0.0
+  and 1.1.0. The earlier non-exact pass is quarantined at
+  `/tmp/akyldash-current-code6.ZAMC37u4ex/qualification.json`: its 240 language
+  responses (304,493,520 bytes) remain unverified and are not reusable. The
+  retained work-tree SHA-256 is
+  `1e874c1d7fab1e00e7e45ff1b16422ea474e22ee13c901ac9a0612f47aa5d24a`.
+- Exact-only code-6 staging is complete at
+  `/tmp/akyldash-current-code6-exact.L4k3AvnOlP`: 201 editions and 402
+  complete language checkpoints, 604,377,885 raw bytes. The raw manifest
+  passes `quick_check`; its document checksum matches the staged directory at
+  `81d8e1a53b2ebef61f08b56d04b3f0a80077d587a4fadae0152bb4b7f13055ea`.
+  Separate normalization at
+  `/tmp/akyldash-current-code6-exact.L4k3AvnOlP-normalized` contains 201
+  editions in official response order, passes `quick_check`, and records
+  normalizer `1.1.0` (2,048,797,024 bytes). The prior `data/minjust-cbd` and
+  `data/minjust-normalized` trees and production were not used as input or
+  modified. The raw staging manifest still records attachments and embedded
+  media as unverified; this single document does not establish corpus
+  completeness or production readiness.
+
 ## Checks not completed
 
 - Docker Engine version from the server API, container and volume inventory,
@@ -149,17 +260,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.

docs/operations/DEPLOYMENT.md

@@ -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,53 @@ 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.
+
+## Release image contract (preparation only)
+
+`infra/deploy/validate_release.py` checks an offline JSON release manifest
+against an explicitly supplied repository allowlist. Schema version 1 requires
+exactly `schema_version`, `source_ref`, `commit`, and `images`; the source ref
+must be `refs/heads/main`, the commit must be a 40-character lowercase Git
+SHA-1, and the images must contain exactly `backend` and `opensearch` as
+`repository@sha256:<64 lowercase hex>` references. The policy has schema
+version 1 and one non-empty repository allowlist for each service. Unknown or
+duplicate JSON keys are rejected.
+
+The validator takes the manifest path and an explicit policy path:
+
+```bash
+python3 infra/deploy/validate_release.py release.json --policy /etc/akyldash/release-policy.json
+```
+
+The policy shape is `{"schema_version":1,"repositories":{"backend":["registry.invalid/akyldash/backend"],"opensearch":["registry.invalid/akyldash/opensearch"]}}`.
+These `registry.invalid` names are format examples only; this change does not
+create a policy file or approve a registry.
+
+The repository name in the policy must be approved for the registry. No real
+registry namespace or root-owned host policy is configured by this change; do
+not use the synthetic `registry.invalid` names in tests as production values.
+The host integration must load policy only from its fixed root-owned path.
+This validator checks format and allowlist only: a manifest can claim a main
+ref and commit, but is not proof of source, build provenance, signature, or
+trusted publication. CI still has no publish credentials, image publication,
+host helper, or deploy trigger.
+
+Compose merges an image override with the existing service `build` entries.
+The regression test renders fake digest references and confirms both entries
+remain; any future deploy command must use `docker compose up --no-build` so
+the host cannot silently build a different image. This does not make digest
+rollback available: the current rollback helper expects local tags. A separate
+host-side deploy/rollback design, owner-approved registry permissions and
+retention, and manual backup/health/rollback end-to-end validation are still
+required before publishing or enabling CD.
 
 ## CI
 
@@ -79,5 +120,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.

docs/operations/ROLLBACK.md

@@ -18,11 +18,18 @@ sudo -n /usr/local/sbin/akyldash-rollback \
 The helper verifies archive integrity, Compose image references, deployment
 file hashes, and the availability of both prior local images. It makes and
 verifies a second cold backup, restores only the prior backend/OpenSearch image
-tags, waits for Compose healthchecks, and checks HTTPS locally. It refuses to
-restore if the current Compose/Caddyfile differs from the saved release. This
-does not restore data or automatically undo a release with an unknown schema
-change. The application has no formal migration framework; compatibility must
-be checked for every change. If it is uncertain, stop and ask the owner.
+tags, waits for Compose healthchecks, and checks HTTPS locally. Digest-pinned
+references are validated against their recorded local image IDs, but rollback
+fails before stopping services or making a backup when either image reference
+uses `@sha256:`. The helper cannot safely switch immutable digest references
+with the current release-recording flow. Do not retag a digest manually; digest
+rollback needs a separately reviewed atomic release-reference mechanism.
+
+It refuses to restore if the current Compose/Caddyfile differs from the saved
+release. This does not restore data or automatically undo a release with an
+unknown schema change. The application has no formal migration framework;
+compatibility must be checked for every change. If it is uncertain, stop and ask
+the owner.
 
 The initial owner-run deployment must record its running backend image ID in
 `/opt/akyldash/deployed-image.txt` with mode `0600`; the rollback helper updates

docs/operations/project-status.md

@@ -5,7 +5,7 @@
 
 - Telegram-бот: `0.2.2`
 - Telegram-бот на Synology: `0.2.1`
-- Backend: `0.11.3`
+- Backend: `0.12.2`
 - Frontend: `0.1.2` — поиск, лаборатория релевантности, аккаунты
 
 ## Краткий итог
@@ -282,4 +282,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.2 · Frontend v0.1.2

docs/product/frontend-search-sps-plan.md

@@ -262,4 +262,4 @@ runtime-зависимостями frontend. Регистрация в стор
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.2 · Frontend v0.1.2

docs/product/minjust-document-normalization-agent-task.md

@@ -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`, иногда
   изображения;

docs/product/project-overview.md

@@ -44,4 +44,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.2 · Frontend v0.1.2

infra/deploy/compare_corpus_inventories.py

@@ -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())

infra/deploy/rollback.py

@@ -19,6 +19,12 @@ ENV_FILE = Path("/etc/akyldash/production.env")
 BACKUP_DIR = Path("/var/backups/akyldash")
 BACKUP_TOOL = Path("/usr/local/lib/akyldash/archive.py")
 APP_SERVICES = {"backend", "opensearch"}
+IMAGE_TAG_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]*")
+IMAGE_DIGEST_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]*@sha256:[0-9a-f]{64}")
+
+
+def is_digest_reference(reference: str) -> bool:
+    return IMAGE_DIGEST_RE.fullmatch(reference) is not None
 
 
 def image_restore_plan(manifest: dict, configured: dict[str, str]) -> dict[str, tuple[str, str]]:
@@ -34,14 +40,48 @@ def image_restore_plan(manifest: dict, configured: dict[str, str]) -> dict[str,
         image_id = image.get("image_id")
         if reference != configured.get(service):
             raise ValueError(f"Compose image reference changed for {service}; refusing rollback")
-        if not isinstance(reference, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:/-]*", reference):
-            raise ValueError(f"{service} image reference is not a local tag")
+        if not isinstance(reference, str) or not (
+            IMAGE_TAG_RE.fullmatch(reference) or is_digest_reference(reference)
+        ):
+            raise ValueError(f"invalid image reference for {service}")
         if not isinstance(image_id, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", image_id):
             raise ValueError(f"invalid previous image ID for {service}")
         plan[service] = (image_id, reference)
     return plan
 
 
+def verify_previous_images(plan: dict[str, tuple[str, str]]) -> None:
+    for image_id, reference in plan.values():
+        try:
+            actual = run(["docker", "image", "inspect", "--format", "{{.Id}}", image_id], capture=True)
+        except subprocess.CalledProcessError as error:
+            raise ValueError(f"previous image is unavailable: {image_id}") from error
+        if actual != image_id:
+            raise ValueError(f"previous image is unavailable: {image_id}")
+        if is_digest_reference(reference):
+            try:
+                referenced_id = run(
+                    ["docker", "image", "inspect", "--format", "{{.Id}}", reference], capture=True
+                )
+            except subprocess.CalledProcessError as error:
+                raise ValueError(f"previous digest reference is unavailable: {reference}") from error
+            if referenced_id != image_id:
+                raise ValueError(f"previous digest reference does not match its recorded image: {reference}")
+
+
+def require_tag_rollback(plan: dict[str, tuple[str, str]]) -> None:
+    if any(is_digest_reference(reference) for _image_id, reference in plan.values()):
+        raise ValueError(
+            "rollback of digest-pinned images is not supported safely; no services or image references were changed"
+        )
+
+
+def tag_previous_images(plan: dict[str, tuple[str, str]]) -> None:
+    require_tag_rollback(plan)
+    for image_id, reference in plan.values():
+        run(["docker", "image", "tag", image_id, reference])
+
+
 def run(command: list[str], *, capture: bool = False) -> str:
     result = subprocess.run(command, check=True, text=True, capture_output=capture)
     return result.stdout.strip() if capture else ""
@@ -96,10 +136,8 @@ def rollback(backup: Path, data_compatible: bool) -> None:
     rendered = json.loads(compose("config", "--format", "json", capture=True))
     configured = {service: rendered["services"][service]["image"] for service in APP_SERVICES}
     plan = image_restore_plan(manifest, configured)
-    for image_id, _reference in plan.values():
-        actual = run(["docker", "image", "inspect", "--format", "{{.Id}}", image_id], capture=True)
-        if actual != image_id:
-            raise ValueError(f"previous image is unavailable: {image_id}")
+    verify_previous_images(plan)
+    require_tag_rollback(plan)
 
     backup_name = f"akyldash-pre-rollback-{dt.datetime.now(dt.timezone.utc):%Y%m%dT%H%M%SZ}.tar.gz"
     backup_output = BACKUP_DIR / backup_name
@@ -123,8 +161,7 @@ def rollback(backup: Path, data_compatible: bool) -> None:
         compose("up", "--detach")
         raise
 
-    for image_id, reference in plan.values():
-        run(["docker", "image", "tag", image_id, reference])
+    tag_previous_images(plan)
     compose("up", "--detach", "--no-build", "--wait", "--wait-timeout", "300")
     run(
         [

infra/deploy/test_compare_corpus_inventories.py

@@ -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()

infra/deploy/test_release_manifest.py

@@ -0,0 +1,131 @@
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from validate_release import validate_manifest
+
+ROOT = Path(__file__).resolve().parents[2]
+REPOSITORIES = {
+    "backend": ["registry.invalid/akyldash/backend"],
+    "opensearch": ["registry.invalid/akyldash/opensearch"],
+}
+
+
+class ReleaseManifestTests(unittest.TestCase):
+    def setUp(self):
+        self.policy = {"schema_version": 1, "repositories": REPOSITORIES}
+        self.manifest = {
+            "schema_version": 1,
+            "source_ref": "refs/heads/main",
+            "commit": "a" * 40,
+            "images": {
+                "backend": "registry.invalid/akyldash/backend@sha256:" + "b" * 64,
+                "opensearch": "registry.invalid/akyldash/opensearch@sha256:" + "c" * 64,
+            },
+        }
+
+    def test_accepts_both_policy_pinned_images(self):
+        self.assertEqual(validate_manifest(self.manifest, self.policy), self.manifest)
+
+    def test_rejects_untrusted_ref(self):
+        self.manifest["source_ref"] = "refs/pull/12/head"
+        with self.assertRaisesRegex(ValueError, "source_ref"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_tag_instead_of_digest(self):
+        self.manifest["images"]["backend"] = "registry.invalid/akyldash/backend:latest"
+        with self.assertRaisesRegex(ValueError, "sha256"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_empty_image_reference(self):
+        self.manifest["images"]["backend"] = ""
+        with self.assertRaisesRegex(ValueError, "sha256"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_invalid_commit(self):
+        self.manifest["commit"] = "not-a-commit"
+        with self.assertRaisesRegex(ValueError, "commit"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_repository_outside_policy(self):
+        self.manifest["images"]["backend"] = "registry.invalid/other/backend@sha256:" + "b" * 64
+        with self.assertRaisesRegex(ValueError, "not allowed"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_missing_or_extra_images(self):
+        self.manifest["images"].pop("opensearch")
+        with self.assertRaisesRegex(ValueError, "exactly backend and opensearch"):
+            validate_manifest(self.manifest, self.policy)
+        self.manifest["images"]["opensearch"] = "registry.invalid/akyldash/opensearch@sha256:" + "c" * 64
+        self.manifest["images"]["extra"] = "registry.invalid/akyldash/extra@sha256:" + "d" * 64
+        with self.assertRaisesRegex(ValueError, "exactly backend and opensearch"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_duplicate_json_keys(self):
+        with tempfile.TemporaryDirectory() as directory:
+            manifest_path = Path(directory) / "manifest.json"
+            policy_path = Path(directory) / "policy.json"
+            manifest_path.write_text(
+                '{"schema_version":1,"source_ref":"refs/heads/main","commit":"' + "a" * 40
+                + '","images":{"backend":"first","backend":"second","opensearch":"third"}}',
+                encoding="utf-8",
+            )
+            policy_path.write_text(json.dumps(self.policy), encoding="utf-8")
+            command = [
+                "python3", str(ROOT / "infra/deploy/validate_release.py"), str(manifest_path),
+                "--policy", str(policy_path),
+            ]
+            result = subprocess.run(command, text=True, capture_output=True)
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("duplicate JSON key: backend", result.stderr)
+
+    def test_rejects_malformed_json(self):
+        with tempfile.TemporaryDirectory() as directory:
+            manifest_path = Path(directory) / "manifest.json"
+            policy_path = Path(directory) / "policy.json"
+            manifest_path.write_text('{"schema_version":', encoding="utf-8")
+            policy_path.write_text(json.dumps(self.policy), encoding="utf-8")
+            result = subprocess.run(
+                ["python3", str(ROOT / "infra/deploy/validate_release.py"), str(manifest_path), "--policy", str(policy_path)],
+                text=True,
+                capture_output=True,
+            )
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("cannot read valid JSON", result.stderr)
+
+    @unittest.skipUnless(shutil.which("docker"), "Docker CLI is unavailable")
+    def test_compose_override_renders_digest_images_and_preserves_build(self):
+        backend = "registry.invalid/akyldash/backend@sha256:" + "b" * 64
+        opensearch = "registry.invalid/akyldash/opensearch@sha256:" + "c" * 64
+        with tempfile.TemporaryDirectory() as directory:
+            temp = Path(directory)
+            secret = temp / "review_secret"
+            secret.touch()
+            override = temp / "release-override.json"
+            override.write_text(json.dumps({"services": {
+                "backend": {"image": backend},
+                "opensearch": {"image": opensearch},
+            }}), encoding="utf-8")
+            command = [
+                "docker", "compose", "--env-file", str(ROOT / "deploy/production/.env.example"),
+                "-f", str(ROOT / "deploy/production/compose.yaml"), "-f", str(override),
+                "config", "--format", "json",
+            ]
+            result = subprocess.run(
+                command, text=True, capture_output=True,
+                env={**os.environ, "REVIEW_SECRET_FILE": str(secret)},
+            )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        services = json.loads(result.stdout)["services"]
+        self.assertEqual(services["backend"]["image"], backend)
+        self.assertEqual(services["opensearch"]["image"], opensearch)
+        self.assertIn("build", services["backend"])
+        self.assertIn("build", services["opensearch"])
+
+
+if __name__ == "__main__":
+    unittest.main()

infra/deploy/test_rollback.py

@@ -1,6 +1,12 @@
+import json
+import subprocess
+import sys
 import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
 
-from rollback import image_restore_plan
+import rollback
+from rollback import image_restore_plan, require_tag_rollback, tag_previous_images, verify_previous_images
 
 
 class RollbackPlanTests(unittest.TestCase):
@@ -28,9 +34,96 @@ class RollbackPlanTests(unittest.TestCase):
     def test_refuses_digest_reference_that_cannot_be_retagged(self):
         self.manifest["images"]["backend"]["configured"] = "registry/backend@sha256:" + "c" * 64
         self.configured["backend"] = self.manifest["images"]["backend"]["configured"]
-        with self.assertRaisesRegex(ValueError, "local tag"):
+        plan = image_restore_plan(self.manifest, self.configured)
+        self.assertEqual(plan["backend"], ("sha256:" + "a" * 64, self.configured["backend"]))
+        with self.assertRaisesRegex(ValueError, "not supported safely"):
+            require_tag_rollback(plan)
+
+    def test_refuses_malformed_digest_reference(self):
+        self.manifest["images"]["backend"]["configured"] = "registry/backend@sha256:xyz"
+        self.configured["backend"] = self.manifest["images"]["backend"]["configured"]
+        with self.assertRaisesRegex(ValueError, "invalid image reference"):
             image_restore_plan(self.manifest, self.configured)
 
+    def test_refuses_invalid_recorded_image_id(self):
+        self.manifest["images"]["backend"]["image_id"] = "sha256:not-a-digest"
+        with self.assertRaisesRegex(ValueError, "invalid previous image ID"):
+            image_restore_plan(self.manifest, self.configured)
+
+    def test_digest_reference_must_resolve_to_recorded_image_id(self):
+        digest = "registry/backend@sha256:" + "c" * 64
+        self.manifest["images"]["backend"]["configured"] = digest
+        self.configured["backend"] = digest
+        plan = image_restore_plan(self.manifest, self.configured)
+        with patch("rollback.run", side_effect=["sha256:" + "a" * 64, "sha256:" + "a" * 64,
+                                                 "sha256:" + "b" * 64]):
+            verify_previous_images(plan)
+        with patch("rollback.run", side_effect=["sha256:" + "a" * 64, "sha256:" + "d" * 64]):
+            with self.assertRaisesRegex(ValueError, "does not match its recorded image"):
+                verify_previous_images(plan)
+
+    def test_missing_previous_image_fails_closed(self):
+        plan = image_restore_plan(self.manifest, self.configured)
+        with patch("rollback.run", side_effect=subprocess.CalledProcessError(1, "docker")):
+            with self.assertRaisesRegex(ValueError, "previous image is unavailable"):
+                verify_previous_images(plan)
+
+    def test_digest_reference_is_never_tagged(self):
+        digest = "registry/backend@sha256:" + "c" * 64
+        plan = {"backend": ("sha256:" + "a" * 64, digest)}
+        with patch("rollback.run") as run_mock:
+            with self.assertRaisesRegex(ValueError, "not supported safely"):
+                tag_previous_images(plan)
+        run_mock.assert_not_called()
+
+    def test_digest_rollback_fails_before_stopping_or_changing_references(self):
+        backend_id = "sha256:" + "a" * 64
+        opensearch_id = "sha256:" + "b" * 64
+        digest = "registry/backend@sha256:" + "c" * 64
+        manifest = {
+            "images": {
+                "backend": {"configured": digest, "image_id": backend_id},
+                "opensearch": self.manifest["images"]["opensearch"],
+            },
+            "deployment_files": {
+                "deploy/production/compose.yaml": "compose-hash",
+                "deploy/production/Caddyfile": "caddy-hash",
+            },
+        }
+        rendered = {
+            "services": {
+                "backend": {"image": digest},
+                "opensearch": {"image": self.configured["opensearch"]},
+            }
+        }
+        fake_archive = SimpleNamespace(
+            digest=lambda path: "compose-hash" if path.name == "compose.yaml" else "caddy-hash",
+            verify_archive=lambda _backup: manifest,
+        )
+        backup = rollback.BACKUP_DIR / "synthetic.tar.gz"
+        with (
+            patch("rollback.Path.is_file", return_value=True),
+            patch("rollback.Path.is_symlink", return_value=False),
+            patch("rollback.Path.stat", return_value=SimpleNamespace(st_uid=0, st_mode=0o100600)),
+            patch("rollback.compose", return_value=json.dumps(rendered)) as compose_mock,
+            patch("rollback.run", side_effect=[backend_id, backend_id, opensearch_id]) as run_mock,
+            patch.dict(sys.modules, {"archive": fake_archive}),
+        ):
+            with self.assertRaisesRegex(ValueError, "not supported safely"):
+                rollback.rollback(backup, data_compatible=True)
+        compose_mock.assert_called_once_with("config", "--format", "json", capture=True)
+        self.assertEqual(run_mock.call_count, 3)
+        self.assertTrue(all(call.args[0][:3] == ["docker", "image", "inspect"]
+                            for call in run_mock.call_args_list))
+
+    def test_tag_rollback_keeps_existing_image_tag_operation(self):
+        plan = image_restore_plan(self.manifest, self.configured)
+        with patch("rollback.run") as run_mock:
+            tag_previous_images(plan)
+        self.assertEqual(run_mock.call_count, 2)
+        run_mock.assert_any_call(["docker", "image", "tag", "sha256:" + "a" * 64,
+                                  "akyldash/backend:local"])
+
 
 if __name__ == "__main__":
     unittest.main()

infra/deploy/test_verify_corpus.py

@@ -0,0 +1,349 @@
+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,
+    NORMALIZER_VERSION,
+    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, normalizer_version=NORMALIZER_VERSION):
+        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": normalizer_version},
+            "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', ?, 'success')", (source_sha, normalizer_version)
+        )
+        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_verifies_legacy_normalizer_1_0_0_manifest(self):
+        normalized, _ = self._nonempty_corpus(normalizer_version="1.0.0")
+        result = verify_corpus(self.raw, normalized, self.base / "artifacts")
+        self.assertEqual(result["status"], "complete", result["finding_examples"])
+        self.assertNotIn(
+            "normalizer_version_mismatch",
+            {item["kind"] for item in result["finding_examples"]},
+        )
+
+    def test_rejects_unsupported_normalizer_version(self):
+        normalized, _ = self._nonempty_corpus(normalizer_version="9.9.9")
+        result = verify_corpus(self.raw, normalized, self.base / "artifacts")
+        self.assertEqual(result["status"], "failed")
+        self.assertIn(
+            "normalizer_version_mismatch",
+            {item["kind"] for item in result["finding_examples"]},
+        )
+
+    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()

infra/deploy/validate_release.py

@@ -0,0 +1,106 @@
+#!/usr/bin/env python3
+"""Validate the shape and repository allowlist of a prepared release manifest."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+
+SERVICES = {"backend", "opensearch"}
+COMMIT_RE = re.compile(r"[0-9a-f]{40}\Z")
+REPOSITORY_RE = re.compile(
+    r"[a-z0-9]+(?:[._-][a-z0-9]+)*(?::[0-9]+)?"
+    r"(?:/[a-z0-9]+(?:[._-][a-z0-9]+)*)*\Z"
+)
+IMAGE_RE = re.compile(r"(.+)@sha256:([0-9a-f]{64})\Z")
+
+
+def unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    result: dict[str, object] = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError(f"duplicate JSON key: {key}")
+        result[key] = value
+    return result
+
+
+def read_json(path: Path) -> object:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=unique_object)
+    except (OSError, UnicodeError, json.JSONDecodeError) as error:
+        raise ValueError(f"cannot read valid JSON from {path}: {error}") from error
+
+
+def validate_policy(policy: object) -> dict[str, list[str]]:
+    if not isinstance(policy, dict) or set(policy) != {"schema_version", "repositories"}:
+        raise ValueError("policy must contain exactly schema_version and repositories")
+    if type(policy["schema_version"]) is not int or policy["schema_version"] != 1:
+        raise ValueError("unsupported policy schema_version")
+    repositories = policy["repositories"]
+    if not isinstance(repositories, dict) or set(repositories) != SERVICES:
+        raise ValueError("policy repositories must contain exactly backend and opensearch")
+    result: dict[str, list[str]] = {}
+    for service in sorted(SERVICES):
+        allowed = repositories[service]
+        if not isinstance(allowed, list) or not allowed:
+            raise ValueError(f"policy repository allowlist for {service} must be a non-empty list")
+        if any(not isinstance(repository, str) or not REPOSITORY_RE.fullmatch(repository) for repository in allowed):
+            raise ValueError(f"policy contains an invalid repository for {service}")
+        if len(set(allowed)) != len(allowed):
+            raise ValueError(f"policy repository allowlist for {service} contains duplicates")
+        result[service] = allowed
+    return result
+
+
+def validate_manifest(manifest: object, policy: object) -> dict[str, object]:
+    allowed_repositories = validate_policy(policy)
+    if not isinstance(manifest, dict) or set(manifest) != {
+        "schema_version", "source_ref", "commit", "images"
+    }:
+        raise ValueError("manifest must contain exactly schema_version, source_ref, commit, and images")
+    if type(manifest["schema_version"]) is not int or manifest["schema_version"] != 1:
+        raise ValueError("unsupported manifest schema_version")
+    if manifest["source_ref"] != "refs/heads/main":
+        raise ValueError("manifest source_ref must be refs/heads/main")
+    commit = manifest["commit"]
+    if not isinstance(commit, str) or not COMMIT_RE.fullmatch(commit):
+        raise ValueError("manifest commit must be a 40-character lowercase Git SHA-1")
+    images = manifest["images"]
+    if not isinstance(images, dict) or set(images) != SERVICES:
+        raise ValueError("manifest images must contain exactly backend and opensearch")
+    for service in sorted(SERVICES):
+        image = images[service]
+        match = IMAGE_RE.fullmatch(image) if isinstance(image, str) else None
+        if not match:
+            raise ValueError(f"{service} image must use repository@sha256:<64 lowercase hex>")
+        repository = match.group(1)
+        if not REPOSITORY_RE.fullmatch(repository):
+            raise ValueError(f"{service} image has an invalid OCI repository")
+        if repository not in allowed_repositories[service]:
+            raise ValueError(f"{service} image repository is not allowed by policy")
+    return manifest
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("manifest", type=Path)
+    parser.add_argument("--policy", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        manifest = validate_manifest(read_json(args.manifest), read_json(args.policy))
+    except ValueError as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+    images = manifest["images"]
+    print(
+        "PASS release manifest "
+        f"commit={manifest['commit']} backend={images['backend']} opensearch={images['opensearch']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

infra/runner/install-runner.sh

@@ -179,7 +179,7 @@ TasksMax=256
 NoNewPrivileges=yes
 PrivateTmp=yes
 ProtectSystem=strict
-ProtectHome=yes
+ProtectHome=read-only
 ProtectKernelTunables=yes
 ProtectKernelModules=yes
 ProtectKernelLogs=yes

infra/runner/test-runner-config.sh

@@ -23,6 +23,10 @@ grep -Fq 'test -w "$HOME" && test -w "$XDG_CONFIG_HOME" && test -w "$XDG_DATA_HO
 grep -Fq 'rootless_env=(HOME="$runner_runtime_home" XDG_CONFIG_HOME="$runner_xdg_config_home" XDG_DATA_HOME="$runner_xdg_data_home"' "$installer"
 grep -Fq 'Environment=XDG_CONFIG_HOME=$runner_xdg_config_home' "$installer"
 grep -Fq 'Environment=XDG_DATA_HOME=$runner_xdg_data_home' "$installer"
+grep -Fq 'Environment=XDG_RUNTIME_DIR=/run/user/$runner_uid' "$installer"
+grep -Fq 'Environment=DOCKER_HOST=unix:///run/user/$runner_uid/docker.sock' "$installer"
+grep -Fq 'ProtectHome=read-only' "$installer"
+assert_absent '^ProtectHome=yes$' "$installer"
 grep -Fq 'runner_home/.config/systemd/user/docker.service.d/environment.conf' "$installer"
 grep -Fq 'Environment=HOME=$runner_runtime_home' "$installer"
 grep -Fq 'systemctl --user restart docker.service' "$installer"
@@ -42,6 +46,7 @@ if grep -Fq 'env HOME="$runner_home"' "$installer"; then
 fi
 grep -Fq 'die '\''stop akyldash-act-runner.service after all jobs finish before rerunning this installer' "$installer"
 grep -Fq 'ReadWritePaths=$runner_runtime_home $runner_xdg_config_home /var/log/akyldash-runner /run/user/$runner_uid' "$installer"
+assert_absent 'ReadWritePaths=.*(/var/run/docker.sock|/run/docker.sock)' "$installer"
 grep -Fq 'os.O_NOFOLLOW | os.O_CLOEXEC' infra/runner/secure-runner-image.py
 grep -Fq 'os.posix_fallocate(image_fd, 0, int(bytes_text))' infra/runner/secure-runner-image.py
 grep -Fq 'pass_fds=(image_fd,)' infra/runner/secure-runner-image.py