From 9b37740dbe940776f8caeab71e39a6e3a9a279e4 Mon Sep 17 00:00:00 2001
From: Adam Hernandez <adam@Hernandez-Partners.local>
Date: Sun, 13 Sep 2026 20:37:46 -0700
Subject: [PATCH] fix(import): restore terminal recovery actions

---
 docs/development/IMPORT_REVIEW_RECOVERY.md    |  6 +++
 src/pullbox/services/health_persistence.py    | 14 ++++--
 src/pullbox/services/health_service.py        |  2 +-
 .../services/import_completed_cleanup.py      | 13 +++--
 .../services/import_known_series_recovery.py  | 12 ++---
 src/pullbox/services/import_orphans.py        |  8 ++-
 src/pullbox/services/import_review_recheck.py |  8 ++-
 .../services/import_terminal_recovery.py      | 19 +++++++
 src/pullbox/ui/import_results_context.py      |  7 +--
 .../ui/templates/partials/import_results.html |  5 +-
 tests/ui/test_import_results_context.py       | 45 +++++++++++++++++
 tests/ui/test_import_ui_routes.py             | 25 +++++++++-
 tests/unit/test_health_checks.py              | 23 +++++++++
 tests/unit/test_health_service_persistence.py | 48 +++++++++++++++++-
 tests/unit/test_import_completed_cleanup.py   | 18 +++++++
 .../unit/test_import_known_series_recovery.py | 21 ++++++++
 tests/unit/test_import_orphans.py             | 10 +++-
 tests/unit/test_import_terminal_recovery.py   | 50 +++++++++++++++++++
 18 files changed, 301 insertions(+), 33 deletions(-)
 create mode 100644 src/pullbox/services/import_terminal_recovery.py
 create mode 100644 tests/unit/test_import_terminal_recovery.py

diff --git a/docs/development/IMPORT_REVIEW_RECOVERY.md b/docs/development/IMPORT_REVIEW_RECOVERY.md
index 17f362f1..56f091ce 100644
--- a/docs/development/IMPORT_REVIEW_RECOVERY.md
+++ b/docs/development/IMPORT_REVIEW_RECOVERY.md
@@ -49,6 +49,12 @@ above-fold count and a direct link to that import's Follow-up workspace.
 
 Available recovery actions are intentionally narrow:
 
+A later task or optional Story Arc failure does not hide these actions after
+the canonical import has durably completed. Pullbox also requires the job to be
+unarchived, idle, and free of rollback work before exposing or applying any
+recovery action. An import that failed before durable completion remains
+ineligible.
+
 - **Dismiss stale Mylar references** marks missing database references skipped.
   It does not delete a review record or touch Mylar's database.
 - **Skip one-page archives** excludes one-page image archives while leaving the
diff --git a/src/pullbox/services/health_persistence.py b/src/pullbox/services/health_persistence.py
index 8f625965..ed98c57e 100644
--- a/src/pullbox/services/health_persistence.py
+++ b/src/pullbox/services/health_persistence.py
@@ -11,7 +11,7 @@
 
 import structlog
 from sqlalchemy import delete, select
