arm-core/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md

Срез 2c — загрузка ОСВ, цикл маппинга, документы (arm-core) — план исполнения

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Branch: feat/14-slice-2c-osv-ingest (worktree .wt/slice-2c/arm-core). Issue: RSHB/arm-core#14. PR: #15 (открыт, спека). Соседний worktree: docs .wt/slice-2c/docs (ветка feat/16-slice-2c-osv-ingest, issue #16, PR #17) — там уже лежит model/data-model-v5.15.dbml; задачи docs-change 1.2 и 2.1 выполняются в Task 1 и Task 12 этого плана. ../fixtures — симлинк, только чтение (golden-снимки парсера golden/expected_osv_parse/**). ../tabular-parser — симлинк на main, только чтение (для e2e через compose). Spec: openspec/changes/slice-2c-osv-ingest/ — proposal.md, design.md (D-1…D-13, таблица решений №1–13), specs/{primary-files,osv-result-mapping,documents-api,recognition-job-exchange,data-model,audit-trail}/spec.md. Правила: ../docs/context/05-domain-rules.md §1.3 (строки, нормализация кода счёта), §1.5 (период), §3 п. 4 (резолв контрагента), §9.2–9.4 (статусы). Спека — единственный источник истины при рулингах; план сгенерирован из tasks.md change'а (номера в скобках).

Goal: файл ОСВ, загруженный через API ядра, проходит очередь, распознаётся tabular-parser'ом и превращается в document + osv_entry без участия человека; статусы файла и документа объяснимы по карточке файла.

Architecture: эндпоинт загрузки → s3_object + stored_file + recognition_job(queued) (outbox существующий) → inbox ставит статусы файла по ack/ошибке → отдельный фоновый цикл маппинга читает результат из S3 и одной транзакцией пишет документы и строки, автозамена прежних версий → API чтения и подтверждения. Чистые функции маппинга — src/domain/osv_mapping.py.

Tech Stack: Python 3.11, FastAPI, SQLAlchemy 2 async, Alembic, PostgreSQL 16, pydantic v2 (контракт v1.1 в src/contract), miniopy-async, pytest-asyncio, httpx.

Global Constraints

