From b54906e3a0c51a74d5d317d52953afe265947306 Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Thu, 13 Aug 2026 16:19:00 -0700 Subject: [PATCH] feat: add single-user database recovery gate --- ace/application/intelligence_builder.py | 14 +- ace/testing/intelligence_builder.py | 9 +- ace/testing/ontology_agent.py | 4 +- ace/testing/watch_brief.py | 5 +- core/engine/cli/commands/recovery.py | 81 ++++ core/engine/core/immutable_records.py | 4 +- core/engine/core/recovery.py | 446 ++++++++++++++++++ docs/evidence/v1-single-user-recovery-gate.md | 77 +++ docs/state-engine-operations.md | 32 +- .../test_builder_database_recovery.py | 268 +++++++++++ tests/test_database_recovery.py | 193 ++++++++ 11 files changed, 1119 insertions(+), 14 deletions(-) create mode 100644 core/engine/cli/commands/recovery.py create mode 100644 core/engine/core/recovery.py create mode 100644 docs/evidence/v1-single-user-recovery-gate.md create mode 100644 tests/intelligence/test_builder_database_recovery.py create mode 100644 tests/test_database_recovery.py diff --git a/ace/application/intelligence_builder.py b/ace/application/intelligence_builder.py index a6a37f1..92cd53f 100644 --- a/ace/application/intelligence_builder.py +++ b/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: diff --git a/ace/testing/intelligence_builder.py b/ace/testing/intelligence_builder.py index 3c430c9..8ac1e1d 100644 --- a/ace/testing/intelligence_builder.py +++ b/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" diff --git a/ace/testing/ontology_agent.py b/ace/testing/ontology_agent.py index 69c873a..dc568a2 100644 --- a/ace/testing/ontology_agent.py +++ b/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,)) diff --git a/ace/testing/watch_brief.py b/ace/testing/watch_brief.py index 1ffbfae..5ef6491 100644 --- a/ace/testing/watch_brief.py +++ b/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,)) diff --git a/core/engine/cli/commands/recovery.py b/core/engine/cli/commands/recovery.py new file mode 100644 index 0000000..93a224f --- /dev/null +++ b/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() diff --git a/core/engine/core/immutable_records.py b/core/engine/core/immutable_records.py index eea56e8..ac648cd 100644 --- a/core/engine/core/immutable_records.py +++ b/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", { diff --git a/core/engine/core/recovery.py b/core/engine/core/recovery.py new file mode 100644 index 0000000..e42de1e --- /dev/null +++ b/core/engine/core/recovery.py @@ -0,0 +1,446 @@ +"""Operational database-state backup and clean-target restore for one ACE store. + +This is runnable recovery, not product-scoped data portability. SurrealDB owns +record serialization; ACE recreates its exact packaged schema before importing +those records because historical migration artifacts can contain stale database +definitions that SurrealDB exports but cannot import into a fresh database. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +import subprocess +import tempfile +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +from surrealdb import AsyncSurreal + +DATABASE_BACKUP_MANIFEST_CONTRACT = "ace.database-backup-manifest/v1" +DATABASE_RESTORE_RECEIPT_CONTRACT = "ace.database-restore-receipt/v1" + + +class DatabaseRecoveryError(RuntimeError): + """A backup or restore could not preserve the declared recovery boundary.""" + + +@dataclass(frozen=True, slots=True) +class DatabaseTarget: + endpoint: str + namespace: str + database: str + username: str + password: str + + +@dataclass(frozen=True, slots=True) +class DatabaseBackupManifest: + contract: str + created_at: str + ace_version: str + surreal_cli_version: str + schema_version: int + namespace: str + database: str + export_filename: str + export_sha256: str + export_size_bytes: int + includes: tuple[str, ...] + excludes: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class DatabaseRestoreReceipt: + contract: str + restored_at: str + ace_version: str + source_ace_version: str + source_schema_version: int + restored_schema_version: int + target_namespace: str + target_database: str + export_sha256: str + export_size_bytes: int + target_was_clean: bool + manifest_verified: bool + + +def _ace_distribution_version() -> str: + try: + return version("ace-core") + except PackageNotFoundError: + return "unknown" + + +def target_from_settings(*, namespace: str | None = None, database: str | None = None) -> DatabaseTarget: + from core.engine.core.config import settings + + return DatabaseTarget( + endpoint=settings.surreal_url, + namespace=namespace or settings.surreal_ns, + database=database or settings.surreal_db, + username=settings.surreal_user, + password=settings.surreal_pass, + ) + + +def _http_endpoint(endpoint: str) -> str: + """Use the CLI's HTTP endpoint while retaining host, port, path, and redaction.""" + + parts = urlsplit(endpoint) + scheme = {"ws": "http", "wss": "https"}.get(parts.scheme, parts.scheme) + host = parts.hostname or "" + if parts.port: + host = f"{host}:{parts.port}" + return urlunsplit((scheme, host, parts.path, parts.query, parts.fragment)) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return f"sha256:{digest.hexdigest()}" + + +def _surreal_binary() -> str: + configured = os.environ.get("ACE_SURREAL_BIN", "").strip() + binary = configured or shutil.which("surreal") + if not binary: + raise DatabaseRecoveryError( + "SurrealDB CLI is required for native backup/restore; install `surreal` or set ACE_SURREAL_BIN" + ) + return binary + + +def _surreal_env(target: DatabaseTarget) -> dict[str, str]: + """Authenticate through the child environment so passwords never enter argv.""" + + return os.environ | { + "SURREAL_USER": target.username, + "SURREAL_PASS": target.password, + "SURREAL_NAMESPACE": target.namespace, + "SURREAL_DATABASE": target.database, + } + + +def _run_native(arguments: list[str], *, target: DatabaseTarget) -> subprocess.CompletedProcess[str]: + try: + return subprocess.run( + [_surreal_binary(), *arguments], + env=_surreal_env(target), + capture_output=True, + text=True, + check=True, + ) + except subprocess.CalledProcessError as exc: + detail = (exc.stderr or exc.stdout or "native SurrealDB operation failed").strip() + raise DatabaseRecoveryError(detail[:2_000]) from exc + + +async def _connect(target: DatabaseTarget) -> AsyncSurreal: + db = AsyncSurreal(target.endpoint) + await db.connect() + await db.signin({"username": target.username, "password": target.password}) + await db.use(target.namespace, target.database) + return db + + +def _one_mapping(result: Any) -> dict[str, Any]: + value = result + while isinstance(value, list) and len(value) == 1: + value = value[0] + return value if isinstance(value, dict) else {} + + +async def database_schema_version(target: DatabaseTarget) -> int: + db = await _connect(target) + try: + result = await db.query("SELECT * FROM config_entry WHERE key = 'schema_version'") + rows = result[0] if isinstance(result, list) and result and isinstance(result[0], list) else result + if not rows: + return 0 + row = rows[0] if isinstance(rows, list) else rows + return int(row.get("value", 0)) if isinstance(row, dict) else 0 + finally: + await db.close() + + +async def database_is_clean(target: DatabaseTarget) -> bool: + """A restore target is clean only when it has no database-level definitions.""" + + db = await _connect(target) + try: + info = _one_mapping(await db.query("INFO FOR DB")) + finally: + await db.close() + if not info: + return True + for key, value in info.items(): + if key in {"name"}: + continue + if value not in ({}, [], (), None, ""): + return False + return True + + +def _manifest_path(export_path: Path, manifest_path: Path | None) -> Path: + return manifest_path or export_path.with_name(f"{export_path.name}.manifest.json") + + +def _write_json_atomically(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle: + temporary = Path(handle.name) + json.dump(payload, handle, indent=2, sort_keys=True) + handle.write("\n") + try: + temporary.replace(path) + except BaseException: + temporary.unlink(missing_ok=True) + raise + + +def _extract_native_records(native_export: Path, records_export: Path) -> None: + """Retain only complete native INSERT statements from one Surreal export.""" + + inserts: list[str] = [] + for line in native_export.read_text(encoding="utf-8").splitlines(): + if line.startswith("INSERT "): + if not line.rstrip().endswith(";"): + raise DatabaseRecoveryError("native export contains an unsupported multiline record statement") + inserts.append(line) + if not inserts: + raise DatabaseRecoveryError("native export contains no database records") + records_export.write_text( + "-- ACE record-complete recovery export; schema is recreated from the exact packaged version.\n" + "OPTION IMPORT;\n\n" + "\n".join(inserts) + "\n", + encoding="utf-8", + ) + + +def _packaged_schema_files() -> tuple[tuple[int, Path], ...]: + from core.engine.core.schema import SCHEMA_DIR + + files: list[tuple[int, Path]] = [] + for path in sorted(SCHEMA_DIR.glob("v*.surql")): + match = re.search(r"v(\d+)", path.name) + if match: + files.append((int(match.group(1)), path)) + if not files: + raise DatabaseRecoveryError("packaged ACE schema migrations are unavailable") + return tuple(files) + + +async def _install_packaged_schema(target: DatabaseTarget, *, expected_version: int) -> None: + from scripts.schema_apply import apply_file, validate_schema + + files = _packaged_schema_files() + code_version = max(version for version, _ in files) + if code_version != expected_version: + raise DatabaseRecoveryError( + "backup schema does not match this ACE installation; restore with the recorded ACE version first" + ) + db = await _connect(target) + try: + for version, path in files: + await apply_file(db, version, path.name, path.read_text(encoding="utf-8")) + await db.query( + "UPSERT config_entry SET key = 'schema_version', value = $version WHERE key = 'schema_version'", + {"version": str(version)}, + ) + await validate_schema(db, expected_version) + info = _one_mapping(await db.query("INFO FOR DB")) + tables = sorted((info.get("tables") or {}).keys()) + if not tables or any(not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) for name in tables): + raise DatabaseRecoveryError("packaged schema exposed an unsafe table identity") + # Migrations can create deterministic seed/config rows. Recovery replaces + # every table's contents with the exact source snapshot after definitions + # are installed, so those temporary rows must not collide with native INSERT. + for name in tables: + await db.query(f"DELETE `{name}`") + except DatabaseRecoveryError: + raise + except Exception as exc: + raise DatabaseRecoveryError( + "packaged schema preparation failed; discard the partial target database before retrying" + ) from exc + finally: + await db.close() + + +def load_backup_manifest(path: Path) -> DatabaseBackupManifest: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + manifest = DatabaseBackupManifest(**payload) + except (OSError, json.JSONDecodeError, TypeError, ValueError) as exc: + raise DatabaseRecoveryError("backup manifest is missing or invalid") from exc + if manifest.contract != DATABASE_BACKUP_MANIFEST_CONTRACT: + raise DatabaseRecoveryError("backup manifest contract is unsupported") + if manifest.schema_version < 1 or manifest.export_size_bytes < 1: + raise DatabaseRecoveryError("backup manifest does not describe a usable database export") + return manifest + + +async def create_database_backup( + export_path: Path, + *, + target: DatabaseTarget, + manifest_path: Path | None = None, + created_at: datetime | None = None, +) -> DatabaseBackupManifest: + """Create one native full-database export plus an adjacent immutable manifest.""" + + export_path = export_path.expanduser().resolve() + manifest_path = _manifest_path(export_path, manifest_path).expanduser().resolve() + if export_path.exists() or manifest_path.exists(): + raise DatabaseRecoveryError("backup output already exists; choose a new path") + export_path.parent.mkdir(parents=True, exist_ok=True) + schema_version = await database_schema_version(target) + if schema_version < 1: + raise DatabaseRecoveryError("source database has no supported ACE schema version") + version = _run_native(["version"], target=target).stdout.strip() + with tempfile.NamedTemporaryFile(dir=export_path.parent, delete=False) as handle: + temporary = Path(handle.name) + with tempfile.NamedTemporaryFile(dir=export_path.parent, delete=False) as handle: + native_temporary = Path(handle.name) + try: + _run_native( + [ + "export", + "--endpoint", + _http_endpoint(target.endpoint), + "--namespace", + target.namespace, + "--database", + target.database, + str(native_temporary), + ], + target=target, + ) + _extract_native_records(native_temporary, temporary) + size = temporary.stat().st_size + if size < 1: + raise DatabaseRecoveryError("native database export is empty") + digest = _sha256(temporary) + temporary.replace(export_path) + except BaseException: + temporary.unlink(missing_ok=True) + raise + finally: + native_temporary.unlink(missing_ok=True) + timestamp = (created_at or datetime.now(UTC)).astimezone(UTC) + manifest = DatabaseBackupManifest( + contract=DATABASE_BACKUP_MANIFEST_CONTRACT, + created_at=timestamp.isoformat(), + ace_version=_ace_distribution_version(), + surreal_cli_version=version, + schema_version=schema_version, + namespace=target.namespace, + database=target.database, + export_filename=export_path.name, + export_sha256=digest, + export_size_bytes=size, + includes=( + "all_surrealdb_table_records", + "ace_packaged_schema_at_recorded_version", + ), + excludes=( + "surrealdb_database_users_and_access_definitions", + "environment_configuration", + "external_connector_credentials", + "external_secret_stores", + "external_source_bodies_not_persisted_in_surrealdb", + ), + ) + try: + _write_json_atomically(manifest_path, asdict(manifest)) + except BaseException: + export_path.unlink(missing_ok=True) + raise + return manifest + + +async def restore_database_backup( + export_path: Path, + *, + manifest_path: Path | None, + target: DatabaseTarget, + restored_at: datetime | None = None, +) -> DatabaseRestoreReceipt: + """Verify and import a native export only into a demonstrably clean database.""" + + export_path = export_path.expanduser().resolve() + manifest_path = _manifest_path(export_path, manifest_path).expanduser().resolve() + if not export_path.is_file(): + raise DatabaseRecoveryError("database export does not exist") + manifest = load_backup_manifest(manifest_path) + if manifest.export_filename != export_path.name: + raise DatabaseRecoveryError("backup manifest names a different export file") + if export_path.stat().st_size != manifest.export_size_bytes or _sha256(export_path) != manifest.export_sha256: + raise DatabaseRecoveryError("database export checksum or size does not match its manifest") + if not await database_is_clean(target): + raise DatabaseRecoveryError("restore target is not clean; use a new empty namespace/database") + await _install_packaged_schema(target, expected_version=manifest.schema_version) + try: + _run_native( + [ + "import", + "--endpoint", + _http_endpoint(target.endpoint), + "--namespace", + target.namespace, + "--database", + target.database, + str(export_path), + ], + target=target, + ) + except DatabaseRecoveryError as exc: + raise DatabaseRecoveryError( + "native restore failed; treat the target database as partial and discard it before retrying" + ) from exc + restored_schema = await database_schema_version(target) + if restored_schema != manifest.schema_version: + raise DatabaseRecoveryError( + "restored schema version does not match the verified backup; quarantine the restored database" + ) + timestamp = (restored_at or datetime.now(UTC)).astimezone(UTC) + return DatabaseRestoreReceipt( + contract=DATABASE_RESTORE_RECEIPT_CONTRACT, + restored_at=timestamp.isoformat(), + ace_version=_ace_distribution_version(), + source_ace_version=manifest.ace_version, + source_schema_version=manifest.schema_version, + restored_schema_version=restored_schema, + target_namespace=target.namespace, + target_database=target.database, + export_sha256=manifest.export_sha256, + export_size_bytes=manifest.export_size_bytes, + target_was_clean=True, + manifest_verified=True, + ) + + +__all__ = [ + "DATABASE_BACKUP_MANIFEST_CONTRACT", + "DATABASE_RESTORE_RECEIPT_CONTRACT", + "DatabaseBackupManifest", + "DatabaseRecoveryError", + "DatabaseRestoreReceipt", + "DatabaseTarget", + "create_database_backup", + "database_is_clean", + "database_schema_version", + "load_backup_manifest", + "restore_database_backup", + "target_from_settings", +] diff --git a/docs/evidence/v1-single-user-recovery-gate.md b/docs/evidence/v1-single-user-recovery-gate.md new file mode 100644 index 0000000..96d9b99 --- /dev/null +++ b/docs/evidence/v1-single-user-recovery-gate.md @@ -0,0 +1,77 @@ +# ACE v1 single-user recovery gate + +**Status:** passed on 2026-08-13 + +**Base:** Core `main` at `0948db68af3f3915132baed35b40549e305a35ea` + +**Topology:** one disposable SurrealDB 3.2.1 SurrealKV store, one product, fresh application service +instances, no model or external network calls + +## Promise tested + +A single operator can persist the domain-neutral Connect → Map → Watch → Brief Builder chain, stop +the database process, reopen the exact chain through fresh services, append a later immutable +`activation_pending` revision, back up the complete ACE database record state, restore into a clean +database, and reopen the same resource page. Database recovery stays distinct from product-scoped +portability and makes no claim that connector credentials or external source bodies are included. + +## Reproduction + +```bash +pytest tests/intelligence/test_builder_database_recovery.py -q --tb=short +pytest tests/test_database_recovery.py \ + tests/intelligence/test_intelligence_builder_connect.py \ + tests/intelligence/test_ontology_agent_map.py \ + tests/intelligence/test_briefing_agent_first_brief.py \ + tests/intelligence/test_intelligence_builder_resource_projection.py \ + -q --tb=short +``` + +Observed result: + +- real restart/append/backup/restore after final restack: `1 passed in 77.02s`; +- focused contracts and failure controls: `34 passed`; +- final restacked focused recovery plus public/kernel boundary gate: `82 passed, 2 skipped`; +- a built `ace_core-0.8.2` wheel contains both recovery modules, and the extracted wheel exposes + `python -m core.engine.cli.commands.recovery --help` with bounded `backup` and `restore` commands; +- Ruff lint and format checks passed for every changed Python file. + +The pre-restack broad non-E2E Core run at base +`7142804e2e3479c8fdbbd062614803300fb7fb4e` reached `7897 passed, 50 skipped, 262 deselected`. It reported one +transient SurrealDB transaction conflict that passed on immediate isolated rerun, plus +`test_real_database_fresh_process_orphans_admission_without_reexecuting`, whose child-process import +fails from a linked worktree. The latter reproduces at exact unmodified base +`7142804e2e3479c8fdbbd062614803300fb7fb4e`; it is not introduced by this recovery lane. + +## Exact assertions + +1. The first persisted page is complete and ends at `first_briefing_ready`. +2. The SurrealDB process stops and restarts against the same SurrealKV path. +3. A fresh `IntelligenceBuilderSessionService` reopens the exact prior revision and a fresh resource + reader emits byte-identical JSON. +4. A later service appends `activation_pending`; every prior resource item remains exactly unchanged + and the new revision names its immediate predecessor. +5. Backup produces a non-empty native record export and adjacent manifest containing ACE/Surreal + versions, schema v177, size, and SHA-256. +6. Restore requires a clean, explicit namespace/database, installs the exact packaged schema, + restores records, and verifies the same schema version. +7. The resource page rebuilt from the restored database is byte-identical to the post-append page. +8. Existing outputs, dirty targets, checksum drift, unsupported manifests, schema mismatch, and + partial native import all fail closed. + +## Defects found by the real-store gate + +- SurrealDB 3.2 selected an inapplicable partial path for the Builder historical read when no index + was named. The immutable-record adapter now selects `immutable_record_scope_key` explicitly for + scoped historical reads and counts. +- Canonical JSON persistence necessarily represents tuples as arrays and datetimes as strings. + Strict Builder contracts now perform JSON-compatible rehydration before their existing digest and + identity validators run; exact material checks remain unchanged. + +## Recovery boundary + +The artifact includes all SurrealDB table records. ACE definitions are recreated from the exact +recorded package version because generated full-schema exports can contain stale historical index +definitions that SurrealDB accepts at runtime but rejects on clean import. Database users/access +definitions, environment configuration, provider or connector credentials, external secret stores, +and non-persisted source bodies are excluded and explicitly named in the manifest. diff --git a/docs/state-engine-operations.md b/docs/state-engine-operations.md index a8e64a0..fc5d826 100644 --- a/docs/state-engine-operations.md +++ b/docs/state-engine-operations.md @@ -96,9 +96,35 @@ not weaken legacy error handling or skip statements manually. ## Backup and restore -Export the database using the supported SurrealDB export path and record file size, duration, -database version, namespace/database, schema head, and a digest. Restore into a clean disposable -store first. Then verify: +For the supported single-user recovery path, use ACE's packaged wrapper rather than replaying a raw +SurrealDB full export directly: + +```bash +python -m core.engine.cli.commands.recovery backup ./ace-backup.surql +python -m core.engine.cli.commands.recovery restore ./ace-backup.surql \ + --target-namespace ace_restore \ + --target-database ace_restore +``` + +`backup` refuses to overwrite either output, serializes every SurrealDB table record with the native +CLI, and writes `ace-backup.surql.manifest.json` with the ACE version, Surreal CLI version, schema +head, byte count, and SHA-256 digest. `restore` verifies that manifest and checksum, requires an +explicit database with no definitions, rebuilds the exact recorded packaged ACE schema, removes +migration seed rows, imports the native record snapshot, and checks the restored schema head. +Pause ACE ingestion and other writers while `backup` runs so the recovery point has one explicit +operational boundary. + +The packaged schema is deliberately authoritative. Historical migrations can leave valid runtime +indexes that reference retired fields; SurrealDB can export those generated definitions but then +reject the same definitions during a clean import. ACE therefore retains SurrealDB's native record +serialization while rebuilding definitions from the exact matching package. A schema-version +mismatch fails before the destination is changed. Any failure after schema preparation makes the +destination partial; discard that destination and retry with a new empty database. + +This is runnable database recovery, not data portability. It does **not** include `.env`, provider or +connector credentials, external secret stores, or external source bodies that were never persisted +in SurrealDB. Back those up through their owning systems. Restore into a clean disposable store +first. Then verify: - exact semantic-family, item, batch, and lineage counts; - lifecycle and authoritative promotion states; diff --git a/tests/intelligence/test_builder_database_recovery.py b/tests/intelligence/test_builder_database_recovery.py new file mode 100644 index 0000000..3697466 --- /dev/null +++ b/tests/intelligence/test_builder_database_recovery.py @@ -0,0 +1,268 @@ +"""Real SurrealKV restart and native backup/restore proof for the Builder chain.""" + +from __future__ import annotations + +import asyncio +import os +import shutil +import socket +import subprocess +import sys +import time +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from surrealdb import AsyncSurreal + +from ace.application import ( + IntelligenceBuilderSessionService, + IntelligenceResourcePlaneService, +) +from ace.application.intelligence_builder_contracts import ( + OnboardingStage, + OnboardingTransitionAuthority, +) +from ace.core.runtime_use import ( + AuthenticatedRuntimeContextV1Alpha1, + AuthorityUseReceiptV1Alpha1, +) +from ace.core.state import GovernedStateHeadPreconditionV1Alpha1 +from ace.intelligence import IntelligenceResourceKind, IntelligenceResourceQueryV1Alpha1 +from ace.testing.watch_brief import exercise_watch_brief_restart +from core.engine.core.immutable_records import SurrealImmutableRecordStore +from core.engine.core.intelligence_resource_plane import intelligence_resource_projection_reader +from core.engine.core.recovery import ( + DatabaseTarget, + create_database_backup, + restore_database_backup, +) + +pytestmark = pytest.mark.e2e + +ROOT = Path(__file__).parents[2] +PRODUCT = "product:intelligence-builder-fixture" +EVALUATED_AT = datetime(2026, 8, 14, 12, 0, tzinfo=UTC) + + +def _port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +async def _wait_port(port: int, process: subprocess.Popen, timeout: float = 20) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError("disposable SurrealDB exited before accepting connections") + try: + _, writer = await asyncio.open_connection("127.0.0.1", port) + writer.close() + await writer.wait_closed() + return + except OSError: + await asyncio.sleep(0.1) + raise RuntimeError("disposable SurrealDB did not accept connections") + + +async def _stop(process: subprocess.Popen | None) -> None: + if process is None or process.poll() is not None: + return + process.terminate() + try: + await asyncio.to_thread(process.wait, 10) + except subprocess.TimeoutExpired: + process.kill() + await asyncio.to_thread(process.wait) + + +class _SingleConnectionPool: + def __init__(self, target: DatabaseTarget) -> None: + self.target = target + self.db: AsyncSurreal | None = None + + async def open(self) -> None: + db = AsyncSurreal(self.target.endpoint) + await db.connect() + await db.signin({"username": self.target.username, "password": self.target.password}) + await db.use(self.target.namespace, self.target.database) + self.db = db + + async def close(self) -> None: + if self.db is not None: + await self.db.close() + self.db = None + + @asynccontextmanager + async def connection(self): + if self.db is None: + raise RuntimeError("fixture pool is closed") + yield self.db + + +class _Authority: + async def resolve_authority_use(self, **kwargs) -> AuthorityUseReceiptV1Alpha1: + return AuthorityUseReceiptV1Alpha1( + product_id=kwargs["context"].product_id, + actor_ref=kwargs["context"].actor_ref, + authenticated_context=kwargs["context"], + use_subject_ref=kwargs["use_subject_ref"], + use_subject_digest=kwargs["use_subject_digest"], + operation=kwargs["operation"], + authority=kwargs["authority"], + grant_ref=kwargs["grant_ref"], + grant_hash="d" * 64, + evaluated_at=kwargs["evaluated_at"], + expires_at=EVALUATED_AT + timedelta(hours=1), + state_head_precondition=GovernedStateHeadPreconditionV1Alpha1( + state_kind="authority_grant", + product_id=kwargs["context"].product_id, + state_id=kwargs["grant_ref"], + sequence=1, + revision_id="authority_revision:recovery-read", + commit_receipt_id="authority_receipt:recovery-read", + ), + ) + + +def _query() -> IntelligenceResourceQueryV1Alpha1: + return IntelligenceResourceQueryV1Alpha1( + authenticated_context=AuthenticatedRuntimeContextV1Alpha1( + product_id=PRODUCT, + actor_ref="principal:fixture-builder", + authentication_receipt_ref="authentication_receipt:recovery-fixture", + authentication_receipt_digest="sha256:" + "a" * 64, + authenticated_at=EVALUATED_AT - timedelta(hours=1), + expires_at=EVALUATED_AT + timedelta(hours=2), + ), + product_id=PRODUCT, + authority_grant_ref="authority_grant:recovery-read", + resource_kinds=(IntelligenceResourceKind.BUILDER_SESSION,), + as_of=EVALUATED_AT, + available_at=EVALUATED_AT, + page_size=200, + ) + + +async def _page(store: SurrealImmutableRecordStore): + return await IntelligenceResourcePlaneService( + reader=intelligence_resource_projection_reader(store), + authority=_Authority(), + ).query(_query(), evaluated_at=EVALUATED_AT) + + +def _surreal_process(binary: str, port: int, store: Path, log) -> subprocess.Popen: + return subprocess.Popen( + [ + binary, + "start", + "--no-banner", + "--username", + "root", + "--password", + "root", + "--bind", + f"127.0.0.1:{port}", + f"surrealkv://{store}", + ], + cwd=ROOT, + stdout=log, + stderr=subprocess.STDOUT, + ) + + +@pytest.mark.asyncio +async def test_builder_chain_reopens_appends_and_restores_from_native_backup(tmp_path): + surreal = os.environ.get("ACE_SURREAL_BIN") or shutil.which("surreal") + if not surreal: + pytest.skip("surreal binary is unavailable") + port = _port() + endpoint = f"ws://127.0.0.1:{port}" + source = DatabaseTarget(endpoint, "ace_builder_recovery", "ace_builder_recovery", "root", "root") + restored = DatabaseTarget(endpoint, "ace_builder_restored", "ace_builder_restored", "root", "root") + database_store = tmp_path / "surrealkv" + log = (tmp_path / "surreal.log").open("wb") + process = _surreal_process(surreal, port, database_store, log) + pool: _SingleConnectionPool | None = None + try: + await _wait_port(port, process) + env = os.environ | { + "SURREAL_URL": endpoint, + "SURREAL_NS": source.namespace, + "SURREAL_DB": source.database, + "SURREAL_USER": source.username, + "SURREAL_PASS": source.password, + "JWT_SECRET": "builder-recovery-fixture-secret-at-least-32-bytes", + "LLM_API_KEY": "sk-test-placeholder", + } + await asyncio.to_thread( + subprocess.run, + [sys.executable, "scripts/schema_apply.py"], + cwd=ROOT, + env=env, + capture_output=True, + text=True, + check=True, + ) + + pool = _SingleConnectionPool(source) + await pool.open() + store = SurrealImmutableRecordStore(pool) + journey = await exercise_watch_brief_restart(store=store) + before_restart = await _page(store) + assert before_restart.state.value == "complete" + assert before_restart.items[-1].payload is not None + assert before_restart.items[-1].payload.parsed_value()["stage"] == "first_briefing_ready" + first_brief_revision = journey.briefing.session.revision + first_page_json = before_restart.model_dump_json() + await pool.close() + pool = None + + await _stop(process) + process = _surreal_process(surreal, port, database_store, log) + await _wait_port(port, process) + + pool = _SingleConnectionPool(source) + await pool.open() + restarted_store = SurrealImmutableRecordStore(pool) + restarted_sessions = IntelligenceBuilderSessionService(store=restarted_store) + reopened = await restarted_sessions.load_latest( + product_id=PRODUCT, + session_id=first_brief_revision.session_id, + available_at=EVALUATED_AT, + ) + assert reopened == first_brief_revision + assert (await _page(restarted_store)).model_dump_json() == first_page_json + + later = await restarted_sessions.advance( + reopened, + stage=OnboardingStage.ACTIVATION_PENDING, + authority=OnboardingTransitionAuthority.AGENT_PROPOSAL, + actor_ref="agent:activation-planner", + occurred_at=datetime(2026, 8, 11, 12, 4, tzinfo=UTC), + ) + after_append = await _page(restarted_store) + assert len(after_append.items) == len(before_restart.items) + 1 + assert after_append.items[:-1] == before_restart.items + assert after_append.items[-1].payload is not None + assert after_append.items[-1].payload.parsed_value()["stage"] == "activation_pending" + assert later.revision.prior_revision_id == reopened.revision_id + await pool.close() + pool = None + + export = tmp_path / "ace-builder.surql" + manifest = await create_database_backup(export, target=source) + receipt = await restore_database_backup(export, manifest_path=None, target=restored) + assert receipt.source_schema_version == receipt.restored_schema_version == manifest.schema_version + + pool = _SingleConnectionPool(restored) + await pool.open() + restored_page = await _page(SurrealImmutableRecordStore(pool)) + assert restored_page.model_dump_json() == after_append.model_dump_json() + finally: + if pool is not None: + await pool.close() + await _stop(process) + log.close() diff --git a/tests/test_database_recovery.py b/tests/test_database_recovery.py new file mode 100644 index 0000000..cb9f2f5 --- /dev/null +++ b/tests/test_database_recovery.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import json +import subprocess +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from core.engine.cli.commands.recovery import recovery +from core.engine.core import recovery as recovery_module +from core.engine.core.recovery import ( + DATABASE_BACKUP_MANIFEST_CONTRACT, + DATABASE_RESTORE_RECEIPT_CONTRACT, + DatabaseRecoveryError, + DatabaseTarget, + create_database_backup, + restore_database_backup, +) + +pytestmark = pytest.mark.unit + +TARGET = DatabaseTarget( + endpoint="ws://127.0.0.1:18001", + namespace="ace_source", + database="ace_source", + username="root", + password="not-in-argv", +) +DESTINATION = DatabaseTarget( + endpoint=TARGET.endpoint, + namespace="ace_restore", + database="ace_restore", + username=TARGET.username, + password=TARGET.password, +) + + +def _native_fixture(calls: list[list[str]]): + def run(arguments: list[str], *, target: DatabaseTarget): + calls.append(arguments) + assert target.password not in arguments + if arguments == ["version"]: + return subprocess.CompletedProcess(arguments, 0, "3.2.3 for fixture\n", "") + if arguments[0] == "export": + Path(arguments[-1]).write_text( + "OPTION IMPORT;\nDEFINE TABLE config_entry;\nINSERT [{ id: config_entry:fixture }];\n", + encoding="utf-8", + ) + return subprocess.CompletedProcess(arguments, 0, "", "") + + return run + + +@pytest.mark.asyncio +async def test_backup_writes_native_export_and_checksum_manifest_without_secrets(tmp_path, monkeypatch): + calls: list[list[str]] = [] + monkeypatch.setattr(recovery_module, "_run_native", _native_fixture(calls)) + + async def schema(_target): + return 177 + + monkeypatch.setattr(recovery_module, "database_schema_version", schema) + export = tmp_path / "ace-backup.surql" + manifest = await create_database_backup( + export, + target=TARGET, + created_at=datetime(2026, 8, 13, 12, 0, tzinfo=UTC), + ) + + assert "INSERT [{ id: config_entry:fixture }];" in export.read_text(encoding="utf-8") + assert "DEFINE TABLE" not in export.read_text(encoding="utf-8") + payload = json.loads((tmp_path / "ace-backup.surql.manifest.json").read_text()) + assert payload["contract"] == DATABASE_BACKUP_MANIFEST_CONTRACT + assert payload["schema_version"] == 177 + assert payload["export_sha256"] == manifest.export_sha256 + assert "external_connector_credentials" in payload["excludes"] + assert TARGET.password not in json.dumps(payload) + assert calls[1][:2] == ["export", "--endpoint"] + assert "http://127.0.0.1:18001" in calls[1] + + +@pytest.mark.asyncio +async def test_backup_refuses_to_overwrite_either_recovery_artifact(tmp_path): + export = tmp_path / "existing.surql" + export.write_text("keep", encoding="utf-8") + with pytest.raises(DatabaseRecoveryError, match="already exists"): + await create_database_backup(export, target=TARGET) + + +@pytest.mark.asyncio +async def test_restore_verifies_checksum_requires_clean_target_and_reports_schema(tmp_path, monkeypatch): + calls: list[list[str]] = [] + monkeypatch.setattr(recovery_module, "_run_native", _native_fixture(calls)) + schema_reads = iter((177,)) + + async def schema(_target): + return next(schema_reads) + + async def clean(_target): + return True + + async def install(_target, *, expected_version): + assert expected_version == 177 + + monkeypatch.setattr(recovery_module, "database_schema_version", schema) + monkeypatch.setattr(recovery_module, "database_is_clean", clean) + monkeypatch.setattr(recovery_module, "_install_packaged_schema", install) + export = tmp_path / "ace-backup.surql" + export.write_text("DEFINE TABLE config_entry;\n", encoding="utf-8") + digest = recovery_module._sha256(export) + (tmp_path / "ace-backup.surql.manifest.json").write_text( + json.dumps( + { + "contract": DATABASE_BACKUP_MANIFEST_CONTRACT, + "created_at": "2026-08-13T12:00:00+00:00", + "ace_version": "1.0.0", + "surreal_cli_version": "3.2.3", + "schema_version": 177, + "namespace": "ace_source", + "database": "ace_source", + "export_filename": export.name, + "export_sha256": digest, + "export_size_bytes": export.stat().st_size, + "includes": ["surrealdb_database_definitions", "surrealdb_database_records"], + "excludes": ["external_connector_credentials"], + } + ), + encoding="utf-8", + ) + + receipt = await restore_database_backup( + export, + manifest_path=None, + target=DESTINATION, + restored_at=datetime(2026, 8, 13, 12, 30, tzinfo=UTC), + ) + + assert receipt.contract == DATABASE_RESTORE_RECEIPT_CONTRACT + assert receipt.target_was_clean is True + assert receipt.manifest_verified is True + assert receipt.restored_schema_version == 177 + assert calls[-1][0] == "import" + + +@pytest.mark.asyncio +async def test_restore_rejects_dirty_target_before_native_import(tmp_path, monkeypatch): + export = tmp_path / "ace-backup.surql" + export.write_text("backup", encoding="utf-8") + (tmp_path / "ace-backup.surql.manifest.json").write_text( + json.dumps( + { + "contract": DATABASE_BACKUP_MANIFEST_CONTRACT, + "created_at": "2026-08-13T12:00:00+00:00", + "ace_version": "1.0.0", + "surreal_cli_version": "3.2.3", + "schema_version": 177, + "namespace": "ace_source", + "database": "ace_source", + "export_filename": export.name, + "export_sha256": recovery_module._sha256(export), + "export_size_bytes": export.stat().st_size, + "includes": [], + "excludes": [], + } + ), + encoding="utf-8", + ) + + async def dirty(_target): + return False + + monkeypatch.setattr(recovery_module, "database_is_clean", dirty) + called = False + + def forbidden(*_args, **_kwargs): + nonlocal called + called = True + + monkeypatch.setattr(recovery_module, "_run_native", forbidden) + with pytest.raises(DatabaseRecoveryError, match="not clean"): + await restore_database_backup(export, manifest_path=None, target=DESTINATION) + assert called is False + + +def test_recovery_cli_exposes_bounded_backup_and_restore_commands(): + runner = CliRunner() + result = runner.invoke(recovery, ["--help"]) + assert result.exit_code == 0 + assert "backup" in result.output + assert "restore" in result.output + assert "connector credentials" in result.output