Add tenant-scoped generation attempt history query

app/generation/repositories/generation.py

@@ -193,6 +193,30 @@ class GenerationRepository:
                 raise GenerationJobNotFoundError("Generation job was not found in this workspace.")
             return job
 
+    async def list_attempts(
+        self, workspace_id: str, job_id: str, *, user_id: str
+    ) -> list[GenerationJobAttempt]:
+        """Return durable retry history only when the logical job belongs to this workspace."""
+        async with self.database.tenant_session(
+            workspace_id=workspace_id, user_id=user_id
+        ) as session:
+            job_exists = await session.scalar(
+                select(GenerationJob.id).where(
+                    GenerationJob.id == job_id,
+                    GenerationJob.workspace_id == workspace_id,
+                )
+            )
+            if job_exists is None:
+                raise GenerationJobNotFoundError(
+                    "Generation job was not found in this workspace."
+                )
+            rows = await session.scalars(
+                select(GenerationJobAttempt)
+                .where(GenerationJobAttempt.generation_job_id == job_id)
+                .order_by(GenerationJobAttempt.attempt_number.asc())
+            )
+            return list(rows.all())
+
     async def cancel(self, workspace_id: str, job_id: str, *, user_id: str) -> GenerationJob:
         """Cancel queued work immediately; request cancellation otherwise.
 
@@ -354,518 +378,4 @@ class GenerationRepository:
                     )
                 if job.provider != provider:
                     raise GenerationProviderJobConflictError(
-                        "Generation worker job belongs to a different provider."
-                    )
-                if GenerationJobStatus(job.status) not in {
-                    GenerationJobStatus.SUBMITTING,
-                    GenerationJobStatus.RUNNING,
-                    GenerationJobStatus.CANCEL_REQUESTED,
-                }:
-                    raise GenerationProviderJobConflictError(
-                        "Generation job is not in a state that can bind a worker job."
-                    )
-                if job.external_job_id is not None:
-                    if job.external_job_id != external_job_id:
-                        raise GenerationProviderJobConflictError(
-                            "Generation job is already bound to a different worker job."
-                        )
-                    return job
-
-                conflicting = await session.scalar(
-                    select(GenerationJob.id).where(
-                        GenerationJob.provider == provider,
-                        GenerationJob.external_job_id == external_job_id,
-                        GenerationJob.id != job.id,
-                    )
-                )
-                if conflicting is not None:
-                    raise GenerationProviderJobConflictError(
-                        "Generation worker job is already bound to another workspace."
-                    )
-                job.external_job_id = external_job_id
-                job.provider_metadata_json = self._merge_safe_metadata(
-                    job.provider_metadata_json, provider_metadata
-                )
-                job.updated_at = datetime.now(timezone.utc)
-                await session.commit()
-                await session.refresh(job)
-                return job
-        except IntegrityError as exc:
-            # The unique database constraint arbitrates a binding race across
-            # independently running worker processes without leaking either
-            # workspace or provider details.
-            raise GenerationProviderJobConflictError(
-                "Generation worker job is already bound to another logical job."
-            ) from exc
-
-    async def complete_with_output_asset(
-        self,
-        workspace_id: str,
-        job_id: str,
-        *,
-        user_id: str,
-        external_job_id: str,
-        output_asset_id: str,
-        provider_metadata: dict[str, object] | None = None,
-    ) -> GenerationJob:
-        """Atomically attach an owned canonical output and complete the job."""
-
-        async with self.database.tenant_session(
-            workspace_id=workspace_id, user_id=user_id
-        ) as session:
-            job = await session.scalar(
-                select(GenerationJob)
-                .where(
-                    GenerationJob.id == job_id,
-                    GenerationJob.workspace_id == workspace_id,
-                )
-                .with_for_update()
-            )
-            if job is None:
-                raise GenerationJobNotFoundError("Generation job was not found in this workspace.")
-            if job.external_job_id != external_job_id:
-                raise GenerationProviderJobConflictError(
-                    "Generation worker job is not bound to this logical job."
-                )
-            asset = await session.scalar(
-                select(CanonicalMediaAsset).where(
-                    CanonicalMediaAsset.id == output_asset_id,
-                    CanonicalMediaAsset.workspace_id == workspace_id,
-                )
-            )
-            if asset is None:
-                raise GenerationOutputConflictError(
-                    "Generation output asset is not owned by this workspace."
-                )
-            if job.output_asset_id is not None:
-                if job.output_asset_id == output_asset_id:
-                    return job
-                raise GenerationOutputConflictError(
-                    "Generation job already has a different canonical output asset."
-                )
-
-            destination = validate_transition(job.status, GenerationJobStatus.SUCCEEDED)
-            now = datetime.now(timezone.utc)
-            job.status = destination.value
-            job.output_asset_id = output_asset_id
-            job.provider_metadata_json = self._merge_safe_metadata(
-                job.provider_metadata_json, provider_metadata
-            )
-            job.error_code = None
-            job.error_message = None
-            job.next_attempt_at = None
-            job.completed_at = now
-            job.updated_at = now
-            await self._set_request_status(
-                session,
-                workspace_id=workspace_id,
-                request_id=job.generation_request_id,
-                status=destination,
-                completed_at=now,
-            )
-            await self._finish_open_attempt(
-                session,
-                generation_job_id=job.id,
-                status=GenerationJobStatus.SUCCEEDED.value,
-                error_code=None,
-                error_message=None,
-                completed_at=now,
-            )
-            await session.commit()
-            await session.refresh(job)
-            return job
-
-    async def start_attempt(
-        self, workspace_id: str, job_id: str, *, user_id: str
-    ) -> GenerationJobAttempt:
-        async with self.database.tenant_session(
-            workspace_id=workspace_id, user_id=user_id
-        ) as session:
-            job = await session.scalar(
-                select(GenerationJob)
-                .where(
-                    GenerationJob.id == job_id,
-                    GenerationJob.workspace_id == workspace_id,
-                )
-                .with_for_update()
-            )
-            if job is None:
-                raise GenerationJobNotFoundError("Generation job was not found in this workspace.")
-            job.attempt_count += 1
-            attempt = GenerationJobAttempt(
-                generation_job_id=job.id,
-                attempt_number=job.attempt_count,
-                status="started",
-            )
-            session.add(attempt)
-            await session.commit()
-            await session.refresh(attempt)
-            return attempt
-
-    async def claim_dispatchable(self, *, limit: int) -> list[GenerationDispatchRecord]:
-        """Atomically claim unbound queued/retry jobs for one submission.
-
-        The worker/service role is the only caller.  ``FOR UPDATE SKIP
-        LOCKED`` prevents two worker processes from submitting the same
-        logical job; SQLite safely serialises writers and ignores the lock
-        clause.  A claim persists ``submitting`` and an attempt before network
-        I/O, so crash recovery can distinguish a never-started job from an
-        ambiguous lost worker response.
-        """
-
-        now = datetime.now(timezone.utc)
-        claimed: list[GenerationDispatchRecord] = []
-        async with self.database.session() as session:
-            rows = await session.execute(
-                select(GenerationJob, GenerationRequest)
-                .join(
-                    GenerationRequest,
-                    GenerationRequest.id == GenerationJob.generation_request_id,
-                )
-                .where(
-                    GenerationJob.external_job_id.is_(None),
-                    GenerationJob.status.in_(
-                        [GenerationJobStatus.QUEUED.value, GenerationJobStatus.RETRYING.value]
-                    ),
-                    or_(
-                        GenerationJob.next_attempt_at.is_(None),
-                        GenerationJob.next_attempt_at <= now,
-                    ),
-                )
-                .order_by(GenerationJob.created_at)
-                .limit(limit)
-                .with_for_update(skip_locked=True)
-            )
-            for job, request in rows.tuples().all():
-                if job.attempt_count >= job.max_attempts:
-                    result = await session.execute(
-                        update(GenerationJob)
-                        .where(
-                            GenerationJob.id == job.id,
-                            GenerationJob.status == job.status,
-                            GenerationJob.external_job_id.is_(None),
-                            GenerationJob.attempt_count == job.attempt_count,
-                        )
-                        .values(
-                            status=GenerationJobStatus.FAILED.value,
-                            error_code="GENERATION_RETRY_LIMIT_EXCEEDED",
-                            error_message="Generation retry limit was reached before submission.",
-                            completed_at=now,
-                            updated_at=now,
-                        )
-                        .execution_options(synchronize_session=False)
-                    )
-                    if result.rowcount != 1:
-                        continue
-                    destination = validate_transition(job.status, GenerationJobStatus.FAILED)
-                    job.status = destination.value
-                    job.error_code = "GENERATION_RETRY_LIMIT_EXCEEDED"
-                    job.error_message = "Generation retry limit was reached before submission."
-                    job.completed_at = now
-                    job.updated_at = now
-                    await self._set_request_status(
-                        session,
-                        workspace_id=job.workspace_id,
-                        request_id=job.generation_request_id,
-                        status=destination,
-                        completed_at=now,
-                    )
-                    continue
-                original_status = job.status
-                original_attempt_count = job.attempt_count
-                destination = validate_transition(original_status, GenerationJobStatus.SUBMITTING)
-                result = await session.execute(
-                    update(GenerationJob)
-                    .where(
-                        GenerationJob.id == job.id,
-                        GenerationJob.status == original_status,
-                        GenerationJob.external_job_id.is_(None),
-                        GenerationJob.attempt_count == original_attempt_count,
-                    )
-                    .values(
-                        status=destination.value,
-                        error_code=None,
-                        error_message=None,
-                        next_attempt_at=None,
-                        updated_at=now,
-                        started_at=job.started_at or now,
-                        attempt_count=original_attempt_count + 1,
-                    )
-                    .execution_options(synchronize_session=False)
-                )
-                if result.rowcount != 1:
-                    continue
-                job.status = destination.value
-                job.error_code = None
-                job.error_message = None
-                job.next_attempt_at = None
-                job.updated_at = now
-                if job.started_at is None:
-                    job.started_at = now
-                job.attempt_count = original_attempt_count + 1
-                attempt = GenerationJobAttempt(
-                    generation_job_id=job.id,
-                    attempt_number=job.attempt_count,
-                    status="submitting",
-                )
-                session.add(attempt)
-                await self._set_request_status(
-                    session,
-                    workspace_id=job.workspace_id,
-                    request_id=job.generation_request_id,
-                    status=destination,
-                    completed_at=None,
-                )
-                claimed.append(
-                    GenerationDispatchRecord(
-                        workspace_id=job.workspace_id,
-                        user_id=request.created_by_user_id,
-                        generation_request_id=request.id,
-                        generation_job_id=job.id,
-                        provider=job.provider,
-                        model_id=request.model_id,
-                        modality=request.modality,
-                        input_asset_id=request.input_asset_id,
-                        project_id=request.project_id,
-                        product_surface=request.product_surface,
-                        spec=dict(request.spec_json or {}),
-                        idempotency_key=request.idempotency_key,
-                        attempt_number=job.attempt_count,
-                        external_job_id=None,
-                        status=destination,
-                    )
-                )
-            await session.commit()
-        return claimed
-
-    async def list_reconcilable(
-        self, *, limit: int, lease_seconds: int
-    ) -> list[GenerationDispatchRecord]:
-        """Return bound active jobs for idempotent status reconciliation."""
-
-        now = datetime.now(timezone.utc)
-        lease_until = now + timedelta(seconds=max(1, lease_seconds))
-        async with self.database.session() as session:
-            rows = await session.execute(
-                select(GenerationJob, GenerationRequest)
-                .join(
-                    GenerationRequest,
-                    GenerationRequest.id == GenerationJob.generation_request_id,
-                )
-                .where(
-                    GenerationJob.external_job_id.is_not(None),
-                    GenerationJob.status.in_(
-                        [
-                            GenerationJobStatus.QUEUED.value,
-                            GenerationJobStatus.SUBMITTING.value,
-                            GenerationJobStatus.RUNNING.value,
-                            GenerationJobStatus.CANCEL_REQUESTED.value,
-                        ]
-                    ),
-                    or_(
-                        GenerationJob.next_attempt_at.is_(None),
-                        GenerationJob.next_attempt_at <= now,
-                    ),
-                )
-                .order_by(GenerationJob.updated_at)
-                .limit(limit)
-            )
-            records: list[GenerationDispatchRecord] = []
-            for job, request in rows.tuples().all():
-                result = await session.execute(
-                    update(GenerationJob)
-                    .where(
-                        GenerationJob.id == job.id,
-                        GenerationJob.status == job.status,
-                        GenerationJob.external_job_id == job.external_job_id,
-                        or_(
-                            GenerationJob.next_attempt_at.is_(None),
-                            GenerationJob.next_attempt_at <= now,
-                        ),
-                    )
-                    .values(next_attempt_at=lease_until, updated_at=now)
-                    .execution_options(synchronize_session=False)
-                )
-                if result.rowcount != 1:
-                    continue
-                job.next_attempt_at = lease_until
-                job.updated_at = now
-                records.append(
-                    GenerationDispatchRecord(
-                        workspace_id=job.workspace_id,
-                        user_id=request.created_by_user_id,
-                        generation_request_id=request.id,
-                        generation_job_id=job.id,
-                        provider=job.provider,
-                        model_id=request.model_id,
-                        modality=request.modality,
-                        input_asset_id=request.input_asset_id,
-                        project_id=request.project_id,
-                        spec=dict(request.spec_json or {}),
-                        idempotency_key=request.idempotency_key,
-                        attempt_number=job.attempt_count,
-                        external_job_id=job.external_job_id,
-                        status=GenerationJobStatus(job.status),
-                    )
-                )
-            await session.commit()
-            return records
-
-    async def set_reconciliation_due(
-        self,
-        workspace_id: str,
-        job_id: str,
-        *,
-        user_id: str,
-        due_at: datetime | None,
-    ) -> None:
-        """Release or defer one bound active job's durable reconciliation lease."""
-
-        async with self.database.tenant_session(
-            workspace_id=workspace_id, user_id=user_id
-        ) as session:
-            job = await session.scalar(
-                select(GenerationJob)
-                .where(
-                    GenerationJob.id == job_id,
-                    GenerationJob.workspace_id == workspace_id,
-                    GenerationJob.external_job_id.is_not(None),
-                    GenerationJob.status.in_(
-                        [
-                            GenerationJobStatus.QUEUED.value,
-                            GenerationJobStatus.SUBMITTING.value,
-                            GenerationJobStatus.RUNNING.value,
-                            GenerationJobStatus.CANCEL_REQUESTED.value,
-                        ]
-                    ),
-                )
-                .with_for_update()
-            )
-            if job is None:
-                return
-            job.next_attempt_at = due_at
-            job.updated_at = datetime.now(timezone.utc)
-            await session.commit()
-
-    async def fail_stale_unbound_submissions(self, *, stale_before: datetime) -> int:
-        """Fail unrecoverably ambiguous submissions that lost their response.
-
-        A remote worker without an idempotency/reconciliation lookup cannot be
-        safely resubmitted after MediaRouter crashes or loses the job ID.  A
-        terminal, auditable failure is preferable to silently making a second
-        video.  This is intentionally provider-neutral; a future idempotent
-        provider may choose a different recovery route before calling it.
-        """
-
-        count = 0
-        async with self.database.session() as session:
-            rows = await session.scalars(
-                select(GenerationJob)
-                .where(
-                    GenerationJob.status == GenerationJobStatus.SUBMITTING.value,
-                    GenerationJob.external_job_id.is_(None),
-                    GenerationJob.updated_at < stale_before,
-                )
-                .with_for_update(skip_locked=True)
-            )
-            now = datetime.now(timezone.utc)
-            for job in rows.all():
-                destination = validate_transition(job.status, GenerationJobStatus.FAILED)
-                job.status = destination.value
-                job.error_code = "GENERATION_SUBMISSION_AMBIGUOUS"
-                job.error_message = (
-                    "Worker submission could not be reconciled safely; it was not retried."
-                )
-                job.completed_at = now
-                job.updated_at = now
-                await self._set_request_status(
-                    session,
-                    workspace_id=job.workspace_id,
-                    request_id=job.generation_request_id,
-                    status=destination,
-                    completed_at=now,
-                )
-                await self._finish_open_attempt(
-                    session,
-                    generation_job_id=job.id,
-                    status="failed_ambiguous",
-                    error_code=job.error_code,
-                    error_message=job.error_message,
-                    completed_at=now,
-                )
-                count += 1
-            await session.commit()
-        return count
-
-    @staticmethod
-    def _merge_safe_metadata(
-        existing: dict[str, object] | None, incoming: dict[str, object] | None
-    ) -> dict[str, object]:
-        """Prevent a worker response from becoming a token/URL secret store."""
-
-        safe_existing = safe_worker_metadata(existing or {})
-        safe_incoming = safe_worker_metadata(incoming or {})
-        assert isinstance(safe_existing, dict)
-        assert isinstance(safe_incoming, dict)
-        return {**safe_existing, **safe_incoming}
-
-    @staticmethod
-    async def _finish_open_attempt(
-        session: AsyncSession,
-        *,
-        generation_job_id: str,
-        status: str,
-        error_code: str | None,
-        error_message: str | None,
-        completed_at: datetime,
-    ) -> None:
-        """Close the newest open attempt without creating a second job."""
-
-        attempt = await session.scalar(
-            select(GenerationJobAttempt)
-            .where(
-                GenerationJobAttempt.generation_job_id == generation_job_id,
-                GenerationJobAttempt.completed_at.is_(None),
-            )
-            .order_by(GenerationJobAttempt.attempt_number.desc())
-            .limit(1)
-            .with_for_update()
-        )
-        if attempt is None:
-            return
-        attempt.status = status[:32]
-        attempt.error_code = error_code
-        attempt.error_message = error_message
-        attempt.completed_at = completed_at
-
-    @staticmethod
-    def _safe_error(code: str | None, message: str | None) -> tuple[str | None, str | None]:
-        """Persist only safe, bounded worker error information."""
-
-        safe_code = code if code and _EXTERNAL_JOB_ID.fullmatch(code) else None
-        safe_message = safe_worker_metadata(message) if message is not None else None
-        if not isinstance(safe_message, str):
-            safe_message = None
-        return safe_code, safe_message[:500] if safe_message is not None else None
-
-    @staticmethod
-    async def _set_request_status(
-        session: AsyncSession,
-        *,
-        workspace_id: str,
-        request_id: str,
-        status: GenerationJobStatus,
-        completed_at: datetime | None,
-    ) -> None:
-        request = await session.scalar(
-            select(GenerationRequest).where(
-                GenerationRequest.id == request_id,
-                GenerationRequest.workspace_id == workspace_id,
-            )
-        )
-        if request is None:
-            raise GenerationRequestNotFoundError(
-                "Generation request was not found in this workspace."
-            )
-        request.status = GenerationRequestStatus(status.value).value
-        request.completed_at = completed_at
+                        "Generation worker job belongs to a different pr