Add single-user database recovery gate

ace/application/intelligence_builder.py

@@ -374,7 +374,11 @@ async def load_artifact(
             raise IntelligenceBuilderSessionError("onboarding artifact is missing or has conflicting records")
         record = matches[0]
         try:
-            artifact = artifact_type.model_validate(record.payload)
+            # ImmutableRecord payloads reopen from canonical JSON in the Surreal
+            # adapter, so strict contracts must first rehydrate tuple/datetime
+            # representations. Their own digest validators still enforce exact
+            # identity after this JSON-compatible coercion.
+            artifact = artifact_type.model_validate(record.payload, strict=False)
             artifact_id, artifact_digest, occurred_at = _artifact_material(artifact)
         except Exception:
             raise IntelligenceBuilderSessionError("persisted onboarding artifact failed revalidation") from None
@@ -413,7 +417,11 @@ async def _replay(
                 record_space=INTELLIGENCE_BUILDER_RECORD_SPACE,
                 record_kind=ONBOARDING_SESSION_REVISION_RECORD_KIND,
             )
-            persisted = None if record is None else IntelligenceBuilderSessionRevisionV1.model_validate(record.payload)
+            persisted = (
+                None
+                if record is None
+                else IntelligenceBuilderSessionRevisionV1.model_validate(record.payload, strict=False)
+            )
         except Exception:
             raise IntelligenceBuilderSessionError("onboarding replay failed exact record validation") from None
         if (
@@ -499,7 +507,7 @@ async def load_latest(
         revisions: list[IntelligenceBuilderSessionRevisionV1] = []
         for record in records:
             try:
-                revision = IntelligenceBuilderSessionRevisionV1.model_validate(record.payload)
+                revision = IntelligenceBuilderSessionRevisionV1.model_validate(record.payload, strict=False)
             except Exception:
                 raise IntelligenceBuilderSessionError("persisted onboarding revision failed revalidation") from None
             if revision.session_id != session_id:

ace/testing/intelligence_builder.py

@@ -25,6 +25,7 @@
     SourceValueKind,
 )
 from ace.core.contracts import canonical_hash
+from ace.core.records import ImmutableRecordStore
 from ace.core.state import ResolvedApprovalReceiptV1
 from ace.testing.immutable_records import InMemoryImmutableRecordStore
 
@@ -125,7 +126,7 @@ class ConnectionAgentReferenceResult:
     restarted_scope: SourceScopeProposalV1
     restarted_profile: SourceProfileProposalV1
     provider: FixtureRegisteredSourceOptionProvider
-    store: InMemoryImmutableRecordStore
+    store: ImmutableRecordStore
 
 
 def provider_free_source_catalog() -> tuple[SourceOptionCatalogV1, tuple[FixtureSourceProfile, ...]]:
@@ -196,10 +197,12 @@ def provider_free_source_catalog() -> tuple[SourceOptionCatalogV1, tuple[Fixture
     return catalog, profiles
 
 
-async def exercise_connection_agent_restart() -> ConnectionAgentReferenceResult:
+async def exercise_connection_agent_restart(
+    *, store: ImmutableRecordStore | None = None
+) -> ConnectionAgentReferenceResult:
     """Run Connect over two fixture sources and reopen the exact durable session."""
 
-    store = InMemoryImmutableRecordStore()
+    store = store or InMemoryImmutableRecordStore()
     catalog, profiles = provider_free_source_catalog()
     provider = FixtureRegisteredSourceOptionProvider(catalog=catalog, profiles=profiles)
     approval_ref = "approval:fixture-source-scope"

ace/testing/ontology_agent.py

@@ -189,10 +189,10 @@ def edited_fixture_proposal(
     )
 
 
-async def exercise_ontology_agent_restart() -> OntologyAgentReferenceResult:
+async def exercise_ontology_agent_restart(*, store: ImmutableRecordStore | None = None) -> OntologyAgentReferenceResult:
     """Connect, Map, edit, approve, and reopen exact proposal/disposition material."""
 
-    connected = await exercise_connection_agent_restart()
+    connected = await exercise_connection_agent_restart(store=store)
     sessions = IntelligenceBuilderSessionService(store=connected.store)
     approval_ref = "approval:fixture-concept-model"
     authority = FixtureCoreAuthorityResolver(approved_receipt_refs=(approval_ref,))

ace/testing/watch_brief.py

@@ -40,6 +40,7 @@
 )
 from ace.application.intelligence_builder import IntelligenceBuilderSessionService
 from ace.application.intelligence_builder_contracts import OnboardingArtifactKind, OnboardingStage
+from ace.core.records import ImmutableRecordStore
 from ace.intelligence.contracts.resources import CanonicalJsonValueV1Alpha1
 from ace.testing.intelligence_builder import FixtureCoreAuthorityResolver
 from ace.testing.ontology_agent import OntologyAgentReferenceResult, exercise_ontology_agent_restart
@@ -477,10 +478,10 @@ class WatchBriefReferenceResult:
     restarted_brief: FirstBriefingPreviewV1
 
 
-async def exercise_watch_brief_restart() -> WatchBriefReferenceResult:
+async def exercise_watch_brief_restart(*, store: ImmutableRecordStore | None = None) -> WatchBriefReferenceResult:
     """Run Connect -> Map -> Watch edit/approve -> Brief -> restart with exact identities."""
 
-    mapped = await exercise_ontology_agent_restart()
+    mapped = await exercise_ontology_agent_restart(store=store)
     sessions = IntelligenceBuilderSessionService(store=mapped.store)
     approval_ref = "approval:fixture-intelligence-model"
     authority = FixtureCoreAuthorityResolver(approved_receipt_refs=(approval_ref,))

core/engine/cli/commands/recovery.py

@@ -0,0 +1,81 @@
+"""`ace recovery` — native full-store backup and clean-target restore."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from dataclasses import asdict
+from pathlib import Path
+
+import click
+
+from core.engine.core.recovery import (
+    DatabaseRecoveryError,
+    create_database_backup,
+    restore_database_backup,
+    target_from_settings,
+)
+
+
+@click.group("recovery")
+def recovery() -> None:
+    """Back up or restore the complete ACE database for single-user recovery.
+
+    These commands do not export environment configuration, connector credentials,
+    external secret stores, or source bodies that ACE did not persist.
+    """
+
+
+@recovery.command("backup")
+@click.argument("output", type=click.Path(path_type=Path, dir_okay=False))
+@click.option("--manifest", type=click.Path(path_type=Path, dir_okay=False))
+def backup(output: Path, manifest: Path | None) -> None:
+    """Write a native full-database export and checksum manifest to new files."""
+
+    try:
+        result = asyncio.run(
+            create_database_backup(
+                output,
+                manifest_path=manifest,
+                target=target_from_settings(),
+            )
+        )
+    except DatabaseRecoveryError as exc:
+        raise click.ClickException(str(exc)) from exc
+    click.echo(json.dumps(asdict(result), indent=2, sort_keys=True))
+
+
+@recovery.command("restore")
+@click.argument("export", type=click.Path(path_type=Path, dir_okay=False, exists=True))
+@click.option("--manifest", type=click.Path(path_type=Path, dir_okay=False, exists=True))
+@click.option("--target-namespace", required=True, help="Explicit empty destination namespace.")
+@click.option("--target-database", required=True, help="Explicit empty destination database.")
+def restore(
+    export: Path,
+    manifest: Path | None,
+    target_namespace: str,
+    target_database: str,
+) -> None:
+    """Verify and import EXPORT only into a clean explicit destination."""
+
+    try:
+        result = asyncio.run(
+            restore_database_backup(
+                export,
+                manifest_path=manifest,
+                target=target_from_settings(
+                    namespace=target_namespace,
+                    database=target_database,
+                ),
+            )
+        )
+    except DatabaseRecoveryError as exc:
+        raise click.ClickException(str(exc)) from exc
+    click.echo(json.dumps(asdict(result), indent=2, sort_keys=True))
+
+
+__all__ = ["recovery"]
+
+
+if __name__ == "__main__":
+    recovery()

core/engine/core/immutable_records.py

@@ -197,6 +197,7 @@ async def read_as_of(
             rows = parse_rows(
                 await db.query(
                     "SELECT payload_json, available_at, stable_id FROM immutable_record "
+                    "WITH INDEX immutable_record_scope_key "
                     "WHERE product = $product "
                     "AND record_space = $record_space AND record_kind = $record_kind "
                     "AND available_at <= $available_at ORDER BY available_at, stable_id",
@@ -244,7 +245,8 @@ async def count_as_of(
         async with self.pool.connection() as db:
             row = parse_one(
                 await db.query(
-                    "SELECT count() AS total FROM immutable_record WHERE product = $product "
+                    "SELECT count() AS total FROM immutable_record "
+                    "WITH INDEX immutable_record_scope_key WHERE product = $product "
                     "AND record_space = $record_space AND record_kind = $record_kind "
                     "AND available_at <= $available_at GROUP ALL",
                     {