feat(agent99): add digest-bound truth-chain index executor
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m54s
CD Pipeline / build-and-deploy (push) Successful in 15m28s
CD Pipeline / post-deploy-checks (push) Successful in 4m16s
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m54s
CD Pipeline / build-and-deploy (push) Successful in 15m28s
CD Pipeline / post-deploy-checks (push) Successful in 4m16s
This commit is contained in:
Your Name
425
ops/agent99-truth-chain-index-executor-overlay.py
Normal file
425
ops/agent99-truth-chain-index-executor-overlay.py
Normal file
@@ -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())
|
||||
315
ops/agent99-truth-chain-index-manager.ps1
Normal file
315
ops/agent99-truth-chain-index-manager.ps1
Normal file
@@ -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
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user