arm-core/openspec/changes/slice-2c-osv-ingest/design.md

Design: Срез 2c — загрузка ОСВ, цикл маппинга, документы

Context

Обследование кода 05.10.2026: загрузки нет; outbox (publish_pending) берёт задания queued через inner join stored_file/s3_object; inbox (handle_message) в одной короткой транзакции под FOR UPDATE фиксирует статус задания, регистрирует s3_object результата и вызывает ResultMapper (NoopMapper); файл результата из S3 не читается (S3Storage.get_object отдаёт сырой response); CommittedJobs.cleanup() не знает document/osv_entry; FakeStorage умеет только put_object. Контракт v1.1 в api/contracts/ (PR #13). DBML v5.14: osv_data_type без NU, document уникален по (stored_file_id, document_type, osv_account_code), stored_file.content_hash без UNIQUE. Макет: загрузка из drawer организации, тип и период не выбираются (40), удаление — иконка в строке (39). Решения владельца 05.10.2026: OQ-01 (все строки, без «Итого»), OQ-02 (квартальные, сверять по данным), OQ-05 (RECOGNIZED нужен), OQ-45 (NU хранить), автозамена при совпадении счёта и периода.

Goals / Non-Goals

Goals: сквозной путь файл → osv_entry без участия человека; inbox остаётся быстрым; повторный запуск маппинга безопасен; все статусы файла и документа объяснимы по карточке файла. Non-Goals: см. proposal.

Decisions

D-1. Маппинг — отдельный цикл, признак «ещё не смапплено» — статус файла

run_mapping_loop(session_factory, storage, settings) в lifespan рядом с публикатором и вахтёром. Кандидат: recognition_job.status IN (done, needs_check) ⋈ stored_file (status = PROCESSING, current_recognition_job_id = job.id), ORDER BY finished_at, id LIMIT 1 FOR UPDATE OF job SKIP LOCKED. Сначала короткая транзакция «взять кандидата и прочитать ключ результата», затем чтение S3 без транзакции, затем одна транзакция маппинга. Новых колонок нет (DBML не меняется ради этого). Альтернатива — маппинг в inbox: блокирует ack на время вставки тысячи строк (OQ-69). Цена ошибки: вернуть вызов маппера в inbox — одна строка.

D-2. Повторы и ошибки маппинга

Счётчик recognition_job.metrics.mapping_attempts (jsonb, как republish_count). Исключение маппинга → rollback, mapping_attempts + 1, лог; ≥ MAPPING_MAX_ATTEMPTS (3) → stored_file.status = ERROR, metrics.mapping_error. Ошибки, которые повторять бессмысленно (ключ результата вне results/{file_id}/{job_id}/, объект не найден, JSON не проходит схему v1.1, в результате нет osvDocument), → сразу ERROR с причиной. Файл в ERROR показывается с error_code/error_message задания или metrics.mapping_error.

D-3. Транзакционность и идемпотентность

Маппинг одного файла — одна транзакция: document×N, osv_entry×M (пакетная вставка insert().values([...])), recognition_job.group_schema, stored_file.status = PROCESSED, автозамена, аудит FILE_REPLACE. Перед вставкой — DELETE osv_entry/document WHERE stored_file_id = file (защита от половинчатого состояния после сбоя; строки ещё не CONFIRMED, потому что файл был PROCESSING). document по индексу (stored_file_id, document_type, osv_account_code) — второй лист с тем же счётом в документ не превращается (D-7).

D-4. Загрузка

Один эндпоинт на несколько файлов; каждый файл — своя транзакция (частичный успех: ответ {"accepted": [...], "rejected": [{"name", "code", "message"}]}). Шаги на файл: safe_filename; размер ≤ UPLOAD_MAX_BYTES (50 МиБ, как PARSE_MAX_BYTES парсера) иначе 413/код FILE_TOO_LARGE; тип по расширению, сверка с сигнатурой (zip/BIFF8/PDF/PNG/ JPEG), несовпадение → UNSUPPORTED_FILE; sha256; дедуп: stored_file того же ЮЛ с тем же content_hash и status != SUPERSEDED → DUPLICATE_FILE с file_id существующего; job_id = uuid4() заранее, ключ uploads/{file_id}/{job_id}/{имя}; S3Object (flush) → put_object → StoredFile(UPLOADED, version_no=1) → RecognitionJob(queued, pipeline по типу, request_subject, schema_version=CONTRACT_VERSION, max_attempts 3/1) → stored_file.current_recognition_job_id. Без storage (dev без S3) — 503 STORAGE_UNAVAILABLE. Аудит FILE_UPLOAD на каждый принятый файл.

D-5. Статусы файла (§9.2) и кто их ставит

Событие	Кто	stored_file.status
загрузка	эндпоинт	UPLOADED
ack сервиса (processing)	inbox	PROCESSING
результат error; вахтёр timeout; публикатор dlq	inbox / вахтёр / публикатор	ERROR
результат done/needs_check	inbox	без изменений (PROCESSING), подхватит цикл
маппинг успешен	цикл	PROCESSED
маппинг невозможен / исчерпаны повторы	цикл	ERROR
автозамена	цикл (после маппинга нового)	SUPERSEDED у прежних
Переходы — только через stored_file.current_recognition_job_id = job.id (чужое/старое задание файл не трогает).

D-6. Маппинг строк (§1.3)

domain/osv_mapping.py (чистые функции): normalize_account_code("60.1") = "60.01" (числовые сегменты → 2 цифры, третий сегмент как есть 76.07.1, буквенные сегменты без изменений 76.АВ; первый сегмент нечисловой — account_root = None, ASSUMPTION(OQ-103)); account_root = первый сегмент (001 для забалансовых); build_entries(doc: OsvDocument, chart: Mapping[str, uuid], resolve: Callable[[row], uuid | None]) -> list[EntryDraft]: пропуск TOTAL; account_code = account_code_raw строки (у не-счётов — унаследованный, парсер его уже проставляет); account_id = chart.get( normalize(account_code)); parent_entry_id через карту row_number → id (id генерируются заранее uuid4); data_type как есть (NU допустим после v5.15); суммы — Decimal из контракта без изменений; counterparty_* как пришли; is_modified = false.

D-7. Документ, предупреждения, статус (§9.3)

На osvDocument: osv_account_code = normalize(header.account_code_raw), period_*, is_cumulative по шапке, затем проверка по данным (D-9), locator. Повтор счёта в том же файле → документ не создаётся, предупреждение sheet=<locator>: счёт <код> уже есть в листе <locator первого>, первый документ → NEEDS_CHECK. Статус: RECOGNIZED, если job.status = done и checks без нарушений и counterparty_check != MISMATCH и нет предупреждений маппинга; иначе NEEDS_CHECK. Предупреждения маппинга пишутся в recognition_job.metrics.mapping_warnings (jsonb, список строк) — отдельной колонки в DBML нет, а карточке файла они нужны.

D-8. Резолв контрагента (§3 п. 4) и проверка шапки

repositories/counterparties.py:CounterpartyIndex — одним запросом на файл: legal_entity(inn, name) + все алиасы → словари inn → id, normalize_alias(name) → {id…} (имя ЮЛ нормализуется той же функцией, что алиас), alias → {id…}. Резолв строки (только COUNTERPARTY, data_type ∈ {BU, SUM}): counterparty_inn → inn; иначе точное имя (после strip) → по точному legal_entity.name; иначе normalize_alias(counterparty_name) → по нормализованным именам и алиасам; на любом шаге > 1 кандидата → None. Без policy (контрагент может быть ЮЛ другого филиала). counterparty_check: header.organization_inn есть → MATCH если равен inn владельца, иначе MISMATCH; ИНН нет → normalize_alias(header.organization_name) ∈ нормализованные имя/алиасы владельца → MATCH, иначе MISMATCH; пустая шапка → UNKNOWN. period_check = UNKNOWN (нет версии консолидации; Срез 4 пересчитает).

D-9. is_cumulative по данным (OQ-02, ASSUMPTION)

Если шапка дала Q2|Q3|Q4 с is_cumulative = false: ищется документ того же ЮЛ, того же osv_account_code, period_year, period_quarter = Q1, файл не SUPERSEDED; если у корневой строки opening_debit/credit совпадают с Q1 (±0.01) — is_cumulative = true и предупреждение период …: нарастающая по данным Q1. Нет Q1 — остаётся по тексту.

D-10. Автозамена и удаление

После PROCESSED: для каждого нового документа — stored_file того же ЮЛ, status != SUPERSEDED, id != new, с document того же document_type, osv_account_code, period_year, period_quarter → SUPERSEDED, superseded_by_file_id = new, updated_at; у нового version_no = max(version_no заменённых) + 1; аудит FILE_REPLACE (object_id старого, after_value новый, description с перечнем заменённых документов и пометкой confirmed=true, если был CONFIRMED; user_id = NULL — системное действие, инициатор загрузки в description). DELETE /files/{id} (DOCUMENT_EDIT, policy): есть CONFIRMED документ → 409 FILE_HAS_CONFIRMED_DOCUMENTS; иначе osv_entry, document, recognition_job (все попытки), stored_file удаляются, s3_object.expires_at = now (объект физически не удаляется — ретеншн OQ-22), аудит FILE_DELETE с before_value. Файл SUPERSEDED удалить можно (он ни на что не влияет). superseded_by_file_id у заменённых обнуляется при удалении нового? — нет: удаление файла, который кого-то заменил, запрещено тем же 409 с кодом FILE_SUPERSEDES_OTHERS (цепочка версий не рвётся).

D-11. API чтения

GET /legal-entities/{id}/files?include_superseded=false → Page[FileOut] (FileOut: файл, job{status, error_code, error_message, metrics.warnings, mapping_warnings}, documents[DocumentBrief]); GET /files/{id} → FileOut; GET /documents/{id} → DocumentOut (+ checks из результата? — результат в S3; документ хранит только статус и проверки, поэтому checks парсера отдаются из recognition_job.metrics — не хранятся; в DocumentOut — period_check, counterparty_check, status, group_schema по locator из job.group_schema); GET /documents/{id}/entries?limit≤2000 → Page[EntryOut] в порядке row_number (дерево восстанавливается по parent_entry_id); POST /documents/{id}/confirm → CONFIRMED, confirmed_by/at, аудит DOCUMENT_CONFIRM; из CONFIRMED повторно — 409. Видимость — через ЮЛ-владельца (legal_entities_visible), чужое → 404.

D-12. S3

S3Storage.get_bytes(key) -> bytes (чтение response.content.read() + закрытие), put_object как есть; FakeStorage получает get_bytes и objects. Чтение результата — по job.result_s3_object.key (бакет сверяется с storage.bucket).

D-13. Тесты

tests/factories.py: make_stored_file, make_job, make_document; CommittedJobs.cleanup() расширен на osv_entry/document. Unit: domain/osv_mapping.py, резолв на словарях. Integration: загрузка (типы, лимит, дубль, частичный успех, policy), статусы по ack/результату/таймауту, цикл маппинга на результате из golden-снимка rb/2024-FY/ОСВ_66 за 2024.xls (FakeStorage с JSON), повторы и ERROR, автозамена с CONFIRMED, удаление, API. fixtures: маппинг всех 82 golden-снимков (../fixtures/golden/expected_osv_parse/**) в БД — инварианты: число строк = rows без TOTAL, у каждого не-корня есть родитель, account_id заполнен у всех счетов из плана, суммы совпадают с снимком. Сквозной прогон через compose (профиль services с реальным tabular-parser) — отдельный integration-тест, помеченный e2e, запускается в make check-full при поднятом профиле (иначе skip с причиной).

Risks / Trade-offs

• [Два цикла и inbox пишут в stored_file] → только по current_recognition_job_id; тесты гонок (вахтёр vs цикл).
• [Удаление и повторная вставка при повторе маппинга] → файл не CONFIRMED в PROCESSING, данных пользователя нет.
• [Резолв по имени даёт ложные совпадения у физлиц/ИП] → только точное имя или алиас, никакой нечёткости; ручное сопоставление — Срез 4.
• [Автозамена трогает CONFIRMED] → решение владельца; предупреждение в аудите и в ответе загрузки нет (асинхронно) — видно в карточке файла.
• [e2e-тест зависит от профиля services] → skip с причиной, не падение.

Migration Plan

0003_dbml_v5_15: ALTER TYPE osv_data_type ADD VALUE 'NU' AFTER 'BU' (downgrade — пересоздание типа, как в 0002); UPDATE app_user SET email = lower(email) (при дублях по регистру миграция падает — осознанно; downgrade данные не возвращает). HEAD в тестах миграций → 0003_dbml_v5_15.

Open Questions

Не блокируют: OQ-22 (ретеншн — expires_at при удалении), OQ-55 (аудит обмена — нет), OQ-73 (replay — нет), OQ-100/103 (допущения), OQ-109 (размер объекта до скачивания — у ядра лимит на загрузке, парсер — как есть).

Decisions made without discussion

№	Дата	Решение	Обоснование	Цена ошибки
1	2026-10-05	D-1: цикл маппинга отделён от inbox, признак — stored_file.status = PROCESSING	OQ-69; без новых колонок	вернуть в inbox
2	2026-10-05	D-2: 3 попытки маппинга, счётчик в metrics; неповторяемые ошибки → ERROR сразу	как republish_count	числа в Settings
3	2026-10-05	D-3: перед вставкой удаляются строки/документы файла (идемпотентность)	файл в PROCESSING не содержит правок	нет
4	2026-10-05	D-4: частичный успех, дедуп по sha256 в пределах ЮЛ, job_id заранее для ключа S3, лимит 50 МиБ	макет: несколько файлов; схема ключей 03-architecture	формат ответа
5	2026-10-05	D-5: ERROR файла ставят inbox/вахтёр/публикатор при финальных ошибках задания	иначе файл вечно PROCESSING	нет
6	2026-10-05	D-6: нормализация кода — числовые сегменты до 2 цифр, буквенные как есть; account_root = None при нечисловом первом сегменте	§1.3; OQ-103	правка функции
7	2026-10-05	D-7: предупреждения маппинга — в metrics.mapping_warnings; повтор счёта в файле → первый лист + NEEDS_CHECK	колонки нет в DBML; уникальность документа	DBML v5.16 с колонкой
8	2026-10-05	D-8: резолв без policy; имя ЮЛ нормализуется как алиас; period_check = UNKNOWN до Среза 4	контрагент может быть чужого филиала; версии нет	пересчёт в Срезе 4
9	2026-10-05	D-9: is_cumulative по данным — только сравнение с Q1 того же года	OQ-02	расширить правило
10	2026-10-05	D-10: удаление запрещено при CONFIRMED и если файл заменил другие; S3-объект получает expires_at	ANL-020, OQ-22	ослабить проверку
11	2026-10-05	D-11: checks парсера в БД не хранятся, карточка документа отдаёт статус и проверки ядра	DBML	колонка/чтение из S3
12	2026-10-05	D-13: e2e через профиль services помечен e2e и пропускается без профиля	нет CI	—
13	2026-10-05	v5.15: email в нижнем регистре вместо expression-индекса lower(email)	сверка ORM↔DBML не умеет expression-индексы	индекс в v5.16
14	2026-10-05	R-1: golden-прогон 82 снимков — второй коммит Task 8 (tests/test_mapping_golden.py)	в плане нет отдельной задачи под tasks 3.4; нужен полный map_file	перенос теста
15	2026-10-05	R-2: make_entries_from_result создаётся в Task 9, не в Task 2	зависит от build_entries	нет
16	2026-10-05	R-3: EntryDraft.counterparty_id, проставляется в цикле маппинга через CounterpartyIndex.resolve для COUNTERPARTY + BU/SUM; build_entries без resolve	спека «Резолв контрагента» не сводилась с сигнатурой плана	сигнатура на callable (D-6)
17	2026-10-05	R-4: inbox при done/needs_check переводит файл UPLOADED → PROCESSING (текущее задание)	core NATS без JetStream: ack может потеряться, иначе файл с готовым результатом не попадёт в маппинг	убрать ветку
18	2026-10-05	R-5: ack/результат ставят PROCESSING только из UPLOADED; ERROR/SUPERSEDED/PROCESSED не перезаписываются	запоздавший ack не воскрешает файл после timeout	параметр only_from
19	2026-10-05	R-6: DocumentOut.warnings — metrics.mapping_warnings задания целиком, group_schema — по locator документа	в DBML нет привязки предупреждения к документу	фильтрация по листу
20	2026-10-05	R-7: в Task 1 правится workspace CLAUDE.md (вне git): только версия DBML	по плану Task 1	откат текста
21	2026-10-05	R-8 (Task 3): 503 STORAGE_UNAVAILABLE через DomainError ({"detail":{"code","message"}}), а не строкой HTTPException; строка остаётся только у 413 внутри read_limited (превращается в rejected[FILE_TOO_LARGE])	спека и Global Constraints плана: ошибки — DomainError; бриф противоречил	формат тела 503
22	2026-10-05	R-9 (Task 3): сбой S3 на одном файле → этот файл в rejected с STORAGE_UNAVAILABLE, откатывается его SAVEPOINT, принятые остаются; если отклонены все — 422	модель частичного успеха D-4; клиент видит причину по файлу	вернуть 503 при отказе всех
23	2026-10-05	R-10 (Task 3): загрузка файла — SAVEPOINT внутри транзакции запроса (не отдельная транзакция на файл, как в тексте спеки); недоменная ошибка БД валит весь запрос	указано в брифе плана; атомарность ответа	отдельные сессии на файл
24	2026-10-05	R-11 (Task 3): pg_advisory_xact_lock(ЮЛ, sha256) против двойного клика — дубль не проскакивает; requested_at = now у задания; size_bytes у файла; .jpeg → JPG; operation_id у маршрута (OpenAPI)	гонка дедупа подтверждена тестом; FIFO outbox идёт по requested_at	убрать блокировку