fix(mapping): категории вместо текста исключений в mapping_error, вставка строк по level

M-1 (= T7.6): metrics.mapping_error виден пользователю (JobBrief.mapping_error) —
ошибка схемы даёт «результат не соответствует схеме recognition-result: ошибок N,
первая: <loc> <type>» без input_value и ссылок pydantic; исчерпавшая попытки
ошибка БД/S3 — «маппинг не удался: ошибка БД/S3 (<ExcType>), попыток N» без
[SQL: …] и параметров. Полный текст исключения — только в логе.

T7.2: черновики osv_entry сортируются устойчиво по level перед вставкой пачками —
родитель раньше потомка на любой границе пачки, независимо от порядка строк
результата.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2

api/src/services/result_mapping.py

@@ -17,6 +17,8 @@ PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recogn
 Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов ОСВ)
 — сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции
 `metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`; следующая попытка — после паузы цикла (R-15).
+`mapping_error` виден пользователю: в нём категория (схема — число ошибок и путь первой; БД/S3 — тип исключения и число
+попыток), без текста исключений pydantic/SQLAlchemy; полный текст — только в логе (M-1).
 Статус файла — через `services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг
 не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — системное событие (`user_id = NULL`).
 """
@@ -175,6 +177,16 @@ def validate_result_key(candidate: Candidate, storage_bucket: str) -> None:
 # --- шаг 2: чтение результата ---------------------------------------------------------------------------------------
 
 
+def _schema_error_summary(exc: ValidationError) -> str:
+    """M-1: для `metrics.mapping_error` (видно пользователю) — число ошибок, путь и тип первой; без `input_value` и
+    ссылок pydantic (полный текст — в логе через `__cause__` у `MappingFatal`)."""
+    errors = exc.errors(include_url=False, include_input=False, include_context=False)
+    first = errors[0] if errors else None
+    where = ".".join(map(str, first["loc"])) if first and first["loc"] else "<корень>"
+    kind = first["type"] if first else "?"
+    return f"результат не соответствует схеме recognition-result: ошибок {exc.error_count()}, первая: {where} {kind}"
+
+
 def _osv_documents(result: RecognitionResult) -> list[OsvDocument]:
     return [document for document in result.documents if isinstance(document, OsvDocument)]
 
@@ -188,7 +200,7 @@ async def _load_result(storage: ObjectStorage, candidate: Candidate) -> Recognit
     try:
         result = RecognitionResult.model_validate_json(content)
     except ValidationError as exc:
-        raise MappingFatal(f"файл результата не соответствует схеме recognition-result: {exc}") from exc
+        raise MappingFatal(_schema_error_summary(exc)) from exc
     if result.job_id != candidate.job_id or result.file_id != candidate.file_id:
         raise MappingFatal(
             f"файл результата относится к другому заданию: job_id={result.job_id}, file_id={result.file_id}"
@@ -405,6 +417,9 @@ async def map_file(  # noqa: PLR0913 - все входы маппинга фай
 
     session.add_all(documents)
     await session.flush()
+    # T7.2: родитель раньше потомка на любой границе пачки (self-FK `parent_entry_id`) — не полагаемся на порядок строк
+    # результата; сортировка устойчивая, внутри уровня порядок строк сохраняется
+    entries.sort(key=lambda values: values["level"])
     for start in range(0, len(entries), ENTRY_BATCH_SIZE):
         await session.execute(insert(OsvEntry).values(entries[start : start + ENTRY_BATCH_SIZE]))
 
@@ -558,15 +573,18 @@ async def _lock_file(session: AsyncSession, file_id: uuid.UUID) -> StoredFile |
     return (await session.execute(stmt)).scalar_one_or_none()
 
 
-async def _fail(session: AsyncSession, candidate: Candidate, message: str, now: datetime) -> None:
+async def _fail(
-    """Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего)."""
+    session: AsyncSession, candidate: Candidate, message: str, now: datetime, *, cause: BaseException | None = None
+) -> None:
+    """Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего).
+    `cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1)."""
     job = await _lock_job(session, candidate.job_id)
     if job is None:
         return
     job.metrics = {**_metrics(job), MAPPING_ERROR_KEY: _short(message)}
     await session.flush()
     changed = await set_file_status_for_job(session, job, FileStatus.ERROR, now=now)
-    logger.error(
+    logger.opt(exception=cause).error(
         "Маппинг результата невозможен: файл переведён в ERROR. error_category=mapping_fatal, job_id={job_id}, "
         "file_id={file_id}, file_changed={changed}, error={error}",
         job_id=candidate.job_id,
@@ -595,8 +613,9 @@ async def _register_failure(
         metrics[MAPPING_ATTEMPTS_KEY] = attempts
         exhausted = attempts >= max_attempts
         if exhausted:
+            # M-1: только категория и тип исключения — текст SQLAlchemy/asyncpg ([SQL: …], параметры) остаётся в логе
             metrics[MAPPING_ERROR_KEY] = _short(
-                f"маппинг не удался после {attempts} попыток: {type(exc).__name__}: {exc}"
+                f"маппинг не удался: ошибка БД/S3 ({type(exc).__name__}), попыток {attempts}"
             )
         job.metrics = metrics
         await session.flush()
@@ -674,7 +693,7 @@ async def process_once(
         await _map_candidate(session_factory, candidate, result, clock)
     except MappingFatal as exc:
         async with session_factory() as session, session.begin():
-            await _fail(session, candidate, str(exc), clock())
+            await _fail(session, candidate, str(exc), clock(), cause=exc.__cause__)
     except Exception as exc:  # сбой S3 или БД: транзакция маппинга откачена, повтор после паузы цикла (R-15)
         return await _register_failure(
             session_factory, candidate, exc, max_attempts=settings.MAPPING_MAX_ATTEMPTS, now=clock()

api/tests/test_result_mapping.py

@@ -21,6 +21,7 @@ from typing import Any
 import pytest
 from loguru import logger
 from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.contract import ack_msg_id, result_msg_id
@@ -378,12 +379,78 @@ async def test_mapping_failure_retries_then_error(
     assert await process_once(maker, storage, _settings()) is True
     job = await _job(maker, job_id)
     assert job.metrics[MAPPING_ATTEMPTS_KEY] == 3  # noqa: PLR2004 - MAPPING_MAX_ATTEMPTS по умолчанию
-    assert "подставная ошибка БД" in job.metrics[MAPPING_ERROR_KEY]
+    # M-1: пользователю — категория и число попыток, текст исключения остаётся только в логе
+    assert job.metrics[MAPPING_ERROR_KEY] == "маппинг не удался: ошибка БД/S3 (RuntimeError), попыток 3"
     assert (await _file(maker, file_id)).status is FileStatus.ERROR
     assert len(calls) == 3  # noqa: PLR2004 - три попытки
     assert await process_once(maker, storage, _settings()) is False
 
 
+_INTERNALS = ("[SQL:", "input_value", "http", "sqlalche.me", "parameters")
+
+
+async def test_schema_error_hides_input_values(committed_sessionmaker: async_sessionmaker[AsyncSession]):
+    """M-1: ошибка схемы результата — в `mapping_error` число ошибок и путь/тип первой, без `input_value` и ссылок
+    pydantic (полный текст — только в логе)."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    result = _minimal()
+    _osv(result)["rows"][1]["opening_debit"] = "секрет http://example.invalid"
+    file_id, job_id, _ = await _prepare(maker, storage, result)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    assert (await _file(maker, file_id)).status is FileStatus.ERROR
+    message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY]
+    assert message.startswith("результат не соответствует схеме recognition-result: ошибок ")
+    assert "первая: documents.0." in message
+    assert "opening_debit" in message
+    assert "секрет" not in message
+    assert not [marker for marker in _INTERNALS if marker in message]
+
+
+async def test_retryable_db_error_hides_sql(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
+):
+    """M-1: исчерпавшая попытки ошибка БД — в `mapping_error` категория, тип исключения и число попыток, без
+    `[SQL: …]`, параметров и ссылок SQLAlchemy."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    file_id, job_id, _ = await _prepare(maker, storage, _minimal())
+
+    async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
+        raise IntegrityError("INSERT INTO osv_entry (id) VALUES ($1)", {"id": "секрет"}, Exception("fk violation"))
+
+    monkeypatch.setattr(rm, "map_file", failing_map_file)
+    settings = _settings().model_copy(update={"MAPPING_MAX_ATTEMPTS": 1})
+
+    assert await process_once(maker, storage, settings) is True
+
+    assert (await _file(maker, file_id)).status is FileStatus.ERROR
+    message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY]
+    assert message == "маппинг не удался: ошибка БД/S3 (IntegrityError), попыток 1"
+    assert not [marker for marker in _INTERNALS if marker in message]
+
+
+async def test_child_before_parent_across_batches(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
+):
+    """T7.2: строки вставляются родитель раньше потомка независимо от порядка строк в результате — потомок перед
+    родителем на границе пачки не нарушает self-FK `parent_entry_id`."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    result = _minimal()
+    rows = _osv(result)["rows"]
+    rows.reverse()  # потомок (level 1) раньше корня (level 0)
+    monkeypatch.setattr(rm, "ENTRY_BATCH_SIZE", 1)  # каждая строка — отдельная пачка
+    file_id, _, _ = await _prepare(maker, storage, result)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    assert (await _file(maker, file_id)).status is FileStatus.PROCESSED
+    [document] = await _documents(maker, file_id)
+    entries = await _entries(maker, document.id)
+    root, child = entries
+    assert child.parent_entry_id == root.id
+
+
 async def test_repeated_mapping_is_idempotent(committed_sessionmaker: async_sessionmaker[AsyncSession]):
     """«Повторный маппинг идемпотентен»: остатки прерванной вставки (документ и половина строк) удаляются, у файла
     ровно один набор документов и строк."""