• DBML v5.15 (../docs/model/data-model-v5.15.dbml, 43 таблицы, 27 enum): osv_data_type + NU, app_user.email в нижнем регистре; полей вне DBML нет. make dbml-check — критерий Task 1 и регрессия в каждой задаче.
• Статусы файла меняет только код, проверивший stored_file.current_recognition_job_id == job.id (D-5): загрузка UPLOADED; inbox ack → PROCESSING; inbox error, вахтёр timeout, публикатор dlq → ERROR; цикл маппинга → PROCESSED/ERROR; автозамена → SUPERSEDED.
• Inbox не читает S3 и не создаёт документы; ResultMapper/NoopMapper удаляются. Маппинг — services/result_mapping.py, чтение S3 вне транзакции, одна транзакция на файл, перед вставкой — очистка строк/документов файла (идемпотентность).
• Строки: все, кроме group_type = TOTAL (OQ-01); data_type как пришёл (включая NU); суммы — Decimal из контракта без изменений; counterparty_* как в результате; is_modified = false; parent_entry_id по parent_row_number.
• Резолв контрагента только для COUNTERPARTY с data_type ∈ {BU, SUM}: ИНН → точное имя → normalize_alias по именам ЮЛ и алиасам; > 1 кандидата → NULL; без policy. Имя ЮЛ нормализуется той же normalize_alias.
• Статус документа: RECOGNIZED только при job.status = done, пустых checks-нарушениях, counterparty_check ≠ MISMATCH и без предупреждений маппинга; иначе NEEDS_CHECK. period_check = UNKNOWN (Срез 4).
• Права и policy: DOCUMENT_EDIT (загрузка, удаление, подтверждение), DOCUMENT_VIEW (чтение); видимость через ЮЛ-владельца (legal_entities_visible); чужое → 404 с тем же телом, что несуществующее.
• Сессии на маршрутах — только SessionDep; фоновые циклы — async_session_maker/session_transaction. audit_event — только AuditService.record/RequestAudit.record (append-only, тест-страж). Ошибки — DomainError ({"detail": {"code", "message"}}), 413 — HTTPException строкой как в импорте.
• Деньги — Decimal/Numeric(18,2); float в домене запрещён.
• Линтеры: ruff (120, E,F,I,UP,B,PL — константы вместо магических чисел, # noqa: PLR0913 - причина), mypy (strict-флаги для src.domain.*, src.models.*). Тесты: маркеры integration, fixtures, новый e2e (объявить в pytest.ini). Команды — из api/ (make из корня). Compose arm-ka поднят (PostgreSQL/NATS/MinIO) — не останавливать.
• Коммиты — русский, по задаче; трейлеры Co-Authored-By: Claude <модель исполнителя> <noreply@anthropic.com> и Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2. Не пушить, не создавать PR, не openspec archive, openspec/specs/ не трогать (кроме как через change). Коммиты в ../docs — в его worktree, те же правила.
• Допущения: # ASSUMPTION(OQ-02) (нарастающая по Q1), # ASSUMPTION(OQ-55) (обмен не аудируется), # ASSUMPTION(OQ-100) (ИНН из текста не извлекается), # ASSUMPTION(OQ-103) (нечисловой первый сегмент → account_root = None), # ASSUMPTION(OQ-22) (expires_at при удалении).

Review Focus

1. Два задания у одного файла (повторная загрузка того же содержимого после SUPERSEDED, или будущий replay): статус файла меняет только текущее задание; результат старого задания после смены current_recognition_job_id файл не трогает — Task 4 (inbox) и Task 7 (цикл): тест «результат чужого задания игнорируется».
2. Результат needs_check при отсутствии документов (парсер по контракту требует documents ≥ 1, но файл может прийти от другого сервиса): цикл маппинга → ERROR «нет документов ОСВ», не исключение — Task 7: тест.
3. Строка COUNTERPARTY с ИНН, которого нет в справочнике, но имя совпадает с алиасом другого ЮЛ — ИНН не найден → переход к имени/алиасу по §3 (ИНН не блокирует) — Task 6: тест.
4. Имя файла с кириллицей, пробелами и .. в ключе S3 — safe_filename + ключ без путей; загрузка файла с пустым именем → UNSUPPORTED_FILE — Task 3: тест.
5. Подтверждение документа файла, который только что заменён автозаменой — гонка «подтвердить» vs «цикл заменил»: confirm перечитывает файл под FOR UPDATE и отвечает 409 FILE_SUPERSEDED — Task 9: тест.

---

Задачи

Task 1: [1.1] DBML v5.15, миграция 0003, email в нижнем регистре

Files:

• Modify: api/src/models/enums.py (OsvDataType.NU после BU), api/src/models/__init__.py (DBML_VERSION = "5.15"), Makefile (DBML ?= ../../docs/model/data-model-v5.15.dbml), api/scripts/dbml_parity.py (DBML_FILE_NAME), api/tests/test_migrations.py (HEAD = "0003_dbml_v5_15" + тест отката на шаг), api/tests/test_enums.py, api/tests/test_check_dbml_script.py (фиктивная v5.16), api/scripts/seed_lib/apply.py (_apply_users: email.strip().lower()), api/src/services/users_admin.py (create_user/update_user: хранить email.strip().lower(); pre-check по lower остаётся), openspec/specs не трогать; docstring'и «v5.14» → «v5.15» там, где речь о текущей версии
• Create: api/alembic/versions/0003_dbml_v5_15.py
• Create (worktree docs .wt/slice-2c/docs): model/entities-v5.15.md (копия v5.14 + «Что изменилось в v5.15»: NU, email), правки context/04-data-model-guide.md, repo-templates/arm-core.CLAUDE.md, workspace /Users/andrey.epifancev/projects/awp-ca/CLAUDE.md (вне git) → v5.15; openspec/changes/slice-2c-osv-ingest/tasks.md 1.2 → [x]; коммит в docs

Interfaces:

• Миграция (стиль 0002): upgrade: ALTER TYPE dkb.osv_data_type ADD VALUE IF NOT EXISTS 'NU' AFTER 'BU'; UPDATE dkb.app_user SET email = lower(email) WHERE email <> lower(email); downgrade: пересоздание osv_data_type со значениями v5.14 (RENAME TYPE → CREATE TYPE → ALTER COLUMN osv_entry.data_type TYPE … USING data_type::text::dkb.osv_data_type → DROP TYPE), email не возвращается.

• OsvDataType: BU, NU, CURRENCY, QUANTITY, SUM (порядок DBML).

• Step 1: Тесты. test_enums ожидает "5.15" и 5 значений osv_data_type; test_migrations: HEAD, test_downgrade_one_step → после downgrade 0002_dbml_v5_14 enum_range(NULL::dkb.osv_data_type) без NU; тест данных: вставить app_user с Email@X.RU на ревизии 0002 → после upgrade head email email@x.ru; test_seed_dev: email dev-пользователей в нижнем регистре; test_api_admin: создание с Qa1@Example.com → хранится qa1@example.com, /me по любому регистру.

• Step 2: падает → Step 3: реализация + docs-правки → Step 4: make dbml-check пуст; make migration-cycle; make check; pytest -m integration tests/test_seed_dev.py tests/test_api_admin.py -q → PASS.

• Step 5: Commit feat(model): DBML v5.15 — osv_data_type NU, email в нижнем регистре, миграция 0003 (+ коммит в docs docs: entities-v5.15, guide, ссылки на DBML v5.15).

---

Task 2: [1.2] S3 get_bytes, фабрики, настройки

Files:

• Modify: api/src/core/clients.py (S3Storage.get_bytes), api/src/services/directory_import.py (ObjectStorage протокол → вынести в api/src/core/storage.py:ObjectStorage(Protocol) с bucket, put_object, get_bytes), api/tests/conftest.py (FakeStorage.get_bytes, KeyError → ObjectNotFoundError), api/tests/factories.py, api/src/core/config.py (UPLOAD_MAX_BYTES: PositiveInt = 50 * 1024 * 1024, MAPPING_POLL_SECONDS: PositiveInt = 5, MAPPING_MAX_ATTEMPTS: PositiveInt = 3), api/.env.example, api/.env.test, api/pytest.ini (маркер e2e)
• Test: api/tests/test_clients.py (unit на get_bytes с заглушкой Minio), api/tests/test_factories.py

Interfaces:

# core/storage.py
class ObjectNotFoundError(Exception): ...
class ObjectStorage(Protocol):
    bucket: str
    async def put_object(self, obj_name: str, file_data: bytes) -> None: ...
    async def get_bytes(self, obj_name: str) -> bytes: ...   # отсутствие объекта → ObjectNotFoundError
# S3Storage.get_bytes: response = await self.client.get_object(self.bucket, obj_name); try: return await response.read()  finally: response.close(); await response.release()
#   (miniopy-async: NoSuchKey → ObjectNotFoundError)
# tests/factories.py
async def make_stored_file(session, *, entity: LegalEntity, status=FileStatus.UPLOADED, file_type=FileType.XLS, original_name="osv.xls",
                           content_hash: str | None = None, job_status=RecognitionStatus.queued, s3_key: str | None = None,
                           uploaded_by: uuid.UUID | None = None) -> tuple[StoredFile, RecognitionJob]   # s3_object + stored_file + job + current_recognition_job_id, flush
async def make_document(session, *, file: StoredFile, job: RecognitionJob, account="60", year=2025, quarter=PeriodQuarter.Q1,
                        status=DocumentStatus.RECOGNIZED, is_cumulative=True) -> Document
async def make_result_object(session, *, job: RecognitionJob, storage: FakeStorage, result: dict) -> S3Object   # кладёт JSON в storage под results/{file_id}/{job_id}/result.json, регистрирует s3_object, job.result_s3_object_id
# CommittedJobs.cleanup(): перед stored_file удалять osv_entry, document (по file_id)

• Step 1: тесты: get_bytes возвращает байты и закрывает response (заглушка с флагами), NoSuchKey → ObjectNotFoundError; фабрики создают цепочку и cleanup удаляет документы.
• Step 2–4: падает → реализация → make check, pytest -m integration tests/test_factories.py -q.
• Step 5: Commit chore: S3 get_bytes, протокол ObjectStorage, фабрики файлов/документов, настройки загрузки и маппинга.

---

Task 3: [2.1] Загрузка файлов

Files:

• Create: api/src/domain/file_types.py, api/src/services/primary_files.py, api/src/schemas/files.py, api/src/api/endpoints/files/__init__.py, api/src/api/endpoints/files/upload.py, api/src/api/dependencies/files.py
• Modify: api/src/api/router.py (include files.router), api/src/services/directory_import.py (safe_filename, CONTENT_TYPES → вынести в api/src/common/filenames.py и переиспользовать)
• Test: api/tests/test_domain_file_types.py, api/tests/test_api_files_upload.py

Interfaces:

# domain/file_types.py
@dataclass(frozen=True)
class DetectedFile: file_type: FileType; pipeline: RecognitionPipeline; content_type: str
def detect_file(name: str, content: bytes) -> DetectedFile          # расширение → тип; сигнатура: xls = b"\xd0\xcf\x11\xe0", xlsx/xlsb = b"PK\x03\x04", pdf = b"%PDF", png = b"\x89PNG", jpg = b"\xff\xd8\xff"; несовпадение/неизвестно → DomainValidationError("UNSUPPORTED_FILE", …)
MAX_ATTEMPTS = {RecognitionPipeline.tabular: 3, RecognitionPipeline.image: 1}
# services/primary_files.py
@dataclass(frozen=True)
class AcceptedFile: file_id: uuid.UUID; job_id: uuid.UUID; name: str
@dataclass(frozen=True)
class RejectedFile: name: str; code: str; message: str; file_id: uuid.UUID | None = None
class PrimaryFilesService:
    def __init__(self, session, audit: RequestAudit, scope: VisibilityScope, storage: ObjectStorage | None, settings: Settings) -> None
    async def upload_one(self, entity: LegalEntity, name: str, content: bytes) -> AcceptedFile      # raises DomainError → эндпоинт превращает в RejectedFile; каждая загрузка — begin_nested() (SAVEPOINT) внутри транзакции запроса, чтобы отказ одного файла не откатил принятые
    async def find_duplicate(self, entity_id, sha256) -> StoredFile | None   # status != SUPERSEDED
# endpoint: POST /legal-entities/{legal_entity_id}/files, files: list[UploadFile] = File(...), require("DOCUMENT_EDIT"); ЮЛ через get_legal_entity(session, scope, id) → 404; storage None → HTTPException(503, "хранилище файлов недоступно") (код STORAGE_UNAVAILABLE в detail-строке); размер — _read_limited(file, settings.UPLOAD_MAX_BYTES) → 413 превращается в RejectedFile(FILE_TOO_LARGE) (не HTTPException на весь запрос); ответ UploadResult(accepted, rejected), 201 если accepted непуст, иначе 422
# ключ S3: f"uploads/{file_id}/{job_id}/{safe_filename(name)}"; S3Object (bucket_role=uploads, sha256, size_bytes, content_type, created_by=f"user:{id}", created_at) → flush → put_object → StoredFile(status=UPLOADED, version_no=1, content_hash=sha256, original_name=safe name, uploaded_by, uploaded_at) → RecognitionJob(file_id, pipeline, request_subject=request_subject(pipeline) из contract.subjects, schema_version=CONTRACT_VERSION, status=queued, max_attempts) → flush → file.current_recognition_job_id = job.id; audit FILE_UPLOAD (object_id=file.id, after_value={original_name, content_hash, job_id, file_type})

• schemas/files.py: UploadResult{accepted: list[AcceptedOut], rejected: list[RejectedOut]}, JobBrief{id, status, error_code, error_message, warnings: list[str], mapping_warnings: list[str], finished_at}, DocumentBrief{id, document_type, osv_account_code, period_year, period_quarter, is_cumulative, status, period_check, counterparty_check, locator}, FileOut{id, legal_entity_id, original_name, file_type, status, version_no, size_bytes, content_hash, uploaded_at, uploaded_by, superseded_by_file_id, job: JobBrief | None, documents: list[DocumentBrief]} (используется Task 9).

• Step 1: Тесты. test_domain_file_types.py: 6 типов по сигнатуре, osv.xlsx с BIFF8 → UNSUPPORTED_FILE, .docx → UNSUPPORTED_FILE, пустое имя → UNSUPPORTED_FILE (Review Focus 4). test_api_files_upload.py (integration, auth_client, app.state.storage = FakeStorage()): сценарии спеки primary-files «Реальная ОСВ принята» (байты ../fixtures/osv/rb/2024-FY/ОСВ_66 за 2024.xls, маркер fixtures; затем publish_pending с FakeNats из tests/test_recognition_exchange.py публикует запрос с ключом uploads/{file_id}/{job_id}/ОСВ_66 за 2024.xls), «Смесь» (xlsx + docx + 51 МиБ через UPLOAD_MAX_BYTES=10 в настройках теста), «Сигнатура», «Чужое ЮЛ» (client_manager → 404), «Дубль» (второй раз → DUPLICATE_FILE с file_id; тот же файл у другого ЮЛ → принят); имя ../..\\ОСВ 60 (2).xls → ключ без .. и слэшей; storage=None → 503; два FILE_UPLOAD в аудите.

• Step 2–4: падает → реализация → pytest -m "integration or fixtures" tests/test_api_files_upload.py -q, make check.

• Step 5: Commit feat(files): загрузка файлов организации — S3, stored_file, recognition_job в outbox, дедуп, аудит.

---

Task 4: [2.2] Статусы файла в обмене, удаление ResultMapper

Files:

• Modify: api/src/services/recognition_exchange.py (_apply_ack → PROCESSING; _apply_result при error → ERROR; expire_deadlines — второй UPDATE stored_file; _process_one при dlq → ERROR; удалить ResultMapper, NoopMapper, _map_result, _MapperError, параметр mapper у handle_message), api/src/app.py (partial(handle_message, session_factory=...)), api/tests/test_recognition_inbox.py, api/tests/test_recognition_exchange.py (RecordingMapper → проверки статуса файла)
• Create: api/src/services/file_status.py

Interfaces:

# services/file_status.py
async def set_file_status_for_job(session, job: RecognitionJob, status: FileStatus, *, now: datetime) -> bool
# UPDATE stored_file SET status=:status, updated_at=:now WHERE id = job.file_id AND current_recognition_job_id = job.id AND status NOT IN (SUPERSEDED, PROCESSED); returns rowcount == 1
# в expire_deadlines: после UPDATE recognition_job … RETURNING id, file_id → UPDATE stored_file SET status='ERROR' WHERE current_recognition_job_id = ANY(ids) AND status NOT IN (SUPERSEDED, PROCESSED)

Ошибка задания (error/timeout/dlq) ставит файлу ERROR; ack — PROCESSING только из UPLOADED; done/needs_check файл не трогают.

• Step 1: Тесты (integration, CommittedJobs + make_stored_file): ack → PROCESSING; error → ERROR; done → файл PROCESSING; timeout вахтёром → ERROR; dlq публикатором → ERROR; результат задания, которое уже не current_recognition_job_id файла → файл не меняется (Review Focus 1); существующие тесты inbox без mapper зелёные.
• Step 2–4: падает → реализация → pytest -m integration tests/test_recognition_inbox.py tests/test_recognition_exchange.py -q, make check.
• Step 5: Commit feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67).

