Add explicit staged import sealing

src/api/imports.py

@@ -1,4 +1,4 @@
-"""REST endpoints for previewing, staging, and rolling back imports."""
+"""REST endpoints for previewing, staging, sealing, and rolling back imports."""
 
 from __future__ import annotations
 
@@ -14,6 +14,7 @@
 from importers.hermes_markdown import HermesMarkdownImporter
 from importers.providers import provider_adapter, provider_descriptors
 from importers.runner import ImportSummary, run_import
+from importers.staging import seal_import_run
 
 router = APIRouter(tags=["imports"])
 
@@ -37,6 +38,21 @@ class ProviderImportRequest(BaseModel):
     resume_run_id: UUID | None = None
 
 
+class ImportSealRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    actor: str = Field(min_length=1, max_length=255)
+    expected_records: int | None = Field(default=None, ge=0)
+
+
+class ImportSealResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    id: UUID
+    status: str
+    config: dict[str, Any]
+
+
 class ImportRollbackRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -118,6 +134,25 @@ async def import_provider_records(request: ProviderImportRequest) -> ImportSumma
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
 
+@router.post(
+    "/imports/{run_id}/seal",
+    response_model=ImportSealResponse,
+    summary="Validate and seal an immutable staged import candidate set",
+)
+async def seal_import(run_id: UUID, request: ImportSealRequest) -> ImportSealResponse:
+    try:
+        result = seal_import_run(
+            run_id,
+            actor=request.actor,
+            expected_records=request.expected_records,
+        )
+        return ImportSealResponse.model_validate(result)
+    except ValueError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
 @router.post(
     "/imports/{run_id}/rollback",
     response_model=ImportRollbackResponse,
@@ -130,4 +165,4 @@ async def rollback_import(run_id: UUID, request: ImportRollbackRequest) -> Impor
     except ValueError as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     except Exception as exc:
-        raise HTTPException(status_code=500, detail=str(exc)) from exc
+        raise HTTPException(status_code=500, detail=str(exc)) from exc

src/importers/runner.py

@@ -39,13 +39,16 @@ def run_import(
 ) -> ImportSummary:
     """Execute one deterministic import pass.
 
-    Candidates are persisted only as import records. Promotion into assertions,
-    projects, tasks, or decisions is deliberately deferred to reconciliation.
+    Candidates are persisted only as import records. Dry runs finish in the
+    ``previewed`` state. Real imports finish in the ``staged`` state and require
+    an explicit validation seal before any later promotion step.
     """
     if resume_run_id:
         existing = get_import_run(resume_run_id)
         if existing is None:
             raise ValueError(f"import run not found: {resume_run_id}")
+        if existing.get("status") in {"sealed", "rolled_back"}:
+            raise ValueError(f"{existing['status']} imports cannot be resumed")
         run_id = UUID(str(existing["id"]))
         config = existing.get("config") or {}
         if isinstance(config, str):
@@ -102,9 +105,10 @@ def run_import(
                 records_rejected=counters["rejected"],
             )
 
+        final_status = "previewed" if dry_run else "staged"
         update_import_run(
             run_id,
-            status="completed",
+            status=final_status,
             cursor_value=cursor,
             records_seen=counters["seen"],
             records_imported=counters["imported"],
@@ -126,11 +130,11 @@ def run_import(
 
     return ImportSummary(
         run_id=run_id,
-        status="completed",
+        status=final_status,
         dry_run=dry_run,
         records_seen=counters["seen"],
         records_imported=counters["imported"],
         records_merged=counters["merged"],
         records_rejected=counters["rejected"],
         cursor=cursor,
-    )
+    )

src/importers/staging.py

@@ -0,0 +1,73 @@
+"""Validation and sealing controls for staged import runs."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from uuid import UUID
+
+from db.connection import get_db_cursor
+
+
+def seal_import_run(run_id: UUID, *, actor: str, expected_records: int | None = None) -> dict[str, Any]:
+    """Seal a completed staging run after validating its immutable candidate set.
+
+    Sealing is a deliberate operator boundary between discovery and promotion. A
+    sealed run cannot accept further candidates or be resumed.
+    """
+    with get_db_cursor() as cursor:
+        cursor.execute("SELECT * FROM import_run WHERE id = %s FOR UPDATE", (run_id,))
+        run = cursor.fetchone()
+        if run is None:
+            raise ValueError(f"import run not found: {run_id}")
+
+        config = run.get("config") or {}
+        if isinstance(config, str):
+            config = json.loads(config)
+        if bool(config.get("dry_run")):
+            raise ValueError("dry-run imports cannot be sealed")
+        if run.get("rolled_back_at") is not None or run.get("status") == "rolled_back":
+            raise ValueError("rolled-back imports cannot be sealed")
+        if run.get("status") == "sealed":
+            return dict(run)
+        if run.get("status") != "staged":
+            raise ValueError("only staged imports can be sealed")
+
+        cursor.execute(
+            """
+            SELECT COUNT(*) AS total,
+                   COUNT(*) FILTER (WHERE result = 'staged' AND rolled_back_at IS NULL) AS valid,
+                   COUNT(*) FILTER (WHERE error IS NOT NULL) AS failed
+            FROM import_record
+            WHERE import_run_id = %s
+            """,
+            (run_id,),
+        )
+        counts = cursor.fetchone()
+        total = int(counts["total"])
+        valid = int(counts["valid"])
+        failed = int(counts["failed"])
+        if failed or valid != total:
+            raise ValueError("import contains invalid, failed, or rolled-back records")
+        if expected_records is not None and total != expected_records:
+            raise ValueError(
+                f"staged record count changed: expected {expected_records}, found {total}"
+            )
+
+        seal = {
+            "actor": actor,
+            "record_count": total,
+            "source_fingerprint": config.get("source_fingerprint"),
+        }
+        cursor.execute(
+            """
+            UPDATE import_run
+            SET status = 'sealed',
+                config = config || jsonb_build_object('seal', %s::jsonb),
+                completed_at = COALESCE(completed_at, NOW())
+            WHERE id = %s
+            RETURNING *
+            """,
+            (json.dumps(seal), run_id),
+        )
+        return dict(cursor.fetchone())

tests/test_import_runner.py

@@ -56,10 +56,31 @@ def test_import_runner_dry_run_records_preview(monkeypatch):
     )
 
     assert result.run_id == run_id
+    assert result.status == "previewed"
     assert result.records_seen == 2
     assert result.records_imported == 2
     assert all(call[1]["operation"] == "preview" for call in recorded)
-    assert updates[-1]["status"] == "completed"
+    assert updates[-1]["status"] == "previewed"
+
+
+def test_import_runner_non_dry_run_requires_sealing(monkeypatch):
+    run_id = uuid4()
+    updates = []
+    monkeypatch.setattr(
+        "importers.runner.create_import_run",
+        lambda *args, **kwargs: {"id": run_id},
+    )
+    monkeypatch.setattr("importers.runner.seen_external_hashes", lambda _: set())
+    monkeypatch.setattr("importers.runner.record_import_candidate", lambda *args, **kwargs: True)
+    monkeypatch.setattr(
+        "importers.runner.update_import_run",
+        lambda run, **kwargs: updates.append(kwargs) or {"id": run},
+    )
+
+    result = run_import(StaticAdapter(["one"]), source_system="hermes.user_memory", dry_run=False)
+
+    assert result.status == "staged"
+    assert updates[-1]["status"] == "staged"
 
 
 def test_import_runner_rejects_resume_after_source_change(monkeypatch):
@@ -68,6 +89,7 @@ def test_import_runner_rejects_resume_after_source_change(monkeypatch):
         "importers.runner.get_import_run",
         lambda _: {
             "id": run_id,
+            "status": "running",
             "config": {"dry_run": True, "source_fingerprint": "a" * 64},
         },
     )
@@ -80,6 +102,21 @@ def test_import_runner_rejects_resume_after_source_change(monkeypatch):
         )
 
 
+def test_import_runner_rejects_sealed_resume(monkeypatch):
+    run_id = uuid4()
+    monkeypatch.setattr(
+        "importers.runner.get_import_run",
+        lambda _: {"id": run_id, "status": "sealed", "config": {"dry_run": False}},
+    )
+
+    with pytest.raises(ValueError, match="sealed imports cannot be resumed"):
+        run_import(
+            StaticAdapter(["one"]),
+            source_system="hermes.user_memory",
+            resume_run_id=run_id,
+        )
+
+
 def test_import_runner_skips_already_recorded_candidate(monkeypatch):
     run_id = uuid4()
     adapter = StaticAdapter(["one"])
@@ -104,6 +141,7 @@ def test_import_runner_skips_already_recorded_candidate(monkeypatch):
 
     result = run_import(adapter, source_system="hermes.user_memory", dry_run=False)
 
+    assert result.status == "staged"
     assert result.records_seen == 1
     assert result.records_imported == 0
-    assert result.records_merged == 1
+    assert result.records_merged == 1