wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(perf): bound truth-chain quality reads
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 4m29s
Details
CD Pipeline / build-and-deploy (push) Successful in 15m43s
Details
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
Details
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 18s
Details
CD Pipeline / post-deploy-checks (push) Successful in 11m37s
Details

Browse Source
This commit is contained in:
Your Name
2026-07-17 04:59:22 +08:00
parent 937134aaa1
commit 0382c120e9
6 changed files with 557 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17.sql Normal file
View File

@@ -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;

14
apps/api/migrations/awooop_truth_chain_structured_reference_indexes_2026-07-17_down.sql Normal file
View File

@@ -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;

494
apps/api/src/services/awooop_truth_chain_service.py
View File

@@ -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(

73
apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py
View File

@@ -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

86
apps/api/tests/test_awooop_truth_chain_service.py
View File

@@ -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