---

Task 5: [3.1a] Домен маппинга — строки, нормализация, проверки

Files:

• Create: api/src/domain/osv_mapping.py
• Test: api/tests/test_domain_osv_mapping.py

Interfaces:

ACCOUNT_SEGMENT_DIGITS = 2
def normalize_account_code(raw: str) -> str          # "60.1"→"60.01", "76.7.1"→"76.07.1", "58.03"→"58.03", "76.АВ"→"76.АВ", "МЦ.04"→"МЦ.04", "001"→"001"; сегменты по '.', числовой сегмент (1–3 цифры) → zfill(2) кроме первого (как есть) и третьего (как есть)
def account_root(code: str) -> str | None            # первый сегмент, если он из цифр; иначе None (ASSUMPTION(OQ-103))
@dataclass(frozen=True)
class EntryDraft: id: uuid.UUID; row_number: int; level: int; parent_id: uuid.UUID | None; row_value: str; group_type: OsvGroupType; group_name: str | None; data_type: OsvDataType | None; account_code: str | None; account_root: str | None; account_id: uuid.UUID | None; counterparty_name: str | None; counterparty_inn: str | None; sums: tuple[Decimal | None, ...]  # opening_debit, opening_credit, turnover_debit, turnover_credit, closing_debit, closing_credit
@dataclass(frozen=True)
class MappedSheet: entries: list[EntryDraft]; root: EntryDraft | None; total_present: bool; orphans: tuple[int, ...]
def build_entries(doc: OsvDocument, chart: Mapping[str, uuid.UUID]) -> MappedSheet
# пропуск TOTAL; id = uuid4 заранее; parent по parent_row_number (нет в карте → parent None; MappedSheet.orphans — список row_number, вызывающий пишет предупреждение); account_code = row.account_code_raw; account_id = chart.get(normalize_account_code(account_code)) if account_code else None; group_type TOTAL не встречается в entries; data_type через OsvDataType(row.data_type) if row.data_type else None; counterparty_* как есть
def document_fields(doc: OsvDocument) -> DocumentDraft    # osv_account_code = normalize_account_code(header.account_code_raw), period_year, period_quarter (PeriodQuarter), is_cumulative, locator
def counterparty_check(header: OsvHeader, owner_inn: str, owner_names_normalized: frozenset[str]) -> DocumentCheckResult
# organization_inn есть → MATCH если == owner_inn (только цифры), иначе MISMATCH; нет ИНН: normalize_alias(organization_name) ∈ owner_names_normalized → MATCH; пустое имя и ИНН → UNKNOWN; иначе MISMATCH
def document_status(*, job_done: bool, checks: OsvChecks | None, counterparty: DocumentCheckResult, warnings: Sequence[str]) -> DocumentStatus
# RECOGNIZED iff job_done and (checks is None or (checks.total_row_matches_root is not False and not checks.balance_identity_violations and not checks.parent_sum_violations)) and counterparty != MISMATCH and not warnings; else NEEDS_CHECK
def opening_matches(a: EntryDraft, b_opening: tuple[Decimal | None, Decimal | None]) -> bool   # ±0.01, None = 0 — для D-9

