feat(agent99): add digest-bound truth-chain index executor

scripts/reboot-recovery/tests/test_agent99_truth_chain_index_executor_contract.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import ast
+import hashlib
+import re
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+OVERLAY = ROOT / "ops/agent99-truth-chain-index-executor-overlay.py"
+MANAGER = ROOT / "ops/agent99-truth-chain-index-manager.ps1"
+MIGRATION = (
+    ROOT
+    / "apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17.sql"
+)
+EXPECTED_MIGRATION_SHA256 = (
+    "657089552ceed8de916d87135164b255c85edc154bf437f870b8c2bbdf14c509"
+)
+
+
+def _assignment(module: ast.Module, name: str):
+    for node in module.body:
+        if (
+            isinstance(node, ast.Assign)
+            and len(node.targets) == 1
+            and isinstance(node.targets[0], ast.Name)
+            and node.targets[0].id == name
+        ):
+            return ast.literal_eval(node.value)
+    raise AssertionError(f"assignment not found: {name}")
+
+
+def _normalize_sql(value: str) -> str:
+    return re.sub(r"\s+", " ", value.strip()).lower()
+
+
+def _migration_statements() -> list[str]:
+    source = MIGRATION.read_text(encoding="utf-8")
+    source = "\n".join(
+        line for line in source.splitlines() if not line.lstrip().startswith("--")
+    )
+    return [part.strip() for part in source.split(";") if part.strip()]
+
+
+def test_fixed_executor_sql_matches_the_digest_bound_migration() -> None:
+    module = ast.parse(OVERLAY.read_text(encoding="utf-8"))
+    extensions = _assignment(module, "_EXTENSION_OPERATIONS")
+    indexes = _assignment(module, "_INDEX_OPERATIONS")
+    embedded = [sql for _, sql in (*extensions, *indexes)]
+
+    assert len(extensions) == 2
+    assert len(indexes) == 12
+    assert [_normalize_sql(sql) for sql in embedded] == [
+        _normalize_sql(sql) for sql in _migration_statements()
+    ]
+    assert hashlib.sha256(MIGRATION.read_bytes()).hexdigest() == (
+        EXPECTED_MIGRATION_SHA256
+    )
+
+
+def test_executor_has_no_raw_sql_or_arbitrary_command_input() -> None:
+    source = OVERLAY.read_text(encoding="utf-8")
+
+    assert 'choices=("check", "apply", "verify")' in source
+    assert "--trace-id" in source
+    assert "--run-id" in source
+    assert "--work-item-id" in source
+    assert "--source-revision" in source
+    assert "--migration-sha256" in source
+    assert "--sql" not in source
+    assert "--command" not in source
+    assert '"raw_sql_accepted": False' in source
+    assert '"raw_sql_emitted": False' in source
+    assert '"query_parameters_emitted": False' in source
+    assert '"secret_value_read": False' in source
+    assert '"arbitrary_command_allowed": False' in source
+    assert '"cross_domain_fallback_allowed": False' in source
+    assert 'isolation_level="AUTOCOMMIT"' in source
+    assert "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in source
+    assert 'text("SET lock_timeout = \'15s\'")' in source
+    assert 'text("SET statement_timeout = \'10min\'")' in source
+
+
+def test_check_success_is_separate_from_target_readiness() -> None:
+    source = OVERLAY.read_text(encoding="utf-8")
+
+    assert '"passed": args.mode == "check" or bool(after["passed"])' in source
+    assert '"target_ready": bool(after["passed"])' in source
+    assert 'terminal = "check_ready"' in source
+    assert 'terminal = "verified_healthy" if after["passed"]' in source
+
+
+def test_manager_is_windows99_bound_and_uses_independent_api_pods() -> None:
+    source = MANAGER.read_text(encoding="utf-8")
+
+    assert 'executor = "host:192.168.0.99"' in source
+    assert '$JumpHost = "192.168.0.110"' in source
+    assert '$ControlPlaneHost = "192.168.0.120"' in source
+    assert '$PodSelector = "app=awoooi-api,environment=prod,system=awoooi"' in source
+    assert "two_ready_api_pods_required" in source
+    assert '$executorPod = [string]$pods[0]' in source
+    assert '$verifierPod = [string]$pods[1]' in source
+    assert 'Invoke-FixedStage -Stage "check" -PodName $executorPod' in source
+    assert 'Invoke-FixedStage -Stage "apply" -PodName $executorPod' in source
+    assert '$verifierStage = if ($Mode -eq "Apply") { "verify" } else { "check" }' in source
+    assert "Get-CatalogSignature $checkReceipt.catalog_after" in source
+    assert "Get-CatalogSignature $verifierReceipt.catalog_after" in source
+    assert "$independentVerifier" in source
+
+
+def test_manager_is_digest_bound_fail_closed_and_cleans_staging() -> None:
+    source = MANAGER.read_text(encoding="utf-8")
+
+    assert "overlay_hash_mismatch" in source
+    assert "stage_receipt_contract_failed" in source
+    assert "target_pod_contract_failed" in source
+    assert "StrictHostKeyChecking=yes" in source
+    assert "StrictHostKeyChecking=no" not in source
+    assert "NumberOfPasswordPrompts=0" in source
+    assert "-RedirectStandardInput $OverlayFullPath" in source
+    assert 'Remove-Item -LiteralPath $OverlayFullPath -Force' in source
+    assert "$stagedOverlayDeleted" in source
+    assert 'rawSqlAccepted = $false' in source
+    assert 'arbitraryCommandAllowed = $false' in source
+    assert 'secretValueRead = $false' in source
+    assert 'telegramNotificationSent = $false' in source
+    assert 'kmRagWritebackPerformed = $false' in source
+    assert '"runtime_verified_controlled_apply"' in source
+