ex-USSR Kinship Research

ex-USSR Kinship Research

An evidence-first genealogy skill for AI agents (Codex, Claude Code and other Agent Skills runtimes). It turns an agent into a careful family-history researcher for the former USSR, the Soviet Union and the Russian Empire. The agent reads church books and civil records, military, deportation and repression files, and unindexed archive scans. It keeps every find traceable, separates records from interpretation and says plainly what is proven and what is not.

Coverage includes all 15 former Soviet republics: Russia, Ukraine, Belarus, Moldova, Estonia, Latvia, Lithuania, Georgia, Armenia, Azerbaijan, Kazakhstan, Kyrgyzstan, Tajikistan, Turkmenistan and Uzbekistan. Archives are chosen by the event's historical jurisdiction and current record custodian; a regional example does not replace local research.

The skill's instructions are written in Russian, but the agent talks to you in your language: English, Ukrainian, Russian, Polish, German and so on. It quotes sources in their original spelling with a translation, and builds search queries in the languages of the records.

---

What it does

• Case intake and scope. One research question, known facts, who may be alive, what is off-limits.
• Reproducible registers. Persons, artifacts, claims, relationships and a full search log (CSV), plus a final report template.
• Historical jurisdiction model. It resolves every event to the state, province, district, parish or civil registry as of the event date, not today's map.
• Name-variant matrices. Pre-reform Russian, Ukrainian, Polish, Romanian, German and Yiddish forms, patronymics, maiden names and transliterations.
• Source strategy. Originals first, then official catalogs, indexes, published document collections, and only then trees and forums as leads.
• Evidence protocol. Claims get explainable statuses, not percentages: hypothesis, single-source, corroborated, strongly-corroborated, contradicted, unresolved. Copies and mirrors of one record share an independence_group, so they never count as independent proof.
• Identity resolution. Namesakes stay separate person_ids until distinguishing evidence appears: house number, parents, godparents, exact birth date or military unit.
• Audit script. It checks referential integrity and flags weak corroboration before you write conclusions. An independence group counts only when a cited artifact belongs to it, so corroboration cannot be declared into existence. It also checks the final report: every cited ID must exist in the registers and every status must match them. A summary shows what is still uncorroborated or marked illegible.
• Final report. Every relationship is laid out edge by edge, with contradictions, failed searches and concrete next steps.

Release history: changelog.

Install

Clone the repository into your agent's skills folder:

# Codex
git clone https://github.com/DenisSergeevitch/ex-ussr-kinship-research ~/.codex/skills/ex-ussr-kinship-research

# Claude Code (personal skills)
git clone https://github.com/DenisSergeevitch/ex-ussr-kinship-research ~/.claude/skills/ex-ussr-kinship-research

For a single project, put the folder in .claude/skills/ inside the repository. The scripts need only Python 3 and the standard library; build_pdf.py also needs a Chromium-based browser.

Use

Describe the family question. The agent works from available information and continues while questions or access requests await a reply. For example:

• "My great-grandfather was exiled to Kazakhstan in the 1930s. Help me find his parents in church records."
• "Here are scans of a marriage certificate and a soldier card. Check whether they are the same person and build the evidence chain."
• "Find the parish and the right FamilySearch film for a village in Podolia in the 1890s and read the births for the surname X."

Keep private case data outside the skill's Git checkout. From the skill folder:

python scripts/init_case.py --case-dir ~/kinship-cases/my-case --research-question "Who were Maria's parents?" --subject-label "Maria"
python scripts/hash_artifact.py ~/kinship-cases/my-case/scans/marriage-1902.jpg   # SHA-256 for every saved artifact
python scripts/audit_evidence.py --case-dir ~/kinship-cases/my-case             # check registers before the final report
python scripts/build_pdf.py --html ~/kinship-cases/my-case/pdf-source/report.html --out ~/kinship-cases/my-case/report.pdf

init_case.py refuses to create a case inside the skill folder, including with --force. Git ignores common legacy case-folder names as an additional safeguard; check staged files before contributing, especially if moving an existing case.

build_pdf.py needs Chrome, Chromium or Edge installed (or CHROME_PATH). With --png-dir it also renders pages to PNG for review if pypdfium2 is available.

Use static HTML with local or embedded images. Active HTML and network image sources are rejected; source links remain clickable. A failed render preserves any existing output PDF.

Repository layout

