diff --git a/polylogue/cli/commands/maintenance/_migrate_tier.py b/polylogue/cli/commands/maintenance/_migrate_tier.py index 7859bdb25c..87f47a4acf 100644 --- a/polylogue/cli/commands/maintenance/_migrate_tier.py +++ b/polylogue/cli/commands/maintenance/_migrate_tier.py @@ -121,6 +121,7 @@ def migrate_tier_command( raise SystemExit(1) from exc result = execution.migration_result if execution is not None else None + receipt = execution.forward_version_receipt if execution is not None else None payload = { "ok": True, "tier": tier, @@ -138,6 +139,19 @@ def migrate_tier_command( "from_version": result.from_version if result is not None else 0 if initialized else None, "to_version": result.to_version if result is not None else initialized_version, "applied_versions": list(result.applied_versions) if result is not None else [], + "forward_version_receipt": ( + { + "tier": receipt.tier.value, + "historical_train_id": receipt.historical_train_id, + "historical_target_version": receipt.historical_target_version, + "current_target_version": receipt.current_target_version, + "observed_live_version": receipt.observed_live_version, + "historical_schema_inventory_sha256": receipt.historical_schema_inventory_sha256, + "archive_identity_digest": receipt.archive_identity_digest, + } + if receipt is not None + else None + ), } if output_format == "json": click.echo(json.dumps(payload, indent=2, sort_keys=True)) @@ -147,6 +161,13 @@ def migrate_tier_command( click.echo(f"Initialized missing {tier} tier at schema version {initialized_version}.") return if result is None: + if receipt is not None: + click.echo( + f"No pending durable migration for {tier}; historical train {receipt.historical_train_id} " + f"is admitted at live schema v{receipt.observed_live_version} " + f"(target v{receipt.current_target_version})." + ) + return click.echo(f"No pending durable migration for {tier}.") return applied = ", ".join(str(version) for version in result.applied_versions) or "none" diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index e1fd4c32bc..00c9204808 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -925,21 +925,32 @@ def _archive_raw_failure_info( ).fetchone()[0] or 0 ) + sample_ids = [ + str(sample["raw_id"]) for sample in lifecycle_snapshot.samples if sample.get("raw_id") is not None + ] lifecycle_by_raw_id = { str(sample["raw_id"]): str(sample["lifecycle"]) for sample in lifecycle_snapshot.samples if sample.get("raw_id") is not None and sample.get("lifecycle") is not None } samples: list[RawFailureSample] = [] - for row in conn.execute( - """ - SELECT r.raw_id, r.origin, r.parse_error, r.validation_status, r.validation_error - FROM raw_sessions AS r - WHERE (parse_error IS NOT NULL AND TRIM(parse_error) != '') OR validation_status = 'failed' - ORDER BY acquired_at_ms DESC, raw_id DESC - LIMIT 50 - """ - ): + rows_by_raw_id: dict[str, sqlite3.Row | tuple[object, ...]] = {} + if sample_ids: + placeholders = ",".join("?" for _ in sample_ids) + rows = conn.execute( + f""" + SELECT r.raw_id, r.origin, r.parse_error, r.validation_status, r.validation_error + FROM raw_sessions AS r + WHERE r.raw_id IN ({placeholders}) + AND ((r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '') OR r.validation_status = 'failed') + """, + sample_ids, + ) + rows_by_raw_id = {str(row[0]): row for row in rows} + for raw_id in sample_ids: + row = rows_by_raw_id.get(raw_id) + if row is None: + continue parse_err = str(row[2] or "") if row[2] else "" val_status = str(row[3] or "") if row[3] else "" val_err = str(row[4] or "") if row[4] else "" @@ -959,7 +970,7 @@ def _archive_raw_failure_info( redacted_error=parse_err or val_err, lifecycle=cast( Literal["deferred", "terminal", "unexplained"], - lifecycle_by_raw_id.get(str(row[0]), "unexplained"), + lifecycle_by_raw_id.get(raw_id, "unexplained"), ), ) ) diff --git a/polylogue/storage/raw_failure_lifecycle.py b/polylogue/storage/raw_failure_lifecycle.py index 7a509e3d00..fa90cc8371 100644 --- a/polylogue/storage/raw_failure_lifecycle.py +++ b/polylogue/storage/raw_failure_lifecycle.py @@ -111,6 +111,7 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra logger.warning("could not open source.db read-only", exc_info=exc) return RawFailureLifecycleSnapshot(False, reason=f"could not open source.db read-only: {exc}") try: + conn.execute("BEGIN") raw_table = conn.execute( "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_sessions'" ).fetchone() @@ -129,43 +130,95 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra conn.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_artifacts'").fetchone() is not None ) - if has_artifacts: - failure_query = """ - SELECT r.raw_id, r.origin, r.validation_status, - ( - SELECT a.artifact_kind - FROM raw_artifacts AS a - WHERE a.raw_id = r.raw_id - AND a.origin = r.origin - AND a.source_path = r.source_path - AND a.source_index = r.source_index - ORDER BY a.last_observed_at_ms DESC, a.artifact_id DESC - LIMIT 1 - ) AS artifact_kind - ,( - SELECT a.support_status - FROM raw_artifacts AS a - WHERE a.raw_id = r.raw_id - AND a.origin = r.origin - AND a.source_path = r.source_path - AND a.source_index = r.source_index - ORDER BY a.last_observed_at_ms DESC, a.artifact_id DESC - LIMIT 1 - ) AS support_status + sample_limit = max(0, sample_limit) + failed_cte = """ + WITH failed AS ( + SELECT r.raw_id, r.origin, r.source_path, r.source_index, + r.validation_status, r.acquired_at_ms FROM raw_sessions AS r WHERE (r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '') OR r.validation_status = 'failed' - ORDER BY r.acquired_at_ms DESC, r.raw_id DESC - """ + ) + """ + latest_artifact_join = """ + LEFT JOIN raw_artifacts AS a + ON a.raw_id = f.raw_id + AND a.origin = f.origin + AND a.source_path = f.source_path + AND a.source_index = f.source_index + AND NOT EXISTS ( + SELECT 1 + FROM raw_artifacts AS newer + WHERE newer.raw_id = a.raw_id + AND newer.origin = a.origin + AND newer.source_path = a.source_path + AND newer.source_index = a.source_index + AND (newer.last_observed_at_ms > a.last_observed_at_ms + OR (newer.last_observed_at_ms = a.last_observed_at_ms + AND newer.artifact_id > a.artifact_id)) + ) + """ + if has_artifacts: + summary_sql = ( + failed_cte + + """ + SELECT f.origin, f.validation_status, a.artifact_kind, a.support_status, + COUNT(*) AS failure_count + FROM failed AS f + """ + + latest_artifact_join + + """ + GROUP BY f.origin, f.validation_status, a.artifact_kind, a.support_status + ORDER BY f.origin, f.validation_status, a.artifact_kind, a.support_status + """ + ) + sample_sql = ( + failed_cte + + """ + , sampled AS ( + SELECT f.raw_id, f.origin, f.validation_status, f.acquired_at_ms, + a.artifact_kind, a.support_status + FROM failed AS f + """ + + latest_artifact_join + + """ + ORDER BY CASE + WHEN f.validation_status = 'failed' THEN 0 + WHEN (a.artifact_kind, a.support_status) IN ( + ('deferred_hot_jsonl_capture', 'partial_decode'), + ('terminal_corrupt_input', 'decode_failed'), + ('terminal_unsupported_shape', 'unsupported_parseable') + ) THEN 1 + ELSE 2 + END, + f.acquired_at_ms DESC, f.raw_id DESC + LIMIT ? + ) + SELECT raw_id, origin, validation_status, artifact_kind, support_status + FROM sampled + """ + ) else: - failure_query = """ - SELECT r.raw_id, r.origin, r.validation_status, NULL AS artifact_kind, NULL AS support_status - FROM raw_sessions AS r - WHERE (r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '') - OR r.validation_status = 'failed' - ORDER BY r.acquired_at_ms DESC, r.raw_id DESC - """ - failed_rows = conn.execute(failure_query).fetchall() + summary_sql = ( + failed_cte + + """ + SELECT f.origin, f.validation_status, NULL, NULL, COUNT(*) AS failure_count + FROM failed AS f + GROUP BY f.origin, f.validation_status + ORDER BY f.origin, f.validation_status + """ + ) + sample_sql = ( + failed_cte + + """ + SELECT raw_id, origin, validation_status, NULL, NULL + FROM failed + ORDER BY acquired_at_ms DESC, raw_id DESC + LIMIT ? + """ + ) + summary_rows = conn.execute(summary_sql).fetchall() + sample_rows = conn.execute(sample_sql, (sample_limit,)).fetchall() except sqlite3.Error as exc: logger.warning("could not read raw failure lifecycle", exc_info=exc) return RawFailureLifecycleSnapshot(False, reason=f"could not read raw failure lifecycle: {exc}") @@ -176,25 +229,33 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra by_artifact_kind: Counter[str] = Counter() counts: Counter[str] = Counter() samples: list[dict[str, str | None]] = [] - for row in failed_rows: + for row in summary_rows: + origin = str(row[0] or "unknown") + artifact_kind = str(row[2]) if row[2] is not None else None + support_status = str(row[3]) if row[3] is not None else None + validation_failed = str(row[1] or "") == "failed" + lifecycle = _lifecycle(artifact_kind, support_status, validation_failed=validation_failed) + count = int(row[4]) + counts[lifecycle] += count + by_origin[origin] += count + by_artifact_kind[artifact_kind or ""] += count + for row in sample_rows: origin = str(row[1] or "unknown") artifact_kind = str(row[3]) if row[3] is not None else None support_status = str(row[4]) if row[4] is not None else None - validation_failed = str(row[2] or "") == "failed" - lifecycle = _lifecycle(artifact_kind, support_status, validation_failed=validation_failed) - counts[lifecycle] += 1 - by_origin[origin] += 1 - by_artifact_kind[artifact_kind or ""] += 1 - if len(samples) < max(0, sample_limit): - samples.append( - { - "raw_id": str(row[0]), - "origin": origin, - "artifact_kind": artifact_kind, - "support_status": support_status, - "lifecycle": lifecycle, - } - ) + samples.append( + { + "raw_id": str(row[0]), + "origin": origin, + "artifact_kind": artifact_kind, + "support_status": support_status, + "lifecycle": _lifecycle( + artifact_kind, + support_status, + validation_failed=str(row[2] or "") == "failed", + ), + } + ) return RawFailureLifecycleSnapshot( available=True, parse_failures=parse_failures, diff --git a/polylogue/storage/sqlite/archive_tiers/bootstrap.py b/polylogue/storage/sqlite/archive_tiers/bootstrap.py index 6353bf310f..c7f78bfe7a 100644 --- a/polylogue/storage/sqlite/archive_tiers/bootstrap.py +++ b/polylogue/storage/sqlite/archive_tiers/bootstrap.py @@ -312,15 +312,69 @@ def initialize_archive_database( def initialize_active_archive_root(root: Path) -> None: """Create or initialize every tier database in an archive root.""" from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation + from polylogue.storage.sqlite.durable_change_train import ( + _record_fresh_durable_bootstrap, + _record_fresh_durable_bootstrap_intent, + _validate_fresh_durable_bootstrap_intent, + ) with OwnedArchiveLocation.acquire( ArchiveLocation.resolve(root), owner_id=f"bootstrap:{os.getpid()}", allow_reentrant=True, ): - reconcile_durable_change_trains_on_startup(root) + # Classify the archive after acquiring ownership. Another process may + # publish a marker or durable train while the probe is in flight. + durable_tier_exists = any( + (root / archive_tier_spec(tier).filename).exists() for tier in DURABLE_MIGRATION_TIERS + ) + manifest_root = root / ".maintenance-state" / "durable-change-trains" + has_durable_train_state = any(manifest_root.glob("*.json")) + has_bootstrap_marker = (manifest_root / ".bootstrap").is_file() + pending_bootstrap_path = manifest_root / ".bootstrap.pending" + has_pending_bootstrap = pending_bootstrap_path.is_file() + if has_pending_bootstrap: + _validate_fresh_durable_bootstrap_intent(root) + if has_durable_train_state: + raise RuntimeError( + "fresh durable bootstrap intent conflicts with durable train state; " + "refusing to guess which authority is current" + ) + fresh_durable_bootstrap = ( + not durable_tier_exists + and not has_durable_train_state + and not has_bootstrap_marker + and not has_pending_bootstrap + ) + recovering_fresh_durable_bootstrap = fresh_durable_bootstrap or ( + has_pending_bootstrap and not has_bootstrap_marker + ) + pre_marker_adoption = ( + (root / archive_tier_spec(ArchiveTier.SOURCE).filename).is_file() + and all((root / archive_tier_spec(tier).filename).is_file() for tier in DURABLE_MIGRATION_TIERS) + and manifest_root.is_dir() + and not has_durable_train_state + and not has_bootstrap_marker + and not has_pending_bootstrap + ) + if fresh_durable_bootstrap: + _record_fresh_durable_bootstrap_intent(root) + if not recovering_fresh_durable_bootstrap and not pre_marker_adoption: + reconcile_durable_change_trains_on_startup(root) for spec in ARCHIVE_TIER_SPECS.values(): initialize_archive_database(root / spec.filename, spec.tier) + if recovering_fresh_durable_bootstrap: + _record_fresh_durable_bootstrap(root) + elif pre_marker_adoption: + from polylogue.storage.sqlite.durable_change_train import _adopt_pre_marker_durable_bootstrap + + _adopt_pre_marker_durable_bootstrap(root) + reconcile_durable_change_trains_on_startup(root) + elif has_pending_bootstrap: + # A crash after publishing the completed marker but before + # removing the intent is harmless. Keep the intent until the + # completed marker has passed normal startup reconciliation. + pending_bootstrap_path.unlink(missing_ok=True) def reconcile_durable_change_trains_on_startup(root: Path) -> tuple[Path, ...]: diff --git a/polylogue/storage/sqlite/durable_change_train.py b/polylogue/storage/sqlite/durable_change_train.py index 2816f3bf2c..64b0f60731 100644 --- a/polylogue/storage/sqlite/durable_change_train.py +++ b/polylogue/storage/sqlite/durable_change_train.py @@ -12,6 +12,7 @@ import sqlite3 import tempfile from collections.abc import Callable, Sequence +from contextlib import closing from dataclasses import dataclass, replace from importlib import resources from pathlib import Path @@ -36,6 +37,7 @@ DurableMigrationClaim, DurableRuntimeConsumerResult, MigrationResult, + _archive_identity_continuity_matches, _assert_durable_database_continuity, _canonical_json_sha256, _require_nonempty, @@ -72,6 +74,9 @@ _MIGRATION_NAME_RE = re.compile(r"^(?P\d{3,})_[a-z0-9_]+\.sql$") _DROP_SQL_RE = re.compile(r"(?is)\bDROP\s+(?:TABLE|INDEX|TRIGGER|VIEW)\b") _SOURCE_CONTINUITY_PENDING_FORMAT = "polylogue.source-continuity-pending.v1" +_FRESH_DURABLE_BOOTSTRAP_FORMAT = "polylogue.durable-bootstrap.v1" +_FRESH_DURABLE_BOOTSTRAP_MARKER = ".bootstrap" +_FRESH_DURABLE_BOOTSTRAP_PENDING_MARKER = ".bootstrap.pending" class DurableSourceTrainMissingError(DurableChangeTrainError): @@ -99,6 +104,30 @@ class DurableChangeTrainExecution: train: DurableChangeTrain | None manifest_path: Path | None migration_result: MigrationResult | None + forward_version_receipt: DurableForwardVersionReceipt | None = None + + +@dataclass(frozen=True, slots=True) +class DurableForwardVersionReceipt: + """Evidence that a historical released train admits a later live tier.""" + + tier: ArchiveTier + historical_train_id: str + historical_target_version: int + current_target_version: int + observed_live_version: int + historical_schema_inventory_sha256: str + archive_identity_digest: str + + +@dataclass(frozen=True, slots=True) +class _DurableForwardVersionEvidence: + """Cached live evidence reused by one no-op maintenance execution.""" + + actual: DurableDatabaseEvidence + integrity_check: tuple[str, ...] + live_inventory: _migration_runner.DurableSchemaInventory + canonical_inventory: _migration_runner.DurableSchemaInventory def durable_migration_sidecar_name(slot: int) -> str: @@ -292,6 +321,213 @@ def durable_change_train_manifest_path(archive_root: Path, tier: ArchiveTier, sl return archive_root / ".maintenance-state" / "durable-change-trains" / f"{tier.value}-{slot:03d}.json" +def _record_fresh_durable_bootstrap(archive_root: Path) -> None: + """Record the versions and identity of a direct, current-schema bootstrap.""" + from polylogue.storage.archive_identity import ArchiveIdentity + + archive_root = archive_root.resolve() + marker_root = archive_root / ".maintenance-state" / "durable-change-trains" + marker_path = marker_root / _FRESH_DURABLE_BOOTSTRAP_MARKER + pending_path = marker_root / _FRESH_DURABLE_BOOTSTRAP_PENDING_MARKER + if marker_path.exists() or any(marker_root.glob("*.json")): + raise DurableChangeTrainError(f"cannot record fresh durable bootstrap over existing train state: {marker_root}") + if pending_path.is_file(): + _validate_fresh_durable_bootstrap_intent(archive_root) + marker_root.mkdir(parents=True, exist_ok=True) + versions: dict[str, int] = {} + for tier in DURABLE_MIGRATION_ADOPTION_FLOORS: + with sqlite3.connect(archive_root / f"{tier.value}.db") as connection: + versions[tier.value] = int(connection.execute("PRAGMA user_version").fetchone()[0]) + payload: dict[str, object] = { + "format": _FRESH_DURABLE_BOOTSTRAP_FORMAT, + "durable_identity_digest": _durable_identity_digest(ArchiveIdentity.resolve(archive_root)), + "versions": versions, + } + payload["marker_digest"] = _bootstrap_marker_digest(payload) + _write_bootstrap_receipt(marker_path, payload) + pending_path.unlink(missing_ok=True) + + +def _record_fresh_durable_bootstrap_intent(archive_root: Path) -> None: + """Record an authenticated intent before creating the first tier file. + + Fresh archive initialization creates several independent SQLite files. A + failure in a later tier can therefore leave a partial fresh archive. The + intent distinguishes that recoverable state from an established archive + whose durable train evidence has been lost. + """ + from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER + + archive_root = archive_root.resolve() + marker_root = archive_root / ".maintenance-state" / "durable-change-trains" + marker_path = marker_root / _FRESH_DURABLE_BOOTSTRAP_MARKER + pending_path = marker_root / _FRESH_DURABLE_BOOTSTRAP_PENDING_MARKER + if marker_path.exists() or any(marker_root.glob("*.json")): + raise DurableChangeTrainError( + f"cannot record fresh durable bootstrap intent over existing train state: {marker_root}" + ) + if pending_path.is_file(): + _validate_fresh_durable_bootstrap_intent(archive_root) + return + versions = {tier.value: ARCHIVE_VERSION_BY_TIER[tier] for tier in DURABLE_MIGRATION_ADOPTION_FLOORS} + payload: dict[str, object] = { + "format": _FRESH_DURABLE_BOOTSTRAP_FORMAT, + "state": "pending", + "durable_identity_digest": _fresh_bootstrap_intent_identity_digest(archive_root), + "versions": versions, + } + payload["marker_digest"] = _bootstrap_marker_digest(payload) + _write_bootstrap_receipt(pending_path, payload) + + +def _validate_fresh_durable_bootstrap_intent(archive_root: Path) -> None: + """Validate the authenticated intent for a recoverable fresh bootstrap.""" + from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER + + archive_root = archive_root.resolve() + marker_path = ( + archive_root / ".maintenance-state" / "durable-change-trains" / _FRESH_DURABLE_BOOTSTRAP_PENDING_MARKER + ) + try: + payload = json.loads(marker_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError(f"invalid fresh durable bootstrap intent: {marker_path}") from exc + if not isinstance(payload, dict) or payload.get("format") != _FRESH_DURABLE_BOOTSTRAP_FORMAT: + raise DurableChangeTrainError(f"fresh durable bootstrap intent format mismatch: {marker_path}") + if payload.get("state") != "pending": + raise DurableChangeTrainError(f"fresh durable bootstrap intent state is invalid: {marker_path}") + if payload.get("durable_identity_digest") != _fresh_bootstrap_intent_identity_digest(archive_root): + raise DurableChangeTrainError("fresh durable bootstrap intent durable identity mismatch") + marker_digest = payload.get("marker_digest") + unsigned_payload = dict(payload) + unsigned_payload.pop("marker_digest", None) + if not isinstance(marker_digest, str) or marker_digest != _bootstrap_marker_digest(unsigned_payload): + raise DurableChangeTrainError("fresh durable bootstrap intent digest mismatch") + raw_versions = payload.get("versions") + if not isinstance(raw_versions, dict): + raise DurableChangeTrainError(f"fresh durable bootstrap intent versions are invalid: {marker_path}") + for tier in DURABLE_MIGRATION_ADOPTION_FLOORS: + if raw_versions.get(tier.value) != ARCHIVE_VERSION_BY_TIER[tier]: + raise DurableChangeTrainError(f"fresh durable bootstrap intent target version is stale: {marker_path}") + + +def _write_bootstrap_receipt(marker_path: Path, payload: dict[str, object]) -> None: + """Atomically publish one bootstrap receipt and fsync its directory.""" + marker_root = marker_path.parent + marker_root.mkdir(parents=True, exist_ok=True) + encoded = (json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8") + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + dir=marker_root, + prefix=f".{marker_path.name}.", + suffix=".tmp", + delete=False, + ) as stream: + temporary = Path(stream.name) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, marker_path) + temporary = None + _migration_runner._fsync_manifest_directory(marker_root) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def _fresh_bootstrap_intent_identity_digest(archive_root: Path) -> str: + """Bind a pre-file bootstrap intent to its root without inode identity.""" + return _canonical_json_sha256( + { + "configured_root": str(archive_root.resolve().absolute()), + "purpose": "fresh-durable-bootstrap", + } + ) + + +def _fresh_durable_bootstrap_versions(archive_root: Path, marker_root: Path) -> dict[ArchiveTier, int]: + """Return direct-bootstrap versions when the marker is authentic.""" + from polylogue.storage.archive_identity import ArchiveIdentity + + archive_root = archive_root.resolve() + marker_root = archive_root / ".maintenance-state" / "durable-change-trains" + marker_path = marker_root / _FRESH_DURABLE_BOOTSTRAP_MARKER + if not marker_path.is_file(): + return {} + try: + payload = json.loads(marker_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError(f"invalid fresh durable bootstrap marker: {marker_path}") from exc + if not isinstance(payload, dict) or payload.get("format") != _FRESH_DURABLE_BOOTSTRAP_FORMAT: + raise DurableChangeTrainError(f"fresh durable bootstrap marker format mismatch: {marker_path}") + if payload.get("durable_identity_digest") != _durable_identity_digest(ArchiveIdentity.resolve(archive_root)): + raise DurableChangeTrainError("fresh durable bootstrap marker durable identity mismatch") + marker_digest = payload.get("marker_digest") + unsigned_payload = dict(payload) + unsigned_payload.pop("marker_digest", None) + if not isinstance(marker_digest, str) or marker_digest != _bootstrap_marker_digest(unsigned_payload): + raise DurableChangeTrainError("fresh durable bootstrap marker digest mismatch") + raw_versions = payload.get("versions") + if not isinstance(raw_versions, dict): + raise DurableChangeTrainError(f"fresh durable bootstrap marker versions are invalid: {marker_path}") + versions: dict[ArchiveTier, int] = {} + for tier in DURABLE_MIGRATION_ADOPTION_FLOORS: + raw_version = raw_versions.get(tier.value) + if not isinstance(raw_version, int) or raw_version < 0: + raise DurableChangeTrainError(f"fresh durable bootstrap marker version is invalid: {marker_path}") + versions[tier] = raw_version + return versions + + +def _durable_identity_digest(identity: object) -> str: + """Digest only the durable source/user identity for bootstrap receipts.""" + from polylogue.storage.archive_identity import ArchiveIdentity + + if not isinstance(identity, ArchiveIdentity): + raise TypeError("durable identity digest requires an ArchiveIdentity") + payload = { + "configured_root": str(identity.configured_root.absolute()), + "durable_id": identity.durable_id, + } + return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def _bootstrap_marker_digest(payload: dict[str, object]) -> str: + """Authenticate bootstrap identity and recorded durable versions together.""" + return _canonical_json_sha256(payload) + + +def _adopt_pre_marker_durable_bootstrap(archive_root: Path) -> None: + """Authenticate a current-schema archive created before bootstrap receipts.""" + from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER + + archive_root = archive_root.resolve() + manifest_root = archive_root / ".maintenance-state" / "durable-change-trains" + if (manifest_root / _FRESH_DURABLE_BOOTSTRAP_MARKER).is_file(): + return + if any(manifest_root.glob("*.json")): + return + for tier in DURABLE_MIGRATION_ADOPTION_FLOORS: + tier_path = archive_root / f"{tier.value}.db" + if not tier_path.is_file(): + continue + with _open_existing_tier(tier_path) as connection: + current_version = int(connection.execute("PRAGMA user_version").fetchone()[0] or 0) + expected_version = ARCHIVE_VERSION_BY_TIER[tier] + if current_version != expected_version: + raise DurableChangeTrainError( + f"pre-marker {tier.value} durable tier is v{current_version}, expected current v{expected_version}" + ) + actual_inventory = _migration_runner.capture_durable_schema_inventory(connection) + expected_inventory = _canonical_schema_inventory(tier, expected_version) + if actual_inventory.sha256 != expected_inventory.sha256: + raise DurableChangeTrainError( + f"pre-marker {tier.value} durable tier schema does not match current canonical DDL" + ) + _record_fresh_durable_bootstrap(archive_root) + + def durable_migration_sidecar_for_slot(tier: ArchiveTier, slot: int) -> DurableMigrationSidecar | None: """Load the package sidecar for the next numbered production migration.""" if tier not in DURABLE_MIGRATION_ADOPTION_FLOORS: @@ -766,9 +1002,11 @@ def _refresh_released_source_train_continuity_locked( caller must first validate the named operation and backup against the pre-mutation live tier. This helper then binds those exact receipt and backup bytes to a separate current-evidence record. The original migration - evidence remains immutable in ``apply_evidence``. + evidence remains immutable in ``apply_evidence``; only the + legacy-authority-digest compatibility path may rewrite that historical + identity field while preserving the migration evidence. """ - from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation + from polylogue.storage.archive_identity import ArchiveIdentity, ArchiveLocation, OwnedArchiveLocation archive_root = archive_root.resolve() mutation_receipt = mutation_receipt.resolve() @@ -844,6 +1082,7 @@ def _refresh_released_source_train_continuity_locked( current, retained_current, label="source continuity retained refresh", + archive_root=archive_root, ) except DurableChangeTrainError: pass @@ -888,16 +1127,27 @@ def _refresh_released_source_train_continuity_locked( pre_mutation_evidence, baseline, label="source continuity pre-mutation", + archive_root=archive_root, ) except DurableChangeTrainError as exc: raise DurableSourceContinuitySemanticError( "source continuity refresh pre-state contains unreceipted content drift" ) from exc - if pre_mutation_evidence.archive_identity_digest != train.apply_evidence.post.archive_identity_digest: + if not _archive_identity_continuity_matches( + pre_mutation_evidence.archive_identity_digest, + train.apply_evidence.post.archive_identity_digest, + archive_root, + ArchiveTier.SOURCE, + ): raise DurableSourceContinuitySemanticError( "source continuity refresh pre-state has the wrong archive identity" ) - if current.archive_identity_digest != train.apply_evidence.post.archive_identity_digest: + if not _archive_identity_continuity_matches( + current.archive_identity_digest, + train.apply_evidence.post.archive_identity_digest, + archive_root, + ArchiveTier.SOURCE, + ): raise DurableSourceContinuitySemanticError("source continuity refresh changed archive identity") if pre_mutation_evidence.quick_check != ("ok",) or current.quick_check != ("ok",): raise DurableSourceContinuitySemanticError( @@ -958,9 +1208,23 @@ def _refresh_released_source_train_continuity_locked( references = _migration_runner._append_proof_refs(train.proof_refs, evidence_ref, refresh_ref) if train.proof is None: raise DurableChangeTrainError("source continuity refresh requires train proof") + retained_apply_evidence = train.apply_evidence + legacy_archive_identity_digest = ArchiveIdentity.resolve(archive_root).authority_identity_digest + if ( + retained_apply_evidence.post.archive_identity_digest == legacy_archive_identity_digest + and current.archive_identity_digest != retained_apply_evidence.post.archive_identity_digest + ): + retained_apply_evidence = replace( + retained_apply_evidence, + post=replace( + retained_apply_evidence.post, + archive_identity_digest=current.archive_identity_digest, + ), + ) updated = replace( train, revision=train.revision + 1, + apply_evidence=retained_apply_evidence, source_continuity_evidence=current, proof_refs=references, ) @@ -1362,29 +1626,88 @@ def _open_existing_tier(tier_path: Path) -> sqlite3.Connection: raise DurableChangeTrainError("durable tier could not be opened without initialization") from exc -def _verify_persisted_live_tier_continuity(conn: sqlite3.Connection, train: DurableChangeTrain) -> None: +def _verify_persisted_live_tier_continuity( + conn: sqlite3.Connection, + train: DurableChangeTrain, + *, + actual: DurableDatabaseEvidence | None = None, +) -> None: """Prove the exact reopened connection still names the persisted durable tier.""" if train.apply_evidence is None: raise DurableChangeTrainError(f"{train.state.value} train lacks post-apply continuity evidence") - actual = capture_durable_database_evidence(conn, train.tier) + actual = actual or capture_durable_database_evidence(conn, train.tier) expected = train.apply_evidence.post if actual.user_version != train.target_version: raise DurableChangeTrainError( f"{train.tier.value} durable tier continuity proof failed; refusing startup initialization/release" ) try: - _assert_durable_database_continuity(actual, expected, label=train.tier.value) + _assert_durable_database_continuity(actual, expected, label=train.tier.value, connection=conn) except DurableChangeTrainError as exc: raise DurableChangeTrainError( f"{train.tier.value} durable tier continuity proof failed; refusing startup initialization/release" ) from exc -def _verify_released_train_live_tier(archive_root: Path, conn: sqlite3.Connection, train: DurableChangeTrain) -> None: +def _historical_schema_evidence(train: DurableChangeTrain) -> DurableFreshDDLParityProof: + """Return the immutable historical schema proof a forward admission needs.""" + if train.apply_evidence is None or train.proof is None or train.fresh_ddl_parity is None: + raise DurableChangeTrainError("released train lacks historical schema evidence for forward-version admission") + historical = train.proof.fresh_ddl_parity + if ( + historical.tier is not train.tier + or historical.target_version != train.target_version + or historical.migrated_version != train.target_version + or historical.fresh_version != train.target_version + or not historical.matches + or historical.missing_objects + or historical.unexpected_objects + or historical.changed_objects + or historical.migrated_inventory_sha256 != train.apply_evidence.post.schema_inventory_sha256 + or historical.fresh_inventory_sha256 != train.fresh_ddl_parity.fresh_inventory_sha256 + ): + raise DurableChangeTrainError( + "released train lacks exact historical schema evidence for forward-version admission" + ) + return historical + + +def _canonical_schema_inventory(tier: ArchiveTier, target_version: int) -> _migration_runner.DurableSchemaInventory: + """Construct the canonical object set for one live durable schema version.""" + try: + normalized_target_version = int(target_version) + except (TypeError, ValueError) as exc: + raise DurableChangeTrainError("canonical schema inventory target version must be an integer") from exc + if isinstance(target_version, bool): + raise DurableChangeTrainError("canonical schema inventory target version must be an integer") + registry = getattr(_migration_runner, "ARCHIVE_DDL_BY_TIER", None) + archive_ddl = registry.get(tier) if isinstance(registry, dict) else None + if not isinstance(archive_ddl, str): + raise DurableChangeTrainError(f"no canonical archive DDL is registered for {tier.value}") + with closing(sqlite3.connect(":memory:")) as fresh: + fresh.execute("PRAGMA foreign_keys = ON") + fresh.executescript(archive_ddl) + fresh.execute(f"PRAGMA user_version = {normalized_target_version}") + _migration_runner._prepare_fresh_connection_for_target(fresh, tier, normalized_target_version) + fresh.commit() + return _migration_runner.capture_durable_schema_inventory(fresh) + + +def _verify_released_train_live_tier( + archive_root: Path, + conn: sqlite3.Connection, + train: DurableChangeTrain, + *, + current_target_version: int | None = None, + actual_evidence: DurableDatabaseEvidence | None = None, + integrity_check: tuple[str, ...] | None = None, + live_inventory: _migration_runner.DurableSchemaInventory | None = None, + canonical_inventory: _migration_runner.DurableSchemaInventory | None = None, +) -> DurableForwardVersionReceipt | None: """Verify a released train remains represented after later trains advance it.""" if train.apply_evidence is None: raise DurableChangeTrainError(f"{train.state.value} train lacks post-apply continuity evidence") - actual = capture_durable_database_evidence(conn, train.tier) + actual = actual_evidence or capture_durable_database_evidence(conn, train.tier) if actual.user_version < train.target_version: raise DurableChangeTrainError( f"{train.tier.value} durable tier continuity proof failed: live version regressed below released train " @@ -1397,15 +1720,174 @@ def _verify_released_train_live_tier(archive_root: Path, conn: sqlite3.Connectio actual, train.source_continuity_evidence, label="source continuity refresh", + connection=conn, ) else: - _verify_persisted_live_tier_continuity(conn, train) - return - integrity = conn.execute("PRAGMA integrity_check").fetchone() - if integrity != ("ok",): + _verify_persisted_live_tier_continuity(conn, train, actual=actual) + return None + historical = _historical_schema_evidence(train) + expected_identity = train.apply_evidence.post.archive_identity_digest + if not _archive_identity_continuity_matches( + actual.archive_identity_digest, + expected_identity, + archive_root, + train.tier, + ): + raise DurableChangeTrainError( + f"{train.tier.value} durable tier immutable archive identity differs from historical train " + f"v{train.target_version} after later train advancement" + ) + if actual.quick_check != ("ok",): raise DurableChangeTrainError( f"{train.tier.value} durable tier integrity check failed after later train advancement" ) + observed_integrity = integrity_check or tuple(str(row[0]) for row in conn.execute("PRAGMA integrity_check")) + if observed_integrity != ("ok",): + raise DurableChangeTrainError( + f"{train.tier.value} durable tier integrity check failed after later train advancement: {observed_integrity}" + ) + live_inventory = live_inventory or _migration_runner.capture_durable_schema_inventory(conn) + if live_inventory.sha256 != actual.schema_inventory_sha256: + raise DurableChangeTrainError( + f"{train.tier.value} durable tier schema inventory changed during forward admission" + ) + expected_inventory = canonical_inventory or _canonical_schema_inventory(train.tier, actual.user_version) + expected_by_ref = {item.object_ref: item for item in expected_inventory.objects} + live_by_ref = {item.object_ref: item for item in live_inventory.objects} + missing = sorted(set(expected_by_ref) - set(live_by_ref)) + unexpected = sorted(set(live_by_ref) - set(expected_by_ref)) + changed = sorted( + object_ref + for object_ref in set(expected_by_ref) & set(live_by_ref) + if expected_by_ref[object_ref].definition_sha256 != live_by_ref[object_ref].definition_sha256 + ) + if missing or unexpected or changed: + raise DurableChangeTrainError( + f"{train.tier.value} durable tier schema differs from the canonical live version: " + f"missing={missing}, unexpected={unexpected}, changed={changed}" + ) + runtime_target = ( + cast(dict[ArchiveTier, int], vars(_migration_runner)["ARCHIVE_VERSION_BY_TIER"])[train.tier] + if current_target_version is None + else current_target_version + ) + if actual.user_version > runtime_target: + raise DurableChangeTrainError( + f"{train.tier.value} durable tier version {actual.user_version} is newer than current target " + f"v{runtime_target}; historical train v{train.target_version} cannot admit it" + ) + return DurableForwardVersionReceipt( + tier=train.tier, + historical_train_id=train.train_id, + historical_target_version=train.target_version, + current_target_version=runtime_target, + observed_live_version=actual.user_version, + historical_schema_inventory_sha256=historical.migrated_inventory_sha256, + archive_identity_digest=actual.archive_identity_digest, + ) + + +def _forward_version_receipt_for_current_tier( + archive_root: Path, + conn: sqlite3.Connection, + tier: ArchiveTier, + *, + current_version: int, + current_target_version: int, + evidence: _DurableForwardVersionEvidence | None = None, +) -> DurableForwardVersionReceipt | None: + """Return the newest released historical-train receipt at the live target.""" + manifest_root = archive_root / ".maintenance-state" / "durable-change-trains" + manifests_by_target = _released_train_manifests_by_target(manifest_root, tier) + historical = [ + train + for train in manifests_by_target.values() + if train.state is DurableChangeTrainState.RELEASED and train.target_version < current_version + ] + if tier in DURABLE_MIGRATION_ADOPTION_FLOORS and current_version > DURABLE_MIGRATION_ADOPTION_FLOORS[tier]: + _require_released_train_chain( + tier, + manifests_by_target, + current_version=current_version, + floor=_chain_floor(tier, _fresh_durable_bootstrap_versions(archive_root, manifest_root)), + ) + if not historical: + return None + if evidence is None: + actual = capture_durable_database_evidence(conn, tier) + evidence = _DurableForwardVersionEvidence( + actual=actual, + integrity_check=tuple(str(row[0]) for row in conn.execute("PRAGMA integrity_check")), + live_inventory=_migration_runner.capture_durable_schema_inventory(conn), + canonical_inventory=_canonical_schema_inventory(tier, actual.user_version), + ) + for train in sorted(historical, key=lambda item: item.target_version, reverse=True): + receipt = _verify_released_train_live_tier( + archive_root, + conn, + train, + current_target_version=current_target_version, + actual_evidence=evidence.actual, + integrity_check=evidence.integrity_check, + live_inventory=evidence.live_inventory, + canonical_inventory=evidence.canonical_inventory, + ) + if receipt is not None: + return receipt + return None + + +def _released_train_manifests_by_target( + manifest_root: Path, + tier: ArchiveTier, +) -> dict[int, DurableChangeTrain]: + """Load one persisted train record per target version for a durable tier.""" + manifests_by_target: dict[int, DurableChangeTrain] = {} + if not manifest_root.is_dir(): + return manifests_by_target + for path in sorted(manifest_root.glob(f"{tier.value}-*.json")): + train = load_durable_change_train_manifest(path) + if train.target_version in manifests_by_target: + raise DurableChangeTrainError( + f"duplicate {tier.value} durable train manifests for target v{train.target_version}" + ) + manifests_by_target[train.target_version] = train + return manifests_by_target + + +def _require_released_train_chain( + tier: ArchiveTier, + manifests_by_target: dict[int, DurableChangeTrain], + *, + current_version: int, + floor: int | None = None, +) -> None: + """Require released, schema-proven evidence for every later version.""" + chain_floor = DURABLE_MIGRATION_ADOPTION_FLOORS[tier] if floor is None else floor + if chain_floor < DURABLE_MIGRATION_ADOPTION_FLOORS[tier] or chain_floor > current_version: + raise DurableChangeTrainError( + f"invalid {tier.value} durable train chain floor v{chain_floor} for live v{current_version}" + ) + missing_targets = [ + version + for version in range(chain_floor + 1, current_version + 1) + if manifests_by_target.get(version) is None + or manifests_by_target[version].state is not DurableChangeTrainState.RELEASED + ] + if missing_targets: + raise DurableChangeTrainError( + f"{tier.value} durable forward admission lacks released train evidence for versions " + f"{missing_targets} from chain floor v{chain_floor} " + f"through live v{current_version}" + ) + for version in range(chain_floor + 1, current_version + 1): + _historical_schema_evidence(manifests_by_target[version]) + + +def _chain_floor(tier: ArchiveTier, bootstrap_versions: dict[ArchiveTier, int]) -> int: + """Return the durable train floor allowed by adoption and bootstrap evidence.""" + adoption_floor = DURABLE_MIGRATION_ADOPTION_FLOORS[tier] + return max(adoption_floor, bootstrap_versions.get(tier, adoption_floor)) def _prove_and_release_persisted_train( @@ -1474,8 +1956,13 @@ def execute_durable_change_train( runtime_consumer_results: Sequence[DurableRuntimeConsumerResult] | None = None, release_archive_ownership: Callable[[], None], ) -> DurableChangeTrainExecution: - """Execute the real maintenance route through every persisted train state.""" - reconcile_durable_change_train_startup(archive_root) + """Execute every persisted train state while the caller holds archive ownership. + + The caller-held lease must cover startup reconciliation and receipt creation so + reused forward-version evidence cannot become stale between those operations. + """ + forward_version_evidence: dict[ArchiveTier, _DurableForwardVersionEvidence] = {} + reconcile_durable_change_train_startup(archive_root, live_evidence_cache=forward_version_evidence) tier_path = archive_root / f"{tier.value}.db" with _open_existing_tier(tier_path) as probe: current_version = int(probe.execute("PRAGMA user_version").fetchone()[0] or 0) @@ -1518,7 +2005,21 @@ def execute_durable_change_train( f"durable migration chain for {tier.value} stops at v{current_version}; " f"runtime requires v{runtime_target_version} and the next train sidecar is missing" ) - return DurableChangeTrainExecution(train=None, manifest_path=None, migration_result=legacy_result) + with _open_existing_tier(tier_path) as live: + forward_version_receipt = _forward_version_receipt_for_current_tier( + archive_root, + live, + tier, + current_version=current_version, + current_target_version=runtime_target_version, + evidence=forward_version_evidence.get(tier), + ) + return DurableChangeTrainExecution( + train=None, + manifest_path=None, + migration_result=legacy_result, + forward_version_receipt=forward_version_receipt, + ) manifest_path = durable_change_train_manifest_path(archive_root, tier, sidecar.slot) if manifest_path.exists(): @@ -1536,8 +2037,28 @@ def execute_durable_change_train( f"released {tier.value} train {train.train_id} expects live v{runtime_target_version}, " f"found v{live_version}; authorize a new execution" ) - _verify_released_train_live_tier(archive_root, live, train) - return DurableChangeTrainExecution(train=train, manifest_path=manifest_path, migration_result=None) + forward_version_receipt = _verify_released_train_live_tier( + archive_root, + live, + train, + current_target_version=runtime_target_version, + actual_evidence=(forward_version_evidence[tier].actual if tier in forward_version_evidence else None), + integrity_check=( + forward_version_evidence[tier].integrity_check if tier in forward_version_evidence else None + ), + live_inventory=( + forward_version_evidence[tier].live_inventory if tier in forward_version_evidence else None + ), + canonical_inventory=( + forward_version_evidence[tier].canonical_inventory if tier in forward_version_evidence else None + ), + ) + return DurableChangeTrainExecution( + train=train, + manifest_path=manifest_path, + migration_result=None, + forward_version_receipt=forward_version_receipt, + ) if train.state is DurableChangeTrainState.DECLARED: previous_revision = train.revision @@ -1617,8 +2138,15 @@ def execute_durable_change_train( return DurableChangeTrainExecution(train=train, manifest_path=manifest_path, migration_result=migration_result) -def reconcile_durable_change_train_startup(archive_root: Path) -> tuple[Path, ...]: - """Reconcile backup-authorized trains left by a crashed maintenance process.""" +def reconcile_durable_change_train_startup( + archive_root: Path, + *, + live_evidence_cache: dict[ArchiveTier, _DurableForwardVersionEvidence] | None = None, +) -> tuple[Path, ...]: + """Reconcile interrupted trains while the caller holds archive ownership. + + The caller-held lease must cover any subsequent use of ``live_evidence_cache``. + """ from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation with OwnedArchiveLocation.acquire( @@ -1626,17 +2154,40 @@ def reconcile_durable_change_train_startup(archive_root: Path) -> tuple[Path, .. owner_id=f"durable-train-recovery:{os.getpid()}", allow_reentrant=True, ): - return _reconcile_durable_change_train_startup_locked(archive_root) + return _reconcile_durable_change_train_startup_locked( + archive_root, + live_evidence_cache=live_evidence_cache, + ) -def _reconcile_durable_change_train_startup_locked(archive_root: Path) -> tuple[Path, ...]: +def _reconcile_durable_change_train_startup_locked( + archive_root: Path, + *, + live_evidence_cache: dict[ArchiveTier, _DurableForwardVersionEvidence] | None = None, +) -> tuple[Path, ...]: """Reconcile persisted trains while the caller holds archive ownership.""" _recover_pending_source_continuity_intents(archive_root) manifest_root = archive_root / ".maintenance-state" / "durable-change-trains" - if not manifest_root.is_dir(): - return () reconciled: list[Path] = [] - for manifest_path in sorted(manifest_root.glob("*.json")): + live_evidence_by_tier: dict[ArchiveTier, DurableDatabaseEvidence] = {} + live_integrity_by_tier: dict[ArchiveTier, tuple[str, ...]] = {} + live_inventory_by_tier: dict[ArchiveTier, _migration_runner.DurableSchemaInventory] = {} + canonical_inventory_by_tier: dict[ArchiveTier, _migration_runner.DurableSchemaInventory] = {} + manifests_by_tier: dict[ArchiveTier, dict[int, DurableChangeTrain]] = {} + validated_tiers: set[ArchiveTier] = set() + manifest_paths = tuple(sorted(manifest_root.glob("*.json"))) + fresh_bootstrap_versions = _fresh_durable_bootstrap_versions(archive_root, manifest_root) + + def record_reconciled(path: Path) -> None: + if path not in reconciled: + reconciled.append(path) + + # Recover every non-released lifecycle state before validating any + # released train. A later committed train may be persisted as + # backup-authorized, applied, or proven while an older released manifest + # is still present. Checking the older train first would reject the + # incomplete chain before startup had a chance to finish that recovery. + for manifest_path in manifest_paths: train = load_durable_change_train_manifest(manifest_path) if train.state is DurableChangeTrainState.FAILED: tier_path = archive_root / f"{train.tier.value}.db" @@ -1648,7 +2199,7 @@ def _reconcile_durable_change_train_startup_locked(archive_root: Path) -> tuple[ writer_release_evidence_ref=f"proof:startup-writer-release:{train.train_id}", ) train = _persist_train_transition(manifest_path, recovered, expected_revision=train.revision) - reconciled.append(manifest_path) + record_reconciled(manifest_path) if train.state is DurableChangeTrainState.ADMITTED: continue if train.state is DurableChangeTrainState.BACKUP_AUTHORIZED: @@ -1665,18 +2216,95 @@ def _reconcile_durable_change_train_startup_locked(archive_root: Path) -> tuple[ _persist_train_transition(manifest_path, exc.failed_train, expected_revision=train.revision) raise train = _persist_train_transition(manifest_path, recovered, expected_revision=train.revision) - if train.state is DurableChangeTrainState.RELEASED: - with _open_existing_tier(archive_root / f"{train.tier.value}.db") as live: - _verify_released_train_live_tier(archive_root, live, train) - reconciled.append(manifest_path) - continue - if train.state not in { + record_reconciled(manifest_path) + + if train.state in { DurableChangeTrainState.APPLIED, DurableChangeTrainState.PROVEN, }: + _prove_and_release_persisted_train(archive_root, manifest_path, train) + record_reconciled(manifest_path) + + # An existing durable tier above its adoption floor must have a complete + # released-train chain even when the manifest directory is absent or empty. + # Fresh database creation does not enter this startup reconciliation route. + # Recovery runs first so an indeterminate persisted failure keeps its + # stronger fail-closed error rather than being masked by chain validation. + for tier, adoption_floor in DURABLE_MIGRATION_ADOPTION_FLOORS.items(): + tier_path = archive_root / f"{tier.value}.db" + if not tier_path.is_file(): + continue + with _open_existing_tier(tier_path) as live: + current_version = int(live.execute("PRAGMA user_version").fetchone()[0] or 0) + if current_version <= adoption_floor: + continue + manifests_by_tier[tier] = _released_train_manifests_by_target(manifest_root, tier) + tier_manifest_paths = tuple(manifest_root.glob(f"{tier.value}-*.json")) + bootstrap_version = fresh_bootstrap_versions.get(tier) + if bootstrap_version is not None and current_version < bootstrap_version: + raise DurableChangeTrainError( + f"{tier.value} durable tier regressed below fresh bootstrap v{bootstrap_version}" + ) + if bootstrap_version == current_version and not tier_manifest_paths: + validated_tiers.add(tier) + continue + _require_released_train_chain( + tier, + manifests_by_tier[tier], + current_version=current_version, + floor=_chain_floor(tier, fresh_bootstrap_versions), + ) + validated_tiers.add(tier) + + for manifest_path in manifest_paths: + train = load_durable_change_train_manifest(manifest_path) + if train.state is not DurableChangeTrainState.RELEASED: continue - _prove_and_release_persisted_train(archive_root, manifest_path, train) - reconciled.append(manifest_path) + with _open_existing_tier(archive_root / f"{train.tier.value}.db") as live: + actual = live_evidence_by_tier.get(train.tier) + if actual is None: + actual = capture_durable_database_evidence(live, train.tier) + live_evidence_by_tier[train.tier] = actual + if ( + actual.user_version > DURABLE_MIGRATION_ADOPTION_FLOORS[train.tier] + and train.tier not in validated_tiers + ): + if train.tier not in manifests_by_tier: + manifests_by_tier[train.tier] = _released_train_manifests_by_target(manifest_root, train.tier) + _require_released_train_chain( + train.tier, + manifests_by_tier[train.tier], + current_version=actual.user_version, + floor=_chain_floor(train.tier, fresh_bootstrap_versions), + ) + if actual.user_version > train.target_version: + if train.tier not in live_integrity_by_tier: + live_integrity_by_tier[train.tier] = tuple( + str(row[0]) for row in live.execute("PRAGMA integrity_check") + ) + if train.tier not in live_inventory_by_tier: + live_inventory_by_tier[train.tier] = _migration_runner.capture_durable_schema_inventory(live) + if train.tier not in canonical_inventory_by_tier: + canonical_inventory_by_tier[train.tier] = _canonical_schema_inventory( + train.tier, actual.user_version + ) + if live_evidence_cache is not None: + live_evidence_cache[train.tier] = _DurableForwardVersionEvidence( + actual=actual, + integrity_check=live_integrity_by_tier[train.tier], + live_inventory=live_inventory_by_tier[train.tier], + canonical_inventory=canonical_inventory_by_tier[train.tier], + ) + _verify_released_train_live_tier( + archive_root, + live, + train, + actual_evidence=actual, + integrity_check=live_integrity_by_tier.get(train.tier), + live_inventory=live_inventory_by_tier.get(train.tier), + canonical_inventory=canonical_inventory_by_tier.get(train.tier), + ) + record_reconciled(manifest_path) return tuple(reconciled) @@ -1697,6 +2325,7 @@ def __getattr__(name: str) -> object: "DurableChangeTrainManifest", "DurableMigrationSidecar", "DurableChangeTrainExecution", + "DurableForwardVersionReceipt", "durable_migration_sidecar_name", "validate_durable_migration_sidecars", "durable_change_train_policy_report", diff --git a/polylogue/storage/sqlite/migration_runner.py b/polylogue/storage/sqlite/migration_runner.py index decd9e3a0d..05c89a9396 100644 --- a/polylogue/storage/sqlite/migration_runner.py +++ b/polylogue/storage/sqlite/migration_runner.py @@ -1607,7 +1607,12 @@ def capture_durable_database_evidence( live_path = _connection_main_path(conn) from polylogue.storage.archive_identity import ArchiveIdentity - archive_identity_digest = ArchiveIdentity.resolve(live_path.parent).authority_identity_digest + # Durable migration evidence must survive replacement of rebuildable + # generations and creation of another durable tier. Bind a train to the + # file it actually migrates, not to the whole durable pair: a source train + # must remain valid when a previously absent user.db is initialized later. + tier_identity = ArchiveIdentity.resolve(live_path.parent).tier(tier.value).stable_id + archive_identity_digest = hashlib.sha256(tier_identity.encode("utf-8")).hexdigest() content_hasher = hashlib.sha256() for statement in conn.iterdump(): content_hasher.update(statement.encode("utf-8")) @@ -1624,18 +1629,50 @@ def capture_durable_database_evidence( ) +def _archive_identity_continuity_matches( + actual_digest: str, + expected_digest: str, + archive_root: Path, + tier: ArchiveTier, +) -> bool: + if actual_digest == expected_digest: + return True + from polylogue.storage.archive_identity import ArchiveIdentity + + identity = ArchiveIdentity.resolve(archive_root.resolve()) + legacy_digest = identity.authority_identity_digest + tier_identity = identity.tier(tier.value).stable_id + durable_digest = hashlib.sha256(tier_identity.encode("utf-8")).hexdigest() + # Manifests written before the durable-tier identity split contain the + # old full-archive digest. Admit that legacy evidence only when the + # current archive still has the same legacy identity and the newly + # captured evidence proves the durable identity is unchanged. + return expected_digest == legacy_digest and actual_digest == durable_digest + + def _assert_durable_database_continuity( actual: DurableDatabaseEvidence, expected: DurableDatabaseEvidence, *, label: str, + archive_root: Path | None = None, + connection: sqlite3.Connection | None = None, ) -> None: """Require the live durable file to retain its authenticated evidence.""" + identity_continuous = actual.archive_identity_digest == expected.archive_identity_digest + if not identity_continuous and (archive_root is not None or connection is not None): + resolved_archive_root = archive_root or _connection_main_path(cast(sqlite3.Connection, connection)).parent + identity_continuous = _archive_identity_continuity_matches( + actual.archive_identity_digest, + expected.archive_identity_digest, + resolved_archive_root, + actual.tier, + ) if ( actual.quick_check != expected.quick_check or actual.quick_check != ("ok",) or actual.user_version != expected.user_version - or actual.archive_identity_digest != expected.archive_identity_digest + or not identity_continuous or actual.content_sha256 != expected.content_sha256 ): raise DurableChangeTrainError(f"{label} durable tier identity/content continuity proof failed") @@ -2368,6 +2405,7 @@ def recover_durable_change_train( capture_durable_database_evidence(conn, train.tier), pre, label="rolled-back recovery", + connection=conn, ) updated = replace( train, diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index b023bbec92..ac3bb4dd88 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -1535,6 +1535,70 @@ def test_migrate_tier_cli_executes_and_persists_a_future_change_train( assert conn.execute("SELECT name FROM sqlite_schema WHERE name='future_items'").fetchone() == ("future_items",) +def test_migrate_tier_cli_exposes_forward_version_receipt( + cli_workspace: dict[str, Path], + cli_runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from polylogue.cli.commands.maintenance import _migrate_tier + from polylogue.storage.sqlite.durable_change_train import ( + DurableChangeTrainExecution, + DurableForwardVersionReceipt, + ) + + source_db = cli_workspace["archive_root"] / "source.db" + if not source_db.exists(): + with sqlite3.connect(source_db) as conn: + conn.execute("PRAGMA user_version = 3") + conn.commit() + receipt = DurableForwardVersionReceipt( + tier=ArchiveTier.SOURCE, + historical_train_id="train:source:v2", + historical_target_version=2, + current_target_version=3, + observed_live_version=3, + historical_schema_inventory_sha256="a" * 64, + archive_identity_digest="b" * 64, + ) + monkeypatch.setattr( + _migrate_tier, + "execute_durable_change_train", + lambda *_args, **_kwargs: DurableChangeTrainExecution( + train=None, + manifest_path=None, + migration_result=None, + forward_version_receipt=receipt, + ), + ) + + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "migrate-tier", "source", "--output-format", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["forward_version_receipt"] == { + "archive_identity_digest": "b" * 64, + "current_target_version": 3, + "historical_schema_inventory_sha256": "a" * 64, + "historical_target_version": 2, + "historical_train_id": "train:source:v2", + "observed_live_version": 3, + "tier": "source", + } + + plain = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "migrate-tier", "source"], + catch_exceptions=False, + ) + assert plain.exit_code == 0, plain.output + assert "historical train train:source:v2 is admitted at live schema v3" in plain.output + assert "(target v3)" in plain.output + + def test_migrate_tier_cli_refuses_live_daemon_before_sql( cli_workspace: dict[str, Path], cli_runner: CliRunner, diff --git a/tests/unit/daemon/test_raw_failure_sample.py b/tests/unit/daemon/test_raw_failure_sample.py index 290876596d..152058f84b 100644 --- a/tests/unit/daemon/test_raw_failure_sample.py +++ b/tests/unit/daemon/test_raw_failure_sample.py @@ -570,6 +570,150 @@ def test_daemon_status_lifecycle_counts_match_the_shared_projection(self, tmp_pa assert info["terminal_rejections"] == snapshot.terminal == 1 assert info["unexplained_failures"] == snapshot.unexplained == 0 + def test_daemon_status_uses_the_lifecycle_sample_rows(self, tmp_path: Path) -> None: + """Status must retain classified rows selected ahead of newer unexplained rows.""" + source_db = tmp_path / "source.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + conn.executemany( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size, + acquired_at_ms, parse_error + ) VALUES (?, ?, ?, ?, 0, ?, 0, ?, ?) + """, + [ + ( + f"raw-unexplained-{index}", + "codex-session", + f"native-{index}", + "/data/unexplained.jsonl", + bytes(index.to_bytes(32, "big")), + 1_770_000_000_100 + index, + "newer unexplained failure", + ) + for index in range(50) + ] + + [ + ( + "raw-terminal", + "unknown-export", + "terminal-native", + "/data/terminal.json", + bytes(32), + 1_770_000_000_000, + "parsed raw payload produced no sessions", + ) + ], + ) + upsert_raw_artifact( + conn, + "raw-terminal", + ArchiveSourceArtifact( + artifact_id="terminal-evidence", + origin="unknown-export", + source_path="/data/terminal.json", + source_index=0, + artifact_kind="terminal_unsupported_shape", + classification_reason="terminal_unsupported_shape", + support_status=ArtifactSupportStatus.UNSUPPORTED_PARSEABLE, + ), + ) + conn.commit() + + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=tmp_path / "index.db"), + ): + info = _raw_failure_info() + + samples = cast(list[RawFailureSample], info["samples"]) + assert len(samples) == 50 + assert any(sample.provider_hint == "unknown-export" and sample.lifecycle == "terminal" for sample in samples) + + def test_lifecycle_samples_are_sql_bounded_for_large_file_backed_failure_sets( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + source_db = tmp_path / "source.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + conn.executemany( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, source_index, blob_hash, + blob_size, acquired_at_ms, parse_error + ) VALUES (?, 'codex-session', ?, '/data/large-failures.jsonl', ?, ?, 0, ?, 'bad input') + """, + [ + (f"raw-{index}", f"native-{index}", index, bytes(32), 1_770_000_000_000 + index) + for index in range(200) + ], + ) + conn.commit() + + statements: list[str] = [] + + def open_traced_readonly(path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(f"file:{path.resolve()}?mode=ro", uri=True) + connection.set_trace_callback(statements.append) + return connection + + monkeypatch.setattr("polylogue.storage.raw_failure_lifecycle.open_readonly_connection", open_traced_readonly) + snapshot = read_raw_failure_lifecycle(source_db, sample_limit=3) + + assert snapshot.parse_failures == snapshot.unexplained == 200 + assert len(snapshot.samples) == 3 + summary_queries = [statement for statement in statements if "GROUP BY f.origin" in statement] + sample_queries = [statement for statement in statements if "FROM sampled" in statement] + assert len(summary_queries) == 1 + assert len(sample_queries) == 1 + assert "ROW_NUMBER()" not in sample_queries[0] + assert "NOT EXISTS" in sample_queries[0] + assert "LIMIT 3" in sample_queries[0] + + def test_lifecycle_samples_are_sql_bounded_without_artifact_table( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + source_db = tmp_path / "source.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + conn.executemany( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, source_index, blob_hash, + blob_size, acquired_at_ms, parse_error + ) VALUES (?, 'codex-session', ?, '/data/no-artifacts.jsonl', ?, ?, 0, ?, 'bad input') + """, + [ + (f"raw-{index}", f"native-{index}", index, bytes(32), 1_770_000_000_000 + index) + for index in range(5) + ], + ) + conn.execute("DROP TABLE raw_artifacts") + conn.commit() + + statements: list[str] = [] + + def open_traced_readonly(path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(f"file:{path.resolve()}?mode=ro", uri=True) + connection.set_trace_callback(statements.append) + return connection + + monkeypatch.setattr("polylogue.storage.raw_failure_lifecycle.open_readonly_connection", open_traced_readonly) + snapshot = read_raw_failure_lifecycle(source_db, sample_limit=3) + + assert snapshot.parse_failures == snapshot.unexplained == 5 + assert snapshot.deferred == snapshot.terminal == 0 + assert len(snapshot.samples) == 3 + assert all(sample["artifact_kind"] is None and sample["support_status"] is None for sample in snapshot.samples) + sample_queries = [statement for statement in statements if "ORDER BY acquired_at_ms DESC" in statement] + assert len(sample_queries) == 1 + assert "LIMIT 3" in sample_queries[0] + @pytest.mark.parametrize("source_state", ["missing", "malformed"]) def test_status_fails_closed_when_source_lifecycle_is_unavailable( self, @@ -680,7 +824,11 @@ def fetchall(self) -> list[object]: def __iter__(self) -> object: normalized = " ".join(self._sql.split()).lower() - if "from raw_sessions as r" in normalized and "limit 50" not in normalized: + if ( + "from raw_sessions as r" in normalized + and "limit 50" not in normalized + and "r.raw_id in (" not in normalized + ): raise AssertionError("raw-failure lifecycle totals must aggregate in SQL") return iter(self._cursor) diff --git a/tests/unit/storage/test_durable_change_train.py b/tests/unit/storage/test_durable_change_train.py index 1a1a506d88..affde4465f 100644 --- a/tests/unit/storage/test_durable_change_train.py +++ b/tests/unit/storage/test_durable_change_train.py @@ -5,6 +5,7 @@ import hashlib import json import os +import shutil import sqlite3 import sys from collections.abc import Callable, Iterator @@ -376,6 +377,8 @@ def test_released_source_train_can_record_an_authorized_mutation_refresh( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: + from polylogue.storage.archive_identity import ArchiveIdentity + db_path = tmp_path / "source.db" _create_current_database(db_path) _install_synthetic_migration(tmp_path, monkeypatch, ArchiveTier.SOURCE) @@ -399,6 +402,17 @@ def test_released_source_train_can_record_an_authorized_mutation_refresh( restart_convergence=restart, ) train = release_durable_change_train(train, evidence_ref="proof:release") + assert train.apply_evidence is not None + train = replace( + train, + apply_evidence=replace( + train.apply_evidence, + post=replace( + train.apply_evidence.post, + archive_identity_digest=ArchiveIdentity.resolve(tmp_path).authority_identity_digest, + ), + ), + ) manifest = tmp_path / ".maintenance-state" / "durable-change-trains" / "source-002.json" manifest.parent.mkdir(parents=True) write_durable_change_train_manifest(manifest, train, expected_revision=-1) @@ -457,7 +471,14 @@ def record_refresh_fsync(path: Path) -> None: assert refreshed.state is DurableChangeTrainState.RELEASED assert refreshed.source_continuity_evidence is not None assert released.apply_evidence is not None - assert refreshed.apply_evidence == released.apply_evidence + assert refreshed.apply_evidence is not None + assert released.apply_evidence is not None + assert refreshed.apply_evidence.pre == released.apply_evidence.pre + assert refreshed.apply_evidence.post.archive_identity_digest != released.apply_evidence.post.archive_identity_digest + assert ( + refreshed.apply_evidence.post.archive_identity_digest + == refreshed.source_continuity_evidence.archive_identity_digest + ) assert refreshed.proof == released.proof assert refreshed.revision == released.revision + 1 assert any(ref.startswith("proof:source-continuity-refresh:") for ref in refreshed.proof_refs) @@ -1115,6 +1136,545 @@ def test_maintenance_route_replays_historical_sidecars_before_current_target( assert conn.execute("PRAGMA user_version").fetchone() == (3,) assert conn.execute("SELECT name FROM sqlite_schema WHERE name='later_items'").fetchone() == ("later_items",) assert released == [True, True] + historical_manifest = durable_change_train_manifest_path(tmp_path, ArchiveTier.SOURCE, 2) + manifest_v3 = durable_change_train_manifest_path(tmp_path, ArchiveTier.SOURCE, 3) + manifest_v3_bytes = manifest_v3.read_bytes() + manifest_v3.unlink() + with pytest.raises(DurableChangeTrainError, match=r"versions \[3\]"): + durable_change_train_module.reconcile_durable_change_train_startup(tmp_path) + assert released == [True, True] + with pytest.raises(DurableChangeTrainError, match="lacks released train evidence"): + execute_durable_change_train( + tmp_path, + ArchiveTier.SOURCE, + backup_manifest=None, + daemon_stopped_evidence_ref="proof:daemon-stopped", + single_writer_evidence_ref="proof:archive-ownership-lock", + release_archive_ownership=lambda: pytest.fail("missing intervening train was admitted"), + ) + manifest_v3.write_bytes(manifest_v3_bytes) + evidence_captures = 0 + schema_inventories = 0 + canonical_inventories = 0 + real_capture = migration_runner.capture_durable_database_evidence + real_schema_inventory = migration_runner.capture_durable_schema_inventory + real_canonical_inventory = durable_change_train_module._canonical_schema_inventory + + def count_evidence_captures( + connection: sqlite3.Connection, tier: ArchiveTier + ) -> migration_runner.DurableDatabaseEvidence: + nonlocal evidence_captures + evidence_captures += 1 + return real_capture(connection, tier) + + def count_schema_inventories( + connection: sqlite3.Connection, + ) -> migration_runner.DurableSchemaInventory: + nonlocal schema_inventories + schema_inventories += 1 + return real_schema_inventory(connection) + + def count_canonical_inventories(tier: ArchiveTier, target_version: int) -> migration_runner.DurableSchemaInventory: + nonlocal canonical_inventories + canonical_inventories += 1 + return real_canonical_inventory(tier, target_version) + + monkeypatch.setattr(durable_change_train_module, "capture_durable_database_evidence", count_evidence_captures) + monkeypatch.setattr(migration_runner, "capture_durable_schema_inventory", count_schema_inventories) + monkeypatch.setattr(durable_change_train_module, "_canonical_schema_inventory", count_canonical_inventories) + third = execute_durable_change_train( + tmp_path, + ArchiveTier.SOURCE, + backup_manifest=None, + daemon_stopped_evidence_ref="proof:daemon-stopped", + single_writer_evidence_ref="proof:archive-ownership-lock", + release_archive_ownership=lambda: released.append(True), + ) + assert third.forward_version_receipt is not None + assert third.forward_version_receipt.historical_target_version == 2 + assert third.forward_version_receipt.observed_live_version == 3 + assert evidence_captures == 1 + # Three inventories: one nested inside the single evidence capture, one + # live inventory read during startup reconciliation, and one canonical + # inventory built from the fresh DDL. The no-op receipt reuses all of them. + assert schema_inventories == 3 + assert canonical_inventories == 1 + + historical_train = load_durable_change_train_manifest(historical_manifest) + with sqlite3.connect(db_path) as conn: + actual = migration_runner.capture_durable_database_evidence(conn, ArchiveTier.SOURCE) + receipt = durable_change_train_module._verify_released_train_live_tier( + tmp_path, + conn, + historical_train, + current_target_version=3, + actual_evidence=actual, + ) + assert receipt is not None + assert receipt.historical_target_version == 2 + assert receipt.current_target_version == 3 + assert receipt.observed_live_version == 3 + assert historical_train.proof is not None + assert ( + receipt.historical_schema_inventory_sha256 == historical_train.proof.fresh_ddl_parity.migrated_inventory_sha256 + ) + with sqlite3.connect(db_path) as conn: + with pytest.raises(DurableChangeTrainError, match="is newer than current target"): + durable_change_train_module._verify_released_train_live_tier( + tmp_path, + conn, + historical_train, + current_target_version=2, + actual_evidence=actual, + ) + + captures = 0 + real_capture = migration_runner.capture_durable_database_evidence + + def count_captures(connection: sqlite3.Connection, tier: ArchiveTier) -> migration_runner.DurableDatabaseEvidence: + nonlocal captures + captures += 1 + return real_capture(connection, tier) + + monkeypatch.setattr(durable_change_train_module, "capture_durable_database_evidence", count_captures) + assert reconcile_durable_change_train_startup(tmp_path) == ( + durable_change_train_manifest_path(tmp_path, ArchiveTier.SOURCE, 2), + durable_change_train_manifest_path(tmp_path, ArchiveTier.SOURCE, 3), + ) + assert captures == 1 + + unrelated_root = tmp_path / "unrelated-archive" + unrelated_root.mkdir() + shutil.copy2(db_path, unrelated_root / "source.db") + unrelated_manifest = durable_change_train_manifest_path(unrelated_root, ArchiveTier.SOURCE, 2) + unrelated_manifest.parent.mkdir(parents=True) + shutil.copy2(historical_manifest, unrelated_manifest) + shutil.copy2(manifest_v3, durable_change_train_manifest_path(unrelated_root, ArchiveTier.SOURCE, 3)) + with sqlite3.connect(unrelated_root / "source.db") as conn: + assert conn.execute("PRAGMA integrity_check").fetchone() == ("ok",) + with pytest.raises(DurableChangeTrainError, match="immutable archive identity differs"): + reconcile_durable_change_train_startup(unrelated_root) + + with sqlite3.connect(db_path) as conn: + conn.execute("DROP TABLE base_items") + conn.commit() + tampered = migration_runner.capture_durable_database_evidence(conn, ArchiveTier.SOURCE) + with pytest.raises(DurableChangeTrainError, match="canonical live version"): + durable_change_train_module._verify_released_train_live_tier( + tmp_path, + conn, + historical_train, + current_target_version=3, + actual_evidence=tampered, + ) + + +def test_continuity_admits_legacy_full_archive_identity_digest(tmp_path: Path) -> None: + from polylogue.storage.archive_identity import ArchiveIdentity + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + with sqlite3.connect(tmp_path / "source.db") as conn: + current = migration_runner.capture_durable_database_evidence(conn, ArchiveTier.SOURCE) + legacy = replace( + current, + archive_identity_digest=ArchiveIdentity.resolve(tmp_path).authority_identity_digest, + ) + migration_runner._assert_durable_database_continuity( + current, + legacy, + label="legacy identity compatibility", + connection=conn, + ) + with pytest.raises(DurableChangeTrainError, match="continuity proof failed"): + migration_runner._assert_durable_database_continuity( + current, + replace(current, archive_identity_digest="a" * 64), + label="foreign identity", + connection=conn, + ) + with pytest.raises(DurableChangeTrainError, match="continuity proof failed"): + migration_runner._assert_durable_database_continuity( + current, + legacy, + label="legacy identity without connection", + ) + + +def test_released_train_chain_is_anchored_at_adoption_floor() -> None: + floor = DURABLE_MIGRATION_ADOPTION_FLOORS[ArchiveTier.SOURCE] + released = cast(DurableChangeTrain, SimpleNamespace(state=DurableChangeTrainState.RELEASED)) + + with pytest.raises(DurableChangeTrainError, match=rf"versions \[{floor + 1}\]"): + durable_change_train_module._require_released_train_chain( + ArchiveTier.SOURCE, + { + floor + 2: released, + floor + 3: released, + }, + current_version=floor + 3, + ) + + +def test_released_train_chain_can_start_at_bootstrap_floor(monkeypatch: pytest.MonkeyPatch) -> None: + floor = DURABLE_MIGRATION_ADOPTION_FLOORS[ArchiveTier.SOURCE] + current_version = floor + 1 + released = cast(DurableChangeTrain, SimpleNamespace(state=DurableChangeTrainState.RELEASED)) + monkeypatch.setattr(durable_change_train_module, "_historical_schema_evidence", lambda _train: None) + + durable_change_train_module._require_released_train_chain( + ArchiveTier.SOURCE, + {current_version: released}, + current_version=current_version, + floor=floor, + ) + + +def test_forward_receipt_checks_missing_chain_before_empty_history( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + floor = DURABLE_MIGRATION_ADOPTION_FLOORS[ArchiveTier.SOURCE] + current_version = floor + 2 + current_train = cast( + DurableChangeTrain, + SimpleNamespace(target_version=current_version, state=DurableChangeTrainState.RELEASED), + ) + monkeypatch.setattr( + durable_change_train_module, + "_released_train_manifests_by_target", + lambda _manifest_root, _tier: {current_version: current_train}, + ) + + with ( + sqlite3.connect(":memory:") as conn, + pytest.raises( + DurableChangeTrainError, + match=rf"versions \[{floor + 1}\]", + ), + ): + durable_change_train_module._forward_version_receipt_for_current_tier( + tmp_path, + conn, + ArchiveTier.SOURCE, + current_version=current_version, + current_target_version=current_version, + ) + + +def test_forward_receipt_skips_non_train_audit_tier(tmp_path: Path) -> None: + with sqlite3.connect(":memory:") as conn: + assert ( + durable_change_train_module._forward_version_receipt_for_current_tier( + tmp_path, + conn, + ArchiveTier.AUDIT, + current_version=1, + current_target_version=1, + ) + is None + ) + + +def test_startup_recovers_later_train_before_released_chain_validation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manifest_root = tmp_path / ".maintenance-state" / "durable-change-trains" + manifest_root.mkdir(parents=True) + first_path = manifest_root / "source-027.json" + later_path = manifest_root / "source-028.json" + first_path.touch() + later_path.touch() + released = cast( + DurableChangeTrain, + SimpleNamespace(state=DurableChangeTrainState.RELEASED, tier=ArchiveTier.SOURCE, target_version=27), + ) + later_released = cast( + DurableChangeTrain, + SimpleNamespace(state=DurableChangeTrainState.RELEASED, tier=ArchiveTier.SOURCE, target_version=28), + ) + backup_authorized = cast( + DurableChangeTrain, + SimpleNamespace( + state=DurableChangeTrainState.BACKUP_AUTHORIZED, + tier=ArchiveTier.SOURCE, + target_version=28, + train_id="train:source:v28", + revision=0, + ), + ) + states = {first_path: released, later_path: backup_authorized} + events: list[tuple[str, int]] = [] + + @contextmanager + def fake_open_tier(_path: Path) -> Iterator[sqlite3.Connection]: + with sqlite3.connect(":memory:") as connection: + yield connection + + def fake_load(path: Path) -> DurableChangeTrain: + return states[path] + + def fake_persist(path: Path, train: DurableChangeTrain, *, expected_revision: int) -> DurableChangeTrain: + states[path] = train + return train + + def fake_recover(*_args: object, **_kwargs: object) -> DurableChangeTrain: + events.append(("recover", 28)) + return later_released + + def fake_capture(*_args: object, **_kwargs: object) -> SimpleNamespace: + return SimpleNamespace(user_version=28) + + monkeypatch.setattr( + durable_change_train_module, + "_recover_pending_source_continuity_intents", + lambda _root: None, + ) + monkeypatch.setattr(durable_change_train_module, "_open_existing_tier", fake_open_tier) + monkeypatch.setattr(durable_change_train_module, "load_durable_change_train_manifest", fake_load) + monkeypatch.setattr(durable_change_train_module, "_persist_train_transition", fake_persist) + monkeypatch.setattr(durable_change_train_module, "reconcile_interrupted_durable_change_train", fake_recover) + monkeypatch.setattr(durable_change_train_module, "capture_durable_database_evidence", fake_capture) + monkeypatch.setattr(durable_change_train_module, "_historical_schema_evidence", lambda _train: None) + monkeypatch.setattr( + migration_runner, + "capture_durable_schema_inventory", + lambda _connection: SimpleNamespace(sha256="inventory"), + ) + monkeypatch.setattr( + durable_change_train_module, + "_canonical_schema_inventory", + lambda _tier, _version: SimpleNamespace(sha256="canonical"), + ) + monkeypatch.setattr( + durable_change_train_module, + "_verify_released_train_live_tier", + lambda _root, _connection, train, **_kwargs: events.append(("verify", train.target_version)), + ) + + durable_change_train_module._reconcile_durable_change_train_startup_locked(tmp_path) + + assert events == [("recover", 28), ("verify", 27), ("verify", 28)] + + +def test_startup_checks_chain_when_only_current_train_remains( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manifest_root = tmp_path / ".maintenance-state" / "durable-change-trains" + manifest_root.mkdir(parents=True) + manifest_path = manifest_root / "source-028.json" + manifest_path.touch() + current = cast( + DurableChangeTrain, + SimpleNamespace(state=DurableChangeTrainState.RELEASED, tier=ArchiveTier.SOURCE, target_version=28), + ) + + @contextmanager + def fake_open_tier(_path: Path) -> Iterator[sqlite3.Connection]: + with sqlite3.connect(":memory:") as connection: + yield connection + + monkeypatch.setattr(durable_change_train_module, "_recover_pending_source_continuity_intents", lambda _root: None) + monkeypatch.setattr(durable_change_train_module, "_open_existing_tier", fake_open_tier) + monkeypatch.setattr(durable_change_train_module, "load_durable_change_train_manifest", lambda _path: current) + monkeypatch.setattr( + durable_change_train_module, + "capture_durable_database_evidence", + lambda _connection, _tier: SimpleNamespace(user_version=28), + ) + monkeypatch.setattr( + durable_change_train_module, + "_released_train_manifests_by_target", + lambda _root, _tier: {28: current}, + ) + + with pytest.raises(DurableChangeTrainError, match="lacks released train evidence"): + durable_change_train_module._reconcile_durable_change_train_startup_locked(tmp_path) + + +def test_startup_checks_chain_when_manifest_directory_is_missing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + (tmp_path / "source.db").touch() + + @contextmanager + def fake_open_tier(_path: Path) -> Iterator[sqlite3.Connection]: + with sqlite3.connect(":memory:") as connection: + connection.execute("PRAGMA user_version = 28") + yield connection + + monkeypatch.setattr(durable_change_train_module, "_recover_pending_source_continuity_intents", lambda _root: None) + monkeypatch.setattr(durable_change_train_module, "_open_existing_tier", fake_open_tier) + + with pytest.raises(DurableChangeTrainError, match="lacks released train evidence"): + durable_change_train_module._reconcile_durable_change_train_startup_locked(tmp_path) + + +def test_fresh_archive_bootstrap_receipt_allows_repeat_startup(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + assert reconcile_durable_change_train_startup(tmp_path) == () + initialize_active_archive_root(tmp_path) + + +def test_fresh_bootstrap_intent_recovers_after_late_tier_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from polylogue.storage.sqlite.archive_tiers import bootstrap + + real_initialize_archive_database = bootstrap.initialize_archive_database + failed = False + + def fail_embeddings_once( + path: Path, + tier: ArchiveTier, + *, + allow_create: bool = True, + expected_version: int | None = None, + ) -> None: + nonlocal failed + if tier is ArchiveTier.EMBEDDINGS and not failed: + failed = True + raise RuntimeError("simulated late fresh-bootstrap failure") + real_initialize_archive_database(path, tier, allow_create=allow_create, expected_version=expected_version) + + monkeypatch.setattr(bootstrap, "initialize_archive_database", fail_embeddings_once) + with pytest.raises(RuntimeError, match="simulated late fresh-bootstrap failure"): + bootstrap.initialize_active_archive_root(tmp_path) + + marker_root = tmp_path / ".maintenance-state" / "durable-change-trains" + assert (marker_root / ".bootstrap.pending").is_file() + assert not (marker_root / ".bootstrap").exists() + assert (tmp_path / "source.db").is_file() + + monkeypatch.setattr(bootstrap, "initialize_archive_database", real_initialize_archive_database) + bootstrap.initialize_active_archive_root(tmp_path) + + assert (marker_root / ".bootstrap").is_file() + assert not (marker_root / ".bootstrap.pending").exists() + assert reconcile_durable_change_train_startup(tmp_path) == () + + +def test_fresh_bootstrap_intent_rejects_tampering_before_recovery( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from polylogue.storage.sqlite.archive_tiers import bootstrap + + real_initialize_archive_database = bootstrap.initialize_archive_database + + def fail_embeddings( + path: Path, + tier: ArchiveTier, + *, + allow_create: bool = True, + expected_version: int | None = None, + ) -> None: + if tier is ArchiveTier.EMBEDDINGS: + raise RuntimeError("simulated late fresh-bootstrap failure") + real_initialize_archive_database(path, tier, allow_create=allow_create, expected_version=expected_version) + + monkeypatch.setattr(bootstrap, "initialize_archive_database", fail_embeddings) + with pytest.raises(RuntimeError, match="simulated late fresh-bootstrap failure"): + bootstrap.initialize_active_archive_root(tmp_path) + + pending = tmp_path / ".maintenance-state" / "durable-change-trains" / ".bootstrap.pending" + payload = json.loads(pending.read_text(encoding="utf-8")) + payload["durable_identity_digest"] = "0" * 64 + pending.write_text(json.dumps(payload), encoding="utf-8") + + with pytest.raises(DurableChangeTrainError, match="intent durable identity mismatch"): + bootstrap.initialize_active_archive_root(tmp_path) + + +def test_pre_marker_current_archive_is_adopted_once(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + marker = tmp_path / ".maintenance-state" / "durable-change-trains" / ".bootstrap" + marker.unlink() + + initialize_active_archive_root(tmp_path) + assert marker.is_file() + + +def test_pre_marker_adoption_refuses_missing_train_directory(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + marker_root = tmp_path / ".maintenance-state" / "durable-change-trains" + (marker_root / ".bootstrap").unlink() + marker_root.rmdir() + + with pytest.raises(DurableChangeTrainError, match="lacks released train evidence"): + initialize_active_archive_root(tmp_path) + + +def test_pre_marker_adoption_requires_all_durable_tiers(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + (tmp_path / ".maintenance-state" / "durable-change-trains" / ".bootstrap").unlink() + (tmp_path / "user.db").unlink() + + with pytest.raises(DurableChangeTrainError, match="lacks released train evidence"): + initialize_active_archive_root(tmp_path) + assert not (tmp_path / "user.db").exists() + + +def test_bootstrap_marker_survives_index_generation_replacement(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + (tmp_path / "index.db").unlink() + + initialize_active_archive_root(tmp_path) + + +def test_fresh_bootstrap_receipt_rejects_archive_identity_mismatch(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + marker = tmp_path / ".maintenance-state" / "durable-change-trains" / ".bootstrap" + payload = json.loads(marker.read_text(encoding="utf-8")) + payload["durable_identity_digest"] = "0" * 64 + marker.write_text(json.dumps(payload), encoding="utf-8") + + with pytest.raises(DurableChangeTrainError, match="durable identity mismatch"): + reconcile_durable_change_train_startup(tmp_path) + + +def test_fresh_bootstrap_receipt_rejects_recorded_version_tampering(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + marker = tmp_path / ".maintenance-state" / "durable-change-trains" / ".bootstrap" + payload = json.loads(marker.read_text(encoding="utf-8")) + cast(dict[str, int], payload["versions"])["source"] += 1 + marker.write_text(json.dumps(payload), encoding="utf-8") + + with pytest.raises(DurableChangeTrainError, match="marker digest mismatch"): + reconcile_durable_change_train_startup(tmp_path) + + +def test_source_train_identity_survives_late_user_tier_initialization(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + + source_path = tmp_path / "source.db" + initialize_archive_database(source_path, ArchiveTier.SOURCE) + with sqlite3.connect(source_path) as conn: + before = migration_runner.capture_durable_database_evidence(conn, ArchiveTier.SOURCE) + + initialize_archive_database(tmp_path / "user.db", ArchiveTier.USER) + with sqlite3.connect(source_path) as conn: + after = migration_runner.capture_durable_database_evidence(conn, ArchiveTier.SOURCE) + + assert after.archive_identity_digest == before.archive_identity_digest def test_future_train_sidecar_hash_and_slot_are_admission_bound(