diff --git a/src/api/imports.py b/src/api/imports.py index bd87eec..4e23be5 100644 --- a/src/api/imports.py +++ b/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 \ No newline at end of file diff --git a/src/importers/runner.py b/src/importers/runner.py index 562c4f6..5eb9483 100644 --- a/src/importers/runner.py +++ b/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, - ) + ) \ No newline at end of file diff --git a/src/importers/staging.py b/src/importers/staging.py new file mode 100644 index 0000000..e2ae5a6 --- /dev/null +++ b/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()) diff --git a/tests/test_import_runner.py b/tests/test_import_runner.py index 4f0b79d..828f96d 100644 --- a/tests/test_import_runner.py +++ b/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 \ No newline at end of file