• Step 1: Тесты (unit): нормализация (таблица из 8 случаев), account_root (60, 001, МЦ.04 → None); build_entries на api/contracts/examples/result-osv-minimal.json (TOTAL отсутствует, parent выставлен, суммы Decimal) и на golden ../fixtures/golden/expected_osv_parse/rb/2024-FY/ОСВ_66 за 2024.json (маркер fixtures; chart — словарь из seed CSV): 9 строк (10 строк снимка минус TOTAL), у «Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ» account_code = "66.03", account_id задан; counterparty_check — ИНН совпал/не совпал, по имени, по алиасу, пустая шапка; document_status — 5 комбинаций; opening_matches на границе 0.01.
• Step 2–4: падает → реализация → pytest tests/test_domain_osv_mapping.py -q, mypy src strict для src.domain.*.
• Step 5: Commit feat(domain): маппинг osvDocument в черновики документа и строк, нормализация кода счёта, проверки (§1.3, §9.3).

---

Task 6: [3.1b] Резолв контрагента

Files:

• Create: api/src/repositories/counterparties.py
• Test: api/tests/test_counterparty_resolve.py

Interfaces:

@dataclass
class CounterpartyIndex:
    by_inn: dict[str, uuid.UUID]
    by_exact_name: dict[str, set[uuid.UUID]]        # strip()
    by_normalized: dict[str, set[uuid.UUID]]        # normalize_alias(name) ∪ normalized_alias алиасов
    @classmethod
    async def load(cls, session) -> "CounterpartyIndex"   # два запроса: legal_entity(id, inn, name), legal_entity_alias(legal_entity_id, normalized_alias); без policy
    def resolve(self, inn: str | None, name: str | None) -> uuid.UUID | None
    # inn (цифры) in by_inn → id; else name.strip() in by_exact_name и ровно один → id (> 1 → None, 0 → дальше); else normalize_alias(name) in by_normalized и ровно один → id; иначе None
    def owner_names(self, entity: LegalEntity) -> frozenset[str]   # normalize_alias(name) ∪ его алиасы — для counterparty_check

