diff --git a/.gitea/workflows/run-migration.yml b/.gitea/workflows/run-migration.yml index cc978cf55..6123f88b2 100644 --- a/.gitea/workflows/run-migration.yml +++ b/.gitea/workflows/run-migration.yml @@ -6,7 +6,8 @@ # 1. 用 MIGRATION_DATABASE_URL (awoooi_migrator 限權帳號) 連 PG # 2. 只跑「新增」的 migration (比對已執行列表) # 3. 跑後寫 asset_discovery_run + automation_operation_log 記錄 -# 4. 失敗自動 rollback (single transaction + ON_ERROR_STOP) +# 4. 一般 migration 使用 single transaction；CONCURRENTLY migration +# 必須保持 transactionless，並由配對 rollback 與 post-verifier 收斂 # # 觸發: 受控手動觸發；apps/api/migrations/ 有變更時先做 source diff / dry-run / rollback verifier @@ -93,13 +94,21 @@ jobs: apply_migration() { local url="$1" local file="$2" - psql "$url" \ - -v ON_ERROR_STOP=1 \ - --single-transaction \ - -f "$file" + if grep -Eiq '^[[:space:]]*(CREATE([[:space:]]+UNIQUE)?|DROP)[[:space:]]+INDEX[[:space:]]+CONCURRENTLY' "$file"; then + if grep -Eiq '^[[:space:]]*(BEGIN|START[[:space:]]+TRANSACTION|COMMIT)([[:space:]]|;|$)' "$file"; then + echo "::error::CONCURRENTLY migration must not contain an explicit transaction: $file" + return 1 + fi + echo "migration_mode=transactionless_concurrently file=$file" + psql "$url" -v ON_ERROR_STOP=1 -f "$file" + return + fi + + echo "migration_mode=single_transaction file=$file" + psql "$url" -v ON_ERROR_STOP=1 --single-transaction -f "$file" } - # 套用每個新檔 (single transaction per file) + # 套用每個新檔；模式由 apply_migration 依 SQL 契約判定。 echo "${{ steps.diff.outputs.new_files }}" | while IFS= read -r file; do [ -z "$file" ] && continue echo "=== Applying: $file ===" diff --git a/apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17.sql b/apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17.sql new file mode 100644 index 000000000..2fbf0b978 --- /dev/null +++ b/apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17.sql @@ -0,0 +1,70 @@ +-- Bound truth-chain quality reads to indexed, structured incident references. +-- Run outside an explicit transaction because CONCURRENTLY requires it. + +CREATE EXTENSION IF NOT EXISTS pg_trgm; +CREATE EXTENSION IF NOT EXISTS btree_gin; + +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; + +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'; + +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'; + +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))); + +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))); + +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))); + +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))); + +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; + +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; + +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; + +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; + +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; diff --git a/apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17_down.sql b/apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17_down.sql new file mode 100644 index 000000000..71995ff5e --- /dev/null +++ b/apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17_down.sql @@ -0,0 +1,14 @@ +-- Roll back only indexes owned by the structured truth-chain migration. + +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_project_content_preview_trgm; +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_top_incident_recent; +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_source_incident_recent; +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_alert_incident_recent; +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_callback_incident_recent; +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_source_code_refs_gin; +DROP INDEX CONCURRENTLY IF EXISTS idx_outbound_source_incident_ids_gin; +DROP INDEX CONCURRENTLY IF EXISTS idx_aol_output_source_incidents_gin; +DROP INDEX CONCURRENTLY IF EXISTS idx_aol_input_source_incidents_gin; +DROP INDEX CONCURRENTLY IF EXISTS idx_aol_output_incident_created; +DROP INDEX CONCURRENTLY IF EXISTS idx_aol_input_incident_created; +DROP INDEX CONCURRENTLY IF EXISTS idx_aol_incident_text_created; diff --git a/apps/api/src/services/awooop_truth_chain_service.py b/apps/api/src/services/awooop_truth_chain_service.py index e2f6d7c12..aac080349 100644 --- a/apps/api/src/services/awooop_truth_chain_service.py +++ b/apps/api/src/services/awooop_truth_chain_service.py @@ -41,6 +41,27 @@ _JSON_TEXT_FIELDS = {"gate_result", "source_envelope"} _QUALITY_SUMMARY_CACHE_TTL_SECONDS = int( os.getenv("AWOOOP_QUALITY_SUMMARY_CACHE_TTL_SECONDS", "120") ) +_QUALITY_SUMMARY_STATEMENT_TIMEOUT_MS = max( + 1_000, + min( + int(os.getenv("AWOOOP_QUALITY_SUMMARY_STATEMENT_TIMEOUT_MS", "12000")), + 30_000, + ), +) +_QUALITY_SUMMARY_MAX_BATCH_ROWS = max( + _MAX_ROWS, + min( + int(os.getenv("AWOOOP_QUALITY_SUMMARY_MAX_BATCH_ROWS", "5000")), + 10_000, + ), +) +_TRUTH_CHAIN_BRANCH_ROWS_PER_INCIDENT = max( + 5, + min( + int(os.getenv("AWOOOP_TRUTH_CHAIN_BRANCH_ROWS_PER_INCIDENT", "20")), + _MAX_ROWS, + ), +) _QUALITY_SUMMARY_OBSERVATIONS: dict[str, dict[str, Any]] = {} @@ -235,6 +256,260 @@ async def _fetch_inbound_conversation_event_rows( )[:limit] +def _truth_chain_branch_limit(source_ids: list[str], limit: int) -> int: + return max( + 1, + min( + int(limit), + _QUALITY_SUMMARY_MAX_BATCH_ROWS, + len(source_ids) * _TRUTH_CHAIN_BRANCH_ROWS_PER_INCIDENT, + ), + ) + + +async def _fetch_automation_operation_rows_for_incidents( + db: Any, + *, + incident_ids: list[str], + limit: int, + cutoff: datetime | None = None, + allow_text_fallback: bool = False, +) -> list[dict[str, Any]]: + """Fetch operation rows through direct, indexable incident references.""" + source_ids = list(dict.fromkeys(str(value) for value in incident_ids if value)) + if not source_ids: + return [] + + columns = """ + op_id, + operation_type, + status, + incident_id, + run_id, + parent_op_id, + actor, + dry_run_result, + error, + duration_ms, + tags, + input ->> 'action' AS input_action, + input ->> 'executor' AS input_executor, + input ->> 'execution_backend' AS input_execution_backend, + input ->> 'catalog_id' AS input_catalog_id, + input ->> 'execution_mode' AS input_execution_mode, + input ->> 'approval_source' AS input_approval_source, + input ->> 'apply_enabled' AS input_apply_enabled, + input ->> 'apply_executed' AS input_apply_executed, + input ->> 'check_mode_executed' AS input_check_mode_executed, + input ->> 'returncode' AS input_returncode, + input ->> 'playbook_id' AS input_playbook_id, + input ->> 'playbook_path' AS input_playbook_path, + input ->> 'ansible_playbook_path' AS input_ansible_playbook_path, + input ->> 'check_mode' AS input_check_mode, + input ->> 'not_used_reason' AS input_not_used_reason, + output ->> 'action' AS output_action, + output ->> 'reason' AS output_reason, + output ->> 'executor' AS output_executor, + output ->> 'execution_backend' AS output_execution_backend, + output ->> 'catalog_id' AS output_catalog_id, + output ->> 'execution_mode' AS output_execution_mode, + output ->> 'approval_source' AS output_approval_source, + output ->> 'apply_enabled' AS output_apply_enabled, + output ->> 'apply_executed' AS output_apply_executed, + output ->> 'check_mode_executed' AS output_check_mode_executed, + output ->> 'returncode' AS output_returncode, + output ->> 'playbook_id' AS output_playbook_id, + output ->> 'playbook_path' AS output_playbook_path, + output ->> 'ansible_playbook_path' AS output_ansible_playbook_path, + output ->> 'check_mode' AS output_check_mode, + output ->> 'not_used_reason' AS output_not_used_reason, + dry_run_result ->> 'returncode' AS dry_run_returncode, + dry_run_result ->> 'apply_executed' AS dry_run_apply_executed, + dry_run_result ->> 'check_mode_executed' + AS dry_run_check_mode_executed, + coalesce(input::text, '') AS input_text, + coalesce(output::text, '') AS output_text, + coalesce(array_to_string(tags, ','), '') AS tags_text, + created_at + """ + bounded_limit = _truth_chain_branch_limit(source_ids, limit) + rows_by_id: dict[str, dict[str, Any]] = {} + time_predicate = "created_at >= :cutoff" if cutoff is not None else "TRUE" + base_params: dict[str, Any] = { + "incident_ids": source_ids, + "incident_tags": source_ids + + [f"incident_id:{incident_id}" for incident_id in source_ids], + } + if cutoff is not None: + base_params["cutoff"] = cutoff + + async def fetch_branch( + where_sql: str, params: dict[str, Any] | None = None + ) -> None: + if len(rows_by_id) >= bounded_limit: + return + rows = await _fetch_all( + db, + f""" + SELECT + {columns} + FROM automation_operation_log + WHERE {time_predicate} + AND ({where_sql}) + ORDER BY created_at DESC + LIMIT :limit + """, + { + **base_params, + **(params or {}), + "limit": bounded_limit - len(rows_by_id), + }, + ) + for row in rows: + op_id = str(row.get("op_id") or "") + if op_id: + rows_by_id.setdefault(op_id, row) + + await fetch_branch("incident_id::text = ANY(CAST(:incident_ids AS text[]))") + await fetch_branch("input ->> 'incident_id' = ANY(CAST(:incident_ids AS text[]))") + await fetch_branch("output ->> 'incident_id' = ANY(CAST(:incident_ids AS text[]))") + await fetch_branch( + "coalesce(input #> '{source_refs,incident_ids}', '[]'::jsonb) " + "?| CAST(:incident_ids AS text[])" + ) + await fetch_branch( + "coalesce(output #> '{source_refs,incident_ids}', '[]'::jsonb) " + "?| CAST(:incident_ids AS text[])" + ) + await fetch_branch("tags && CAST(:incident_tags AS text[])") + + if allow_text_fallback and len(source_ids) == 1: + fallback_params = {"needle": f"%{source_ids[0]}%"} + await fetch_branch("coalesce(input::text, '') LIKE :needle", fallback_params) + await fetch_branch("coalesce(output::text, '') LIKE :needle", fallback_params) + + return sorted( + rows_by_id.values(), + key=lambda row: str(row.get("created_at") or ""), + reverse=True, + )[:bounded_limit] + + +async def _fetch_outbound_rows_for_incidents( + db: Any, + *, + project_id: str, + incident_ids: list[str], + limit: int, + cutoff: datetime | None = None, + allow_text_fallback: bool = False, +) -> list[dict[str, Any]]: + """Fetch outbound receipts without casting the full envelope to text.""" + source_ids = list(dict.fromkeys(str(value) for value in incident_ids if value)) + if not source_ids: + return [] + + columns = """ + message_id, + project_id, + run_id, + channel_type, + message_type, + content_hash, + content_preview, + content_redacted, + redaction_version, + source_envelope, + provider_message_id, + send_status, + queued_at, + sent_at, + triggered_by_state + """ + bounded_limit = _truth_chain_branch_limit(source_ids, limit) + rows_by_id: dict[str, dict[str, Any]] = {} + time_predicate = "queued_at >= :cutoff" if cutoff is not None else "TRUE" + valid_run_ids: list[str] = [] + for source_id in source_ids: + try: + valid_run_ids.append(str(UUID(source_id))) + except ValueError: + continue + base_params: dict[str, Any] = { + "project_id": project_id, + "incident_ids": source_ids, + } + if cutoff is not None: + base_params["cutoff"] = cutoff + + async def fetch_branch( + where_sql: str, params: dict[str, Any] | None = None + ) -> None: + if len(rows_by_id) >= bounded_limit: + return + rows = await _fetch_all( + db, + f""" + SELECT + {columns} + FROM awooop_outbound_message + WHERE project_id = :project_id + AND {time_predicate} + AND ({where_sql}) + ORDER BY queued_at DESC + LIMIT :limit + """, + { + **base_params, + **(params or {}), + "limit": bounded_limit - len(rows_by_id), + }, + ) + for row in rows: + message_id = str(row.get("message_id") or "") + if message_id: + rows_by_id.setdefault(message_id, row) + + if valid_run_ids: + await fetch_branch( + "run_id = ANY(CAST(:run_ids AS uuid[]))", + {"run_ids": valid_run_ids}, + ) + for source_ref_path in ( + "{source_refs,incident_ids}", + "{source_refs,code_refs}", + ): + await fetch_branch( + f"coalesce(source_envelope #> '{source_ref_path}', '[]'::jsonb) " + "?| CAST(:incident_ids AS text[])" + ) + for direct_path in ( + "{callback_reply,incident_id}", + "{alert_notification,incident_id}", + "{source_refs,incident_id}", + ): + await fetch_branch( + f"source_envelope #>> '{direct_path}' " + "= ANY(CAST(:incident_ids AS text[]))" + ) + await fetch_branch( + "source_envelope ->> 'incident_id' " + "= ANY(CAST(:incident_ids AS text[]))" + ) + + if allow_text_fallback and len(source_ids) == 1: + await fetch_branch( + "content_preview ILIKE :needle", + {"needle": f"%{source_ids[0]}%"}, + ) + + return sorted( + rows_by_id.values(), + key=lambda row: str(row.get("queued_at") or ""), + reverse=True, + )[:bounded_limit] + + def _source_type(source_id: str, incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> str: if incident is not None: return "incident" @@ -1698,65 +1973,11 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[ """, {"incident_id": incident_id, "limit": _MAX_ROWS}, ) - automation_ops = await _fetch_all( + automation_ops = await _fetch_automation_operation_rows_for_incidents( db, - """ - SELECT - op_id, - operation_type, - status, - incident_id, - run_id, - parent_op_id, - actor, - dry_run_result, - error, - duration_ms, - tags, - input ->> 'action' AS input_action, - input ->> 'executor' AS input_executor, - input ->> 'execution_backend' AS input_execution_backend, - input ->> 'catalog_id' AS input_catalog_id, - input ->> 'execution_mode' AS input_execution_mode, - input ->> 'approval_source' AS input_approval_source, - input ->> 'apply_enabled' AS input_apply_enabled, - input ->> 'apply_executed' AS input_apply_executed, - input ->> 'check_mode_executed' AS input_check_mode_executed, - input ->> 'returncode' AS input_returncode, - input ->> 'playbook_id' AS input_playbook_id, - input ->> 'playbook_path' AS input_playbook_path, - input ->> 'ansible_playbook_path' AS input_ansible_playbook_path, - input ->> 'check_mode' AS input_check_mode, - input ->> 'not_used_reason' AS input_not_used_reason, - output ->> 'action' AS output_action, - output ->> 'reason' AS output_reason, - output ->> 'executor' AS output_executor, - output ->> 'execution_backend' AS output_execution_backend, - output ->> 'catalog_id' AS output_catalog_id, - output ->> 'execution_mode' AS output_execution_mode, - output ->> 'approval_source' AS output_approval_source, - output ->> 'apply_enabled' AS output_apply_enabled, - output ->> 'apply_executed' AS output_apply_executed, - output ->> 'check_mode_executed' AS output_check_mode_executed, - output ->> 'returncode' AS output_returncode, - output ->> 'playbook_id' AS output_playbook_id, - output ->> 'playbook_path' AS output_playbook_path, - output ->> 'ansible_playbook_path' AS output_ansible_playbook_path, - output ->> 'check_mode' AS output_check_mode, - output ->> 'not_used_reason' AS output_not_used_reason, - dry_run_result ->> 'returncode' AS dry_run_returncode, - dry_run_result ->> 'apply_executed' AS dry_run_apply_executed, - dry_run_result ->> 'check_mode_executed' AS dry_run_check_mode_executed, - created_at - FROM automation_operation_log - WHERE incident_id::text = :incident_id - OR coalesce(input::text, '') LIKE :needle - OR coalesce(output::text, '') LIKE :needle - OR coalesce(array_to_string(tags, ','), '') LIKE :needle - ORDER BY created_at DESC - LIMIT :limit - """, - {"incident_id": incident_id, "needle": f"%{incident_id}%", "limit": _MAX_ROWS}, + incident_ids=[incident_id], + limit=_MAX_ROWS, + allow_text_fallback=True, ) auto_repair_executions = await _fetch_all( db, @@ -1871,42 +2092,12 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[ fingerprint_value=fingerprint_value, limit=_MAX_ROWS, ) - outbound_rows = await _fetch_all( + outbound_rows = await _fetch_outbound_rows_for_incidents( db, - """ - SELECT - message_id, - project_id, - run_id, - channel_type, - message_type, - content_hash, - content_preview, - content_redacted, - redaction_version, - source_envelope, - provider_message_id, - send_status, - queued_at, - sent_at, - triggered_by_state - FROM awooop_outbound_message - WHERE project_id = :project_id - AND ( - run_id::text = :source_id - OR content_preview ILIKE :needle - OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id - OR coalesce(source_envelope #> '{source_refs,code_refs}', '[]'::jsonb) ? :source_id - ) - ORDER BY queued_at DESC - LIMIT :limit - """, - { - "source_id": source_id, - "project_id": project_id, - "needle": f"%{source_id}%", - "limit": _MAX_ROWS, - }, + project_id=project_id, + incident_ids=[source_id], + limit=_MAX_ROWS, + allow_text_fallback=True, ) source_type = _source_type(source_id, incident, drift) @@ -2244,6 +2435,10 @@ async def fetch_automation_quality_summary( cutoff = datetime.now(UTC) - timedelta(hours=bounded_hours) async with get_db_context(normalized_project_id) as db: + await db.execute( + text("SELECT set_config('statement_timeout', :timeout, true)"), + {"timeout": f"{_QUALITY_SUMMARY_STATEMENT_TIMEOUT_MS}ms"}, + ) incidents = await _fetch_all( db, """ @@ -2313,7 +2508,11 @@ async def fetch_automation_quality_summary( if not incident_ids: records = [] else: - batch_params = {"incident_ids": incident_ids, "limit": _MAX_ROWS * len(incident_ids)} + batch_row_limit = min( + _MAX_ROWS * len(incident_ids), + _QUALITY_SUMMARY_MAX_BATCH_ROWS, + ) + batch_params = {"incident_ids": incident_ids, "limit": batch_row_limit} approvals = await _fetch_all( db, """ @@ -2415,72 +2614,11 @@ async def fetch_automation_quality_summary( """, batch_params, ) - automation_ops = await _fetch_all( + automation_ops = await _fetch_automation_operation_rows_for_incidents( db, - """ - SELECT - op_id, - operation_type, - status, - incident_id, - run_id, - parent_op_id, - actor, - dry_run_result, - error, - duration_ms, - tags, - input ->> 'action' AS input_action, - input ->> 'executor' AS input_executor, - input ->> 'execution_backend' AS input_execution_backend, - input ->> 'catalog_id' AS input_catalog_id, - input ->> 'execution_mode' AS input_execution_mode, - input ->> 'approval_source' AS input_approval_source, - input ->> 'apply_enabled' AS input_apply_enabled, - input ->> 'apply_executed' AS input_apply_executed, - input ->> 'check_mode_executed' AS input_check_mode_executed, - input ->> 'returncode' AS input_returncode, - input ->> 'playbook_id' AS input_playbook_id, - input ->> 'playbook_path' AS input_playbook_path, - input ->> 'ansible_playbook_path' AS input_ansible_playbook_path, - input ->> 'check_mode' AS input_check_mode, - input ->> 'not_used_reason' AS input_not_used_reason, - output ->> 'action' AS output_action, - output ->> 'reason' AS output_reason, - output ->> 'executor' AS output_executor, - output ->> 'execution_backend' AS output_execution_backend, - output ->> 'catalog_id' AS output_catalog_id, - output ->> 'execution_mode' AS output_execution_mode, - output ->> 'approval_source' AS output_approval_source, - output ->> 'apply_enabled' AS output_apply_enabled, - output ->> 'apply_executed' AS output_apply_executed, - output ->> 'check_mode_executed' AS output_check_mode_executed, - output ->> 'returncode' AS output_returncode, - output ->> 'playbook_id' AS output_playbook_id, - output ->> 'playbook_path' AS output_playbook_path, - output ->> 'ansible_playbook_path' AS output_ansible_playbook_path, - output ->> 'check_mode' AS output_check_mode, - output ->> 'not_used_reason' AS output_not_used_reason, - dry_run_result ->> 'returncode' AS dry_run_returncode, - dry_run_result ->> 'apply_executed' AS dry_run_apply_executed, - dry_run_result ->> 'check_mode_executed' AS dry_run_check_mode_executed, - coalesce(input::text, '') AS input_text, - coalesce(output::text, '') AS output_text, - coalesce(array_to_string(tags, ','), '') AS tags_text, - created_at - FROM automation_operation_log - WHERE incident_id::text = ANY(CAST(:incident_ids AS text[])) - OR EXISTS ( - SELECT 1 - FROM unnest(CAST(:incident_ids AS text[])) AS source_ids(incident_id) - WHERE coalesce(input::text, '') LIKE '%' || source_ids.incident_id || '%' - OR coalesce(output::text, '') LIKE '%' || source_ids.incident_id || '%' - OR coalesce(array_to_string(tags, ','), '') LIKE '%' || source_ids.incident_id || '%' - ) - ORDER BY created_at DESC - LIMIT :limit - """, - batch_params, + incident_ids=incident_ids, + limit=batch_row_limit, + cutoff=cutoff, ) auto_repair_executions = await _fetch_all( db, @@ -2550,40 +2688,12 @@ async def fetch_automation_quality_summary( """, {**batch_params, "project_id": normalized_project_id}, ) - outbound_rows = await _fetch_all( + outbound_rows = await _fetch_outbound_rows_for_incidents( db, - """ - SELECT - message_id, - project_id, - run_id, - channel_type, - message_type, - content_hash, - content_preview, - content_redacted, - redaction_version, - source_envelope, - provider_message_id, - send_status, - queued_at, - sent_at, - triggered_by_state - FROM awooop_outbound_message - WHERE project_id = :project_id - AND ( - run_id::text = ANY(CAST(:incident_ids AS text[])) - OR EXISTS ( - SELECT 1 - FROM unnest(CAST(:incident_ids AS text[])) AS source_ids(incident_id) - WHERE coalesce(content_preview, '') ILIKE '%' || source_ids.incident_id || '%' - OR coalesce(source_envelope::text, '') LIKE '%' || source_ids.incident_id || '%' - ) - ) - ORDER BY queued_at DESC - LIMIT :limit - """, - {**batch_params, "project_id": normalized_project_id}, + project_id=normalized_project_id, + incident_ids=incident_ids, + limit=batch_row_limit, + cutoff=cutoff, ) approvals_by_incident = _group_rows_by_incident_reference( diff --git a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py index 27d746919..5d3938807 100644 --- a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py +++ b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py @@ -3,12 +3,15 @@ from pathlib import Path from src.services.platform_operator_service import _provider_event_id_prefix_bounds ROOT = Path(__file__).resolve().parents[1] +REPO_ROOT = ROOT.parents[1] MIGRATION = "awooop_conversation_event_hot_path_indexes_2026-07-01.sql" ROLLBACK = "awooop_conversation_event_hot_path_indexes_2026-07-01_down.sql" CONTENT_PREVIEW_MIGRATION = "awooop_conversation_event_content_preview_trgm_2026-07-01.sql" CONTENT_PREVIEW_ROLLBACK = "awooop_conversation_event_content_preview_trgm_2026-07-01_down.sql" AUTOMATION_LOG_MIGRATION = "automation_operation_log_truth_chain_hot_path_indexes_2026-07-01.sql" AUTOMATION_LOG_ROLLBACK = "automation_operation_log_truth_chain_hot_path_indexes_2026-07-01_down.sql" +STRUCTURED_REFERENCE_MIGRATION = "awooop_truth_chain_structured_reference_indexes_2026-07-17.sql" +STRUCTURED_REFERENCE_ROLLBACK = "awooop_truth_chain_structured_reference_indexes_2026-07-17_down.sql" def _read(path: str) -> str: @@ -39,6 +42,14 @@ def _automation_log_rollback_sql() -> str: return _read(f"migrations/{AUTOMATION_LOG_ROLLBACK}") +def _structured_reference_migration_sql() -> str: + return _read(f"migrations/{STRUCTURED_REFERENCE_MIGRATION}") + + +def _structured_reference_rollback_sql() -> str: + return _read(f"migrations/{STRUCTURED_REFERENCE_ROLLBACK}") + + def test_hot_path_indexes_are_concurrent_and_transactionless() -> None: sql = _migration_sql() @@ -173,7 +184,7 @@ def test_automation_operation_log_indexes_bound_truth_chain_lookups() -> None: assert "incident_id::text" in truth_chain_source assert "coalesce(input::text, '') LIKE" in truth_chain_source assert "coalesce(output::text, '') LIKE" in truth_chain_source - assert "coalesce(array_to_string(tags, ','), '') LIKE" in truth_chain_source + assert "tags && CAST(:incident_tags AS text[])" in truth_chain_source for index_name in ( "idx_automation_operation_log_output_text_trgm", @@ -182,3 +193,63 @@ def test_automation_operation_log_indexes_bound_truth_chain_lookups() -> None: ): assert f"DROP INDEX CONCURRENTLY IF EXISTS {index_name};" in rollback assert "DROP EXTENSION" not in rollback + + +def test_structured_truth_chain_indexes_match_split_reference_branches() -> None: + sql = _structured_reference_migration_sql() + source = _read("src/services/awooop_truth_chain_service.py") + + assert "BEGIN;" not in sql + assert "COMMIT;" not in sql + assert sql.count("CREATE INDEX CONCURRENTLY IF NOT EXISTS") == 12 + assert "idx_aol_incident_text_created" in sql + assert "idx_aol_input_incident_created" in sql + assert "idx_aol_output_incident_created" in sql + assert "idx_aol_input_source_incidents_gin" in sql + assert "idx_aol_output_source_incidents_gin" in sql + assert "idx_outbound_source_incident_ids_gin" in sql + assert "idx_outbound_source_code_refs_gin" in sql + assert "idx_outbound_callback_incident_recent" in sql + assert "idx_outbound_alert_incident_recent" in sql + assert "idx_outbound_source_incident_recent" in sql + assert "idx_outbound_top_incident_recent" in sql + assert "idx_outbound_project_content_preview_trgm" in sql + + assert "_fetch_automation_operation_rows_for_incidents" in source + assert "_fetch_outbound_rows_for_incidents" in source + assert "source_envelope::text" not in source + assert "FROM unnest(CAST(:incident_ids AS text[]))" not in source + + +def test_structured_truth_chain_rollback_only_drops_owned_indexes() -> None: + rollback = _structured_reference_rollback_sql() + + assert "BEGIN;" not in rollback + assert "COMMIT;" not in rollback + assert rollback.count("DROP INDEX CONCURRENTLY IF EXISTS") == 12 + assert "DROP EXTENSION" not in rollback + for index_name in ( + "idx_outbound_project_content_preview_trgm", + "idx_outbound_top_incident_recent", + "idx_outbound_source_incident_recent", + "idx_outbound_alert_incident_recent", + "idx_outbound_callback_incident_recent", + "idx_outbound_source_code_refs_gin", + "idx_outbound_source_incident_ids_gin", + "idx_aol_output_source_incidents_gin", + "idx_aol_input_source_incidents_gin", + "idx_aol_output_incident_created", + "idx_aol_input_incident_created", + "idx_aol_incident_text_created", + ): + assert f"DROP INDEX CONCURRENTLY IF EXISTS {index_name};" in rollback + + +def test_migration_workflow_keeps_concurrent_indexes_outside_transactions() -> None: + workflow = (REPO_ROOT / ".gitea/workflows/run-migration.yml").read_text() + + assert "migration_mode=transactionless_concurrently" in workflow + assert "CONCURRENTLY migration must not contain an explicit transaction" in workflow + assert "psql \"$url\" -v ON_ERROR_STOP=1 -f \"$file\"" in workflow + assert "migration_mode=single_transaction" in workflow + assert "--single-transaction -f \"$file\"" in workflow diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index dbda93b3b..a89e694e6 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -11,6 +11,7 @@ from unittest.mock import AsyncMock import pytest +import src.services.awooop_truth_chain_service as truth_chain_service from src.core.config import settings from src.jobs.awooop_ansible_candidate_backfill_job import ( enqueue_ai_decision_ansible_candidate, @@ -150,14 +151,90 @@ def test_quality_summary_includes_recent_ansible_operation_incidents() -> None: def test_quality_summary_uses_batched_truth_chain_inputs() -> None: source = inspect.getsource(fetch_automation_quality_summary) + outbound_helper = inspect.getsource( + truth_chain_service._fetch_outbound_rows_for_incidents + ) assert "fetch_truth_chain(" not in source assert "approval_records" in source assert "incident_evidence" in source - assert "awooop_outbound_message" in source + assert "_fetch_outbound_rows_for_incidents" in source + assert "awooop_outbound_message" in outbound_helper assert "_build_summary_quality_records" in source +def test_quality_summary_bounds_database_work_and_uses_structured_helpers() -> None: + source = inspect.getsource(fetch_automation_quality_summary) + + assert "_QUALITY_SUMMARY_STATEMENT_TIMEOUT_MS" in source + assert "_QUALITY_SUMMARY_MAX_BATCH_ROWS" in source + assert "_fetch_automation_operation_rows_for_incidents" in source + assert "_fetch_outbound_rows_for_incidents" in source + assert "FROM unnest(CAST(:incident_ids AS text[]))" not in source + assert "source_envelope::text" not in source + + +@pytest.mark.asyncio +async def test_batch_operation_lookup_splits_indexable_reference_branches( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[str, dict[str, object]]] = [] + + async def fake_fetch_all( + _db: object, sql: str, params: dict[str, object] + ) -> list[dict[str, object]]: + calls.append((sql, params)) + return [] + + monkeypatch.setattr(truth_chain_service, "_fetch_all", fake_fetch_all) + cutoff = datetime.now(UTC) - timedelta(hours=24) + rows = await truth_chain_service._fetch_automation_operation_rows_for_incidents( + object(), + incident_ids=["INC-20260717-AAAAAA", "INC-20260717-BBBBBB"], + limit=200, + cutoff=cutoff, + ) + + assert rows == [] + assert len(calls) == 6 + assert all("FROM automation_operation_log" in sql for sql, _ in calls) + assert all("created_at >= :cutoff" in sql for sql, _ in calls) + assert all(params["cutoff"] == cutoff for _, params in calls) + assert all("FROM unnest" not in sql for sql, _ in calls) + assert all("LIKE '%' ||" not in sql for sql, _ in calls) + + +@pytest.mark.asyncio +async def test_batch_outbound_lookup_never_casts_full_envelope_to_text( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[str, dict[str, object]]] = [] + + async def fake_fetch_all( + _db: object, sql: str, params: dict[str, object] + ) -> list[dict[str, object]]: + calls.append((sql, params)) + return [] + + monkeypatch.setattr(truth_chain_service, "_fetch_all", fake_fetch_all) + cutoff = datetime.now(UTC) - timedelta(hours=24) + rows = await truth_chain_service._fetch_outbound_rows_for_incidents( + object(), + project_id="awoooi", + incident_ids=["INC-20260717-AAAAAA"], + limit=100, + cutoff=cutoff, + ) + + assert rows == [] + assert len(calls) == 6 + assert all("FROM awooop_outbound_message" in sql for sql, _ in calls) + assert all("queued_at >= :cutoff" in sql for sql, _ in calls) + assert all("source_envelope::text" not in sql for sql, _ in calls) + assert all("FROM unnest" not in sql for sql, _ in calls) + assert all("content_preview ILIKE" not in sql for sql, _ in calls) + + def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() -> None: decision_source = inspect.getsource(record_ansible_decision_audit) terminal_source = verified_ansible_candidate_terminal_sql("candidate") @@ -195,8 +272,13 @@ def test_ansible_transport_cooldown_uses_asyncpg_safe_interval_parameter() -> No def test_fetch_truth_chain_returns_inbound_redacted_envelope_fields() -> None: source = inspect.getsource(fetch_truth_chain) helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows) + outbound_helper = inspect.getsource( + truth_chain_service._fetch_outbound_rows_for_incidents + ) - assert "content_redacted" in source + assert "_fetch_outbound_rows_for_incidents" in source + assert "content_redacted" in helper_source + assert "content_redacted" in outbound_helper assert "source_envelope" in helper_source assert "source_refs,event_ids" in helper_source assert "source_refs,incident_ids" in helper_source