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

ops/agent99-truth-chain-index-executor-overlay.py

@@ -0,0 +1,425 @@
+#!/usr/bin/env python3
+"""Apply the fixed truth-chain index set without accepting raw SQL."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import hashlib
+import io
+import json
+import os
+import re
+import socket
+import sys
+import time
+from contextlib import redirect_stderr, redirect_stdout
+from typing import Any
+
+from sqlalchemy import text
+
+from src.db.base import get_engine
+
+
+_SCHEMA_VERSION = "agent99_truth_chain_index_executor_v1"
+_COMMAND_ID = "truth_chain_structured_indexes_v1"
+_CANONICAL_ASSET = "database:awoooi:prod/truth-chain-structured-indexes"
+_MIGRATION_SHA256 = (
+    "657089552ceed8de916d87135164b255c85edc154bf437f870b8c2bbdf14c509"
+)
+_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$")
+
+_EXTENSION_OPERATIONS = (
+    ("pg_trgm", "CREATE EXTENSION IF NOT EXISTS pg_trgm"),
+    ("btree_gin", "CREATE EXTENSION IF NOT EXISTS btree_gin"),
+)
+
+_INDEX_OPERATIONS = (
+    (
+        "idx_aol_incident_text_created",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_aol_incident_text_created
+            ON automation_operation_log ((incident_id::text), created_at DESC)
+            WHERE incident_id IS NOT NULL
+        """,
+    ),
+    (
+        "idx_aol_input_incident_created",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_aol_input_incident_created
+            ON automation_operation_log ((input ->> 'incident_id'), created_at DESC)
+            WHERE input ? 'incident_id'
+        """,
+    ),
+    (
+        "idx_aol_output_incident_created",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_aol_output_incident_created
+            ON automation_operation_log ((output ->> 'incident_id'), created_at DESC)
+            WHERE output ? 'incident_id'
+        """,
+    ),
+    (
+        "idx_aol_input_source_incidents_gin",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_aol_input_source_incidents_gin
+            ON automation_operation_log
+            USING GIN ((COALESCE(input #> '{source_refs,incident_ids}', '[]'::jsonb)))
+        """,
+    ),
+    (
+        "idx_aol_output_source_incidents_gin",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_aol_output_source_incidents_gin
+            ON automation_operation_log
+            USING GIN ((COALESCE(output #> '{source_refs,incident_ids}', '[]'::jsonb)))
+        """,
+    ),
+    (
+        "idx_outbound_source_incident_ids_gin",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_source_incident_ids_gin
+            ON awooop_outbound_message
+            USING GIN ((COALESCE(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb)))
+        """,
+    ),
+    (
+        "idx_outbound_source_code_refs_gin",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_source_code_refs_gin
+            ON awooop_outbound_message
+            USING GIN ((COALESCE(source_envelope #> '{source_refs,code_refs}', '[]'::jsonb)))
+        """,
+    ),
+    (
+        "idx_outbound_callback_incident_recent",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_callback_incident_recent
+            ON awooop_outbound_message (
+                project_id,
+                (source_envelope #>> '{callback_reply,incident_id}'),
+                queued_at DESC
+            )
+            WHERE source_envelope #>> '{callback_reply,incident_id}' IS NOT NULL
+        """,
+    ),
+    (
+        "idx_outbound_alert_incident_recent",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_alert_incident_recent
+            ON awooop_outbound_message (
+                project_id,
+                (source_envelope #>> '{alert_notification,incident_id}'),
+                queued_at DESC
+            )
+            WHERE source_envelope #>> '{alert_notification,incident_id}' IS NOT NULL
+        """,
+    ),
+    (
+        "idx_outbound_source_incident_recent",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_source_incident_recent
+            ON awooop_outbound_message (
+                project_id,
+                (source_envelope #>> '{source_refs,incident_id}'),
+                queued_at DESC
+            )
+            WHERE source_envelope #>> '{source_refs,incident_id}' IS NOT NULL
+        """,
+    ),
+    (
+        "idx_outbound_top_incident_recent",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_top_incident_recent
+            ON awooop_outbound_message (
+                project_id,
+                (source_envelope ->> 'incident_id'),
+                queued_at DESC
+            )
+            WHERE source_envelope ->> 'incident_id' IS NOT NULL
+        """,
+    ),
+    (
+        "idx_outbound_project_content_preview_trgm",
+        """
+        CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_outbound_project_content_preview_trgm
+            ON awooop_outbound_message
+            USING GIN (project_id, content_preview gin_trgm_ops)
+            WHERE content_preview IS NOT NULL
+        """,
+    ),
+)
+
+_EXPECTED_EXTENSIONS = tuple(name for name, _ in _EXTENSION_OPERATIONS)
+_EXPECTED_INDEXES = tuple(name for name, _ in _INDEX_OPERATIONS)
+
+_CATALOG_SQL = text(
+    """
+    SELECT index_class.relname AS index_name,
+           index_state.indisvalid,
+           index_state.indisready
+    FROM pg_catalog.pg_class AS index_class
+    JOIN pg_catalog.pg_index AS index_state
+      ON index_state.indexrelid = index_class.oid
+    WHERE index_class.relname IN (
+        'idx_aol_incident_text_created',
+        'idx_aol_input_incident_created',
+        'idx_aol_output_incident_created',
+        'idx_aol_input_source_incidents_gin',
+        'idx_aol_output_source_incidents_gin',
+        'idx_outbound_source_incident_ids_gin',
+        'idx_outbound_source_code_refs_gin',
+        'idx_outbound_callback_incident_recent',
+        'idx_outbound_alert_incident_recent',
+        'idx_outbound_source_incident_recent',
+        'idx_outbound_top_incident_recent',
+        'idx_outbound_project_content_preview_trgm'
+    )
+    """
+)
+
+
+def _normalized_sql(value: str) -> str:
+    return re.sub(r"\s+", " ", value.strip()).lower()
+
+
+def _sha256(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _operation_digest() -> str:
+    statements = [sql for _, sql in (*_EXTENSION_OPERATIONS, *_INDEX_OPERATIONS)]
+    return _sha256("\n".join(_normalized_sql(sql) for sql in statements))
+
+
+def _safe_error(exc: BaseException) -> dict[str, Any]:
+    database_error = getattr(exc, "orig", None)
+    sqlstate = str(
+        getattr(database_error, "sqlstate", None)
+        or getattr(database_error, "pgcode", None)
+        or ""
+    )
+    return {
+        "error_class": exc.__class__.__name__,
+        "database_error_class": (
+            database_error.__class__.__name__ if database_error else None
+        ),
+        "sqlstate": sqlstate or None,
+    }
+
+
+async def _catalog(connection: Any) -> dict[str, Any]:
+    extension_result = await connection.execute(
+        text(
+            """
+            SELECT extname
+            FROM pg_catalog.pg_extension
+            WHERE extname IN ('pg_trgm', 'btree_gin')
+            ORDER BY extname
+            """
+        )
+    )
+    extensions = {
+        str(row.get("extname") or "")
+        for row in extension_result.mappings().all()
+    }
+    index_result = await connection.execute(_CATALOG_SQL)
+    index_rows = list(index_result.mappings().all())
+    present = {str(row.get("index_name") or "") for row in index_rows}
+    valid_ready = {
+        str(row.get("index_name") or "")
+        for row in index_rows
+        if row.get("indisvalid") and row.get("indisready")
+    }
+    missing_extensions = sorted(set(_EXPECTED_EXTENSIONS) - extensions)
+    missing_indexes = sorted(set(_EXPECTED_INDEXES) - present)
+    invalid_or_unready = sorted(present - valid_ready)
+    return {
+        "expected_extension_count": len(_EXPECTED_EXTENSIONS),
+        "extension_count": len(extensions),
+        "missing_extensions": missing_extensions,
+        "expected_index_count": len(_EXPECTED_INDEXES),
+        "present_index_count": len(present),
+        "valid_ready_index_count": len(valid_ready),
+        "missing_indexes": missing_indexes,
+        "invalid_or_unready_indexes": invalid_or_unready,
+        "passed": not missing_extensions
+        and not missing_indexes
+        and not invalid_or_unready,
+    }
+
+
+async def _execute_fixed_operation(
+    connection: Any,
+    *,
+    operation_type: str,
+    name: str,
+    sql: str,
+) -> dict[str, Any]:
+    started = time.perf_counter()
+    try:
+        await connection.execute(text(sql))
+    except Exception as exc:
+        return {
+            "type": operation_type,
+            "name": name,
+            "statement_sha256": _sha256(_normalized_sql(sql)),
+            "duration_ms": round((time.perf_counter() - started) * 1000),
+            "success": False,
+            **_safe_error(exc),
+        }
+    return {
+        "type": operation_type,
+        "name": name,
+        "statement_sha256": _sha256(_normalized_sql(sql)),
+        "duration_ms": round((time.perf_counter() - started) * 1000),
+        "success": True,
+        "error_class": None,
+        "database_error_class": None,
+        "sqlstate": None,
+    }
+
+
+async def _run(args: argparse.Namespace) -> dict[str, Any]:
+    for identity in (args.trace_id, args.run_id, args.work_item_id):
+        if not _SAFE_ID.fullmatch(identity):
+            raise ValueError("identity_invalid")
+    if not re.fullmatch(r"[0-9a-f]{40}", args.source_revision):
+        raise ValueError("source_revision_invalid")
+    if args.migration_sha256 != _MIGRATION_SHA256:
+        raise ValueError("migration_sha256_mismatch")
+
+    started = time.perf_counter()
+    operations: list[dict[str, Any]] = []
+    engine = get_engine()
+    async with engine.connect() as raw_connection:
+        connection = await raw_connection.execution_options(
+            isolation_level="AUTOCOMMIT"
+        )
+        before = await _catalog(connection)
+        if args.mode == "apply" and before["invalid_or_unready_indexes"]:
+            after = before
+            terminal = "blocked_invalid_or_unready_index"
+        else:
+            if args.mode == "apply":
+                await connection.execute(text("SET lock_timeout = '15s'"))
+                await connection.execute(text("SET statement_timeout = '10min'"))
+                for name, sql in _EXTENSION_OPERATIONS:
+                    if name not in before["missing_extensions"]:
+                        continue
+                    operation = await _execute_fixed_operation(
+                        connection,
+                        operation_type="create_extension",
+                        name=name,
+                        sql=sql,
+                    )
+                    operations.append(operation)
+                    if not operation["success"]:
+                        break
+                if not operations or operations[-1]["success"]:
+                    for name, sql in _INDEX_OPERATIONS:
+                        if name not in before["missing_indexes"]:
+                            continue
+                        operation = await _execute_fixed_operation(
+                            connection,
+                            operation_type="create_index_concurrently",
+                            name=name,
+                            sql=sql,
+                        )
+                        operations.append(operation)
+                        if not operation["success"]:
+                            break
+            after = await _catalog(connection)
+            failed_operation = any(not operation["success"] for operation in operations)
+            if args.mode == "check":
+                terminal = "check_ready"
+            elif args.mode == "verify":
+                terminal = "verified_healthy" if after["passed"] else "verification_failed"
+            elif failed_operation:
+                terminal = "apply_failed"
+            else:
+                terminal = (
+                    "apply_verified_local"
+                    if after["passed"]
+                    else "verification_failed"
+                )
+
+    process_identity = ":".join(
+        [socket.gethostname(), str(os.getpid()), args.run_id, args.mode]
+    )
+    write_attempted = bool(operations)
+    writes_performed = any(operation["success"] for operation in operations)
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "command_id": _COMMAND_ID,
+        "canonical_asset": _CANONICAL_ASSET,
+        "mode": args.mode,
+        "trace_id": args.trace_id,
+        "run_id": args.run_id,
+        "work_item_id": args.work_item_id,
+        "source_revision": args.source_revision,
+        "migration_sha256": args.migration_sha256,
+        "operation_digest_sha256": _operation_digest(),
+        "pod_name": socket.gethostname(),
+        "process_identity_sha256": _sha256(process_identity),
+        "elapsed_ms": round((time.perf_counter() - started) * 1000),
+        "catalog_before": before,
+        "catalog_after": after,
+        "operations": operations,
+        "write_attempted": write_attempted,
+        "writes_performed": writes_performed,
+        "raw_sql_accepted": False,
+        "raw_sql_emitted": False,
+        "query_parameters_emitted": False,
+        "secret_value_read": False,
+        "arbitrary_command_allowed": False,
+        "cross_domain_fallback_allowed": False,
+        "verifier": {
+            "passed": args.mode == "check" or bool(after["passed"]),
+            "target_ready": bool(after["passed"]),
+        },
+        "terminal": terminal,
+    }
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--mode", choices=("check", "apply", "verify"), required=True)
+    parser.add_argument("--trace-id", required=True)
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--work-item-id", required=True)
+    parser.add_argument("--source-revision", required=True)
+    parser.add_argument("--migration-sha256", required=True)
+    return parser.parse_args()
+
+
+def main() -> int:
+    try:
+        application_output = io.StringIO()
+        with redirect_stdout(application_output), redirect_stderr(application_output):
+            receipt = asyncio.run(_run(_parse_args()))
+    except Exception as exc:
+        receipt = {
+            "schema_version": _SCHEMA_VERSION,
+            "command_id": _COMMAND_ID,
+            "terminal": "failed",
+            **_safe_error(exc),
+            "write_attempted": False,
+            "writes_performed": False,
+            "raw_sql_accepted": False,
+            "raw_sql_emitted": False,
+            "query_parameters_emitted": False,
+            "secret_value_read": False,
+            "arbitrary_command_allowed": False,
+            "cross_domain_fallback_allowed": False,
+        }
+    print(json.dumps(receipt, separators=(",", ":"), sort_keys=True))
+    return 0 if receipt.get("terminal") in {
+        "check_ready",
+        "apply_verified_local",
+        "verified_healthy",
+    } else 3
+
+
+if __name__ == "__main__":
+    sys.exit(main())

ops/agent99-truth-chain-index-manager.ps1

@@ -0,0 +1,315 @@
+[CmdletBinding()]
+param(
+  [ValidateSet("Check", "Apply")][string]$Mode = "Check",
+  [Parameter(Mandatory = $true)][ValidatePattern("^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$")][string]$TraceId,
+  [Parameter(Mandatory = $true)][ValidatePattern("^[0-9a-f-]{36}$")][string]$RunId,
+  [Parameter(Mandatory = $true)][ValidatePattern("^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$")][string]$WorkItemId,
+  [Parameter(Mandatory = $true)][ValidatePattern("^[0-9a-f]{40}$")][string]$SourceRevision,
+  [Parameter(Mandatory = $true)][ValidatePattern("^[0-9a-f]{64}$")][string]$MigrationSha256,
+  [Parameter(Mandatory = $true)][ValidatePattern("^[0-9a-f]{64}$")][string]$ExpectedOverlaySha256,
+  [string]$StagedOverlayPath = "C:\Windows\Temp\Agent99\truth-chain-index-executor-overlay.py",
+  [Parameter(Mandatory = $true)][string]$ResultPath,
+  [string]$AgentRoot = "C:\Wooo\Agent99"
+)
+
+Set-StrictMode -Version Latest
+$ErrorActionPreference = "Stop"
+$ProgressPreference = "SilentlyContinue"
+[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new($false)
+
+$CommandId = "truth_chain_structured_indexes_v1"
+$CanonicalAsset = "database:awoooi:prod/truth-chain-structured-indexes"
+$JumpHost = "192.168.0.110"
+$ControlPlaneHost = "192.168.0.120"
+$RemoteUser = "wooo"
+$Kubectl = "/usr/local/bin/kubectl"
+$Namespace = "awoooi-prod"
+$PodSelector = "app=awoooi-api,environment=prod,system=awoooi"
+$Container = "api"
+$IdentityFile = Join-Path $AgentRoot "keys\agent99_ed25519"
+$ApiPodPattern = "^awoooi-api-[a-z0-9][a-z0-9-]{0,126}$"
+$EvidenceRoot = [System.IO.Path]::GetFullPath((Join-Path $AgentRoot "evidence\"))
+$StagingRoot = [System.IO.Path]::GetFullPath("C:\Windows\Temp\Agent99\")
+$ResultFullPath = [System.IO.Path]::GetFullPath($ResultPath)
+$OverlayFullPath = [System.IO.Path]::GetFullPath($StagedOverlayPath)
+$startedAt = Get-Date
+$checkReceipt = $null
+$applyReceipt = $null
+$verifierReceipt = $null
+$executorPod = $null
+$verifierPod = $null
+$errorCode = $null
+$transportAnomalies = @()
+$transientCleanupPassed = $true
+$stagedOverlayDeleted = $false
+
+if (-not $ResultFullPath.StartsWith($EvidenceRoot, [System.StringComparison]::OrdinalIgnoreCase)) { throw "result_path_outside_evidence" }
+if (-not $OverlayFullPath.StartsWith($StagingRoot, [System.StringComparison]::OrdinalIgnoreCase)) { throw "overlay_path_outside_staging" }
+
+function Get-Sha256 {
+  param([string]$Path)
+  (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant()
+}
+
+function Write-AtomicJson {
+  param([string]$Path, [object]$Value)
+  [System.IO.Directory]::CreateDirectory((Split-Path -Parent $Path)) | Out-Null
+  $temporaryPath = "$Path.tmp"
+  [System.IO.File]::WriteAllText($temporaryPath, ($Value | ConvertTo-Json -Depth 20), [System.Text.UTF8Encoding]::new($false))
+  Move-Item -LiteralPath $temporaryPath -Destination $Path -Force
+}
+
+function Get-ReadyApiPods {
+  $remoteCommand = (
+    "ssh -o BatchMode=yes -o IdentitiesOnly=yes -o StrictHostKeyChecking=yes " +
+    "-o NumberOfPasswordPrompts=0 -o ConnectTimeout=8 -l $RemoteUser " +
+    "$ControlPlaneHost sudo -n $Kubectl get pods " +
+    "--namespace $Namespace --selector '$PodSelector' " +
+    "--field-selector 'status.phase=Running' --output name"
+  )
+  $arguments = @(
+    "-i", $IdentityFile,
+    "-o", "BatchMode=yes",
+    "-o", "IdentitiesOnly=yes",
+    "-o", "StrictHostKeyChecking=yes",
+    "-o", "NumberOfPasswordPrompts=0",
+    "-o", "ConnectTimeout=8",
+    "-o", "ServerAliveInterval=20",
+    "-o", "ServerAliveCountMax=3",
+    "-l", $RemoteUser,
+    $JumpHost,
+    $remoteCommand
+  )
+  $token = [Guid]::NewGuid().ToString("N")
+  $stdoutPath = Join-Path $env:TEMP "agent99-index-pods-$token.out"
+  $stderrPath = Join-Path $env:TEMP "agent99-index-pods-$token.err"
+  try {
+    $process = Start-Process -FilePath "ssh.exe" -ArgumentList $arguments -NoNewWindow `
+      -RedirectStandardOutput $stdoutPath -RedirectStandardError $stderrPath -PassThru
+    if (-not $process.WaitForExit(120 * 1000)) {
+      try { & taskkill.exe /PID $process.Id /T /F 2>&1 | Out-Null } catch {}
+      throw "pod_inventory_timeout"
+    }
+    $process.WaitForExit() | Out-Null
+    $process.Refresh()
+    $exitCode = if ($null -ne $process.ExitCode) { [int]$process.ExitCode } else { 255 }
+    $stdout = if (Test-Path -LiteralPath $stdoutPath -PathType Leaf) { [string](Get-Content -LiteralPath $stdoutPath -Raw) } else { "" }
+    $pods = @()
+    foreach ($line in @($stdout -split "`r?`n" | Where-Object { $_.Trim() })) {
+      if ($line.Trim() -notmatch "^pod/(.+)$") { throw "pod_inventory_shape_invalid" }
+      $podName = [string]$Matches[1]
+      if ($podName -notmatch $ApiPodPattern) { throw "pod_inventory_shape_invalid" }
+      $pods += $podName
+    }
+    $pods = @($pods | Sort-Object -Unique)
+    if ($pods.Count -lt 2) { throw "two_ready_api_pods_required" }
+    if ($exitCode -notin @(0, 255)) { throw "pod_inventory_transport_failed" }
+    if ($exitCode -eq 255) {
+      $script:transportAnomalies += [ordered]@{ stage = "pod_inventory"; exitCode = 255; acceptedBecause = "strict_pod_inventory_contract" }
+    }
+    return $pods
+  } finally {
+    Remove-Item -LiteralPath $stdoutPath, $stderrPath -Force -ErrorAction SilentlyContinue
+    if ((Test-Path -LiteralPath $stdoutPath) -or (Test-Path -LiteralPath $stderrPath)) { $script:transientCleanupPassed = $false }
+  }
+}
+
+function Test-StageReceipt {
+  param([object]$Receipt, [string]$ExpectedMode, [string]$ExpectedPod)
+  if (-not $Receipt) { return $false }
+  $allowedTerminal = switch ($ExpectedMode) {
+    "check" { [string]$Receipt.terminal -eq "check_ready" }
+    "apply" { [string]$Receipt.terminal -eq "apply_verified_local" }
+    "verify" { [string]$Receipt.terminal -eq "verified_healthy" }
+    default { $false }
+  }
+  return [bool](
+    [string]$Receipt.schema_version -eq "agent99_truth_chain_index_executor_v1" -and
+    [string]$Receipt.command_id -eq $CommandId -and
+    [string]$Receipt.canonical_asset -eq $CanonicalAsset -and
+    [string]$Receipt.mode -eq $ExpectedMode -and
+    [string]$Receipt.trace_id -eq $TraceId -and
+    [string]$Receipt.run_id -eq $RunId -and
+    [string]$Receipt.work_item_id -eq $WorkItemId -and
+    [string]$Receipt.source_revision -eq $SourceRevision -and
+    [string]$Receipt.migration_sha256 -eq $MigrationSha256 -and
+    [string]$Receipt.pod_name -eq $ExpectedPod -and
+    [string]$Receipt.process_identity_sha256 -match "^[0-9a-f]{64}$" -and
+    [string]$Receipt.operation_digest_sha256 -match "^[0-9a-f]{64}$" -and
+    $Receipt.raw_sql_accepted -eq $false -and
+    $Receipt.raw_sql_emitted -eq $false -and
+    $Receipt.query_parameters_emitted -eq $false -and
+    $Receipt.secret_value_read -eq $false -and
+    $Receipt.arbitrary_command_allowed -eq $false -and
+    $Receipt.cross_domain_fallback_allowed -eq $false -and
+    $Receipt.verifier.passed -eq $true -and
+    $allowedTerminal
+  )
+}
+
+function Get-CatalogSignature {
+  param([object]$Catalog)
+  if (-not $Catalog) { return "" }
+  [ordered]@{
+    expectedExtensionCount = [int]$Catalog.expected_extension_count
+    extensionCount = [int]$Catalog.extension_count
+    missingExtensions = @($Catalog.missing_extensions)
+    expectedIndexCount = [int]$Catalog.expected_index_count
+    presentIndexCount = [int]$Catalog.present_index_count
+    validReadyIndexCount = [int]$Catalog.valid_ready_index_count
+    missingIndexes = @($Catalog.missing_indexes)
+    invalidOrUnreadyIndexes = @($Catalog.invalid_or_unready_indexes)
+    passed = [bool]$Catalog.passed
+  } | ConvertTo-Json -Depth 6 -Compress
+}
+
+function Invoke-FixedStage {
+  param([ValidateSet("check", "apply", "verify")][string]$Stage, [string]$PodName)
+  if ($PodName -notmatch $ApiPodPattern) { throw "target_pod_contract_failed" }
+  $remoteCommand = (
+    "ssh -o BatchMode=yes -o IdentitiesOnly=yes -o StrictHostKeyChecking=yes " +
+    "-o NumberOfPasswordPrompts=0 -o ConnectTimeout=8 -l $RemoteUser " +
+    "$ControlPlaneHost sudo -n $Kubectl exec --stdin " +
+    "--namespace $Namespace --container $Container $PodName -- " +
+    "python -B - --mode $Stage --trace-id $TraceId --run-id $RunId " +
+    "--work-item-id $WorkItemId --source-revision $SourceRevision " +
+    "--migration-sha256 $MigrationSha256"
+  )
+  $arguments = @(
+    "-i", $IdentityFile,
+    "-o", "BatchMode=yes",
+    "-o", "IdentitiesOnly=yes",
+    "-o", "StrictHostKeyChecking=yes",
+    "-o", "NumberOfPasswordPrompts=0",
+    "-o", "ConnectTimeout=8",
+    "-o", "ServerAliveInterval=20",
+    "-o", "ServerAliveCountMax=3",
+    "-l", $RemoteUser,
+    $JumpHost,
+    $remoteCommand
+  )
+  $token = [Guid]::NewGuid().ToString("N")
+  $stdoutPath = Join-Path $env:TEMP "agent99-index-$Stage-$token.out"
+  $stderrPath = Join-Path $env:TEMP "agent99-index-$Stage-$token.err"
+  try {
+    $process = Start-Process -FilePath "ssh.exe" -ArgumentList $arguments -NoNewWindow `
+      -RedirectStandardInput $OverlayFullPath -RedirectStandardOutput $stdoutPath `
+      -RedirectStandardError $stderrPath -PassThru
+    $timeoutSeconds = if ($Stage -eq "apply") { 1800 } else { 180 }
+    if (-not $process.WaitForExit($timeoutSeconds * 1000)) {
+      try { & taskkill.exe /PID $process.Id /T /F 2>&1 | Out-Null } catch {}
+      throw "stage_timeout:$Stage"
+    }
+    $process.WaitForExit() | Out-Null
+    $process.Refresh()
+    $exitCode = if ($null -ne $process.ExitCode) { [int]$process.ExitCode } else { 255 }
+    try {
+      $receipt = Get-Content -LiteralPath $stdoutPath -Raw | ConvertFrom-Json
+    } catch {
+      throw "stage_receipt_parse_failed:$Stage"
+    }
+    if (-not (Test-StageReceipt -Receipt $receipt -ExpectedMode $Stage -ExpectedPod $PodName)) { throw "stage_receipt_contract_failed:$Stage" }
+    if ($exitCode -notin @(0, 255)) { throw "stage_transport_failed:$Stage" }
+    if ($exitCode -eq 255) {
+      $script:transportAnomalies += [ordered]@{ stage = $Stage; pod = $PodName; exitCode = 255; acceptedBecause = "strict_stage_receipt_contract" }
+    }
+    return $receipt
+  } finally {
+    Remove-Item -LiteralPath $stdoutPath, $stderrPath -Force -ErrorAction SilentlyContinue
+    if ((Test-Path -LiteralPath $stdoutPath) -or (Test-Path -LiteralPath $stderrPath)) { $script:transientCleanupPassed = $false }
+  }
+}
+
+try {
+  foreach ($path in @($OverlayFullPath, $IdentityFile)) {
+    if (-not (Test-Path -LiteralPath $path -PathType Leaf)) { throw "required_asset_missing:$([System.IO.Path]::GetFileName($path))" }
+  }
+  if ((Get-Sha256 $OverlayFullPath) -ne $ExpectedOverlaySha256) { throw "overlay_hash_mismatch" }
+  $pods = Get-ReadyApiPods
+  $executorPod = [string]$pods[0]
+  $verifierPod = [string]$pods[1]
+  $checkReceipt = Invoke-FixedStage -Stage "check" -PodName $executorPod
+  if ($Mode -eq "Apply" -and -not $checkReceipt.catalog_after.passed) {
+    $applyReceipt = Invoke-FixedStage -Stage "apply" -PodName $executorPod
+  }
+  $verifierStage = if ($Mode -eq "Apply") { "verify" } else { "check" }
+  $verifierReceipt = Invoke-FixedStage -Stage $verifierStage -PodName $verifierPod
+} catch {
+  $errorCode = [string]$_.Exception.Message
+} finally {
+  Remove-Item -LiteralPath $OverlayFullPath -Force -ErrorAction SilentlyContinue
+  $stagedOverlayDeleted = -not (Test-Path -LiteralPath $OverlayFullPath)
+}
+
+$independentVerifier = [bool](
+  $checkReceipt -and $verifierReceipt -and
+  [string]$checkReceipt.pod_name -ne [string]$verifierReceipt.pod_name -and
+  [string]$checkReceipt.process_identity_sha256 -ne [string]$verifierReceipt.process_identity_sha256
+)
+$writesExpected = [bool]($Mode -eq "Apply" -and $checkReceipt -and -not $checkReceipt.catalog_after.passed)
+$applyContractPassed = [bool](
+  -not $writesExpected -or
+  ($applyReceipt -and $applyReceipt.write_attempted -eq $true -and $applyReceipt.verifier.passed -eq $true)
+)
+$catalogAgreement = [bool](
+  $checkReceipt -and $verifierReceipt -and
+  (Get-CatalogSignature $checkReceipt.catalog_after) -eq
+    (Get-CatalogSignature $verifierReceipt.catalog_after)
+)
+$targetReady = [bool]($verifierReceipt -and $verifierReceipt.catalog_after.passed -eq $true)
+$modeContractPassed = [bool](
+  ($Mode -eq "Check" -and $catalogAgreement) -or
+  ($Mode -eq "Apply" -and $targetReady)
+)
+$passed = [bool](
+  -not $errorCode -and $checkReceipt -and $verifierReceipt -and
+  $independentVerifier -and $applyContractPassed -and $modeContractPassed -and
+  $transientCleanupPassed -and $stagedOverlayDeleted
+)
+$finishedAt = Get-Date
+$terminal = if ($passed) {
+  if ($Mode -eq "Check") { "verified_check_no_write" }
+  elseif ($writesExpected) { "runtime_verified_controlled_apply" }
+  else { "runtime_verified_healthy_no_write" }
+} else {
+  "degraded_with_safe_next_action"
+}
+$receipt = [ordered]@{
+  schemaVersion = "agent99_truth_chain_index_manager_v1"
+  traceId = $TraceId
+  runId = $RunId
+  workItemId = $WorkItemId
+  generatedAt = $finishedAt.ToString("o")
+  executor = "host:192.168.0.99"
+  target = $CanonicalAsset
+  sourceRevision = $SourceRevision
+  migrationSha256 = $MigrationSha256
+  dispatchPath = "windows99_digest_bound_stdin_to_host110_to_host120_to_two_api_pods"
+  risk = [ordered]@{ level = if ($Mode -eq "Apply") { "medium" } else { "low" }; rawSqlAccepted = $false; arbitraryCommandAllowed = $false; secretValueRead = $false }
+  check = [ordered]@{ overlaySha256 = $ExpectedOverlaySha256; receipt = $checkReceipt }
+  execution = [ordered]@{ mode = $Mode; performed = $writesExpected; receipt = $applyReceipt; errorCode = $errorCode; elapsedMs = [math]::Round(($finishedAt - $startedAt).TotalMilliseconds) }
+  verifier = [ordered]@{ receipt = $verifierReceipt; independent = $independentVerifier; catalogAgreement = $catalogAgreement; targetReady = $targetReady; passed = $passed }
+  cleanup = [ordered]@{ stagedOverlayDeleted = $stagedOverlayDeleted; transientOutputDeleted = $transientCleanupPassed }
+  rollback = [ordered]@{ automatic = $false; sourceAsset = "awooop_truth_chain_structured_reference_indexes_2026-07-17_down.sql" }
+  transportAnomalies = @($transportAnomalies)
+  telegramNotificationSent = $false
+  kmRagWritebackPerformed = $false
+  terminal = $terminal
+}
+Write-AtomicJson -Path $ResultFullPath -Value $receipt
+
+[ordered]@{
+  resultPath = $ResultFullPath
+  resultSha256 = Get-Sha256 $ResultFullPath
+  terminal = $terminal
+  writesExpected = $writesExpected
+  writeCount = if ($applyReceipt) { @($applyReceipt.operations).Count } else { 0 }
+  validReadyIndexCount = if ($verifierReceipt) { [int]$verifierReceipt.catalog_after.valid_ready_index_count } else { 0 }
+  targetReady = $targetReady
+  independentVerifier = $independentVerifier
+  stagedOverlayDeleted = $stagedOverlayDeleted
+  errorCode = $errorCode
+} | ConvertTo-Json -Depth 8 -Compress
+
+if (-not $passed) { exit 3 }
+exit 0

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
+