fix(statements): очистка по свежему плану, аудит удаления шаблонной строки, сброс подтверждения и родителя при правке строк (T4.2, T4.4, T13.2, M-4, M-10, M-12)

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

api/scripts/purge_statements.py

@@ -91,11 +91,13 @@ class PurgePlan:
 
 
 class PurgeRefusedError(Exception):
-    """Очистка отклонена: есть документы, замороженные в верифицированных версиях."""
+    """Очистка отклонена: есть документы, замороженные в верифицированных версиях. `plan` — свежий план,
+    построенный под блокировками (его счётчики и `blocking` описывают состояние базы на момент отказа)."""
 
-    def __init__(self, blocking: Sequence[BlockingVersion]) -> None:
+    def __init__(self, plan: PurgePlan) -> None:
         super().__init__("очистка отклонена")
-        self.blocking = list(blocking)
+        self.plan = plan
+        self.blocking = list(plan.blocking)
 
 
 def _frozen_version() -> ColumnElement[bool]:
@@ -258,15 +260,22 @@ async def apply_purge(session: AsyncSession, plan: PurgePlan, *, now: datetime)
 
     Блокирует файлы плана (по `id`), заново проверяет отказ под блокировками и только потом меняет данные: при
     `PurgeRefusedError` база не изменена."""
-    whole = list(plan.form_only_files) + (list(plan.mixed_files) if plan.include_mixed else [])
-    partial = [] if plan.include_mixed else list(plan.mixed_files)
-    for file_id in sorted([*whole, *partial]):
-        await files_repo.lock_jobs(session, file_id)
-        await files_repo.lock_file(session, file_id)
-        await files_repo.lock_file_documents(session, file_id)
-    fresh = await plan_purge(session, include_mixed=plan.include_mixed)
+    locked: set[uuid.UUID] = set()
+
+    async def lock(file_ids: Sequence[uuid.UUID]) -> None:
+        for file_id in sorted(set(file_ids) - locked):
+            await files_repo.lock_jobs(session, file_id)
+            await files_repo.lock_file(session, file_id)
+            await files_repo.lock_file_documents(session, file_id)
+            locked.add(file_id)
+
+    await lock([*plan.form_only_files, *plan.mixed_files])
+    fresh = await plan_purge(session, include_mixed=plan.include_mixed)  # T4.2: списки файлов — из свежего плана
+    await lock([*fresh.form_only_files, *fresh.mixed_files])  # файлы, появившиеся после первого плана
     if fresh.blocking:
-        raise PurgeRefusedError(fresh.blocking)
+        raise PurgeRefusedError(fresh)
+    whole = list(fresh.form_only_files) + (list(fresh.mixed_files) if fresh.include_mixed else [])
+    partial = [] if fresh.include_mixed else list(fresh.mixed_files)
     audit = AuditService(session)
     for file_id in sorted(whole):
         await _purge_whole_file(session, audit, file_id, now)
@@ -323,7 +332,7 @@ async def run(argv: Sequence[str], *, transaction: Transaction = session_transac
                 return EXIT_REFUSED
             await apply_purge(session, plan, now=utc_now())
     except PurgeRefusedError as error:
-        print(render_report(PurgePlan(blocking=error.blocking)), file=sys.stderr)
+        print(render_report(error.plan), file=sys.stderr)  # T4.4: свежий план с реальными счётчиками
         print("очистка отклонена, база не изменена", file=sys.stderr)
         return EXIT_REFUSED
     print("очистка выполнена")

api/tests/test_purge_statements.py

@@ -458,3 +458,33 @@ async def test_reupload_after_purge_not_duplicate(world: World, seeded: AsyncSes
     service = PrimaryFilesService.__new__(PrimaryFilesService)
     service._session = session  # type: ignore[attr-defined]  # find_duplicate читает только сессию
     assert await service.find_duplicate(entity_id, "ab" * 32) is None
+
+
+async def test_apply_uses_fresh_plan_for_file_lists(world: World) -> None:
+    """T4.2: файл, появившийся после построения плана, удаляется по свежему плану под блокировками."""
+    session = world.session
+    first, first_job = await make_stored_file(session, entity=world.entity, original_name="first.pdf")
+    await world.form_document(first, first_job, DocumentType.BALANCE)
+    stale = await purge_statements.plan_purge(session, include_mixed=False)
+    late, late_job = await make_stored_file(session, entity=world.entity, original_name="late.pdf")
+    await world.form_document(late, late_job, DocumentType.BALANCE)
+    first_id, late_id = first.id, late.id
+
+    await purge_statements.apply_purge(session, stale, now=datetime.datetime.now(datetime.UTC))
+
+    assert await _count(session, StoredFile, StoredFile.id.in_([first_id, late_id])) == 0
+
+
+async def test_refusal_carries_fresh_plan_counters(world: World) -> None:
+    """T4.4: отказ под блокировками несёт свежий план — счётчики отчёта не нулевые."""
+    objects = await _mixed_and_form_only(world)
+    stale = await purge_statements.plan_purge(world.session, include_mixed=False)
+    version = await world.version(ConsolidationStatus.VERIFIED)
+    await world.bind(objects["income"], version)  # type: ignore[arg-type]
+
+    with pytest.raises(purge_statements.PurgeRefusedError) as raised:
+        await purge_statements.apply_purge(world.session, stale, now=datetime.datetime.now(datetime.UTC))
+
+    refused = raised.value.plan
+    assert refused.form_documents > 0 and refused.blocking
+    assert "документов форм: 3" in purge_statements.render_report(refused)