feat(mapping): автозамена прежних версий файла после маппинга (FILE_REPLACE)

`supersede_previous` вызывается в `map_file` после `PROCESSED`, в той же транзакции (D-10):
файлы того же ЮЛ (не SUPERSEDED, не сам новый) с документом тех же type/счёт/год/квартал →
SUPERSEDED, `superseded_by_file_id`, `updated_at`; новому — `version_no` = max + 1;
одно системное событие `FILE_REPLACE` на заменённый файл (user_id NULL, инициатор загрузки,
перечень документов и `confirmed=true|false` в description). Документы и строки заменённых
файлов сохраняются. Автозамены одного ЮЛ сериализуются advisory-блокировкой, заменяемые
файлы блокируются FOR UPDATE в порядке id, документы перечитываются после блокировки.

Тесты: сценарий спеки «Новая ОСВ 60 за 1 кв. 2025» (CONFIRMED и RECOGNIZED), другой
счёт/год/квартал, другое ЮЛ, идемпотентность после сбоя, цепочка v1→v2→v3, одно событие на файл.

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

api/src/services/result_mapping.py

@@ -11,13 +11,14 @@ PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recogn
 3. транзакция маппинга: задание и файл перечитываются под блокировкой (порядок «задание → файл», как у inbox,
    вахтёра и публикатора, — без взаимоблокировок); файл уже не `PROCESSING` или задание уже не текущее (результат
    старого задания, Review Focus 1) — ничего не меняется. Иначе `map_file`: очистка документов и строк файла,
