AURORA-1442: Правки по раунду 1

README.md

@@ -55,10 +55,11 @@ PostgreSQL 16, core NATS без JetStream), `web/` — фронтенд (React +
 OCR, `error` сервиса и ошибки маппинга не повторяются. Ошибки: 409 `FILE_NOT_RETRYABLE`,
 `MAX_ATTEMPTS_REACHED`, `VERSION_NOT_EDITABLE`.
 
-Пакет включает файлы с `stored_file.consolidation_id` выбранной консолидации и ЮЛ из периметра выбранной
+Пакет включает только файлы `ERROR` с `stored_file.consolidation_id` выбранной консолидации и ЮЛ из периметра выбранной
 версии; файлы другой консолидации и загрузки без контекста не включаются. Версия должна быть «в работе»
 (`INITIATED`, `DOCUMENTS_PROCESSING`, `DATA_CORRECTION`), пользовательская блокировка не нужна.
-Версии одной консолидации могут видеть общий файл. Неподходящие файлы пропускаются:
+Версии одной консолидации могут видеть общий файл. Другие статусы файлов не блокируются и не попадают
+в ответ. Неподходящие `ERROR`-файлы пропускаются:
 `accepted: [{file_id, job_id, attempt}]`, `skipped: [{file_id, code}]`. Пустой пакет также отвечает 202.
 Для одиночного файла консолидации требуется хотя бы одна рабочая версия с ЮЛ файла в периметре.
 Повтор не меняет статусы версий. Два конкурентных запроса не создают две новые попытки.

api/src/api/dependencies/files.py

@@ -66,10 +66,11 @@ StatementServiceDep = Annotated[StatementService, Depends(get_statement_service)
 
 def get_file_retry_service(
     session: SessionDep,
+    app_settings: SettingsDep,
     audit: Annotated[RequestAudit, Depends(get_request_audit)],
     scope: Annotated[VisibilityScope, Depends(get_scope)],
 ) -> FileRetryService:
-    return FileRetryService(session, audit, scope)
+    return FileRetryService(session, audit, scope, app_settings)
 
 
 FileRetryServiceDep = Annotated[FileRetryService, Depends(get_file_retry_service)]

api/src/api/endpoints/files/retry.py

@@ -11,13 +11,19 @@ from src.schemas.common import ErrorDetail
 from src.schemas.files import FileOut, RetryBatchOut
 
 router = APIRouter(dependencies=[Depends(require("DOCUMENT_EDIT"))])
-RESPONSES: dict[int | str, dict[str, Any]] = {
+FILE_RESPONSES: dict[int | str, dict[str, Any]] = {
     404: {"model": ErrorDetail, "description": "Нет или вне видимости"},
     409: {"model": ErrorDetail, "description": "FILE_NOT_RETRYABLE, MAX_ATTEMPTS_REACHED, VERSION_NOT_EDITABLE"},
 }
 
 
-@router.post("/files/{file_id}/retry", response_model=FileOut, status_code=202, responses=RESPONSES)
+BATCH_RESPONSES: dict[int | str, dict[str, Any]] = {
+    404: {"model": ErrorDetail, "description": "Нет или вне видимости"},
+    409: {"model": ErrorDetail, "description": "VERSION_NOT_EDITABLE"},
+}
+
+
+@router.post("/files/{file_id}/retry", response_model=FileOut, status_code=202, responses=FILE_RESPONSES)
 async def retry_file(file_id: uuid.UUID, service: FileRetryServiceDep) -> FileOut:
     """Новая попытка tabular после TIMEOUT или NO_ACK, до max_attempts."""
     return await service.retry(file_id)
@@ -27,7 +33,7 @@ async def retry_file(file_id: uuid.UUID, service: FileRetryServiceDep) -> FileOu
     "/consolidations/versions/{version_id}/files/retry",
     response_model=RetryBatchOut,
     status_code=202,
-    responses=RESPONSES,
+    responses=BATCH_RESPONSES,
 )
 async def retry_version_files(version_id: uuid.UUID, service: FileRetryServiceDep) -> RetryBatchOut:
     """Повтор подходящих файлов консолидации по периметру выбранной версии."""

api/src/repositories/files.py

@@ -18,7 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from src.domain.policy import VisibilityScope, legal_entities_visible
 from src.models.access import AppUser
 from src.models.audit_ex import Comment
-from src.models.consolidation import ConsolidationDocument, ConsolidationVersion
+from src.models.consolidation import ConsolidationDocument, ConsolidationMember, ConsolidationVersion
 from src.models.directories import LegalEntity
 from src.models.enums import DocumentStatus, FileStatus, S3BucketRole
 from src.models.files import Document, RecognitionJob, S3Object, StoredFile
@@ -298,3 +298,22 @@ async def list_desk_files(  # noqa: PLR0913 - фильтр и пагинация
     )
     files = list((await session.execute(page_stmt)).scalars().all())
     return files, total
+
+
+async def retry_candidates(
+    session: AsyncSession, scope: VisibilityScope, consolidation_id: uuid.UUID, version_id: uuid.UUID
+) -> list[StoredFile]:
+    """ERROR-файлы консолидации по видимому периметру версии; без блокировки нормальных файлов."""
+    members = select(ConsolidationMember.legal_entity_id).where(
+        ConsolidationMember.consolidation_version_id == version_id
+    )
+    stmt = (
+        _visible_files(select(StoredFile), scope)
+        .where(
+            StoredFile.consolidation_id == consolidation_id,
+            StoredFile.legal_entity_id.in_(members),
+            StoredFile.status == FileStatus.ERROR,
+        )
+        .order_by(StoredFile.id)
+    )
+    return list((await session.scalars(stmt)).all())

api/src/services/file_retry.py

@@ -9,6 +9,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from src.api.dependencies.audit import RequestAudit
 from src.common.clock import utc_now
 from src.common.exceptions import ConflictError, NotFoundError
+from src.contract import CONTRACT_VERSION, request_subject
+from src.core.config import Settings, get_settings
 from src.domain.policy import IN_WORK_STATUSES, VisibilityScope
 from src.models.consolidation import ConsolidationMember, ConsolidationVersion
 from src.models.enums import AuditAction, AuditSection, FileStatus, RecognitionPipeline, RecognitionStatus
@@ -20,8 +22,11 @@ from src.services.documents import file_out
 
 
 class FileRetryService:
-    def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None:
+    def __init__(
+        self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, settings: Settings | None = None
+    ) -> None:
         self.session, self.audit, self.scope = session, audit, scope
+        self.settings = settings if settings is not None else get_settings()
 
     async def _lock_files(self, files: list[StoredFile]) -> list[tuple[StoredFile, RecognitionJob | None]]:
         expected = {file.id: file.current_recognition_job_id for file in files}
@@ -63,7 +68,6 @@ class FileRetryService:
             job is None
             or file.status is not FileStatus.ERROR
             or job.pipeline is not RecognitionPipeline.tabular
-            # Мб достаточно только по RecognitionStatus?
             or (job.status, job.error_code)
             not in {(RecognitionStatus.timeout, "TIMEOUT"), (RecognitionStatus.dlq, "NO_ACK")}
             or await self.session.scalar(select(exists().where(Document.stored_file_id == file.id)))
@@ -78,8 +82,8 @@ class FileRetryService:
         new = RecognitionJob(
             file_id=file.id,
             pipeline=job.pipeline,
-            request_subject=job.request_subject,
+            request_subject=request_subject(job.pipeline.value, self.settings.OPENBAO.NATS.request_subjects()),
-            schema_version=job.schema_version,
+            schema_version=CONTRACT_VERSION,
             status=RecognitionStatus.queued,
             attempt=job.attempt + 1,
             max_attempts=job.max_attempts,
@@ -125,9 +129,10 @@ class FileRetryService:
                         select(ConsolidationVersion)
                         .where(
                             ConsolidationVersion.consolidation_id == file.consolidation_id,
+                            ConsolidationVersion.status.in_(IN_WORK_STATUSES),
                         )
                         .order_by(ConsolidationVersion.id)
-                        .with_for_update()
+                        .with_for_update(read=True)
                         .execution_options(populate_existing=True)
                     )
                 ).all()
@@ -156,15 +161,7 @@ class FileRetryService:
         members = select(ConsolidationMember.legal_entity_id).where(
             ConsolidationMember.consolidation_version_id == version_id
         )
-        candidates = list(
+        candidates = await files_repo.retry_candidates(self.session, self.scope, consolidation.id, version_id)
-            (
-                await self.session.scalars(
-                    files_repo._visible_files(select(StoredFile), self.scope)
-                    .where(StoredFile.consolidation_id == consolidation.id, StoredFile.legal_entity_id.in_(members))
-                    .order_by(StoredFile.id)
-                )
-            ).all()
-        )
         locked = await self._lock_files(candidates)
         version = await versions_repo.lock_version(self.session, version_id)
         if version is None:
@@ -174,7 +171,11 @@ class FileRetryService:
         current_members = set((await self.session.scalars(members)).all())
         result = RetryBatchOut(accepted=[], skipped=[])
         for file, job in locked:
-            if file.legal_entity_id not in current_members or file.consolidation_id != consolidation.id:
+            if (
+                file.legal_entity_id not in current_members
+                or file.consolidation_id != consolidation.id
+                or file.status is not FileStatus.ERROR
+            ):
                 continue
             reason = await self._reason(file, job)
             if reason:

api/tests/test_api_file_retry.py

@@ -9,17 +9,19 @@ from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.api.dependencies.audit import RequestAudit
+from src.api.dependencies.directories import get_app_settings
 from src.common.exceptions import ConflictError
+from src.contract import CONTRACT_VERSION
 from src.domain.policy import VisibilityScope
 from src.models.audit_ex import AuditEvent
-from src.models.consolidation import ConsolidationMember, ConsolidationVersion
+from src.models.consolidation import Consolidation, ConsolidationMember, ConsolidationVersion
 from src.models.enums import ConsolidationStatus, FileStatus, RecognitionPipeline, RecognitionStatus
-from src.models.files import RecognitionJob
+from src.models.files import RecognitionJob, StoredFile
 from src.services.audit import AuditService
 from src.services.file_retry import FileRetryService
 from src.services.recognition_exchange import handle_message, publish_pending
 from tests.conftest import AuthClientFactory
-from tests.factories import make_document, make_stored_file
+from tests.factories import make_document, make_stored_file, make_version
 from tests.test_api_consolidation_members import _user, _version
 from tests.test_api_files_read import KA_RF, _rostov_entity
 from tests.test_recognition_exchange import FakeNats
@@ -109,7 +111,7 @@ async def test_batch_filters_perimeter_and_skips_ineligible(
     assert response.status_code == HTTPStatus.ACCEPTED, response.text
     body = response.json()
     assert [row["file_id"] for row in body["accepted"]] == [str(good.id)]
-    assert body["skipped"] == [{"file_id": str(bad.id), "code": "FILE_NOT_RETRYABLE"}]
+    assert body["skipped"] == []
     event = (
         await seeded.execute(
             select(AuditEvent).where(AuditEvent.object_id == good.id, AuditEvent.action == "FILE_REPROCESS")
@@ -236,7 +238,7 @@ async def test_concurrent_batch_retry(committed_sessionmaker) -> None:
 
     results = await asyncio.wait_for(asyncio.gather(retry(), retry()), timeout=10)
     assert sorted(len(result.accepted) for result in results) == [0, 1]
-    assert sorted(len(result.skipped) for result in results) == [0, 1]
+    assert all(result.skipped == [] for result in results)
 
 
 @pytest.mark.parametrize("change", ["status", "perimeter"])
@@ -343,3 +345,77 @@ async def test_batch_no_ack_at_attempt_limit_is_max_attempts_reached(
     assert response.json() == {"accepted": [], "skipped": [{"file_id": str(file.id), "code": "MAX_ATTEMPTS_REACHED"}]}
     jobs = await seeded.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file.id))
     assert [item.id for item in jobs.all()] == [job.id]
+
+
+async def test_retry_uses_current_contract_and_subject(
+    seeded: AsyncSession, auth_client: AuthClientFactory, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    entity = await _rostov_entity(seeded)
+    file, old = await make_stored_file(
+        seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
+    )
+    old.error_code, old.max_attempts = "TIMEOUT", 3
+    old.schema_version, old.request_subject = "0.9", "old.tabular.subject"
+    monkeypatch.setattr(get_app_settings().OPENBAO.NATS, "NATS_TABULAR_REQUEST_SUBJECT", "current.tabular.subject")
+    await seeded.flush()
+    client = await auth_client(KA_RF)
+    response = await client.post(f"/files/{file.id}/retry")
+    assert response.status_code == HTTPStatus.ACCEPTED, response.text
+    await seeded.refresh(file)
+    new = await seeded.get(RecognitionJob, file.current_recognition_job_id)
+    assert new is not None
+    assert new.schema_version == CONTRACT_VERSION and new.request_subject == "current.tabular.subject"
+    await seeded.refresh(old)
+    assert old.schema_version == "0.9" and old.request_subject == "old.tabular.subject"
+
+
+@pytest.mark.parametrize(
+    "normal_status", [FileStatus.UPLOADED, FileStatus.PROCESSING, FileStatus.PROCESSED, FileStatus.SUPERSEDED]
+)
+async def test_batch_does_not_lock_normal_files(committed_sessionmaker, normal_status: FileStatus) -> None:
+    async with committed_sessionmaker() as session, session.begin():
+        version, members, _, _ = await _version(session, members=1, locked_by=None)
+        normal, job = await make_stored_file(
+            session, entity=members[0], status=normal_status, job_status=RecognitionStatus.processing
+        )
+        normal.consolidation_id = version.consolidation_id
+        user = await _user(session, KA_RF)
+        version_id, file_id, job_id, user_id = version.id, normal.id, job.id, user.id
+
+    async def retry():
+        async with committed_sessionmaker() as session, session.begin():
+            audit = RequestAudit(AuditService(session), user_id, None, None)
+            return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id)
+
+    async with committed_sessionmaker() as holder, holder.begin():
+        await holder.execute(select(RecognitionJob).where(RecognitionJob.id == job_id).with_for_update())
+        await holder.execute(select(StoredFile).where(StoredFile.id == file_id).with_for_update())
+        result = await asyncio.wait_for(retry(), timeout=5)
+        assert result.accepted == [] and result.skipped == []
+
+
+@pytest.mark.parametrize("closed_status", [ConsolidationStatus.VERIFIED, ConsolidationStatus.CANCELLED])
+async def test_single_retry_does_not_lock_closed_versions(committed_sessionmaker, closed_status) -> None:
+    async with committed_sessionmaker() as session, session.begin():
+        working, members, _, _ = await _version(session, members=1, locked_by=None)
+        consolidation = await session.get(Consolidation, working.consolidation_id)
+        user = await _user(session, KA_RF)
+        closed = await make_version(
+            session, consolidation=consolidation, version_no=2, members=members, initiator=user, status=closed_status
+        )
+        file, job = await make_stored_file(
+            session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout
+        )
+        file.consolidation_id = working.consolidation_id
+        job.error_code, job.max_attempts = "TIMEOUT", 3
+        file_id, user_id, closed_id = file.id, user.id, closed.id
+
+    async def retry():
+        async with committed_sessionmaker() as session, session.begin():
+            audit = RequestAudit(AuditService(session), user_id, None, None)
+            return await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id)
+
+    async with committed_sessionmaker() as holder, holder.begin():
+        await holder.execute(select(ConsolidationVersion).where(ConsolidationVersion.id == closed_id).with_for_update())
+        card = await asyncio.wait_for(retry(), timeout=5)
+        assert card.status == FileStatus.UPLOADED

api/tests/test_openapi_routes.py

@@ -275,3 +275,12 @@ def test_version_member_declares_type_and_osv_accounts(schema: dict[str, Any]) -
     """front-api-gaps D-3: участник периметра — тип ЮЛ и счета ОСВ к периоду версии."""
     member = schema["components"]["schemas"]["MemberOut"]
     assert {"entity_type", "osv_accounts", "checklist"} <= set(member["required"])
+
+
+def test_retry_routes_declare_different_conflicts(schema: dict[str, Any]) -> None:
+    single = schema["paths"]["/api/v1/files/{file_id}/retry"]["post"]["responses"]["409"]["description"]
+    batch = schema["paths"]["/api/v1/consolidations/versions/{version_id}/files/retry"]["post"]["responses"]["409"][
+        "description"
+    ]
+    assert "FILE_NOT_RETRYABLE" in single and "MAX_ATTEMPTS_REACHED" in single
+    assert batch == "VERSION_NOT_EDITABLE"

openspec/changes/retry-recognition-and-edit-osv-header/design.md

@@ -38,8 +38,9 @@ DocumentService умеет confirm; привязки централизован
 
 ### D2. Новая попытка вместо обновления прежнего job
 
-Создать новый UUID job, queued, attempt + 1; перенести только file_id, pipeline, max_attempts и настройки
+Создать новый UUID job, queued, attempt + 1; перенести file_id, pipeline, max_attempts.
-контракта/маршрута, необходимые публикатору. Ошибки, метрики, result_s3_object_id, service metadata, message ids
+schema_version брать из CONTRACT_VERSION, request_subject — из текущих настроек NATS, как при загрузке;
+версию контракта и subject прежнего задания не копировать. Ошибки, метрики, result_s3_object_id, service metadata, message ids
 и времена старой попытки не копируются. requested_at — сейчас, reply_deadline формируется публикатором.
 Переключить current_recognition_job_id и вернуть файл из ERROR в UPLOADED. Прежний job неизменен.
 Это соответствует уточнённой задаче и не требует изменения дедупа парсера по старому job_id (OQ-73).
@@ -47,7 +48,9 @@ DocumentService умеет confirm; привязки централизован
 
 ### D3. Пакет, частичные отказы и версии
 
-Выбрать файлы выбранной консолидации по stored_file.consolidation_id, с ЮЛ из периметра версии и policy ЮЛ.
+Отбор repositories/files.py::retry_candidates выбирает только ERROR-файлы выбранной консолидации
+по stored_file.consolidation_id, с ЮЛ из периметра версии и policy ЮЛ, до любых блокировок.
+Остальные статусы не блокируются и не попадают в skipped; после блокировки статус ERROR перепроверяется.
 Файл без документов не отбирается через consolidation_document. Дедуп по file_id.
 Согласованный пользователем общий набор у нескольких версий сохраняется; повтор не создаёт версионной копии файла.
 
@@ -63,11 +66,15 @@ DocumentService умеет confirm; привязки централизован
 ### D4. Блокировки и гонки
 
 Retry следует порядку работающего обмена: текущее задание FOR UPDATE → файл FOR UPDATE → версии по id.
+В одиночном повторе только рабочие версии берутся FOR SHARE: статус и периметр защищены от изменений,
+обычное чтение не блокируется, VERIFIED/CANCELLED вообще не блокируются.
 После блокировки файла перечитать current_recognition_job_id; если изменился, одиночный запрос получает
 FILE_NOT_RETRYABLE, пакет пропускает файл. Не захватывать новое текущее задание, уже удерживая файл:
 это инвертирует порядок обмена. Лимит и отсутствие документов проверять заново под блокировкой.
 Для пакета сначала блокировать все отобранные прежние jobs по id, затем файлы по id, затем версии по id;
 после захвата версий перечитать статус и состав периметра. Не начинать пакет с FOR UPDATE версии.
+OpenAPI одиночного повтора описывает FILE_NOT_RETRYABLE/MAX_ATTEMPTS_REACHED/VERSION_NOT_EDITABLE;
+для пакетного 409 только VERSION_NOT_EDITABLE, остальные причины возвращаются в skipped.
 
 Для PATCH — файл → документ FOR NO KEY UPDATE → существующая advisory-блокировка привязок ГК → версии
 по id, как при confirm/delete. Проверить связанные статусы под блокировками и не использовать

openspec/changes/retry-recognition-and-edit-osv-header/specs/primary-files/spec.md

@@ -18,11 +18,11 @@
 - **THEN** 409 MAX_ATTEMPTS_REACHED, состояние не меняется
 
 ### Requirement: Пакет повторной обработки версии
-`POST /api/v1/consolidations/versions/{version_id}/files/retry` SHALL рассматривать видимые файлы в статусе `ERROR` с
+`POST /api/v1/consolidations/versions/{version_id}/files/retry` SHALL выбирать видимые ERROR-файлы
-consolidation_id консолидации версии и ЮЛ из периметра версии; файлы других статусов в пакет не входят и не блокируются.
+консолидации по периметру версии. Другие статусы не входят в пакет и не блокируются.
-Для каждого допустимого файла SHALL создаваться новая попытка. Успех — 202 с accepted[{file_id, job_id, attempt}] и
+Ответ — 202 с accepted[{file_id, job_id, attempt}] и skipped[{file_id, code}]. В skipped SHALL входить
-skipped[{file_id, code}], где skipped — рассмотренные файлы, которые повторить нельзя (`FILE_NOT_RETRYABLE`,
+только ERROR-файлы, которые нельзя повторить (FILE_NOT_RETRYABLE, MAX_ATTEMPTS_REACHED).
-`MAX_ATTEMPTS_REACHED`). Файлы без контекста консолидации MUST NOT включаться по одному лишь совпадению ЮЛ.
+Файлы без контекста консолидации MUST NOT включаться по совпадению ЮЛ.
 
 #### Scenario: Смешанный пакет
 - **WHEN** у версии есть файл timeout/TIMEOUT, OCR, успешный файл и файл с исчерпанным лимитом

openspec/changes/retry-recognition-and-edit-osv-header/tasks.md

@@ -79,4 +79,11 @@
 - [x] 7.1 Ревьюер: `api/.env.test` — порт PostgreSQL 5432 по compose (свой порт — переменной `OPENBAO__DATABASE__DATABASE_URL`).
 - [x] 7.2 Ревьюер: дельта data-model переименована на v5.21, `openspec/config.yaml` — DBML v5.21, README/CLAUDE.md — миграции 0008/0009.
 - [x] 7.3 Ревьюер: синтетические тесты PATCH шапки ОСВ без `fixtures`, смешанный пакет и NO_ACK на пределе.
-- [ ] 7.4 Автор: замечания комментария ревью (пакет — только файлы в `ERROR`, блокировки версий, `CONTRACT_VERSION`, репозиторий, OpenAPI).
+- [x] 7.4 Автор: замечания комментария ревью (пакет — только файлы в `ERROR`, блокировки версий, `CONTRACT_VERSION`, репозиторий, OpenAPI).
+
+Раунд 1: все шесть замечаний к коду исправлены. Пакет отбирается репозиторием только по ERROR до блокировок,
+статус перепроверяется под блокировкой; одиночный повтор берёт FOR SHARE только рабочих версий.
+Новая попытка использует CONTRACT_VERSION и subject из SettingsDep. OpenAPI-конфликты разделены;
+проверка пары status/error_code сохранена, вопросительный комментарий удалён.
+Проверки: 101 passed (file_retry, синтетическая шапка ОСВ, OpenAPI), make lint,
+openspec validate --strict, git diff --check — успешно. .env.test не менялся.