SKILL.md                         main instructions (workflow, hard rules, language policy)
references/
  source-strategy.md             source hierarchy, query patterns, catalogs, archive requests
  evidence-protocol.md           artifact assessment, statuses, identity resolution, conflicts
  ex-ussr-locale-guide.md        jurisdictions, name variants, calendars, record types by period
  browser-and-captcha.md         browsing, sign-in, CAPTCHA hand-off, rate limits
  privacy-and-safety.md          living people, minors, DNA, sensitive findings
  seed-sources.md                curated starting resources with reliability classes
  field-lessons.md               practical lessons from a real case
  changelog.md                   release history
  photo-search.md                finding, identifying, dating and storing photos
  habsburg-galicia-bukovina.md   Galicia and Bukovina under the Habsburgs and after
assets/                          case intake, CSV registers, final report template,
                                 photos README template, readable PDF template (report-pdf.template.html)
scripts/                         init_case.py, hash_artifact.py, audit_evidence.py, build_pdf.py
evals/discovery-and-logic.yaml   trigger and reasoning test cases
tests/                           unit tests (python3 -m unittest discover -s tests)

Responsible use

• Family history, not surveillance. The skill refuses to look up current addresses, phones or accounts of living people, and it minimizes their data in reports.
• No bypassing. No CAPTCHA solving, no paywall, geo-block or rate-limit evasion, no leaked databases. Sign-ins are done by the user.
• No uploads of your documents or DNA data to third-party services without your explicit consent.
• Sensitive findings (deportation, repression, adoption, DNA surprises) are reported neutrally and only as far as the question requires.
• Archive access rules change. Always check the current official procedure before sending a request; the skill only drafts requests, it does not send them.

references/changelog.md

Release history

1.5.0 - 2026-10-25

• Incorporate PR #17's gravestone-to-military-family route and Soviet civilian employment, awards and evacuation research, with regional examples and separate identity and relationship conclusions.
• Consolidate access and collection-coverage guidance in existing references; distinguish digitization from availability, contributor observations from verified policies, and dependent copies from independent evidence.
• Add official archive starting points for the Caucasus and Central Asia, make coverage of all 15 former Soviet republics explicit, and use «страны Балтии» in Russian documentation.

1.4.0 - 2026-10-25

• Autonomous, resumable research continues while questions, logins or access requests await a reply. Intake asks only what materials are already available and reuses supplied information.
• Routine public historical searches, permitted evidence downloads and local processing need no repeated per-file approval. Sources and query variants are chosen for the question rather than an exhaustive checklist.
• Reports and templates adapt to available evidence. Partial findings remain useful; PDF is optional, and ordinary English words and archive identifiers no longer block PDF generation.
• Evidence audit normalizes independence-group labels, checks direct evidence and provenance on the same artifact, and flags contradictory evidence and invalid person references. Existing research cases are reused.
• Private cases stay outside the skill checkout; source links and FamilySearch source categories are clarified. Release history lives here instead of expanding the README.
• Available skill updates are offered to the user without pausing research; no custom updater is required.

1.3.0

General rules for a complete first pass, photo search and a readable PDF for the family. They were prompted by a real case (a WWII officer), where a family memorial page with a certified service-record extract and photos turned up only after the report was "done", but they apply to any line: peasant, urban, repressed, emigrant or Jewish families, and women's lines.