-   документы и строки, `group_schema`, предупреждения, `PROCESSED`.
+   документы и строки, `group_schema`, предупреждения, `PROCESSED`, автозамена прежних версий (`supersede_previous`,
+   D-10, аудит `FILE_REPLACE`).
 
 Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов ОСВ)
 — сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции
 `metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`; следующая попытка — после паузы цикла (R-15).
 Статус файла — через `services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг
-не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — отдельно.
+не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — системное событие (`user_id = NULL`).
 """
 
 import asyncio
@@ -31,7 +32,7 @@ from typing import Any
 
 from loguru import logger
 from pydantic import ValidationError
-from sqlalchemy import Integer, delete, func, insert, select
+from sqlalchemy import Integer, Select, delete, func, insert, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.contract.result import OsvDocument, RecognitionResult
@@ -48,7 +49,10 @@ from src.domain.osv_mapping import (
 from src.models.accounting import ChartOfAccount
 from src.models.directories import LegalEntity
 from src.models.enums import (
+    AuditAction,
+    AuditSection,
     DocumentCheckResult,
+    DocumentStatus,
     DocumentType,
     FileStatus,
     OsvDataType,
@@ -59,6 +63,7 @@ from src.models.enums import (
 from src.models.files import Document, RecognitionJob, S3Object, StoredFile
 from src.models.osv import OsvEntry
 from src.repositories.counterparties import CounterpartyIndex
+from src.services.audit import AuditService
 from src.services.file_status import set_file_status_for_job
 
 __all__ = [
@@ -72,6 +77,7 @@ __all__ = [
     "pick_candidate",
     "process_once",
     "run_mapping_loop",
+    "supersede_previous",
     "validate_result_key",
 ]
 
@@ -85,6 +91,8 @@ MAPPABLE_JOB_STATUSES = (RecognitionStatus.done, RecognitionStatus.needs_check)
 CUMULATIVE_CHECK_QUARTERS = frozenset({PeriodQuarter.Q2, PeriodQuarter.Q3, PeriodQuarter.Q4})
 RESOLVED_DATA_TYPES = frozenset({OsvDataType.BU, OsvDataType.SUM})  # резолв контрагента — §3 п. 4, R-3
 ROOT_LEVEL = 0
+STORED_FILE_OBJECT_TYPE = "stored_file"  # `audit_event.object_type` события FILE_REPLACE
+SUPERSEDE_LOCK_PREFIX = "supersede"  # пространство ключей advisory-блокировки автозамены в пределах ЮЛ
 _FORBIDDEN_KEY_SEGMENTS = frozenset({"", ".", ".."})
 
 Clock = Callable[[], datetime]
@@ -315,11 +323,11 @@ async def map_file(  # noqa: PLR0913 - все входы маппинга фай
     index: CounterpartyIndex,
     now: datetime,
 ) -> list[str]:
-    """Маппинг результата в транзакции вызывающего (D-3, D-6…D-9); возвращает предупреждения маппинга файла.
+    """Маппинг результата в транзакции вызывающего (D-3, D-6…D-10); возвращает предупреждения маппинга файла.
 
     Очистка прежних документов/строк файла → документ на каждый лист ОСВ (повтор счёта — предупреждение) → строки
     пакетами по `ENTRY_BATCH_SIZE` → `job.group_schema`, `metrics.mapping_warnings` → файл `PROCESSED` (только из
-    `PROCESSING` и только для текущего задания). Без commit.
+    `PROCESSING` и только для текущего задания) → автозамена прежних версий (`supersede_previous`). Без commit.
     """
     sheets = _osv_documents(result)
     if not sheets:
@@ -408,9 +416,128 @@ async def map_file(  # noqa: PLR0913 - все входы маппинга фай
     ):
         raise RuntimeError(f"файл {file.id} не переведён в PROCESSED: не PROCESSING или задание {job.id} не текущее")
     await session.refresh(file, ["status", "updated_at"])
+    await supersede_previous(session, new_file=file, new_documents=documents, uploader_id=file.uploaded_by, now=now)
     return warnings
 
 
+# --- автозамена прежних версий --------------------------------------------------------------------------------------
+
+
+DocumentKey = tuple[DocumentType, str | None, int, PeriodQuarter]
+
+
+def _document_key(document: Document) -> DocumentKey | None:
+    """Ключ совпадения документов для автозамены (D-10); документ без года или квартала в замене не участвует."""
+    if document.period_year is None or document.period_quarter is None:
+        return None
+    return (document.document_type, document.osv_account_code, document.period_year, document.period_quarter)
+
+
+def _describe_document(document: Document) -> str:
+    period = f"{document.period_year} {document.period_quarter.value if document.period_quarter else ''}".strip()
+    return f"{document.document_type.value} {document.osv_account_code or ''} {period} ({document.status.value})"
+
+
+async def supersede_previous(
+    session: AsyncSession,
+    *,
+    new_file: StoredFile,
+    new_documents: Sequence[Document],
+    uploader_id: uuid.UUID | None,
+    now: datetime,
+) -> list[StoredFile]:
+    """D-10, spec primary-files «Автозамена прежних версий»: файлы того же ЮЛ (не `SUPERSEDED`, не сам новый) с
+    документом тех же `document_type`, `osv_account_code`, `period_year`, `period_quarter` → `SUPERSEDED`,
+    `superseded_by_file_id = new_file.id`; у нового файла `version_no` = максимум заменённых + 1. На каждый
+    заменённый файл — одно системное событие `FILE_REPLACE` (`user_id = NULL`, инициатор загрузки — в `description`,
+    `confirmed=true`, если среди заменённых документов есть `CONFIRMED`; решение владельца 05.10.2026). Документы и
+    строки заменённых файлов не трогаются (история). Возвращает заменённые файлы; в транзакции вызывающего, без commit.
+
+    Блокировки: advisory-блокировка ЮЛ сериализует автозамены одного ЮЛ (два новых файла одного документа, смапленные
+    параллельно, иначе не увидели бы друг друга и оба остались бы действующими); заменяемые файлы — `FOR UPDATE` в
+    порядке `id` (сериализация с подтверждением документа, которое перечитывает файл под `FOR UPDATE`).
+    """
+    keys = {key for key in map(_document_key, new_documents) if key is not None}
+    if not keys:
+        return []
+    lock_key = func.hashtextextended(f"{SUPERSEDE_LOCK_PREFIX}:{new_file.legal_entity_id}", 0)
+    await session.execute(select(func.pg_advisory_xact_lock(lock_key)))
+
+    def matching(stmt: Select[tuple[Document]]) -> Select[tuple[Document]]:
+        return stmt.join(StoredFile, StoredFile.id == Document.stored_file_id).where(
+            StoredFile.legal_entity_id == new_file.legal_entity_id,
+            StoredFile.id != new_file.id,
+            StoredFile.status != FileStatus.SUPERSEDED,
+            Document.document_type.in_(sorted({key[0] for key in keys})),
+            Document.period_year.in_(sorted({key[2] for key in keys})),
+        )
+
+    candidates = (await session.execute(matching(select(Document)))).scalars().all()
+    file_ids = sorted({document.stored_file_id for document in candidates if _document_key(document) in keys})
+    if not file_ids:
+        return []
+    locked = (
+        (
+            await session.execute(
+                select(StoredFile)
+                .where(StoredFile.id.in_(file_ids), StoredFile.status != FileStatus.SUPERSEDED)
+                .order_by(StoredFile.id)
+                .with_for_update()
+                .execution_options(populate_existing=True)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    if not locked:
+        return []
+    ids = [old.id for old in locked]
+    # документы перечитываются ПОСЛЕ блокировки файлов: подтверждение, зафиксированное до неё, попадёт в `confirmed`
+    fresh = await session.execute(
+        matching(select(Document)).where(Document.stored_file_id.in_(ids)).execution_options(populate_existing=True)
+    )
+    replaced_documents: dict[uuid.UUID, list[Document]] = {}
+    for document in fresh.scalars():
+        if _document_key(document) in keys:
+            replaced_documents.setdefault(document.stored_file_id, []).append(document)
+    await session.execute(
+        update(StoredFile)
+        .where(StoredFile.id.in_(ids), StoredFile.status != FileStatus.SUPERSEDED)
+        .values(status=FileStatus.SUPERSEDED, superseded_by_file_id=new_file.id, updated_at=now)
+        .execution_options(synchronize_session=False)
+    )
+    new_file.version_no = max(old.version_no for old in locked) + 1
+    new_file.updated_at = now
+    await session.flush()
+
+    audit = AuditService(session)
+    uploader = f"user:{uploader_id}" if uploader_id is not None else "неизвестен"
+    for old in locked:
+        await session.refresh(old, ["status", "superseded_by_file_id", "updated_at"])
+        documents = sorted(replaced_documents.get(old.id, []), key=_describe_document)
+        confirmed = any(document.status is DocumentStatus.CONFIRMED for document in documents)
+        await audit.record(
+            action=AuditAction.FILE_REPLACE,
+            section=AuditSection.PRIMARY_DOCUMENTS,
+            object_type=STORED_FILE_OBJECT_TYPE,
+            object_id=old.id,
+            user_id=None,
+            after_value={"file_id": new_file.id, "version_no": new_file.version_no},
+            description=(
+                f"заменён файлом {new_file.original_name} (загрузил {uploader}); "
+                f"документы: {', '.join(map(_describe_document, documents))}; "
+                f"confirmed={'true' if confirmed else 'false'}"
+            ),
+        )
+    logger.info(
+        "Автозамена: прежние версии файла заменены. file_id={file_id}, version_no={version_no}, replaced={replaced}",
+        file_id=new_file.id,
+        version_no=new_file.version_no,
+        replaced=[str(old_id) for old_id in ids],
+    )
+    return list(locked)
+
+
 # --- проход и цикл --------------------------------------------------------------------------------------------------
 
 

api/tests/test_result_mapping.py

@@ -24,9 +24,13 @@ from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.core.config import Settings, get_settings
+from src.models.access import AppUser
 from src.models.accounting import ChartOfAccount
+from src.models.audit_ex import AuditEvent
 from src.models.directories import LegalEntity
 from src.models.enums import (
+    AuditAction,
+    AuditSection,
     DocumentCheckResult,
     DocumentStatus,
     DocumentType,
@@ -47,6 +51,7 @@ from src.services.result_mapping import (
     MappingFatal,
     process_once,
     run_mapping_loop,
+    supersede_previous,
     validate_result_key,
 )
 from tests.conftest import FakeStorage
@@ -104,13 +109,16 @@ async def _prepare(  # noqa: PLR0913 - каждое поле подготовк
     job_status: RecognitionStatus = RecognitionStatus.done,
     file_status: FileStatus = FileStatus.PROCESSING,
     key: str | None = None,
+    uploaded_by: uuid.UUID | None = None,
 ) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]:
     """ЮЛ-владелец, файл `PROCESSING` с текущим заданием `done`, результат в `storage`; `(file_id, job_id, entity_id)`.
 
     `job_id`/`file_id` результата подставляются; `key` — подменить ключ `s3_object` результата (OQ-70)."""
     async with maker() as session, session.begin():
         owner = entity or await make_legal_entity(session, name=owner_name, inn=owner_inn)
-        file, job = await make_stored_file(session, entity=owner, status=file_status, job_status=job_status)
+        file, job = await make_stored_file(
+            session, entity=owner, status=file_status, job_status=job_status, uploaded_by=uploaded_by
+        )
         job.finished_at = _now()
         job.metrics = copy.deepcopy(result.get("metrics"))  # как inbox: метрики сервиса из конверта
         payload = copy.deepcopy(result) | {"job_id": str(job.id), "file_id": str(file.id)}
@@ -722,3 +730,253 @@ async def test_run_mapping_loop_survives_errors(monkeypatch: pytest.MonkeyPatch)
 
     assert sleeps == [settings.MAPPING_POLL_SECONDS, 0]
     assert len(records) == 1
+
+
+# --- автозамена прежних версий (spec primary-files «Автозамена прежних версий», audit-trail; design D-10) ----------
+
+OldDocument = tuple[str, int, PeriodQuarter, DocumentStatus]  # счёт, год, квартал, статус документа прежнего файла
+OSV_60_2025_Q1: OldDocument = ("60", 2025, PeriodQuarter.Q1, DocumentStatus.CONFIRMED)
+
+
+async def _old_file(  # noqa: PLR0913 - каждое поле прежнего файла задаётся тестом явно
+    maker: async_sessionmaker[AsyncSession],
+    entity_id: uuid.UUID,
+    documents: list[OldDocument],
+    *,
+    version_no: int = 1,
+    status: FileStatus = FileStatus.PROCESSED,
+    superseded_by: uuid.UUID | None = None,
+) -> uuid.UUID:
+    """Прежний файл ЮЛ с документами ОСВ и строкой ОСВ на каждый документ (история должна сохраниться)."""
+    async with maker() as session, session.begin():
+        entity = await session.get(LegalEntity, entity_id)
+        assert entity is not None
+        file, job = await make_stored_file(
+            session, entity=entity, status=status, job_status=RecognitionStatus.done, original_name="old.xls"
+        )
+        file.version_no = version_no
+        file.superseded_by_file_id = superseded_by
+        for account, year, quarter, document_status in documents:
+            document = await make_document(
+                session, file=file, job=job, account=account, year=year, quarter=quarter, status=document_status
+            )
+            session.add(
+                OsvEntry(
+                    recognition_job_id=job.id,
+                    document_id=document.id,
+                    owner_legal_entity_id=entity_id,
+                    row_number=1,
+                    level=0,
+                    row_value=account,
+                )
+            )
+        return file.id
+
+
+async def _entity(
+    maker: async_sessionmaker[AsyncSession], name: str = OWNER_NAME, inn: str = "7700000000"
+) -> uuid.UUID:
+    async with maker() as session, session.begin():
+        return (await make_legal_entity(session, name=name, inn=inn)).id
+
+
+async def _any_user(maker: async_sessionmaker[AsyncSession]) -> uuid.UUID:
+    async with maker() as session:
+        return (await session.execute(select(AppUser.id).order_by(AppUser.email).limit(1))).scalar_one()
+
+
+async def _replace_events(maker: async_sessionmaker[AsyncSession]) -> list[AuditEvent]:
+    async with maker() as session:
+        rows = await session.execute(
+            select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_REPLACE).order_by(AuditEvent.occurred_at)
+        )
+        return list(rows.scalars())
+
+
+async def _active_files(maker: async_sessionmaker[AsyncSession], entity_id: uuid.UUID) -> set[uuid.UUID]:
+    """Файлы ЮЛ, которые список файлов показывает по умолчанию (без `include_superseded`)."""
+    async with maker() as session:
+        rows = await session.execute(
+            select(StoredFile.id).where(
+                StoredFile.legal_entity_id == entity_id, StoredFile.status != FileStatus.SUPERSEDED
+            )
+        )
+        return set(rows.scalars())
+
+
+@pytest.mark.parametrize(
+    ("old_status", "confirmed"),
+    [(DocumentStatus.CONFIRMED, "confirmed=true"), (DocumentStatus.RECOGNIZED, "confirmed=false")],
+)
+async def test_new_osv_60_supersedes_previous_file(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], old_status: DocumentStatus, confirmed: str
+):
+    """«Новая ОСВ 60 за 1 кв. 2025»: прежний файл (документ `CONFIRMED` тоже, решение владельца 05.10.2026) —
+    `SUPERSEDED`, `superseded_by_file_id` = новый, у нового `version_no = 2`; `FILE_REPLACE` — системное событие
+    (`user_id` NULL, инициатор загрузки — в `description`); документы и строки прежнего файла сохранены."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    entity_id = await _entity(maker)
+    uploader = await _any_user(maker)
+    old_id = await _old_file(maker, entity_id, [("60", 2025, PeriodQuarter.Q1, old_status)])
+    async with maker() as session:
+        entity = await session.get(LegalEntity, entity_id)
+    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=uploader)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    old, new = await _file(maker, old_id), await _file(maker, new_id)
+    assert old.status is FileStatus.SUPERSEDED
+    assert old.superseded_by_file_id == new_id
+    assert old.updated_at is not None
+    assert (new.status, new.version_no, new.superseded_by_file_id) == (FileStatus.PROCESSED, 2, None)
+    assert await _active_files(maker, entity_id) == {new_id}
+    [old_document] = await _documents(maker, old_id)  # история: документ и строки прежнего файла на месте
+    assert old_document.status is old_status
+    assert len(await _entries(maker, old_document.id)) == 1
+
+    [event] = await _replace_events(maker)
+    assert event.user_id is None
+    assert event.section is AuditSection.PRIMARY_DOCUMENTS
+    assert (event.object_type, event.object_id) == ("stored_file", old_id)
+    assert event.after_value == {"file_id": str(new_id), "version_no": 2}
+    assert event.description is not None
+    assert f"user:{uploader}" in event.description
+    assert f"OSV 60 2025 Q1 ({old_status.value})" in event.description
+    assert confirmed in event.description
+
+
+@pytest.mark.parametrize(
+    "document",
+    [
+        ("62", 2025, PeriodQuarter.Q1, DocumentStatus.CONFIRMED),  # другой счёт
+        ("60", 2024, PeriodQuarter.Q1, DocumentStatus.CONFIRMED),  # тот же счёт, другой год
+        ("60", 2025, PeriodQuarter.Q2, DocumentStatus.CONFIRMED),  # тот же счёт, другой квартал
+    ],
+)
+async def test_other_document_is_not_superseded(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], document: OldDocument
+):
+    """Два разных счёта (или периода) не заменяют друг друга: прежний файл `PROCESSED`, у нового `version_no = 1`."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    entity_id = await _entity(maker)
+    old_id = await _old_file(maker, entity_id, [document])
+    async with maker() as session:
+        entity = await session.get(LegalEntity, entity_id)
+    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    old, new = await _file(maker, old_id), await _file(maker, new_id)
+    assert (old.status, old.superseded_by_file_id) == (FileStatus.PROCESSED, None)
+    assert (new.status, new.version_no) == (FileStatus.PROCESSED, 1)
+    assert await _replace_events(maker) == []
+
+
+async def test_other_legal_entity_is_not_superseded(committed_sessionmaker: async_sessionmaker[AsyncSession]):
+    """Тот же документ у другого ЮЛ не затрагивается: замена — только в пределах ЮЛ-владельца."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    other_id = await _entity(maker, name="ООО «Другое ЮЛ»", inn="7799999999")
+    old_id = await _old_file(maker, other_id, [OSV_60_2025_Q1])
+    new_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000")
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    assert (await _file(maker, old_id)).status is FileStatus.PROCESSED
+    assert (await _file(maker, new_id)).version_no == 1
+    assert await _replace_events(maker) == []
+
+
+async def test_supersede_is_idempotent(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
+):
+    """Сбой после автозамены откатывает и замену, и аудит; повтор маппинга даёт ровно одно `FILE_REPLACE`. Повторный
+    вызов `supersede_previous` для того же файла ничего не находит (заменённый уже `SUPERSEDED`)."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    entity_id = await _entity(maker)
+    old_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1])
+    async with maker() as session:
+        entity = await session.get(LegalEntity, entity_id)
+    new_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)
+    real_map_file = rm.map_file
+    failures = [RuntimeError("сбой БД после автозамены")]
+
+    async def flaky_map_file(session: AsyncSession, **kwargs: Any) -> list[str]:
+        warnings = await real_map_file(session, **kwargs)
+        if failures:
+            raise failures.pop()
+        return warnings
+
+    monkeypatch.setattr(rm, "map_file", flaky_map_file)
+
+    assert await process_once(maker, storage, _settings()) is False  # повторяемая неудача: всё откачено
+    assert (await _file(maker, old_id)).status is FileStatus.PROCESSED
+    assert await _replace_events(maker) == []
+
+    assert await process_once(maker, storage, _settings()) is True
+    assert (await _file(maker, old_id)).status is FileStatus.SUPERSEDED
+    assert len(await _replace_events(maker)) == 1
+
+    async with maker() as session, session.begin():
+        new = await session.get(StoredFile, new_id)
+        assert new is not None
+        documents = list((await session.execute(select(Document).where(Document.stored_file_id == new_id))).scalars())
+        replaced = await supersede_previous(
+            session, new_file=new, new_documents=documents, uploader_id=None, now=_now()
+        )
+    assert replaced == []
+    assert len(await _replace_events(maker)) == 1
+    assert (await _file(maker, new_id)).version_no == 2  # noqa: PLR2004 - версия не растёт от повторного вызова
+
+
+async def test_version_chain(committed_sessionmaker: async_sessionmaker[AsyncSession]):
+    """Цепочка v1 → v2 → v3: третий файл заменяет только действующий v2 и получает `version_no = 3`; v1 остаётся
+    заменённым файлом v2."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    entity_id = await _entity(maker)
+    v2_id = await _old_file(maker, entity_id, [OSV_60_2025_Q1], version_no=2)
+    v1_id = await _old_file(
+        maker, entity_id, [OSV_60_2025_Q1], version_no=1, status=FileStatus.SUPERSEDED, superseded_by=v2_id
+    )
+    async with maker() as session:
+        entity = await session.get(LegalEntity, entity_id)
+    v3_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    v1, v2, v3 = await _file(maker, v1_id), await _file(maker, v2_id), await _file(maker, v3_id)
+    assert (v1.status, v1.superseded_by_file_id) == (FileStatus.SUPERSEDED, v2_id)
+    assert (v2.status, v2.superseded_by_file_id) == (FileStatus.SUPERSEDED, v3_id)
+    assert v3.version_no == 3  # noqa: PLR2004 - третья версия
+    [event] = await _replace_events(maker)
+    assert event.object_id == v2_id
+    assert event.after_value == {"file_id": str(v3_id), "version_no": 3}
+
+
+async def test_one_event_per_replaced_file(committed_sessionmaker: async_sessionmaker[AsyncSession]):
+    """Новый файл с ОСВ 60 и 62 заменяет прежний файл с теми же двумя документами одним событием `FILE_REPLACE`,
+    в `description` — оба документа."""
+    maker, storage = committed_sessionmaker, FakeStorage()
+    entity_id = await _entity(maker)
+    old_id = await _old_file(
+        maker,
+        entity_id,
+        [OSV_60_2025_Q1, ("62", 2025, PeriodQuarter.Q1, DocumentStatus.NEEDS_CHECK)],
+    )
+    result = _minimal()
+    second = copy.deepcopy(_osv(result)) | {"locator": "sheet:Лист_2"}
+    second["header"]["account_code_raw"] = "62"
+    result["documents"].append(second)
+    async with maker() as session:
+        entity = await session.get(LegalEntity, entity_id)
+    new_id, _, _ = await _prepare(maker, storage, result, entity=entity)
+
+    assert await process_once(maker, storage, _settings()) is True
+
+    assert (await _file(maker, old_id)).superseded_by_file_id == new_id
+    assert len(await _documents(maker, new_id)) == 2  # noqa: PLR2004 - два листа, два счёта
+    [event] = await _replace_events(maker)
+    assert event.object_id == old_id
+    assert event.description is not None
+    assert "OSV 60 2025 Q1 (CONFIRMED)" in event.description
+    assert "OSV 62 2025 Q1 (NEEDS_CHECK)" in event.description
+    assert "confirmed=true" in event.description

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

@@ -247,9 +247,9 @@ async def supersede_previous(session, *, new_file: StoredFile, new_documents: Se
 ```
 Вызывается в `map_file` после установки `PROCESSED`, в той же транзакции.
 