-from sqlalchemy.exc import OperationalError
+from sqlalchemy.exc import OperationalError, PendingRollbackError
 from sqlalchemy.ext.asyncio import async_sessionmaker
 
 from pullbox.core.sqlite_lock import (
@@ -206,16 +206,22 @@ async def persist_health_outcomes(
                 await active_session.commit()
                 await session.commit()
             return
-        except OperationalError as exc:
+        except (OperationalError, PendingRollbackError) as exc:
             await active_session.rollback()
-            if not is_sqlite_locked_error(exc) or attempt == lock_retry_attempts:
+            recoverable = isinstance(exc, PendingRollbackError) or is_sqlite_locked_error(exc)
+            if not recoverable or attempt == lock_retry_attempts:
                 raise
             delay_seconds = retry_delay(attempt)
             logger.warning(
-                "health_result_persist_retrying_after_sqlite_lock",
+                (
+                    "health_result_persist_retrying_after_pending_rollback"
+                    if isinstance(exc, PendingRollbackError)
+                    else "health_result_persist_retrying_after_sqlite_lock"
+                ),
                 attempt=attempt,
                 max_attempts=lock_retry_attempts,
                 delay_seconds=delay_seconds,
+                failure_type=type(exc).__name__,
             )
         finally:
             if manage_commit:
diff --git a/src/pullbox/services/health_service.py b/src/pullbox/services/health_service.py
index b39cbccf..ccd2a8e2 100644
--- a/src/pullbox/services/health_service.py
+++ b/src/pullbox/services/health_service.py
@@ -528,7 +528,7 @@ async def _safe_run(
     @staticmethod
     async def _rollback_failed_check_session(session: AsyncSession | None) -> None:
         """Restore a failed check transaction before later checks or persistence."""
-        if session is None or session.is_active:
+        if session is None:
             return
         try:
             await session.rollback()
diff --git a/src/pullbox/services/import_completed_cleanup.py b/src/pullbox/services/import_completed_cleanup.py
index 35a7b0ce..fffb89ca 100644
--- a/src/pullbox/services/import_completed_cleanup.py
+++ b/src/pullbox/services/import_completed_cleanup.py
@@ -20,7 +20,6 @@
 from pullbox.core.name_matcher import NameMatcher
 from pullbox.models.audit_log import AuditEventType
 from pullbox.models.import_job import (
-    ImportControlRequest,
     ImportedFile,
     ImportedFileStatus,
     ImportedSeries,
@@ -40,6 +39,7 @@
 from pullbox.services.import_story_arc_resolution import (
     refresh_story_arc_entries_for_import_files,
 )
+from pullbox.services.import_terminal_recovery import allows_terminal_import_recovery
 
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import AsyncSession
@@ -638,12 +638,11 @@ async def _load_completed_job(session: AsyncSession, job_id: int) -> ImportJob:
     job = await session.get(ImportJob, job_id, populate_existing=True)
     if job is None:
         raise NotFoundError("ImportJob", job_id)
-    if job.status is not ImportJobStatus.COMPLETED:
-        raise ValidationError("Job must be in COMPLETED state for recovery cleanup")
-    if job.control_request is not ImportControlRequest.NONE:
-        raise ValidationError("The import job has a pending control request")
-    if job.archived_at is not None:
-        raise ValidationError("Archived import jobs must be restored before cleanup")
+    if not allows_terminal_import_recovery(job):
+        raise ValidationError(
+            "Job must have a COMPLETED canonical import with no pending control or rollback "
+            "work for recovery cleanup"
+        )
     return job
 
 
diff --git a/src/pullbox/services/import_known_series_recovery.py b/src/pullbox/services/import_known_series_recovery.py
index 59f0946e..d077924d 100644
--- a/src/pullbox/services/import_known_series_recovery.py
+++ b/src/pullbox/services/import_known_series_recovery.py
@@ -18,7 +18,6 @@
     ImportedFileStatus,
     ImportedSeries,
     ImportJob,
-    ImportJobStatus,
     ImportSeriesStatus,
     ImportSourceType,
 )
@@ -31,6 +30,7 @@
     build_import_metadata_conflict,
     source_metadata_for_import_file,
 )
+from pullbox.services.import_terminal_recovery import allows_terminal_import_recovery
 
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import AsyncSession
@@ -213,14 +213,8 @@ async def load_known_series_recovery(
     job_id: int,
 ) -> tuple[KnownSeriesRecovery, ...]:
     """Return only exact recoverable file identities without modifying the session."""
-    job = (
-        await session.execute(
-            select(ImportJob.status, ImportJob.source_type).where(
-                ImportJob.id == job_id,
-            )
-        )
-    ).one_or_none()
-    if job is None or job.status is not ImportJobStatus.COMPLETED:
+    job = await session.get(ImportJob, job_id)
+    if job is None or not allows_terminal_import_recovery(job):
         return ()
     plans: list[KnownSeriesRecovery] = []
     matched_local_ids: dict[int, int | None] = {}
diff --git a/src/pullbox/services/import_orphans.py b/src/pullbox/services/import_orphans.py
index 70f3ad5c..041c734f 100644
--- a/src/pullbox/services/import_orphans.py
+++ b/src/pullbox/services/import_orphans.py
@@ -27,6 +27,7 @@
 )
 from pullbox.services.import_retry_helpers import require_retained_import_destination
 from pullbox.services.import_review_recheck import prepare_retryable_failed_sources_for_retry
+from pullbox.services.import_terminal_recovery import allows_terminal_import_recovery
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable, Sequence
@@ -527,8 +528,11 @@ async def retry_failed_series(
     if job is None:
         raise NotFoundError("ImportJob", job_id)
 
-    if job.status != ImportJobStatus.COMPLETED:
-        raise ValidationError(f"Job must be in COMPLETED state to retry (current: {job.status})")
+    if not allows_terminal_import_recovery(job):
+        raise ValidationError(
+            "Job must have a COMPLETED canonical import with no pending control or rollback "
+            f"work to retry (current: {job.status})"
+        )
 
     require_retained_import_destination(job)
 
diff --git a/src/pullbox/services/import_review_recheck.py b/src/pullbox/services/import_review_recheck.py
index 5c4ecaaa..9219f8b9 100644
--- a/src/pullbox/services/import_review_recheck.py
+++ b/src/pullbox/services/import_review_recheck.py
@@ -41,6 +41,7 @@
 from pullbox.services.import_safety_diagnostics import build_import_safety_diagnostics
 from pullbox.services.import_series_match_state import clear_auto_cv_match_fields
 from pullbox.services.import_source_metadata import source_metadata_for_import_file
+from pullbox.services.import_terminal_recovery import allows_terminal_import_recovery
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -453,8 +454,11 @@ async def prepare_completed_import_file_recheck(
     job = await session.get(ImportJob, job_id)
     if job is None:
         raise NotFoundError("ImportJob", job_id)
-    if job.status != ImportJobStatus.COMPLETED or job.control_request != ImportControlRequest.NONE:
-        raise ValidationError("Job must be idle in COMPLETED before a failed-file recheck")
+    if not allows_terminal_import_recovery(job):
+        raise ValidationError(
+            "Job must have a COMPLETED canonical import with no pending control or rollback "
+            "work before a failed-file recheck"
+        )
     if not source_roots:
         raise ValidationError("At least one explicit source root is required")
     roots = [(path.expanduser().absolute(), resolve_preview_source(path)) for path in source_roots]
diff --git a/src/pullbox/services/import_terminal_recovery.py b/src/pullbox/services/import_terminal_recovery.py
new file mode 100644
index 00000000..fe77456b
--- /dev/null
+++ b/src/pullbox/services/import_terminal_recovery.py
@@ -0,0 +1,19 @@
+"""Shared eligibility rules for safe actions after a durable import completion."""
+
+from __future__ import annotations
+
+from pullbox.models.import_job import ImportControlRequest, ImportJob, ImportJobStatus
+
+
+def allows_terminal_import_recovery(job: ImportJob) -> bool:
+    """Return whether post-import actions may safely operate on this job."""
+    if (
+        job.archived_at is not None
+        or job.control_request is not ImportControlRequest.NONE
+        or job.story_arc_rollback_waiting_work_id is not None
+        or dict(job.progress_snapshot or {}).get("mode") == "rollback"
+    ):
+        return False
+    if job.status is ImportJobStatus.COMPLETED:
+        return True
+    return job.status is ImportJobStatus.FAILED and job.import_completed_at is not None
diff --git a/src/pullbox/ui/import_results_context.py b/src/pullbox/ui/import_results_context.py
index 82042c33..19559de4 100644
--- a/src/pullbox/ui/import_results_context.py
+++ b/src/pullbox/ui/import_results_context.py
@@ -45,6 +45,7 @@
     ImportSafetyCategory,
     import_safety_category_label,
 )
+from pullbox.services.import_terminal_recovery import allows_terminal_import_recovery
 from pullbox.services.import_workflow_state import import_control_state_for_job
 
 if TYPE_CHECKING:
@@ -996,10 +997,9 @@ async def load_import_results_context(
     safety_category_summaries = (
         await _load_safety_category_summaries(session, job_id) if files_safety_blocked > 0 else []
     )
+    recovery_actions_available = allows_terminal_import_recovery(job)
     cleanup_action_summaries = (
-        await _load_cleanup_action_summaries(session, job_id)
-        if job.status is ImportJobStatus.COMPLETED and job.archived_at is None
-        else []
+        await _load_cleanup_action_summaries(session, job_id) if recovery_actions_available else []
     )
     misplaced_source_restore_count = 0
     misplaced_source_duplicate_count = 0
@@ -1137,6 +1137,7 @@ async def load_import_results_context(
         ),
         "safety_category_summaries": safety_category_summaries,
         "cleanup_action_summaries": cleanup_action_summaries,
+        "recovery_actions_available": recovery_actions_available,
         "recommended_conflict_groups": recommended_conflict_groups,
         "recommended_conflict_files": recommended_conflict_files,
         "already_owned_conflict_files": already_owned_conflict_files,
diff --git a/src/pullbox/ui/templates/partials/import_results.html b/src/pullbox/ui/templates/partials/import_results.html
index 16afc257..b51e62ef 100644
--- a/src/pullbox/ui/templates/partials/import_results.html
+++ b/src/pullbox/ui/templates/partials/import_results.html
@@ -31,6 +31,7 @@
       - rollback_actions_rolled_back: int
 -#}
 {% from "components/settings_shell.html" import inline_alert %}
+{% set recovery_actions_available = recovery_actions_available|default(job.status.value == 'completed' and not job.archived_at) %}
 
 <div
   x-data="importResultsData({
@@ -49,7 +50,7 @@
   <div data-testid="import-follow-up-job-action-bar" class="rounded-xl border border-pb-border bg-pb-card/60 px-4 py-3">
     <div class="flex flex-wrap items-center justify-between gap-2">
       <div class="flex flex-wrap items-center gap-2">
-        {% if (failed_count > 0 or files_failed > 0) and not job.removed_library_root_snapshot %}
+        {% if recovery_actions_available and (failed_count > 0 or files_failed > 0) and not job.removed_library_root_snapshot %}
         <button
           type="button"
           data-testid="import-results-retry-action"
@@ -300,7 +301,7 @@ <h2 class="text-2xl font-semibold tracking-tight text-pb-text">
   {% set safety_category_summaries = safety_category_summaries|default([]) %}
   {% set cleanup_safe_action_count = cleanup_safe_action_count|default(0) %}
   {% set cleanup_needs_review_count = cleanup_needs_review_count|default(0) %}
-  {% if job.status.value == 'completed' and not job.archived_at and (cleanup_action_summaries or cleanup_needs_review_count > 0) %}
+  {% if recovery_actions_available and (cleanup_action_summaries or cleanup_needs_review_count > 0) %}
   <section data-testid="import-results-recovery-dashboard" class="section-card overflow-hidden">
     <div class="section-header">
       <div class="section-header-copy">
diff --git a/tests/ui/test_import_results_context.py b/tests/ui/test_import_results_context.py
index bd959cc8..afc7cf4b 100644
--- a/tests/ui/test_import_results_context.py
+++ b/tests/ui/test_import_results_context.py
@@ -181,6 +181,51 @@ async def test_results_count_retryable_failed_inspection_as_safe_next_step(db_se
     assert context["cleanup_action_summaries"][0]["action"] == "retry_source_inspection"
 
 
+@pytest.mark.asyncio
+async def test_results_keep_recovery_actions_for_post_completion_failure(db_session) -> None:  # type: ignore[no-untyped-def]
+    from pullbox.ui.import_results_context import load_import_results_context
+
+    job = ImportJob(
+        source_path="/tmp/comics",
+        source_type=ImportSourceType.FILESYSTEM,
+        status=ImportJobStatus.FAILED,
+        import_completed_at=datetime.now(UTC),
+        error_message="Optional follow-up failed after canonical import completed.",
+    )
+    db_session.add(job)
+    await db_session.flush()
+    imported_series = ImportedSeries(
+        import_job_id=job.id,
+        raw_series_name="Recoverable import",
+        status=ImportSeriesStatus.IMPORTED,
+    )
+    db_session.add(imported_series)
+    await db_session.flush()
+    db_session.add(
+        ImportedFile(
+            import_job_id=job.id,
+            import_series_id=imported_series.id,
+            file_path="/tmp/comics/missing.cbz",
+            file_name="missing.cbz",
+            file_size=1024,
+            file_format="cbz",
+            status=ImportedFileStatus.SAFETY_BLOCKED,
+            diagnostics={
+                "safety_block": build_import_safety_diagnostics(
+                    ImportSafetyCategory.SOURCE_MISSING.value,
+                    code=ImportSafetyCategory.SOURCE_MISSING.value,
+                )
+            },
+        )
+    )
+    await db_session.flush()
+
+    context = await load_import_results_context(db_session, job)
+
+    assert context["recovery_actions_available"] is True
+    assert context["cleanup_action_summaries"][0]["action"] == "dismiss_missing_references"
+
+
 @pytest.mark.asyncio
 async def test_load_import_results_context_reports_changed_sources_separately(
     db_session,
diff --git a/tests/ui/test_import_ui_routes.py b/tests/ui/test_import_ui_routes.py
index ec09e917..4f40dbfc 100644
--- a/tests/ui/test_import_ui_routes.py
+++ b/tests/ui/test_import_ui_routes.py
@@ -1116,7 +1116,11 @@ def test_failed_results_template_preserves_bounded_safety_actions(self) -> None:
             diagnostics={"safety_block": {"overrideable": True, "reason": "Too large"}},
         )
         html = templates.env.get_template("partials/import_results.html").render(
-            job=SimpleNamespace(id=35, status=SimpleNamespace(value="failed")),
+            job=SimpleNamespace(
+                id=35,
+                status=SimpleNamespace(value="failed"),
+                archived_at=None,
+            ),
             can_rollback=False,
             imported_count=0,
             failed_count=1,
@@ -1142,6 +1146,23 @@ def test_failed_results_template_preserves_bounded_safety_actions(self) -> None:
             files_safety_blocked=105,
             safety_blocked_files=[blocked_file],
             safety_blocked_files_truncated=5,
+            recovery_actions_available=True,
+            cleanup_action_summaries=[
+                {
+                    "action": "dismiss_missing_references",
+                    "label": "Dismiss stale Mylar references",
+                    "description": "Clear stale references without deleting files.",
+                    "button_label": "Dismiss references",
+                    "tone": "neutral",
+                    "affected_count": 1,
+                    "affected_file_count": 1,
+                    "item_unit": "file",
+                    "examples": ("missing.cbz",),
+                }
+            ],
+            safety_category_summaries=[],
+            cleanup_safe_action_count=1,
+            cleanup_needs_review_count=0,
             resume_step=5,
             resume_job_id=35,
             resume_progress_snapshot={},
@@ -1152,6 +1173,8 @@ def test_failed_results_template_preserves_bounded_safety_actions(self) -> None:
         assert "large-tpb.cbz" in html
         assert "5 more safety exceptions are not shown" in html
         assert 'data-testid="import-results-safety-retry-91"' in html
+        assert 'data-testid="import-results-recovery-dashboard"' in html
+        assert "Dismiss stale Mylar references" in html
 
     def test_results_template_explains_changed_source_failures(self) -> None:
         from types import SimpleNamespace
diff --git a/tests/unit/test_health_checks.py b/tests/unit/test_health_checks.py
index 72fb8ccd..e3b46f28 100644
--- a/tests/unit/test_health_checks.py
+++ b/tests/unit/test_health_checks.py
@@ -2540,6 +2540,29 @@ async def exploding_check() -> CheckOutcome:
         assert outcomes[0].status == HealthStatus.UNHEALTHY
         assert "boom" in outcomes[0].message.lower()
 
+    @pytest.mark.asyncio
+    async def test_failed_check_always_rolls_back_before_persistence(
+        self,
+        settings: MagicMock,
+    ) -> None:
+        service = _make_service(settings)
+        session = MagicMock()
+        session.is_active = True
+        session.rollback = AsyncMock()
+
+        async def cancelled_database_check() -> CheckOutcome:
+            raise TimeoutError
+
+        outcomes = await service._safe_run(
+            cancelled_database_check(),
+            "database",
+            "connectivity",
+            session=session,
+        )
+
+        assert outcomes[0].status == HealthStatus.UNHEALTHY
+        session.rollback.assert_awaited_once_with()
+
     @pytest.mark.asyncio
     async def test_actionable_guidance_present(self, db_session: AsyncSession) -> None:
         """Non-healthy outcomes should have non-empty guidance."""
diff --git a/tests/unit/test_health_service_persistence.py b/tests/unit/test_health_service_persistence.py
index c846cc7a..62e63e26 100644
--- a/tests/unit/test_health_service_persistence.py
+++ b/tests/unit/test_health_service_persistence.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import pytest
-from sqlalchemy.exc import OperationalError
+from sqlalchemy.exc import OperationalError, PendingRollbackError
 
 from pullbox.models.health import HealthStatus
 from pullbox.services import health_service
@@ -59,6 +59,29 @@ def all(self) -> list[object]:
         return []
 
 
+class _PendingRollbackHealthSession(_FakeHealthSession):
+    async def execute(self, _stmt):  # type: ignore[no-untyped-def]
+        factory = self._factory
+        assert isinstance(factory, _PendingRollbackHealthSessionFactory)
+        factory.execute_attempts += 1
+        if factory.execute_attempts == 1:
+            raise PendingRollbackError("connection invalidated during the health check")
+        return _FakeExecuteResult()
+
+    async def flush(self) -> None:
+        self._factory.flush_attempts += 1
+
+
+class _PendingRollbackHealthSessionFactory(_FakeHealthSessionFactory):
+    def __init__(self) -> None:
+        super().__init__()
+        self.execute_attempts = 0
+
+    def __call__(self) -> _PendingRollbackHealthSession:
+        self.session_count += 1
+        return _PendingRollbackHealthSession(self)
+
+
 @pytest.mark.asyncio
 async def test_health_persist_outcomes_retries_with_fresh_sessions(monkeypatch) -> None:
     """Locked writes should reopen a fresh session before retrying."""
@@ -83,3 +106,26 @@ async def test_health_persist_outcomes_retries_with_fresh_sessions(monkeypatch)
     assert factory.flush_attempts == 2
     assert factory.rollback_attempts == 1
     assert factory.commit_attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_health_persist_outcomes_recovers_from_pending_rollback(monkeypatch) -> None:
+    factory = _PendingRollbackHealthSessionFactory()
+    monkeypatch.setattr(health_service, "get_session_factory", lambda: factory)
+    monkeypatch.setattr(health_service, "sqlite_lock_retry_delay", lambda _attempt: 0.0)
+    outcomes = [
+        CheckOutcome(
+            component="database",
+            check_name="connectivity",
+            status=HealthStatus.UNHEALTHY,
+            message="Check timed out",
+        )
+    ]
+    initial_session = factory()
+
+    await HealthService._persist_outcomes(initial_session, outcomes)
+
+    assert factory.session_count == 2
+    assert factory.rollback_attempts == 1
+    assert factory.flush_attempts == 1
+    assert factory.commit_attempts == 2
diff --git a/tests/unit/test_import_completed_cleanup.py b/tests/unit/test_import_completed_cleanup.py
index 10626091..66c707a0 100644
--- a/tests/unit/test_import_completed_cleanup.py
+++ b/tests/unit/test_import_completed_cleanup.py
@@ -945,6 +945,24 @@ async def test_cleanup_rejects_non_completed_job(db_session: AsyncSession) -> No
         )
 
 
+@pytest.mark.asyncio
+async def test_cleanup_rejects_failed_job_without_durable_completion(
+    db_session: AsyncSession,
+) -> None:
+    job, _imported_series = await _seed_job(db_session)
+    job.status = ImportJobStatus.FAILED
+    job.import_completed_at = None
+    await db_session.commit()
+
+    with pytest.raises(ValidationError, match=r"(?i)completed"):
+        await preview_completed_import_cleanup(
+            db_session,
+            job.id,
+            CompletedImportCleanupAction.DISMISS_MISSING_REFERENCES,
+            actor_id=42,
+        )
+
+
 @pytest.mark.asyncio
 async def test_cleanup_file_review_is_bounded_and_paginated(
     db_session: AsyncSession,
diff --git a/tests/unit/test_import_known_series_recovery.py b/tests/unit/test_import_known_series_recovery.py
index de44f8c0..6aa6d880 100644
--- a/tests/unit/test_import_known_series_recovery.py
+++ b/tests/unit/test_import_known_series_recovery.py
@@ -1,6 +1,7 @@
 """Legacy completed imports retain good files without overriding real conflicts."""
 
 from copy import deepcopy
+from datetime import UTC, datetime
 
 import pytest
 
@@ -279,6 +280,26 @@ async def test_known_series_action_is_visible_in_existing_follow_up_and_file_pre
     assert [item.id for item in page.items] == [file.id]
 
 
+async def test_known_series_recovery_survives_post_completion_job_failure(db_session):
+    job, _series, file = await seed_recovery(db_session)
+    db_session.add(User(id=42, username="operator", password_hash="unused"))
+    job.status = ImportJobStatus.FAILED
+    job.import_completed_at = datetime.now(UTC)
+    job.error_message = "Optional follow-up failed after canonical import completed."
+    await db_session.commit()
+
+    plans = await load_known_series_recovery(db_session, job.id)
+    preview = await preview_completed_import_cleanup(
+        db_session,
+        job.id,
+        CompletedImportCleanupAction.RECOVER_KNOWN_SERIES,
+        actor_id=42,
+    )
+
+    assert [plan.file_id for plan in plans] == [file.id]
+    assert preview.affected_file_count == 1
+
+
 async def test_known_series_action_rejects_changed_parent_evidence(db_session):
     job, series, _file = await seed_recovery(db_session)
     action = CompletedImportCleanupAction.RECOVER_KNOWN_SERIES
diff --git a/tests/unit/test_import_orphans.py b/tests/unit/test_import_orphans.py
index 8de13460..8c5b92b4 100644
--- a/tests/unit/test_import_orphans.py
+++ b/tests/unit/test_import_orphans.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING
 from unittest.mock import AsyncMock
 
@@ -953,11 +954,18 @@ async def test_retry_failed_series_resets_duplicate_file_failures(
     assert failed_file.error_message is None
 
 
+@pytest.mark.parametrize(
+    "job_status",
+    [ImportJobStatus.COMPLETED, ImportJobStatus.FAILED],
+)
 async def test_retry_failed_series_resets_partial_import_file_failures(
     db_session: AsyncSession,
+    job_status: ImportJobStatus,
 ) -> None:
     service = _make_service()
-    job = await _create_job_row(db_session)
+    job = await _create_job_row(db_session, status=job_status)
+    if job_status is ImportJobStatus.FAILED:
+        job.import_completed_at = datetime.now(UTC)
     imported = await _create_imported_series(
         db_session,
         job,
diff --git a/tests/unit/test_import_terminal_recovery.py b/tests/unit/test_import_terminal_recovery.py
new file mode 100644
index 00000000..370d6fb3
--- /dev/null
+++ b/tests/unit/test_import_terminal_recovery.py
@@ -0,0 +1,50 @@
+"""Eligibility boundaries for post-import recovery actions."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+
+from pullbox.models.import_job import (
+    ImportControlRequest,
+    ImportJob,
+    ImportJobStatus,
+    ImportSourceType,
+)
+from pullbox.services.import_terminal_recovery import allows_terminal_import_recovery
+
+
+def _job(status: ImportJobStatus) -> ImportJob:
+    return ImportJob(
+        source_path="/imports/mylar.db",
+        source_type=ImportSourceType.MYLAR3,
+        status=status,
+        control_request=ImportControlRequest.NONE,
+        progress_snapshot={},
+    )
+
+
+def test_completed_and_durably_completed_failed_jobs_allow_recovery() -> None:
+    completed = _job(ImportJobStatus.COMPLETED)
+    failed_after_completion = _job(ImportJobStatus.FAILED)
+    failed_after_completion.import_completed_at = datetime.now(UTC)
+
+    assert allows_terminal_import_recovery(completed) is True
+    assert allows_terminal_import_recovery(failed_after_completion) is True
+
+
+@pytest.mark.parametrize("guard", ["no_completion", "archived", "control", "rollback"])
+def test_incomplete_or_mutating_terminal_jobs_reject_recovery(guard: str) -> None:
+    job = _job(ImportJobStatus.FAILED)
+    job.import_completed_at = datetime.now(UTC)
+    if guard == "no_completion":
+        job.import_completed_at = None
+    elif guard == "archived":
+        job.archived_at = datetime.now(UTC)
+    elif guard == "control":
+        job.control_request = ImportControlRequest.CANCEL
+    else:
+        job.progress_snapshot = {"mode": "rollback"}
+
+    assert allows_terminal_import_recovery(job) is False