fix(perf): recover invalid telegram receipt index

agent99-contract-check.ps1

@@ -139,7 +139,31 @@ Add-Check "transport:dedicated_identity" ($control.Contains("sshIdentityFile") -
 Add-Check "transport:jump_first" ($control.Contains("preferJumpHost") -and $control.Contains("ssh_jump_fallback")) "preferred jump with bounded fallback is present"
 Add-Check "transport:stale_process_guard" ($control.Contains("Invoke-AgentStaleSshProcessGuard") -and $control.Contains("taskkill.exe /PID") -and $control.Contains("ssh_stale_process_guard")) "timed-out and stale SSH clients are bounded"
 Add-Check "transport:cross_process_serialization" ($control.Contains("Enter-AgentSshTransportLock") -and $control.Contains("[IO.FileShare]::None") -and $control.Contains("ssh_transport_lock_failed")) "cross-process SSH sessions are serialized"
-Add-Check "database:fixed_bounded_executor" ($dbBoundedExecutor.Contains('$CommandId = "telegram_receipt_index_apply_v1"') -and $dbBoundedExecutor.Contains('$CanonicalAsset = "public.awooop_outbound_message"') -and $dbBoundedExecutor.Contains('$Kubectl = "/usr/local/bin/kubectl"') -and $dbBoundedExecutor.Contains('domain = "database"') -and $dbBoundedExecutor.Contains('dispatchOnly = $true') -and $dbBoundedExecutor.Contains('Invoke-Agent99ReadyApiPodReadback') -and $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "check" $ExecutorPod') -and $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "apply" $ExecutorPod') -and $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "verify" $VerifierPod') -and $dbBoundedExecutor.Contains('sudo -n $Kubectl exec') -and -not $dbBoundedExecutor.Contains("Invoke-Expression") -and -not $dbBoundedExecutor.Contains("-EncodedCommand") -and -not $dbBoundedExecutor.Contains("/usr/local/sbin/awoooi-db-bounded-executor")) "Agent99 dispatches one fixed database command through the allowlisted host120 kubectl path and a second Ready API pod verifier"
+$dbExecutorIdentityContract = [bool](
+  $dbBoundedExecutor.Contains('$CommandId = "telegram_receipt_index_v2_apply_v1"') -and
+  $dbBoundedExecutor.Contains('$CanonicalAsset = "public.awooop_outbound_message"') -and
+  $dbBoundedExecutor.Contains('$NormalizedReindexSqlSha256 = "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7"')
+)
+$dbExecutorRecoveryContract = [bool](
+  $dbBoundedExecutor.Contains('awoooi_db_bounded_executor_receipt_v2') -and
+  $dbBoundedExecutor.Contains('--normalized-reindex-sql-sha256 $NormalizedReindexSqlSha256') -and
+  $dbBoundedExecutor.Contains('"reindex_index_concurrently"') -and
+  $dbBoundedExecutor.Contains('Receipt.database_sqlstate')
+)
+$dbExecutorTransportContract = [bool](
+  $dbBoundedExecutor.Contains('$Kubectl = "/usr/local/bin/kubectl"') -and
+  $dbBoundedExecutor.Contains('domain = "database"') -and
+  $dbBoundedExecutor.Contains('dispatchOnly = $true') -and
+  $dbBoundedExecutor.Contains('Invoke-Agent99ReadyApiPodReadback') -and
+  $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "check" $ExecutorPod') -and
+  $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "apply" $ExecutorPod') -and
+  $dbBoundedExecutor.Contains('Invoke-Agent99FixedDbStage "verify" $VerifierPod') -and
+  $dbBoundedExecutor.Contains('sudo -n $Kubectl exec') -and
+  -not $dbBoundedExecutor.Contains("Invoke-Expression") -and
+  -not $dbBoundedExecutor.Contains("-EncodedCommand") -and
+  -not $dbBoundedExecutor.Contains("/usr/local/sbin/awoooi-db-bounded-executor")
+)
+Add-Check "database:fixed_bounded_executor" ($dbExecutorIdentityContract -and $dbExecutorRecoveryContract -and $dbExecutorTransportContract) "Agent99 dispatches one digest-bound create/reindex command through the allowlisted host120 kubectl path, validates safe receipts, and requires a second Ready API pod verifier"
 Add-Check "recovery:auto_trigger" ($control.Contains("Get-AgentBootState") -and $control.Contains("Start-AgentRecoveryFromObservation") -and $control.Contains("recovery_already_queued")) "boot and host-down recovery use a single-flight queue"
 Add-Check "recovery:durable_boot_event" ($control.Contains("pendingRecovery") -and $control.Contains("host_reboot_recovery_pending") -and $control.Contains("Update-AgentBootRecoveryState") -and $control.Contains("recovered_late")) "boot event remains pending until verified recovery terminal"
 Add-Check "recovery:terminal_evidence_budget" ($taskRegistration.Contains('$recoverySettings') -and $taskRegistration.Contains('New-TimeSpan -Minutes 20')) "startup recovery outlives the ten-minute SLO long enough to write terminal evidence"

agent99-db-bounded-executor.ps1

@@ -14,10 +14,11 @@ param(
 
 $ErrorActionPreference = "Stop"
 
-$CommandId = "telegram_receipt_index_apply_v1"
+$CommandId = "telegram_receipt_index_v2_apply_v1"
 $CanonicalAsset = "public.awooop_outbound_message"
-$MigrationSha256 = "50aa05dea1718475051bdf4d45b1361291ec2335512b4d5c7d0a7eb77f935f5e"
+$MigrationSha256 = "ca36ef38097de2f4a88e950e136ae175fbe1c292ad4a2c30f3a9547c08be1dc8"
-$NormalizedSqlSha256 = "7e2f18b618c5c97bf7b428e1b224d4ef117c5551d5dc66b84838b14e546bf63a"
+$NormalizedSqlSha256 = "a10e88b55b66eaac7c09933acf33b13f3c9ada73f89c4a099e7eccaefaf1d0eb"
+$NormalizedReindexSqlSha256 = "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7"
 $JumpHost = "192.168.0.110"
 $ControlPlaneHost = "192.168.0.120"
 $RemoteUser = "wooo"
@@ -65,6 +66,7 @@ function Write-Agent99DbTerminal {
     sourceRevision = $SourceRevision
     migrationSha256 = $MigrationSha256
     normalizedSqlSha256 = $NormalizedSqlSha256
+    normalizedReindexSqlSha256 = $NormalizedReindexSqlSha256
     terminal = $Terminal
     errorCode = if ($ErrorCode) { $ErrorCode } else { $null }
     checkReceipt = $CheckReceipt
@@ -111,6 +113,16 @@ function Test-Agent99DbReceipt {
   } else {
     [bool]($Receipt.write_attempted -eq $false -and $Receipt.writes_performed -eq $false)
   }
+  $operation = [string]$Receipt.operation
+  $operationContract = if ($ExpectedStage -ne "apply") {
+    [bool]($operation -eq "none")
+  } elseif ($operation -eq "no_write") {
+    [bool]($Receipt.write_attempted -eq $false -and $Receipt.writes_performed -eq $false)
+  } elseif ($operation -in @("create_index_concurrently", "reindex_index_concurrently")) {
+    [bool]($Receipt.write_attempted -eq $true)
+  } else {
+    [bool]($operation -eq "none" -and $Receipt.write_attempted -eq $false)
+  }
   $catalogContract = [bool](
     ($terminal -ne "check_ready" -or $Receipt.catalog.apply_ready -eq $true) -and
     ($terminal -notin @("apply_verified_local", "verified_healthy") -or (
@@ -121,8 +133,9 @@ function Test-Agent99DbReceipt {
   return [bool](
     $terminalAllowed -and
     $writeContract -and
+    $operationContract -and
     $catalogContract -and
-    [string]$Receipt.schema_version -eq "awoooi_db_bounded_executor_receipt_v1" -and
+    [string]$Receipt.schema_version -eq "awoooi_db_bounded_executor_receipt_v2" -and
     [string]$Receipt.command_id -eq $CommandId -and
     [string]$Receipt.canonical_asset -eq $CanonicalAsset -and
     [string]$Receipt.domain -eq "database" -and
@@ -135,9 +148,11 @@ function Test-Agent99DbReceipt {
     [string]$Receipt.source_revision -eq $SourceRevision -and
     [string]$Receipt.migration_sha256 -eq $MigrationSha256 -and
     [string]$Receipt.normalized_sql_sha256 -eq $NormalizedSqlSha256 -and
+    [string]$Receipt.normalized_reindex_sql_sha256 -eq $NormalizedReindexSqlSha256 -and
     [string]$Receipt.pod_name -eq $ExpectedPod -and
     [string]$Receipt.pod_name -match $ApiPodPattern -and
     [string]$Receipt.process_identity_sha256 -match "^[0-9a-f]{64}$" -and
+    ($null -eq $Receipt.database_sqlstate -or [string]$Receipt.database_sqlstate -match "^[0-9A-Z]{5}$") -and
     $Receipt.raw_sql_emitted -eq $false -and
     $Receipt.secret_value_read -eq $false -and
     $Receipt.arbitrary_command_allowed -eq $false -and
@@ -229,6 +244,7 @@ function Invoke-Agent99FixedDbStage {
     "python -B -m src.services.db_bounded_executor " +
     "--mode $Stage --command-id $CommandId --canonical-asset $CanonicalAsset " +
     "--migration-sha256 $MigrationSha256 --normalized-sql-sha256 $NormalizedSqlSha256 " +
+    "--normalized-reindex-sql-sha256 $NormalizedReindexSqlSha256 " +
     "--trace-id $TraceId --run-id $RunId --work-item-id $WorkItemId --source-revision $SourceRevision"
   )
   $arguments = @(

apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql

@@ -0,0 +1,14 @@
+-- awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql
+-- Add the corrected descending hot-path index under a new name so recovery is
+-- additive and never drops or reindexes the drifted invalid predecessor.
+--
+-- CONCURRENTLY must run outside a transaction block.
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest_v2
+    ON awooop_outbound_message (project_id, queued_at DESC)
+    INCLUDE (send_status)
+    WHERE channel_type = 'telegram'
+      AND (
+        source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result'
+        OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent'
+      );

apps/api/src/services/db_bounded_executor.py

@@ -24,20 +24,20 @@ from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 from src.db.base import get_engine
 
-SCHEMA_VERSION = "awoooi_db_bounded_executor_receipt_v1"
+SCHEMA_VERSION = "awoooi_db_bounded_executor_receipt_v2"
-COMMAND_ID = "telegram_receipt_index_apply_v1"
+COMMAND_ID = "telegram_receipt_index_v2_apply_v1"
 CANONICAL_ASSET = "public.awooop_outbound_message"
 DOMAIN = "database"
 EXECUTOR = "db_bounded_executor"
 VERIFIER = "db_independent_verifier"
-INDEX_NAME = "idx_awooop_outbound_msg_telegram_receipt_latest"
+INDEX_NAME = "idx_awooop_outbound_msg_telegram_receipt_latest_v2"
 MIGRATION_FILE = (
-    "awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql"
+    "awooop_outbound_message_telegram_receipt_hot_path_index_v2_2026-07-17.sql"
 )
-MIGRATION_SHA256 = "50aa05dea1718475051bdf4d45b1361291ec2335512b4d5c7d0a7eb77f935f5e"
+MIGRATION_SHA256 = "ca36ef38097de2f4a88e950e136ae175fbe1c292ad4a2c30f3a9547c08be1dc8"
 
 CREATE_INDEX_SQL = """\
-CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest_v2
     ON awooop_outbound_message (project_id, queued_at DESC)
     INCLUDE (send_status)
     WHERE channel_type = 'telegram'
@@ -48,7 +48,15 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt
 """
 NORMALIZED_SQL = " ".join(CREATE_INDEX_SQL.split())
 NORMALIZED_SQL_SHA256 = (
-    "7e2f18b618c5c97bf7b428e1b224d4ef117c5551d5dc66b84838b14e546bf63a"
+    "a10e88b55b66eaac7c09933acf33b13f3c9ada73f89c4a099e7eccaefaf1d0eb"
+)
+REINDEX_INDEX_SQL = (
+    "REINDEX INDEX CONCURRENTLY "
+    "public.idx_awooop_outbound_msg_telegram_receipt_latest_v2;"
+)
+NORMALIZED_REINDEX_SQL = " ".join(REINDEX_INDEX_SQL.split())
+NORMALIZED_REINDEX_SQL_SHA256 = (
+    "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7"
 )
 
 _SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$")
@@ -56,17 +64,17 @@ _SOURCE_REVISION = re.compile(r"^[0-9a-f]{40}$")
 _API_POD = re.compile(r"^awoooi-api-[a-z0-9][a-z0-9-]{0,126}$")
 _LOCK_SQL = """
 SELECT pg_try_advisory_lock(
-    hashtextextended('awoooi/db/telegram_receipt_index_apply_v1', 0)
+    hashtextextended('awoooi/db/telegram_receipt_index_v2_apply_v1', 0)
 ) AS acquired
 """
 _UNLOCK_SQL = """
 SELECT pg_advisory_unlock(
-    hashtextextended('awoooi/db/telegram_receipt_index_apply_v1', 0)
+    hashtextextended('awoooi/db/telegram_receipt_index_v2_apply_v1', 0)
 ) AS released
 """
 _SESSION_GUARDS_SQL = (
     "SET search_path TO pg_catalog, public",
-    "SET lock_timeout TO '5s'",
+    "SET lock_timeout TO '30s'",
     "SET statement_timeout TO '10min'",
 )
 _CATALOG_SQL = f"""
@@ -94,6 +102,18 @@ WITH target AS (
         pg_get_indexdef(i.indexrelid, 1, true) AS key_1,
         pg_get_indexdef(i.indexrelid, 2, true) AS key_2,
         pg_get_indexdef(i.indexrelid, 3, true) AS include_1,
+        pg_index_column_has_property(i.indexrelid, 1, 'asc')
+            AS key_1_ascending,
+        pg_index_column_has_property(i.indexrelid, 1, 'nulls_last')
+            AS key_1_nulls_last,
+        pg_index_column_has_property(i.indexrelid, 2, 'asc')
+            AS key_2_ascending,
+        pg_index_column_has_property(i.indexrelid, 2, 'desc')
+            AS key_2_descending,
+        pg_index_column_has_property(i.indexrelid, 2, 'nulls_first')
+            AS key_2_nulls_first,
+        pg_index_column_has_property(i.indexrelid, 2, 'nulls_last')
+            AS key_2_nulls_last,
         pg_get_expr(i.indpred, i.indrelid) AS predicate
     FROM target t
     JOIN pg_index i ON i.indrelid = t.table_oid
@@ -144,6 +164,18 @@ SELECT
     COALESCE((SELECT key_1 FROM index_catalog), '') AS key_1,
     COALESCE((SELECT key_2 FROM index_catalog), '') AS key_2,
     COALESCE((SELECT include_1 FROM index_catalog), '') AS include_1,
+    COALESCE((SELECT key_1_ascending FROM index_catalog), false)
+        AS key_1_ascending,
+    COALESCE((SELECT key_1_nulls_last FROM index_catalog), false)
+        AS key_1_nulls_last,
+    COALESCE((SELECT key_2_ascending FROM index_catalog), false)
+        AS key_2_ascending,
+    COALESCE((SELECT key_2_descending FROM index_catalog), false)
+        AS key_2_descending,
+    COALESCE((SELECT key_2_nulls_first FROM index_catalog), false)
+        AS key_2_nulls_first,
+    COALESCE((SELECT key_2_nulls_last FROM index_catalog), false)
+        AS key_2_nulls_last,
     COALESCE((SELECT predicate FROM index_catalog), '') AS predicate
 FROM target t
 """
@@ -177,6 +209,7 @@ def validate_execution_contract(
     canonical_asset: str,
     migration_sha256: str,
     normalized_sql_sha256: str,
+    normalized_reindex_sql_sha256: str,
     trace_id: str,
     run_id: str,
     work_item_id: str,
@@ -194,6 +227,10 @@ def validate_execution_contract(
             normalized_sql_sha256,
             NORMALIZED_SQL_SHA256,
         ),
+        "normalized_reindex_sql_sha256": (
+            normalized_reindex_sql_sha256,
+            NORMALIZED_REINDEX_SQL_SHA256,
+        ),
     }
     for field, (actual, expected) in exact.items():
         if actual != expected:
@@ -215,6 +252,11 @@ def validate_execution_contract(
         raise BoundedExecutorContractError("runtime_not_allowlisted_api_pod")
     if hashlib.sha256(NORMALIZED_SQL.encode()).hexdigest() != NORMALIZED_SQL_SHA256:
         raise BoundedExecutorContractError("embedded_sql_digest_mismatch")
+    if (
+        hashlib.sha256(NORMALIZED_REINDEX_SQL.encode()).hexdigest()
+        != NORMALIZED_REINDEX_SQL_SHA256
+    ):
+        raise BoundedExecutorContractError("embedded_reindex_sql_digest_mismatch")
 
 
 def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]:
@@ -226,21 +268,56 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]:
         _normalize_catalog_expression(str(row.get("predicate") or ""))
         == _EXPECTED_PREDICATE
     )
-    index_shape_exact = bool(
+    index_non_unique = row.get("index_non_unique") is True
+    key_count_exact = row.get("key_count_exact") is True
+    total_attribute_count_exact = row.get("total_attribute_count_exact") is True
+    access_method_exact = row.get("access_method_exact") is True
+    key_1_expression_exact = (
+        _normalize_catalog_expression(str(row.get("key_1") or "")) == "project_id"
+    )
+    key_1_order_exact = bool(
+        row.get("key_1_ascending") is True and row.get("key_1_nulls_last") is True
+    )
+    key_1_exact = key_1_expression_exact and key_1_order_exact
+    key_2_expression_exact = (
+        _normalize_catalog_expression(str(row.get("key_2") or "")) == "queued_at"
+    )
+    key_2_ascending = row.get("key_2_ascending") is True
+    key_2_descending = row.get("key_2_descending") is True
+    key_2_nulls_first = row.get("key_2_nulls_first") is True
+    key_2_nulls_last = row.get("key_2_nulls_last") is True
+    if not key_2_expression_exact:
+        key_2_variant = "unknown_expression"
+    elif key_2_descending and key_2_nulls_first:
+        key_2_variant = "descending_nulls_first"
+    elif key_2_descending and key_2_nulls_last:
+        key_2_variant = "descending_nulls_last"
+    elif key_2_ascending and key_2_nulls_first:
+        key_2_variant = "ascending_nulls_first"
+    elif key_2_ascending and key_2_nulls_last:
+        key_2_variant = "ascending_nulls_last"
+    else:
+        key_2_variant = "unknown_order"
+    key_2_order_exact = key_2_descending and key_2_nulls_first
+    key_2_exact = key_2_expression_exact and key_2_order_exact
+    include_1_exact = (
+        _normalize_catalog_expression(str(row.get("include_1") or "")) == "send_status"
+    )
+    index_definition_exact_except_validity = bool(
         index_exists
         and row.get("index_ready") is True
-        and row.get("index_valid") is True
+        and index_non_unique
-        and row.get("index_non_unique") is True
+        and key_count_exact
-        and row.get("key_count_exact") is True
+        and total_attribute_count_exact
-        and row.get("total_attribute_count_exact") is True
+        and access_method_exact
-        and row.get("access_method_exact") is True
+        and key_1_exact
-        and _normalize_catalog_expression(str(row.get("key_1") or "")) == "project_id"
+        and key_2_exact
-        and _normalize_catalog_expression(str(row.get("key_2") or ""))
+        and include_1_exact
-        == "queued_atdesc"
-        and _normalize_catalog_expression(str(row.get("include_1") or ""))
-        == "send_status"
         and predicate_exact
     )
+    index_shape_exact = bool(
+        index_definition_exact_except_validity and row.get("index_valid") is True
+    )
     base_ready = bool(
         row.get("table_exists") is True
         and row.get("columns_complete") is True
@@ -250,11 +327,22 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]:
         and row.get("schema_usage") is True
         and row.get("table_owner_privilege_inherited") is True
     )
-    drift = bool(index_count > 1 or (index_exists and not index_shape_exact))
+    reindex_ready = bool(
+        base_ready
+        and index_definition_exact_except_validity
+        and row.get("index_valid") is not True
+    )
+    drift = bool(
+        index_count > 1 or (index_exists and not index_definition_exact_except_validity)
+    )
     apply_ready = bool(
         base_ready
         and not drift
-        and (index_shape_exact or row.get("schema_create") is True)
+        and (
+            index_shape_exact
+            or reindex_ready
+            or (not index_exists and row.get("schema_create") is True)
+        )
     )
     return {
         "table_exists": row.get("table_exists") is True,
@@ -272,8 +360,28 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]:
         "index_exists": index_exists,
         "index_ready": row.get("index_ready") is True,
         "index_valid": row.get("index_valid") is True,
+        "index_non_unique": index_non_unique,
+        "key_count_exact": key_count_exact,
+        "total_attribute_count_exact": total_attribute_count_exact,
+        "access_method_exact": access_method_exact,
+        "key_1_expression_exact": key_1_expression_exact,
+        "key_1_order_exact": key_1_order_exact,
+        "key_1_exact": key_1_exact,
+        "key_2_expression_exact": key_2_expression_exact,
+        "key_2_order_exact": key_2_order_exact,
+        "key_2_ascending": key_2_ascending,
+        "key_2_descending": key_2_descending,
+        "key_2_nulls_first": key_2_nulls_first,
+        "key_2_nulls_last": key_2_nulls_last,
+        "key_2_exact": key_2_exact,
+        "key_2_variant": key_2_variant,
+        "include_1_exact": include_1_exact,
+        "index_definition_exact_except_validity": (
+            index_definition_exact_except_validity
+        ),
         "index_shape_exact": index_shape_exact,
         "predicate_exact": predicate_exact,
+        "reindex_ready": reindex_ready,
         "catalog_drift": drift,
         "apply_ready": apply_ready,
     }
@@ -297,7 +405,9 @@ def _receipt(
     catalog: Mapping[str, Any],
     writes_performed: bool | None,
     write_attempted: bool = False,
+    operation: str = "none",
     error_code: str | None = None,
+    database_sqlstate: str | None = None,
 ) -> dict[str, Any]:
     role = {
         "check": "precheck_executor",
@@ -325,19 +435,31 @@ def _receipt(
         "migration_file": MIGRATION_FILE,
         "migration_sha256": MIGRATION_SHA256,
         "normalized_sql_sha256": NORMALIZED_SQL_SHA256,
+        "normalized_reindex_sql_sha256": NORMALIZED_REINDEX_SQL_SHA256,
         "terminal": terminal,
         "catalog": dict(catalog),
         "writes_performed": writes_performed,
         "write_attempted": write_attempted,
+        "operation": operation,
         "raw_sql_emitted": False,
         "secret_value_read": False,
         "arbitrary_command_allowed": False,
         "cross_domain_fallback_allowed": False,
-        "rollback": "no_write_or_create_index_concurrently_only",
+        "rollback": "no_write_or_fixed_concurrent_index_operation_only",
         "error_code": error_code,
+        "database_sqlstate": database_sqlstate,
     }
 
 
+def _safe_database_sqlstate(exc: SQLAlchemyError) -> str | None:
+    """Return only a five-character SQLSTATE, never backend error text."""
+
+    original = getattr(exc, "orig", None)
+    value = getattr(original, "sqlstate", None) or getattr(original, "pgcode", None)
+    candidate = str(value or "").upper()
+    return candidate if re.fullmatch(r"[0-9A-Z]{5}", candidate) else None
+
+
 async def execute_bounded_db_command(
     *,
     mode: str,
@@ -345,6 +467,7 @@ async def execute_bounded_db_command(
     canonical_asset: str,
     migration_sha256: str,
     normalized_sql_sha256: str,
+    normalized_reindex_sql_sha256: str,
     trace_id: str,
     run_id: str,
     work_item_id: str,
@@ -361,6 +484,7 @@ async def execute_bounded_db_command(
         canonical_asset=canonical_asset,
         migration_sha256=migration_sha256,
         normalized_sql_sha256=normalized_sql_sha256,
+        normalized_reindex_sql_sha256=normalized_reindex_sql_sha256,
         trace_id=trace_id,
         run_id=run_id,
         work_item_id=work_item_id,
@@ -371,6 +495,7 @@ async def execute_bounded_db_command(
     db_engine = engine
     writes_performed = False
     write_attempted = False
+    operation = "none"
     lock_acquired = False
     try:
         if db_engine is None:
@@ -453,7 +578,15 @@ async def execute_bounded_db_command(
                         writes_performed=False,
                         error_code="locked_precheck_failed",
                     )
-                if not locked["index_shape_exact"]:
+                if locked["index_shape_exact"]:
+                    operation = "no_write"
+                elif locked["reindex_ready"]:
+                    operation = "reindex_index_concurrently"
+                    write_attempted = True
+                    await connection.execute(text(REINDEX_INDEX_SQL))
+                    writes_performed = True
+                else:
+                    operation = "create_index_concurrently"
                     write_attempted = True
                     await connection.execute(text(CREATE_INDEX_SQL))
                     writes_performed = True
@@ -472,6 +605,7 @@ async def execute_bounded_db_command(
                     catalog=after,
                     writes_performed=writes_performed,
                     write_attempted=write_attempted,
+                    operation=operation,
                     error_code=(
                         None if verified else "local_catalog_verification_failed"
                     ),
@@ -493,9 +627,10 @@ async def execute_bounded_db_command(
             catalog={},
             writes_performed=None if write_attempted else writes_performed,
             write_attempted=write_attempted,
+            operation=operation,
             error_code="database_timeout",
         )
-    except SQLAlchemyError:
+    except SQLAlchemyError as exc:
         return _receipt(
             mode=mode,
             trace_id=trace_id,
@@ -507,7 +642,9 @@ async def execute_bounded_db_command(
             catalog={},
             writes_performed=None if write_attempted else writes_performed,
             write_attempted=write_attempted,
+            operation=operation,
             error_code="database_operation_failed",
+            database_sqlstate=_safe_database_sqlstate(exc),
         )
     except Exception:
         return _receipt(
@@ -521,6 +658,7 @@ async def execute_bounded_db_command(
             catalog={},
             writes_performed=None if write_attempted else writes_performed,
             write_attempted=write_attempted,
+            operation=operation,
             error_code="executor_internal_error",
         )
 
@@ -532,6 +670,7 @@ def _parser() -> argparse.ArgumentParser:
     parser.add_argument("--canonical-asset", required=True)
     parser.add_argument("--migration-sha256", required=True)
     parser.add_argument("--normalized-sql-sha256", required=True)
+    parser.add_argument("--normalized-reindex-sql-sha256", required=True)
     parser.add_argument("--trace-id", required=True)
     parser.add_argument("--run-id", required=True)
     parser.add_argument("--work-item-id", required=True)
@@ -550,6 +689,7 @@ async def _async_main() -> int:
             canonical_asset=args.canonical_asset,
             migration_sha256=args.migration_sha256,
             normalized_sql_sha256=args.normalized_sql_sha256,
+            normalized_reindex_sql_sha256=args.normalized_reindex_sql_sha256,
             trace_id=args.trace_id,
             run_id=args.run_id,
             work_item_id=args.work_item_id,

apps/api/tests/test_db_bounded_executor.py

@@ -20,8 +20,11 @@ from src.services.db_bounded_executor import (
     CREATE_INDEX_SQL,
     MIGRATION_FILE,
     MIGRATION_SHA256,
+    NORMALIZED_REINDEX_SQL,
+    NORMALIZED_REINDEX_SQL_SHA256,
     NORMALIZED_SQL,
     NORMALIZED_SQL_SHA256,
+    REINDEX_INDEX_SQL,
     BoundedExecutorContractError,
     evaluate_catalog_snapshot,
     execute_bounded_db_command,
@@ -63,6 +66,12 @@ def _absent_catalog() -> dict:
         "key_count_exact": False,
         "total_attribute_count_exact": False,
         "access_method_exact": False,
+        "key_1_ascending": False,
+        "key_1_nulls_last": False,
+        "key_2_ascending": False,
+        "key_2_descending": False,
+        "key_2_nulls_first": False,
+        "key_2_nulls_last": False,
         "key_1": "",
         "key_2": "",
         "include_1": "",
@@ -80,8 +89,12 @@ def _exact_catalog() -> dict:
         "key_count_exact": True,
         "total_attribute_count_exact": True,
         "access_method_exact": True,
+        "key_1_ascending": True,
+        "key_1_nulls_last": True,
+        "key_2_descending": True,
+        "key_2_nulls_first": True,
         "key_1": "project_id",
-        "key_2": "queued_at DESC",
+        "key_2": "queued_at",
         "include_1": "send_status",
         "predicate": """
             ((channel_type = 'telegram'::text) AND
@@ -93,6 +106,10 @@ def _exact_catalog() -> dict:
     }
 
 
+def _invalid_exact_catalog() -> dict:
+    return {**_exact_catalog(), "index_valid": False}
+
+
 def _contract(mode: str = "check") -> dict:
     return {
         "mode": mode,
@@ -100,6 +117,7 @@ def _contract(mode: str = "check") -> dict:
         "canonical_asset": CANONICAL_ASSET,
         "migration_sha256": MIGRATION_SHA256,
         "normalized_sql_sha256": NORMALIZED_SQL_SHA256,
+        "normalized_reindex_sql_sha256": NORMALIZED_REINDEX_SQL_SHA256,
         **IDENTITY,
         "source_revision": SOURCE_REVISION,
         "runtime_source_revision": SOURCE_REVISION,
@@ -113,6 +131,10 @@ def test_fixed_migration_and_embedded_sql_digests_match() -> None:
     assert hashlib.sha256(migration).hexdigest() == MIGRATION_SHA256
     assert _normalized_migration_sql() == NORMALIZED_SQL
     assert hashlib.sha256(NORMALIZED_SQL.encode()).hexdigest() == NORMALIZED_SQL_SHA256
+    assert (
+        hashlib.sha256(NORMALIZED_REINDEX_SQL.encode()).hexdigest()
+        == NORMALIZED_REINDEX_SQL_SHA256
+    )
 
 
 def test_embedded_sql_is_one_fixed_transactionless_create_only() -> None:
@@ -133,6 +155,25 @@ def test_embedded_sql_is_one_fixed_transactionless_create_only() -> None:
         assert forbidden not in normalized
 
 
+def test_embedded_reindex_is_one_fixed_transactionless_recovery_only() -> None:
+    normalized = REINDEX_INDEX_SQL.upper()
+
+    assert normalized.count("REINDEX INDEX CONCURRENTLY") == 1
+    assert normalized.count(";") == 1
+    assert "PUBLIC.IDX_AWOOOP_OUTBOUND_MSG_TELEGRAM_RECEIPT_LATEST" in normalized
+    for forbidden in (
+        "DROP ",
+        "TRUNCATE ",
+        "DELETE ",
+        "UPDATE ",
+        "INSERT ",
+        "ALTER ",
+        "BEGIN",
+        "COMMIT",
+    ):
+        assert forbidden not in normalized
+
+
 def test_precheck_requires_immediately_effective_owner_and_plain_table() -> None:
     source = (ROOT / "src/services/db_bounded_executor.py").read_text()
 
@@ -152,6 +193,11 @@ def test_precheck_requires_immediately_effective_owner_and_plain_table() -> None
             "f" * 64,
             "normalized_sql_sha256_contract_mismatch",
         ),
+        (
+            "normalized_reindex_sql_sha256",
+            "e" * 64,
+            "normalized_reindex_sql_sha256_contract_mismatch",
+        ),
         ("mode", "rollback", "mode_not_allowlisted"),
         ("pod_name", "awoooi-worker-1", "runtime_not_allowlisted_api_pod"),
         (
@@ -176,7 +222,24 @@ def test_contract_rejects_any_scope_or_runtime_drift(
 def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None:
     absent = evaluate_catalog_snapshot(_absent_catalog())
     exact = evaluate_catalog_snapshot(_exact_catalog())
-    drift_row = {**_exact_catalog(), "key_2": "queued_at"}
+    invalid_exact = evaluate_catalog_snapshot(_invalid_exact_catalog())
+    wrong_null_order = evaluate_catalog_snapshot(
+        {
+            **_invalid_exact_catalog(),
+            "key_2_nulls_first": False,
+            "key_2_nulls_last": True,
+        }
+    )
+    ascending_order = evaluate_catalog_snapshot(
+        {
+            **_invalid_exact_catalog(),
+            "key_2_ascending": True,
+            "key_2_descending": False,
+            "key_2_nulls_first": False,
+            "key_2_nulls_last": True,
+        }
+    )
+    drift_row = {**_exact_catalog(), "key_2": "created_at"}
     drift = evaluate_catalog_snapshot(drift_row)
     non_owner = evaluate_catalog_snapshot(
         {**_absent_catalog(), "table_owner_privilege_inherited": False}
@@ -192,6 +255,17 @@ def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None:
     assert absent["index_exists"] is False
     assert exact["index_shape_exact"] is True
     assert exact["apply_ready"] is True
+    assert invalid_exact["index_shape_exact"] is False
+    assert invalid_exact["index_definition_exact_except_validity"] is True
+    assert invalid_exact["reindex_ready"] is True
+    assert invalid_exact["catalog_drift"] is False
+    assert invalid_exact["apply_ready"] is True
+    assert invalid_exact["key_2_variant"] == "descending_nulls_first"
+    assert wrong_null_order["key_2_exact"] is False
+    assert wrong_null_order["key_2_variant"] == "descending_nulls_last"
+    assert ascending_order["key_2_exact"] is False
+    assert ascending_order["key_2_variant"] == "ascending_nulls_last"
+    assert wrong_null_order["catalog_drift"] is True
     assert drift["catalog_drift"] is True
     assert drift["apply_ready"] is False
     assert non_owner["apply_ready"] is False
@@ -239,6 +313,8 @@ class _Connection:
             return _Result(scalar=True)
         if "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql:
             return _Result()
+        if "REINDEX INDEX CONCURRENTLY" in sql:
+            return _Result()
         raise AssertionError(f"unexpected SQL: {sql}")
 
 
@@ -284,6 +360,7 @@ async def test_apply_uses_autocommit_fixed_sql_and_same_session_lock_release() -
 
     assert receipt["terminal"] == "apply_verified_local"
     assert receipt["writes_performed"] is True
+    assert receipt["operation"] == "create_index_concurrently"
     assert receipt["domain"] == "database"
     assert receipt["cross_domain_fallback_allowed"] is False
     assert receipt["secret_value_read"] is False
@@ -291,7 +368,7 @@ async def test_apply_uses_autocommit_fixed_sql_and_same_session_lock_release() -
     assert engine.connect_count == 1
     assert connection.statements[:3] == [
         "SET search_path TO pg_catalog, public",
-        "SET lock_timeout TO '5s'",
+        "SET lock_timeout TO '30s'",
         "SET statement_timeout TO '10min'",
     ]
     assert sum("CREATE INDEX CONCURRENTLY" in sql for sql in connection.statements) == 1
@@ -314,9 +391,55 @@ async def test_apply_is_idempotent_when_exact_index_already_exists() -> None:
 
     assert receipt["terminal"] == "apply_verified_local"
     assert receipt["writes_performed"] is False
+    assert receipt["operation"] == "no_write"
     assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements)
 
 
+@pytest.mark.asyncio
+async def test_apply_recovers_only_exact_invalid_index_with_concurrent_reindex() -> (
+    None
+):
+    connection = _Connection(
+        [_invalid_exact_catalog(), _invalid_exact_catalog(), _exact_catalog()]
+    )
+    engine = _Engine(connection)
+
+    receipt = await execute_bounded_db_command(
+        **_contract("apply"),
+        engine=engine,
+    )
+
+    assert receipt["terminal"] == "apply_verified_local"
+    assert receipt["writes_performed"] is True
+    assert receipt["operation"] == "reindex_index_concurrently"
+    assert connection.statements.count(REINDEX_INDEX_SQL) == 1
+    assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements)
+
+
+@pytest.mark.asyncio
+async def test_apply_blocks_invalid_index_with_wrong_definition_without_write() -> None:
+    wrong = {
+        **_invalid_exact_catalog(),
+        "key_2_ascending": True,
+        "key_2_descending": False,
+        "key_2_nulls_first": False,
+        "key_2_nulls_last": True,
+    }
+    connection = _Connection([wrong])
+    engine = _Engine(connection)
+
+    receipt = await execute_bounded_db_command(
+        **_contract("apply"),
+        engine=engine,
+    )
+
+    assert receipt["terminal"] == "blocked"
+    assert receipt["writes_performed"] is False
+    assert receipt["operation"] == "none"
+    assert all("REINDEX INDEX" not in sql for sql in connection.statements)
+    assert all("CREATE INDEX" not in sql for sql in connection.statements)
+
+
 @pytest.mark.asyncio
 async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None:
     connection = _Connection([_exact_catalog()])
@@ -329,6 +452,7 @@ async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None:
 
     assert receipt["terminal"] == "verified_healthy"
     assert receipt["executor_role"] == "independent_catalog_verifier"
+    assert receipt["operation"] == "none"
     assert receipt["writes_performed"] is False
     assert all("advisory_lock" not in sql for sql in connection.statements)
 
@@ -351,3 +475,33 @@ async def test_failed_create_reports_unknown_write_state_without_backend_detail(
     assert receipt["error_code"] == "database_operation_failed"
     assert "sensitive backend detail" not in json.dumps(receipt)
     assert any("pg_advisory_unlock" in sql for sql in connection.statements)
+
+
+class _SafeSqlstateFailureConnection(_Connection):
+    async def execute(self, statement):
+        sql = str(statement)
+        if "REINDEX INDEX CONCURRENTLY" in sql:
+            self.statements.append(sql)
+            error = SQLAlchemyError("backend text must not escape")
+            error.orig = type("DriverError", (), {"sqlstate": "55P03"})()
+            raise error
+        return await super().execute(statement)
+
+
+@pytest.mark.asyncio
+async def test_failed_reindex_returns_safe_sqlstate_without_backend_detail() -> None:
+    connection = _SafeSqlstateFailureConnection(
+        [_invalid_exact_catalog(), _invalid_exact_catalog()]
+    )
+    engine = _Engine(connection)
+
+    receipt = await execute_bounded_db_command(
+        **_contract("apply"),
+        engine=engine,
+    )
+
+    assert receipt["terminal"] == "failed"
+    assert receipt["operation"] == "reindex_index_concurrently"
+    assert receipt["writes_performed"] is None
+    assert receipt["database_sqlstate"] == "55P03"
+    assert "backend text" not in json.dumps(receipt)

scripts/ops/tests/test_db_bounded_executor_runtime_entrypoint.py

@@ -30,8 +30,9 @@ def test_agent99_is_dispatch_only_and_apply_requires_independent_verifier() -> N
     source = AGENT99_ENTRYPOINT.read_text()
 
     for fixed in (
-        '$CommandId = "telegram_receipt_index_apply_v1"',
+        '$CommandId = "telegram_receipt_index_v2_apply_v1"',
         '$CanonicalAsset = "public.awooop_outbound_message"',
+        '$NormalizedReindexSqlSha256 = "1857625d45032fba6fea52ebfa6cd0961ffe170b8c09488c773c60a63c75b1b7"',
         '$JumpHost = "192.168.0.110"',
         '$ControlPlaneHost = "192.168.0.120"',
         '$Kubectl = "/usr/local/bin/kubectl"',
@@ -50,6 +51,11 @@ def test_agent99_is_dispatch_only_and_apply_requires_independent_verifier() -> N
     assert 'Invoke-Agent99FixedDbStage "check" $ExecutorPod' in source
     assert 'Invoke-Agent99FixedDbStage "apply" $ExecutorPod' in source
     assert 'Invoke-Agent99FixedDbStage "verify" $VerifierPod' in source
+    assert "--normalized-reindex-sql-sha256 $NormalizedReindexSqlSha256" in source
+    assert (
+        '$operation -in @("create_index_concurrently", "reindex_index_concurrently")'
+        in source
+    )
     apply_flow = source[source.index("# Apply is the only write path") :]
     assert apply_flow.index('Invoke-Agent99FixedDbStage "check"') < apply_flow.index(
         'Invoke-Agent99FixedDbStage "apply"'
@@ -112,6 +118,10 @@ def test_agent99_atomic_bundle_owns_the_dispatch_entrypoint() -> None:
     for source in (deploy, bootstrap, contract, transport, receiver):
         assert '"agent99-db-bounded-executor.ps1"' in source
     assert 'Add-Check "database:fixed_bounded_executor"' in contract
+    assert "$dbExecutorIdentityContract" in contract
+    assert "$dbExecutorRecoveryContract" in contract
+    assert "awoooi_db_bounded_executor_receipt_v2" in contract
+    assert "Receipt.database_sqlstate" in contract
     assert 'expectedRuntimeFileCount": 16' in transport
     assert "expectedRuntimeFileCount=16" in transport
     assert "expectedRuntimeFileCount -ne 16" in receiver