-- [ ] **Step 1: Тесты:** сценарий спеки «Новая ОСВ 60 за 1 кв. 2025» (старый `CONFIRMED` → `SUPERSEDED`, `version_no = 2`, аудит с `confirmed=true`); два разных счёта не заменяют друг друга; тот же счёт другого года не заменяет; заменённый файл не появляется в `GET files` без `include_superseded` (проверяется в Task 9, здесь — по БД).
+- [x] **Step 1: Тесты:** сценарий спеки «Новая ОСВ 60 за 1 кв. 2025» (старый `CONFIRMED` → `SUPERSEDED`, `version_no = 2`, аудит с `confirmed=true`); два разных счёта не заменяют друг друга; тот же счёт другого года не заменяет; заменённый файл не появляется в `GET files` без `include_superseded` (проверяется в Task 9, здесь — по БД).
-- [ ] **Step 2–4:** падает → реализация → тесты, `make check`.
+- [x] **Step 2–4:** падает → реализация → тесты, `make check`.
-- [ ] **Step 5: Commit** `feat(mapping): автозамена прежних версий файла после маппинга (FILE_REPLACE)`.
+- [x] **Step 5: Commit** `feat(mapping): автозамена прежних версий файла после маппинга (FILE_REPLACE)`.
 
 ---
 

openspec/changes/slice-2c-osv-ingest/tasks.md

@@ -41,7 +41,7 @@
       повторы `mapping_attempts` → `ERROR`, `run_mapping_loop` в lifespan (`ExchangeTasks`). Интеграционные тесты: сценарии спеки
       `osv-result-mapping` на `FakeStorage` с golden JSON (`ОСВ_66`), ключ вне папки, сбой с повтором (monkeypatch), идемпотентность,
       два листа одного счёта, MISMATCH шапки, нарастающая по Q1, RECOGNIZED.
-- [ ] 3.3 **Автозамена.** В `map_file` после `PROCESSED`: поиск прежних файлов по документам, `SUPERSEDED`, `version_no`, аудит
+- [x] 3.3 **Автозамена.** В `map_file` после `PROCESSED`: поиск прежних файлов по документам, `SUPERSEDED`, `version_no`, аудит
       `FILE_REPLACE` (системное событие). Тесты: сценарий спеки (CONFIRMED заменяется, `confirmed=true` в description), два
       разных счёта не заменяют друг друга, `include_superseded`.
 - [ ] 3.4 **Golden-прогон.** `tests/fixtures/test_mapping_golden.py` (`fixtures`): все `../fixtures/golden/expected_osv_parse/**/*.json`