• Step 1: Тесты (integration на db_session с фабриками make_legal_entity + алиасы через LegalEntityAlias): сценарий спеки «По ИНН, по алиасу, неоднозначно»; ИНН не найден, имя совпадает с алиасом другого ЮЛ → по алиасу (Review Focus 3); закрытое ЮЛ (close_date) тоже резолвится; owner_names.
• Step 2–4: падает → реализация → тесты, make check.
• Step 5: Commit feat(mapping): индекс и резолв контрагентов по ИНН/имени/алиасу (§3 п. 4).

---

Task 7: [3.2] Цикл маппинга

Files:

• Create: api/src/services/result_mapping.py
• Modify: api/src/services/recognition_exchange.py (ExchangeTasks.start принимает storage и добавляет задачу run_mapping_loop), api/src/app.py (_start_exchange(nats, storage); цикл маппинга стартует вместе с обменом, при NATS_ENABLED=false не стартует)
• Test: api/tests/test_result_mapping.py

Interfaces:

MAPPING_ATTEMPTS_KEY = "mapping_attempts"; MAPPING_ERROR_KEY = "mapping_error"; MAPPING_WARNINGS_KEY = "mapping_warnings"
class MappingFatal(Exception): """ошибка без повтора (ключ, объект, схема, нет документов)"""
@dataclass(frozen=True)
class Candidate: job_id: uuid.UUID; file_id: uuid.UUID; bucket: str; key: str
async def pick_candidate(session) -> Candidate | None
# SELECT job.id, job.file_id, s3.bucket, s3.key FROM recognition_job job JOIN stored_file f ON f.id=job.file_id AND f.current_recognition_job_id=job.id JOIN s3_object s3 ON s3.id=job.result_s3_object_id
#   WHERE job.status IN (done, needs_check) AND f.status='PROCESSING' AND coalesce((job.metrics->>'mapping_attempts')::int,0) < :max ORDER BY job.finished_at, job.id LIMIT 1 FOR UPDATE OF job SKIP LOCKED
def validate_result_key(candidate: Candidate, storage_bucket: str) -> None   # MappingFatal при бакете ≠ storage.bucket или ключе не вида results/{file_id}/{job_id}/…
async def map_file(session, *, job: RecognitionJob, file: StoredFile, result: RecognitionResult, chart: Mapping[str, uuid], index: CounterpartyIndex, now: datetime) -> list[str]  # предупреждения
# в транзакции: DELETE osv_entry WHERE document_id IN (документы файла); DELETE document WHERE stored_file_id; по osvDocument: document_fields → если osv_account_code уже встречался в этом файле → warning "sheet=<locator>: счёт <код> уже есть в листе <locator первого>", первый документ помечается needs_check; иначе Document(...); build_entries → insert(OsvEntry).values([...]) пакетами по 1000; D-9: для Q2–Q4 с is_cumulative False — ищем Document того же ЮЛ/счёта/года Q1 с файлом != SUPERSEDED, корень = osv_entry с row_number минимальным level 0 → opening_matches → is_cumulative True + warning; counterparty_check(header, file.legal_entity.inn, index.owner_names(entity)); status = document_status(...); job.group_schema = [{"locator", "groups":[{"type","name","has_inn"}]}]; job.metrics[mapping_warnings] = warnings; file.status = PROCESSED, updated_at; вызов supersede_previous добавляет Task 8; в этой задаче map_file завершается установкой PROCESSED
async def process_once(session_factory, storage: ObjectStorage, settings, *, now=None) -> bool
# 1) короткая транзакция: pick_candidate → нет → False; validate_result_key (MappingFatal → в той же транзакции файл ERROR, metrics.mapping_error, commit, return True)
# 2) без транзакции: content = await storage.get_bytes(key) (ObjectNotFoundError → fatal); RecognitionResult.model_validate_json (ValidationError → fatal); нет osvDocument → fatal "нет документов ОСВ"
# 3) транзакция: заново SELECT job FOR UPDATE и file; если file.status != PROCESSING или current_job != job → return True (кто-то опередил); chart = {code: id} одним запросом; index = CounterpartyIndex.load; map_file; commit
# исключение в 3): rollback; отдельная транзакция: metrics.mapping_attempts += 1; если >= MAPPING_MAX_ATTEMPTS → file ERROR + mapping_error; лог с traceback
async def run_mapping_loop(session_factory, storage, settings, *, sleep=asyncio.sleep) -> None   # while True: processed = await process_once(...); await sleep(0 if processed else settings.MAPPING_POLL_SECONDS); исключения цикла логируются, цикл не умирает