• Ask the family first. Before searching, the agent asks once whether relatives have researched this person before, which trees and pages the family keeps (FamilySearch, Geni, MyHeritage, VGD forum threads, and memorial bases that fit the case: Immortal Regiment, Bessmertny Barak, Posledny Adres, Yad Vashem testimonies) and who holds documents, letters, albums and family records. The answer goes to prior_family_research in the intake.
• Coverage checklists. references/source-strategy.md section 12 has checklists for church books, revision lists and censuses, civil registration after 1918, repression and deportation, evacuation, POWs and Ostarbeiter, emigration, Jewish communities and military service. Every item must be closed in search-log.csv as found, not-found, blocked or skipped before the report. A rich first database no longer ends the search.
• Name variants in every database. е/ё, -ев/-ов, female forms, maiden and married surnames, Polish, Yiddish, German and Romanian spellings, church and everyday first names.
• Site-internal search for family and memorial pages. An empty web search does not cover a site's internal database; the agent uses the site's own search when available.
• Grade materials, not websites. A scan of an original on a family page is an image-of-original artifact with its own provenance; the page's text is family testimony.
• Photos in the main pass. references/photo-search.md: where to look (documents, albums, family pages, school and workplace groups, gravestones, files that may contain photos), identification by caption and provenance only (no face recognition), dating by studio stamps and mounts or by uniform as a hypothesis, maximum available resolution, not-found limited to the sources actually checked, a photos/ folder with a README (assets/photos-readme.template.md). init_case.py now creates photos/.
• Readable PDF for the family. assets/report-pdf.template.html and scripts/build_pdf.py: a clean A4 layout with plain-language reliability labels instead of internal statuses. Required sections: summary, the person and their family, a "how we are related" chain with a label on every parent-child link, timeline, documents, contradictions, where we searched without result, next steps and sources. Optional blocks: places and jurisdictions, military service with awards, repression and rehabilitation, migration and emigration, photos and document scans. The script refuses to build if internal IDs, English statuses or unfilled placeholders remain in the visible text, then prints via headless Chrome, Chromium or Edge.
• Military database notes (field lessons): award sheets have higher commanders' resolutions on the reverse side; 1944-1957 long-service awards are distinguished from combat awards by the decree title.

1.2.0

The seed catalog previously leaned heavily on Russia; this release expands coverage of the other claimed jurisdictions; archive coverage and current access still need checking for each case:

