diff --git a/ops/agent99-truth-chain-index-executor-overlay.py b/ops/agent99-truth-chain-index-executor-overlay.py new file mode 100644 index 000000000..d18f47bd3 --- /dev/null +++ b/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()) diff --git a/ops/agent99-truth-chain-index-manager.ps1 b/ops/agent99-truth-chain-index-manager.ps1 new file mode 100644 index 000000000..38d1cf730 --- /dev/null +++ b/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 diff --git a/scripts/reboot-recovery/tests/test_agent99_truth_chain_index_executor_contract.py b/scripts/reboot-recovery/tests/test_agent99_truth_chain_index_executor_contract.py new file mode 100644 index 000000000..79c36e7bc --- /dev/null +++ b/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 +