ExchangeTasks.start(session_factory, nats, settings, storage) — третья задача recognition-result-mapper только при storage is not None (иначе warning «S3 не подключено: маппинг результатов не запущен»).

• Step 1: Тесты (integration, committed_sessionmaker + FakeStorage; результат — golden JSON ОСВ_66 за 2024.json с подставленными job_id/file_id): сценарии спеки osv-result-mapping «Результат смапплен» (документ 66/2024/Q4/cumulative, locator, 9 строк (10 строк снимка минус TOTAL), group_schema), «Ключ вне папки» → ERROR, «Сбой с повтором» (monkeypatch map_file бросает 2 раза → attempts, 3-й → ERROR), «Идемпотентность» (вставить половину строк вручную → process_once → ровно один набор), «Два листа одного счёта» (JSON с двумя документами 60) → один документ NEEDS_CHECK + warning, «Шапка другой организации» → MISMATCH/NEEDS_CHECK, «Нарастающая по данным» (сначала документ Q1 через фабрики с корнем, затем Q3 с теми же opening) → is_cumulative = True, «Чистый результат» → RECOGNIZED; результат без документов (status: error-подобный JSON с documents: [] не валиден по контракту → fatal) → ERROR (Review Focus 2); результат для задания, которое уже не текущее → не трогает файл (Review Focus 1); run_mapping_loop с заглушкой sleep обрабатывает два кандидата подряд.
• Step 2–4: падает → реализация → pytest -m "integration or fixtures" tests/test_result_mapping.py -q, make check.
• Step 5: Commit feat(mapping): фоновый цикл маппинга результата в document/osv_entry (OQ-69, OQ-70).