• Central imperial/Soviet archives of Russia: RGIA (St. Petersburg), RGADA, RGVIA, RGAVMF, GARF, plus digitized provincial address-calendars (Presidential Library).
• Ukraine and Belarus: CDIAL (Lviv), CDIAK (Kyiv), Ukraine's national archives portal, and the National Historical Archive of Belarus (Minsk).
• Moldova: Bessarabian metric books 1820-1924 at the National Archives, with online certificate requests (the domain changed to arhiva.gov.md).
• The Baltics: Latvia's Raduraksti (church books, revision lists, 1935 census), Lithuania's ePaveldas, Estonia's Saaga.
• Poland and the Habsburg layer: Szukajwarchiwach, Geneteka (volunteer index covering Polish, Lithuanian, Belarusian and western-Ukrainian parishes), AGAD (historical eastern-territory collections), Matricula Online, Mapire (Habsburg cadastral and military maps), Kriegsarchiv and successor-state archives (military personnel records; custody and survival depend on period and origin).
• Jewish families of the region: JewishGen, JRI-Poland, Yad Vashem, Routes to Roots Foundation (which archive holds which town's surviving records).

New reference references/habsburg-galicia-bukovina.md: the jurisdiction chain 1772-1945, Austrian-era metric books and consistorial copies, calendar verification by confession, period, and individual register (a 12- or 13-day discrepancy is a hypothesis to investigate), Habsburg conscription records, Galician land cadastres (Josephine and Franciscan records - household and landholding evidence that helps disambiguate namesakes), and a storage map for each record type. New eval cases cover the Galician jurisdiction chain and the 13-day date conflict.

1.1.0

Version 1.1.0 adds references/field-lessons.md, distilled from a real multi-week research case. Personal data from that case is not included.

• Unindexed scans. Most ex-USSR church books on FamilySearch are images only. The file shows how to find the right film (including archive-reference-to-film mapping), map it, skim the key column and zoom on candidates. It also covers what to log.
• Reading cues that link records.
  • An exact birth date in a marriage register can match a baptism day for day.
  • House numbers persist for decades.
  • Stillbirths are listed among births.
  • Patronymic formulas: Ukrainian possessive forms, Romanian fiul lui / n., Church Slavonic, German.
  • Clerk name swaps, such as Grigorie and Gheorghe.
• Marginal annotations. Later Soviet-era notes on old registers show when a certificate was issued, which is evidence that the person or their family was alive and requesting documents.
• Repression and deportation records.
  • Published document collections and their typical contents.
  • How rehabilitation books group people.
  • How to read a "basis" column such as a 25-year sentence.
  • Why a family is linked to a deportation operation, not to the local incident used to justify it.
  • How to draft an archive request.
• Military databases. WWII loss records (survivors are absent), "surname nests" built from birthplace clusters, and which portals put up CAPTCHAs. Austro-Hungarian WWI casualty lists help for Galicia and Bukovina.
• Places that no longer exist. Liquidated hamlets, redrawn districts, and the difference between "born in" and "grew up in". Also archives in occupied or war-damaged areas.
• Family materials as sources. Passports, oral-history recordings with timestamps, and the user's own notes, all tagged by source type.
• Browser technique. Deep-zoom tile crops, composite column views for fast skimming, keeping long jobs in their own tab, polite crawling and retries.
• Knowing when to stop. When public sources are exhausted, the agent says so once, prepares questions for living relatives and request drafts for archives, and does not invent or loop.

Also in 1.1.0:

• The agent replies in the user's language; the intake template defaults to requested_output_language: auto.
• The skill description is bilingual, so it triggers for English-speaking users too.
• CAPTCHAs are handed to a human. The agent never solves or bypasses them, and the user signs in to sites themselves.
• More seed sources, each with a reliability class and role.
• New eval cases: language matching, unindexed scans, house-number identity, birthplace conflicts, exhausted sources, repression lists.

scripts/init_case.py

#!/usr/bin/env python3
"""Create a portable case directory from the bundled assets."""

from __future__ import annotations

import argparse
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path

ASSET_FILES = {
    "case-intake.template.yaml": "case-intake.yaml",
    "person-register.csv": "person-register.csv",
    "artifact-register.csv": "artifact-register.csv",
    "claims-register.csv": "claims-register.csv",
    "relationship-register.csv": "relationship-register.csv",
    "search-log.csv": "search-log.csv",
    "final-report.template.md": "final-report.md",
}
PHOTOS_README = "photos-readme.template.md"


def quote_yaml(value: str) -> str:
    return '"' + value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", " ") + '"'


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Создаёт структуру дела для исследования родственных связей."
    )
    parser.add_argument("--case-dir", required=True, help="Директория нового дела")
    parser.add_argument("--research-question", required=True, help="Главный исследовательский вопрос")
    parser.add_argument("--subject-label", default="", help="Краткая метка основного человека")
    parser.add_argument(
        "--force",
        action="store_true",
        help="Перезаписать только файлы шаблонов; не удаляет иные файлы",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    script_dir = Path(__file__).resolve().parent
    assets_dir = script_dir.parent / "assets"
    case_dir = Path(args.case_dir).expanduser().resolve()

    if case_dir == script_dir.parent or script_dir.parent in case_dir.parents:
        print(
            "Ошибка: храните дело вне папки навыка, например ~/kinship-cases/my-case.",
            file=sys.stderr,
        )
        return 2

    missing = [name for name in ASSET_FILES if not (assets_dir / name).is_file()]
    if missing:
        print(
            "Ошибка: в skill отсутствуют шаблоны: " + ", ".join(missing),
            file=sys.stderr,
        )
        return 2

    if case_dir.exists() and not case_dir.is_dir():
        print(f"Ошибка: {case_dir} существует и не является директорией.", file=sys.stderr)
        return 2

    case_dir.mkdir(parents=True, exist_ok=True)
    conflicts = [dest for dest in ASSET_FILES.values() if (case_dir / dest).exists()]
    if conflicts and not args.force:
        print(
            "Ошибка: целевые файлы уже существуют: "
            + ", ".join(conflicts)
            + ". Продолжайте работу с существующим делом. Для намеренного сброса шаблонов есть --force.",
            file=sys.stderr,
        )
        return 2

    for source_name, dest_name in ASSET_FILES.items():
        shutil.copyfile(assets_dir / source_name, case_dir / dest_name)

    photos_dir = case_dir / "photos"
    photos_dir.mkdir(exist_ok=True)
    photos_readme = photos_dir / "README.md"
    if not photos_readme.exists() and (assets_dir / PHOTOS_README).is_file():
        shutil.copyfile(assets_dir / PHOTOS_README, photos_readme)

    intake_path = case_dir / "case-intake.yaml"
    text = intake_path.read_text(encoding="utf-8")
    created_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
    case_id = case_dir.name
    replacements = {
        'case_id: ""': f"case_id: {quote_yaml(case_id)}",
        'created_at: ""': f"created_at: {quote_yaml(created_at)}",
        'research_question: ""': f"research_question: {quote_yaml(args.research_question)}",
        'subject_label: ""': f"subject_label: {quote_yaml(args.subject_label)}",
        '    label: ""': f"    label: {quote_yaml(args.subject_label)}",
    }
    for old, new in replacements.items():
        if old not in text:
            print(f"Ошибка: маркер шаблона не найден: {old}", file=sys.stderr)
            return 3
        text = text.replace(old, new, 1)
    intake_path.write_text(text, encoding="utf-8")

    print(f"Дело создано: {case_dir}")
    for dest_name in ASSET_FILES.values():
        print(f"- {dest_name}")
    print("- photos/README.md")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

scripts/hash_artifact.py

#!/usr/bin/env python3
"""Compute a SHA-256 checksum and basic metadata for a local artifact."""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Вычисляет SHA-256 локального артефакта.")
    parser.add_argument("path", help="Путь к файлу")
    parser.add_argument("--json", action="store_true", help="Вывести JSON")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    path = Path(args.path).expanduser().resolve()
    if not path.exists():
        print(f"Ошибка: файл не найден: {path}", file=sys.stderr)
        return 2
    if not path.is_file():
        print(f"Ошибка: путь не является файлом: {path}", file=sys.stderr)
        return 2

    stat = path.stat()
    result = {
        "path": str(path),
        "filename": path.name,
        "size_bytes": stat.st_size,
        "modified_at_utc": datetime.fromtimestamp(stat.st_mtime, timezone.utc)
        .replace(microsecond=0)
        .isoformat(),
        "sha256": sha256_file(path),
    }
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        for key, value in result.items():
            print(f"{key}: {value}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

scripts/audit_evidence.py

#!/usr/bin/env python3
"""Audit cross-references and corroboration claims without assigning truth."""

from __future__ import annotations

import argparse
import csv
import json
import re
import sys
from collections import Counter
from pathlib import Path
from typing import Iterable

REQUIRED_FILES = {
    "artifact-register.csv": {
        "artifact_id",
        "independence_group",
        "directness",
        "originality",
        "related_claim_ids",
        "status",
    },
    "claims-register.csv": {
        "claim_id",
        "artifact_ids",
        "supporting_independence_groups",
        "contradicting_artifact_ids",
        "status",
        "rationale",
    },
    "relationship-register.csv": {
        "relationship_id",
        "person_a_id",
        "person_b_id",
        "claim_ids",
        "supporting_artifact_ids",
        "contradicting_artifact_ids",
        "independence_groups",
        "status",
        "rationale",
    },
    "person-register.csv": {"person_id"},
}

CORROBORATED = {"corroborated", "strongly-corroborated"}
VALID_CLAIM_STATUSES = {
    "",
    "hypothesis",
    "single-source",
    "corroborated",
    "strongly-corroborated",
    "contradicted",
    "unresolved",
}


def split_ids(value: str) -> list[str]:
    if not value:
        return []
    normalized = value.replace(",", ";")
    return [part.strip() for part in normalized.split(";") if part.strip()]


def read_csv(path: Path, required: set[str]) -> tuple[list[dict[str, str]], list[str]]:
    errors: list[str] = []
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as handle:
            reader = csv.DictReader(handle)
            if reader.fieldnames is None:
                return [], [f"{path.name}: отсутствует строка заголовка"]
            missing = sorted(required - set(reader.fieldnames))
            if missing:
                errors.append(f"{path.name}: отсутствуют колонки: {', '.join(missing)}")
            rows = [{k: (v or "").strip() for k, v in row.items()} for row in reader]
            return rows, errors
    except OSError as exc:
        return [], [f"{path.name}: ошибка чтения: {exc}"]


def duplicate_ids(rows: Iterable[dict[str, str]], field: str) -> list[str]:
    ids = [row.get(field, "") for row in rows if row.get(field, "")]
    return sorted(key for key, count in Counter(ids).items() if count > 1)


def artifact_groups(artifact_ids: Iterable[str], artifact_by_id: dict[str, dict[str, str]]) -> set[str]:
    return {
        artifact_by_id[artifact_id]["independence_group"].strip().casefold()
        for artifact_id in artifact_ids
        if artifact_id in artifact_by_id
        and artifact_by_id[artifact_id].get("independence_group", "").strip()
    }


def has_direct_record(artifact_ids: Iterable[str], artifact_by_id: dict[str, dict[str, str]]) -> bool:
    """Одна и та же запись должна быть прямой и иметь описанное происхождение.

    Непустое описание и локатор — проверка заполнения, не доказательство
    подлинности документа или истинности заключения.
    """
    for artifact_id in artifact_ids:
        row = artifact_by_id.get(artifact_id, {})
        provenance = row.get("provenance_level", "").strip().casefold()
        if (
            row.get("directness", "").casefold() == "direct"
            and row.get("originality", "").casefold()
            in {"original", "image-of-original", "official-extract"}
            and provenance not in {"", "unknown", "неизвестно", "не установлено"}
            and (row.get("source_url_or_path", "").strip() or row.get("archival_reference", "").strip())
        ):
            return True
    return False


def id_prefix(value: str) -> str:
    """Префикс ID без номера: 'A-0001' -> 'A-'."""
    return value.rstrip("0123456789")


REPORT_MARKERS = ("[неразборчиво]", "[предположительно")
HEADING_RE = re.compile(r"^#{2,6}\s")


def audit_report(
    text: str,
    known_ids: dict[str, set[str]],
    statuses: dict[str, str],
) -> tuple[list[str], dict[str, int]]:
    """Сверить ID и статусы итогового отчёта с реестрами.

    known_ids: префикс -> ID этого префикса из реестров. Стандартные префиксы
    проверяются и при пустых реестрах; плейсхолдеры (`A-....`) не содержат номера.
    statuses: ID утверждения или связи -> статус в реестре.
    """
    errors: list[str] = []
    for prefix, ids in sorted(known_ids.items()):
        pattern = re.compile(rf"(?<![\w-]){re.escape(prefix)}\d+(?!\w)")
        for cited in sorted(set(pattern.findall(text)) - ids):
            errors.append(f"final-report.md: ссылка на {cited}, которого нет в реестрах")

    valid = VALID_CLAIM_STATUSES - {""}
    section_id = ""
    section_checked = False
    for number, line in enumerate(text.splitlines(), start=1):
        stripped = line.strip()
        if HEADING_RE.match(stripped):
            words = re.findall(r"[\w-]+", stripped)
            section_id = next((word for word in words if word in statuses), "")
            section_checked = False
            continue
        if stripped.startswith("|"):
            # Строка таблицы с ID в первой ячейке и статусом в одной из ячеек.
            cells = [cell.strip().strip("`") for cell in stripped.strip("|").split("|")]
            row_id, row_statuses = cells[0], [cell for cell in cells[1:] if cell in valid]
            if row_id in statuses and row_statuses and statuses[row_id] not in row_statuses:
                errors.append(
                    f"final-report.md:{number}: {row_id} — в отчёте {row_statuses[0]}, в реестре {statuses[row_id] or 'пусто'}"
                )
            continue
        # Первое поле вида **Статус:** value в секции связи/утверждения.
        match = re.match(r"^\*\*[^*]+:\*\*\s*`?([\w-]+)`?", stripped)
        if section_id and not section_checked and match and match.group(1) in valid:
            section_checked = True
            if match.group(1) != statuses[section_id]:
                errors.append(
                    f"final-report.md:{number}: {section_id} — в отчёте {match.group(1)}, в реестре {statuses[section_id] or 'пусто'}"
                )

    markers = {marker.strip("[]"): text.count(marker) for marker in REPORT_MARKERS}
    return errors, markers


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Проверяет целостность реестров и минимальные условия кроссподтверждения."
    )
    parser.add_argument("--case-dir", required=True, help="Директория дела")
    parser.add_argument("--json", action="store_true", help="Вывести JSON")
    parser.add_argument(
        "--strict",
        action="store_true",
        help="Возвращать ненулевой код также при предупреждениях",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    case_dir = Path(args.case_dir).expanduser().resolve()
    errors: list[str] = []
    warnings: list[str] = []

    if not case_dir.is_dir():
        print(f"Ошибка: директория дела не найдена: {case_dir}", file=sys.stderr)
        return 2

    data: dict[str, list[dict[str, str]]] = {}
    for filename, required in REQUIRED_FILES.items():
        path = case_dir / filename
        if not path.is_file():
            errors.append(f"Отсутствует обязательный файл: {filename}")
            data[filename] = []
            continue
        rows, file_errors = read_csv(path, required)
        errors.extend(file_errors)
        data[filename] = rows

    artifacts = data["artifact-register.csv"]
    claims = data["claims-register.csv"]
    relationships = data["relationship-register.csv"]
    persons = data["person-register.csv"]

    for rows, field, label in [
        (artifacts, "artifact_id", "артефакта"),
        (claims, "claim_id", "утверждения"),
        (relationships, "relationship_id", "связи"),
        (persons, "person_id", "лица"),
    ]:
        duplicates = duplicate_ids(rows, field)
        if duplicates:
            errors.append(f"Дублирующиеся ID {label}: {', '.join(duplicates)}")
        for index, row in enumerate(rows, start=2):
            if not row.get(field, ""):
                warnings.append(f"{field}: пустой ID в строке {index}")

    artifact_by_id = {row["artifact_id"]: row for row in artifacts if row.get("artifact_id")}
    claim_by_id = {row["claim_id"]: row for row in claims if row.get("claim_id")}
    person_ids = {row["person_id"] for row in persons if row.get("person_id")}

    for artifact_id, row in artifact_by_id.items():
        if not row.get("independence_group", "").strip():
            warnings.append(f"{artifact_id}: не указан independence_group")
        for person_id in split_ids(row.get("person_ids", "")):
            if person_id not in person_ids:
                errors.append(f"{artifact_id}: неизвестный person_id {person_id}")
        for claim_id in split_ids(row.get("related_claim_ids", "")):
            if claim_id not in claim_by_id:
                errors.append(f"{artifact_id}: неизвестный related claim_id {claim_id}")

    for claim_id, row in claim_by_id.items():
        status = row.get("status", "")
        if status not in VALID_CLAIM_STATUSES:
            warnings.append(f"{claim_id}: неизвестный статус {status!r}")
        for person_id in split_ids(row.get("subject_person_ids", "")):
            if person_id not in person_ids:
                errors.append(f"{claim_id}: неизвестный subject_person_id {person_id}")
        referenced = split_ids(row.get("artifact_ids", ""))
        contradictory = split_ids(row.get("contradicting_artifact_ids", ""))
        for artifact_id in referenced + contradictory:
            if artifact_id not in artifact_by_id:
                errors.append(f"{claim_id}: неизвестный artifact_id {artifact_id}")

        groups = artifact_groups(referenced, artifact_by_id)
        for group in sorted(set(split_ids(row.get("supporting_independence_groups", "")))):
            if group.casefold() not in groups:
                errors.append(
                    f"{claim_id}: independence_group {group} заявлен, но ни один из artifact_ids к нему не относится"
                )

        if status in CORROBORATED and len(groups) < 2:
            warnings.append(
                f"{claim_id}: статус {status}, но найдено независимых групп: {len(groups)}"
            )
        if status in CORROBORATED and not row.get("rationale"):
            warnings.append(f"{claim_id}: для {status} отсутствует rationale")
        if status in CORROBORATED and contradictory:
            warnings.append(
                f"{claim_id}: статус {status}, но указаны противоречащие артефакты; "
                "проверьте и объясните, разрешён ли существенный конфликт"
            )
        if status == "single-source" and len(groups) > 1:
            warnings.append(
                f"{claim_id}: несколько независимых групп, но статус всё ещё single-source; проверьте конфликт/полноту"
            )
        if status == "strongly-corroborated" and not has_direct_record(referenced, artifact_by_id):
            warnings.append(
                f"{claim_id}: strongly-corroborated без единой прямой записи "
                "с указанным происхождением и локатором оригинала, образа или официальной выписки"
            )

    for row in relationships:
        relation_id = row.get("relationship_id", "<без ID>")
        status = row.get("status", "")
        if status not in VALID_CLAIM_STATUSES:
            warnings.append(f"{relation_id}: неизвестный статус {status!r}")
        for person_field in ("person_a_id", "person_b_id"):
            person_id = row.get(person_field, "")
            if person_id and person_id not in person_ids:
                errors.append(f"{relation_id}: неизвестный {person_field} {person_id}")
        for claim_id in split_ids(row.get("claim_ids", "")):
            if claim_id not in claim_by_id:
                errors.append(f"{relation_id}: неизвестный claim_id {claim_id}")
        artifact_refs = split_ids(row.get("supporting_artifact_ids", "")) + split_ids(
            row.get("contradicting_artifact_ids", "")
        )
        for artifact_id in artifact_refs:
            if artifact_id not in artifact_by_id:
                errors.append(f"{relation_id}: неизвестный artifact_id {artifact_id}")
        # Группа засчитывается, только если за ней стоит артефакт: прямой или через claim_ids.
        backing_ids = split_ids(row.get("supporting_artifact_ids", ""))
        contradictory = set(split_ids(row.get("contradicting_artifact_ids", "")))
        for claim_id in split_ids(row.get("claim_ids", "")):
            if claim_id in claim_by_id:
                backing_ids += split_ids(claim_by_id[claim_id].get("artifact_ids", ""))
                contradictory.update(split_ids(claim_by_id[claim_id].get("contradicting_artifact_ids", "")))
        groups = artifact_groups(backing_ids, artifact_by_id)
        for group in sorted(set(split_ids(row.get("independence_groups", "")))):
            if group.casefold() not in groups:
                errors.append(
                    f"{relation_id}: independence_group {group} заявлен, но ни один из артефактов связи к нему не относится"
                )
        if status in CORROBORATED:
            if len(groups) < 2:
                warnings.append(
                    f"{relation_id}: статус {row.get('status')}, но указано независимых групп: {len(groups)}"
                )
            if not row.get("rationale"):
                warnings.append(f"{relation_id}: отсутствует rationale")
            if contradictory:
                warnings.append(
                    f"{relation_id}: статус {status}, но указаны противоречащие артефакты; "
                    "проверьте и объясните, разрешён ли существенный конфликт"
                )
        if status == "strongly-corroborated" and not has_direct_record(backing_ids, artifact_by_id):
            warnings.append(
                f"{relation_id}: strongly-corroborated без единой прямой записи "
                "с указанным происхождением и локатором оригинала, образа или официальной выписки"
            )

    # Итоговый отчёт: ссылки и статусы должны совпадать с реестрами.
    markers: dict[str, int] = {}
    report = case_dir / "final-report.md"
    if report.is_file():
        known_ids: dict[str, set[str]] = {
            prefix: set() for prefix in ("A-", "C-", "R-", "P-")
        }
        for rows, field in [
            (artifacts, "artifact_id"),
            (claims, "claim_id"),
            (relationships, "relationship_id"),
            (persons, "person_id"),
        ]:
            for row in rows:
                if row.get(field, "") and id_prefix(row[field]) != row[field]:
                    known_ids.setdefault(id_prefix(row[field]), set()).add(row[field])
        statuses = {claim_id: row.get("status", "") for claim_id, row in claim_by_id.items()}
        statuses.update(
            (row["relationship_id"], row.get("status", ""))
            for row in relationships
            if row.get("relationship_id")
        )
        report_errors, markers = audit_report(
            report.read_text(encoding="utf-8-sig"), known_ids, statuses
        )
        errors.extend(report_errors)

    # Сводка готовности: что ещё не снято. Не ошибки — для решения, выдавать ли отчёт.
    summary = {
        "claims_by_status": dict(Counter(row.get("status", "") or "пусто" for row in claims)),
        "relationships_by_status": dict(
            Counter(row.get("status", "") or "пусто" for row in relationships)
        ),
        "report_markers": markers,
    }

    result = {
        "case_dir": str(case_dir),
        "counts": {
            "persons": len(persons),
            "artifacts": len(artifacts),
            "claims": len(claims),
            "relationships": len(relationships),
        },
        "summary": summary,
        "errors": errors,
        "warnings": warnings,
    }

    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(f"Проверено дело: {case_dir}")
        print(
            "Записей: "
            f"лица={len(persons)}, артефакты={len(artifacts)}, "
            f"утверждения={len(claims)}, связи={len(relationships)}"
        )
        open_statuses = ("hypothesis", "single-source", "unresolved", "contradicted", "пусто")
        print("\nСВОДКА:")
        for label, key in [("утверждения", "claims_by_status"), ("связи", "relationships_by_status")]:
            counts = summary[key]
            line = ", ".join(f"{status}={counts[status]}" for status in sorted(counts))
            pending = sum(counts.get(status, 0) for status in open_statuses)
            print(f"- {label}: {line or 'нет'}; без кроссподтверждения: {pending}")
        if markers:
            print("- в отчёте: " + ", ".join(f"[{name}]={count}" for name, count in markers.items()))
        if errors:
            print("\nОШИБКИ:")
            for item in errors:
                print(f"- {item}")
        if warnings:
            print("\nПРЕДУПРЕЖДЕНИЯ:")
            for item in warnings:
                print(f"- {item}")
        if not errors and not warnings:
            print("Ошибок и предупреждений не найдено.")

    if errors:
        return 1
    if warnings and args.strict:
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

[Content truncated due to size limits]