---

Task 8: [3.3] Автозамена

Files:

• Modify: api/src/services/result_mapping.py (supersede_previous), api/src/services/audit.py — без изменений (системное событие user_id=None)
• Test: api/tests/test_result_mapping.py (+ сценарии)

Interfaces:

async def supersede_previous(session, *, new_file: StoredFile, new_documents: Sequence[Document], uploader_id: uuid.UUID | None, now) -> list[StoredFile]
# для каждого нового документа: SELECT stored_file JOIN document WHERE stored_file.legal_entity_id = new.legal_entity_id AND stored_file.id != new.id AND stored_file.status != SUPERSEDED AND document.(type, osv_account_code, period_year, period_quarter) совпадают; distinct файлы → status SUPERSEDED, superseded_by_file_id = new.id, updated_at; new.version_no = max(v.version_no) + 1 (если нашлись); AuditService(session).record(action=FILE_REPLACE, section=PRIMARY_DOCUMENTS, object_type="stored_file", object_id=old.id, user_id=None, after_value={"file_id": new.id, "version_no": new.version_no}, description=f"заменён файлом {new.original_name} (загрузил user:{uploader_id}); документы: OSV 60 2025 Q1 (CONFIRMED)…; confirmed={true|false}")

Вызывается в map_file после установки PROCESSED, в той же транзакции.

• Step 1: Тесты: сценарий спеки «Новая ОСВ 60 за 1 кв. 2025» (старый CONFIRMED → SUPERSEDED, version_no = 2, аудит с confirmed=true); два разных счёта не заменяют друг друга; тот же счёт другого года не заменяет; заменённый файл не появляется в GET files без include_superseded (проверяется в Task 9, здесь — по БД).
• Step 2–4: падает → реализация → тесты, make check.
• Step 5: Commit feat(mapping): автозамена прежних версий файла после маппинга (FILE_REPLACE).

---

Task 9: [4.1] API чтения и подтверждение документа

Files:

• Create: api/src/repositories/files.py, api/src/services/documents.py, api/src/schemas/documents.py, api/src/api/endpoints/files/read.py, api/src/api/endpoints/files/documents.py
• Modify: api/src/api/endpoints/files/__init__.py, api/src/api/dependencies/files.py, api/tests/test_openapi_routes.py
• Test: api/tests/test_api_files_read.py, api/tests/test_api_documents.py

Interfaces:

# repositories/files.py
async def list_files(session, scope, entity_id, *, include_superseded: bool, limit, offset) -> tuple[list[StoredFile], int]   # с joinedload job (current) и documents (selectin)
async def get_file(session, scope, file_id) -> StoredFile | None       # через legal_entities_visible по владельцу
async def get_document(session, scope, document_id) -> Document | None
async def list_entries(session, document_id, *, limit, offset) -> tuple[list[OsvEntry], int]   # ORDER BY row_number
# services/documents.py
class DocumentService:
    async def confirm(self, document_id) -> Document   # SELECT document JOIN stored_file FOR UPDATE; file.status == SUPERSEDED → ConflictError("FILE_SUPERSEDED"); status == CONFIRMED → ConflictError("ALREADY_CONFIRMED"); status ∉ {RECOGNIZED, NEEDS_CHECK} → ConflictError("DOCUMENT_NOT_CONFIRMABLE"); set CONFIRMED, confirmed_by, confirmed_at; audit DOCUMENT_CONFIRM before/after status
# schemas/documents.py: DocumentOut{id, file_id, legal_entity_id, document_type, osv_account_code, period_year, period_quarter, is_cumulative, locator, status, period_check, counterparty_check, confirmed_by, confirmed_at, group_schema: list[GroupOut] | None, warnings: list[str]}; EntryOut{id, row_number, level, parent_entry_id, row_value, group_type, group_name, data_type, account_code, account_root, account_id, counterparty_name, counterparty_inn, counterparty_id, opening_debit…closing_credit (Decimal → str)}; EntriesPageParams(limit ≤ 2000, default 500)
# маршруты: GET /legal-entities/{id}/files → Page[FileOut]; GET /files/{file_id} → FileOut; GET /documents/{id} → DocumentOut; GET /documents/{id}/entries → Page[EntryOut]; POST /documents/{id}/confirm → DocumentOut

• Step 1: Тесты: сценарии спеки documents-api (список с PROCESSED+документом и ERROR, include_superseded, карточка, 404 для чужого, строки деревом на данных из golden через make_entries_from_result, limit=2001 → 422, подтверждение с аудитом, повтор → 409, документ SUPERSEDED файла → 409; гонка Review Focus 5: файл переведён в SUPERSEDED между чтением и confirm — имитация прямым UPDATE перед вызовом → 409); test_openapi_routes — новые пути.
• Step 2–4: падает → реализация → тесты, make check.
• Step 5: Commit feat(files): API файлов, документов и строк ОСВ; подтверждение документа.

---

Task 10: [4.2] Удаление файла

Files:

• Modify: api/src/services/primary_files.py (delete), api/src/api/endpoints/files/read.py (DELETE /files/{file_id}), api/src/repositories/files.py
• Test: api/tests/test_api_files_delete.py

Interfaces:

async def delete(self, file_id) -> None   # get_file(scope) → 404; есть document CONFIRMED → ConflictError("FILE_HAS_CONFIRMED_DOCUMENTS"); EXISTS stored_file WHERE superseded_by_file_id = id → ConflictError("FILE_SUPERSEDES_OTHERS"); before = FileOut-словарь; DELETE osv_entry (по документам файла), document, recognition_job WHERE file_id (сначала file.current_recognition_job_id = NULL + flush — FK цикл), stored_file; s3_object.expires_at = now (ASSUMPTION(OQ-22)); audit FILE_DELETE before_value=before

• Step 1: Тесты: сценарии спеки (ошибочный файл → 204, записей нет, expires_at, аудит; CONFIRMED → 409; файл, заменивший другой → 409; SUPERSEDED → 204; чужой → 404; PROCESSED без подтверждения → 204 со строками).
• Step 2–4: падает → реализация → тесты, make check.
• Step 5: Commit feat(files): удаление файла с ограничениями, expires_at у объекта S3, FILE_DELETE.

---

Task 11: [5.1] Сквозной прогон через compose

Files:

• Create: api/tests/e2e/__init__.py, api/tests/e2e/test_osv_ingest_e2e.py, api/tests/e2e/conftest.py
• Modify: Makefile (e2e: cd api && pytest -m e2e, check-full включает e2e), README.md (как поднять профиль services)

Interfaces: тест (маркеры integration, e2e, fixtures): предусловия — curl localhost:8001/healthcheck tabular-parser из профиля services (docker compose -f ../docs/dev/docker-compose.yml --profile services up -d --build tabular-parser) и NATS_ENABLED=true у тестового приложения; иначе pytest.skip("профиль services не поднят"). Шаги: полное приложение src.app.app (lifespan с реальными NATS/MinIO из .env.test, NATS_ENABLED=true через monkeypatch settings) на отдельной БД (committed_sessionmaker + seed) → POST /legal-entities/{id}/files для ЮЛ, созданного фабрикой с ИНН 3528052718 (АО «Русский бисквит», владелец ОСВ 66) с rb/2024-FY/ОСВ_66 за 2024.xls → ждать до 60 с GET /files/{id} → PROCESSED → документ 66/2024/Q4, 9 строк (10 строк снимка минус TOTAL), counterparty_check = MATCH, status = RECOGNIZED; очистка: удалить файл через API.

• Step 1: тест; Step 2: поднять профиль (make up + --profile services для tabular-parser), прогнать make e2e → PASS; без профиля → skip; make check-full зелёный.
• Step 3: Commit test(e2e): сквозной прогон загрузки ОСВ через compose с реальным tabular-parser.

---

Task 12: [5.2] Документы и make check-full

Files:

• Modify: README.md (разделы «Первичные документы: загрузка», «Статусы файла и документа», «Цикл маппинга» с настройками, «API файлов и документов», «Удаление и замена», «Ограничения»), CLAUDE.md (DBML v5.15, domain/osv_mapping.py, services/primary_files.py, services/result_mapping.py, три фоновых цикла, маркер e2e), openspec/changes/slice-2c-osv-ingest/tasks.md (все [x] кроме 5.3), openspec/changes/slice-2c-osv-ingest/design.md (рулинги контроллера)

• Modify (worktree docs): context/05-domain-rules.md (§1.3, §1.5, §3 п. 4, §9.2–9.3, новый §9.5 по факту), context/03-architecture.md (цикл маппинга), context/07-open-questions.md (закрыть OQ-67/69/70/59/109; новые OQ: удаление после консолидаций, ретеншн при удалении), context/09-delivery-plan.md (2c по факту), dev/qa-guide.md (раздел «Загрузка ОСВ и просмотр строк через API»: curl-сценарий, ожидания на ОСВ_66), openspec/changes/slice-2c-osv-ingest/tasks.md 2.1 → [x]

• Step 1: таблица «требование спеки → сценарий → тест» в отчёте задачи; пропуски — добавить тесты.

• Step 2: документы; openspec validate --strict в обоих репозиториях.

• Step 3: make check-full (compose поднят; профиль services — по возможности, иначе e2e skip) → зелёный; вывод — в ledger.

• Step 4: Commit здесь docs: README/CLAUDE.md Среза 2c, tasks.md; в docs docs: §1.3/§1.5/§3/§9 по факту маппинга, архитектура (цикл маппинга), OQ, план срезов, qa-guide.

---

Завершение

Финальное ревью всей ветки (opus): спека ↔ код по шести дельтам; инварианты (поля вне DBML v5.15, audit_event append-only, код роли, float, первичка не мутируется — osv_entry пишется один раз и только маппером); безопасность (policy на всех маршрутах файлов/документов, 404 vs 403, safe_filename, ключ результата OQ-70, лимит размера, 503 без S3); гонки (два цикла + inbox на stored_file, confirm vs автозамена); производительность маппинга (1045 строк ОСВ_60 за 2024.xls — цель < 2 с). Находки C/I/M со статусами; deferred — в PR и follow-up issue при закрытии.