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

fix(agent): degrade runtime receipt query timeouts
Some checks failed
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 35s
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-07-01 20:39:43 +08:00
parent 17b6bf36ce
commit cf2822529f
2 changed files with 149 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -849,6 +849,7 @@ def _build_runtime_receipt_readback_recovery(
db_read_status: str,
log_integration_taxonomy: Mapping[str, Any],
error_type: str | None,
partial_query_failures: Iterable[Mapping[str, Any]] = (),
) -> dict[str, Any]:
"""Expose the next controlled action when live receipt/log evidence is absent."""
@@ -867,9 +868,22 @@ def _build_runtime_receipt_readback_recovery(
recent_classified_event_total = _int_value(
taxonomy_rollups.get("recent_classified_event_total")
)
db_unavailable = db_read_status != "ok"
db_unavailable = db_read_status in {"not_queried", "unavailable"}
partial_failures = [dict(item) for item in partial_query_failures]
db_partial = db_read_status == "partial"
no_live_log_events = classified_event_total <= 0
if db_unavailable:
if db_partial and not no_live_log_events:
status = "degraded_runtime_receipt_partial_query_failures"
safe_next_action_id = (
"repair_partial_runtime_receipt_queries_then_rerun_log_integration_taxonomy"
)
safe_next_action_stage = "runtime_receipt_partial_query_recovery"
safe_next_action = (
"Optimize or fallback the failed receipt queries while keeping successful "
"LOG source families visible to KM/RAG/MCP/PlayBook consumers."
)
blocker_fields = ["partial_query_failures"]
elif db_unavailable:
status = "blocked_runtime_receipt_db_readback_unavailable"
safe_next_action_id = (
"repair_runtime_receipt_db_readback_then_rerun_log_integration_taxonomy"
@@ -918,6 +932,7 @@ def _build_runtime_receipt_readback_recovery(
taxonomy_rollups.get("active_source_family_count")
),
"inactive_source_family_ids": inactive_source_family_ids,
"partial_query_failures": partial_failures,
"blocker_fields": blocker_fields,
"operation_boundaries": {
"readback_only": True,
@@ -2141,7 +2156,7 @@ def _build_work_item_progress(
"completed"
if runtime_truth_complete
else "blocked"
if db_read_status != "ok"
if db_read_status in {"not_queried", "unavailable"}
else "in_progress"
),
"exit_criteria": "production API reports db_read_status=ok and live executor receipts",
@@ -2149,7 +2164,9 @@ def _build_work_item_progress(
if runtime_truth_complete
else (
"runtime_receipt_db_readback_unavailable"
if db_read_status != "ok"
if db_read_status in {"not_queried", "unavailable"}
else "runtime_receipt_partial_query_failures"
if db_read_status == "partial"
else "live_log_classified_event_total_zero"
),
"db_read_status": db_read_status,
@@ -3043,6 +3060,7 @@ def build_runtime_receipt_readback_from_rows(
grouped_alert_event_count_rows: Iterable[Mapping[str, Any] | Any] = (),
log_controlled_writeback_consumer: Mapping[str, Any] | None = None,
error_type: str | None = None,
partial_query_failures: Iterable[Mapping[str, Any]] = (),
) -> dict[str, Any]:
"""Build the live executor receipt readback from already-fetched rows."""
@@ -3135,6 +3153,7 @@ def build_runtime_receipt_readback_from_rows(
db_read_status=db_read_status,
log_integration_taxonomy=log_integration_taxonomy,
error_type=error_type,
partial_query_failures=partial_query_failures,
)
agent_decision_wiring = _build_agent_decision_wiring(
operation_summary=operation_summary,
@@ -3198,6 +3217,7 @@ def build_runtime_receipt_readback_from_rows(
"project_id": project_id,
"lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)),
"db_read_status": db_read_status,
"partial_query_failures": [dict(item) for item in partial_query_failures],
"writes_on_read": False,
"ansible_operations": {
"counts": operation_summary,
@@ -3972,11 +3992,15 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
"limit": max(1, int(limit or 20)),
}
log_controlled_writeback_consumer: dict[str, Any] | None = None
partial_query_failures: list[dict[str, str]] = []
try:
async with get_db_context(project_id) as db:
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
async def _set_statement_timeout() -> None:
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
async def _safe_aux_rows(
await _set_statement_timeout()
async def _safe_rows(
query_name: str,
sql: str,
fallback_sql: str | None = None,
@@ -3984,89 +4008,131 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
try:
return (await db.execute(text(sql), params)).mappings().all()
except Exception as exc: # pragma: no cover - depends on live schema drift
partial_query_failures.append({
"query_name": query_name,
"error_type": type(exc).__name__,
})
logger.warning(
"ai_agent_autonomous_runtime_trace_aux_read_failed",
"ai_agent_autonomous_runtime_trace_read_failed",
project_id=project_id,
query_name=query_name,
error_type=type(exc).__name__,
)
rollback = getattr(db, "rollback", None)
if callable(rollback):
try:
await rollback()
await _set_statement_timeout()
except Exception as reset_exc: # pragma: no cover - live DB state
logger.warning(
"ai_agent_autonomous_runtime_trace_reset_failed",
project_id=project_id,
query_name=query_name,
error_type=type(reset_exc).__name__,
)
if fallback_sql:
try:
return (await db.execute(text(fallback_sql), params)).mappings().all()
except Exception as fallback_exc: # pragma: no cover - live schema drift
partial_query_failures.append({
"query_name": f"{query_name}_fallback",
"error_type": type(fallback_exc).__name__,
})
logger.warning(
"ai_agent_autonomous_runtime_trace_aux_fallback_failed",
project_id=project_id,
query_name=query_name,
error_type=type(fallback_exc).__name__,
)
rollback = getattr(db, "rollback", None)
if callable(rollback):
try:
await rollback()
await _set_statement_timeout()
except Exception as reset_exc: # pragma: no cover
logger.warning(
"ai_agent_autonomous_runtime_trace_reset_failed",
project_id=project_id,
query_name=f"{query_name}_fallback",
error_type=type(reset_exc).__name__,
)
return []
operation_counts = (
await db.execute(text(_RUNTIME_OPERATION_COUNTS_SQL), params)
).mappings().all()
operation_latest = (
await db.execute(text(_RUNTIME_OPERATION_LATEST_SQL), params)
).mappings().all()
auto_repair_counts = (
await db.execute(text(_RUNTIME_AUTO_REPAIR_COUNTS_SQL), params)
).mappings().all()
auto_repair_latest = (
await db.execute(text(_RUNTIME_AUTO_REPAIR_LATEST_SQL), params)
).mappings().all()
verifier_counts = (
await db.execute(text(_RUNTIME_VERIFIER_COUNTS_SQL), params)
).mappings().all()
verifier_latest = (
await db.execute(text(_RUNTIME_VERIFIER_LATEST_SQL), params)
).mappings().all()
km_counts = (
await db.execute(text(_RUNTIME_KM_COUNTS_SQL), params)
).mappings().all()
km_latest = (
await db.execute(text(_RUNTIME_KM_LATEST_SQL), params)
).mappings().all()
telegram_counts = (
await db.execute(text(_RUNTIME_TELEGRAM_COUNTS_SQL), params)
).mappings().all()
telegram_latest = (
await db.execute(text(_RUNTIME_TELEGRAM_LATEST_SQL), params)
).mappings().all()
mcp_gateway_counts = await _safe_aux_rows(
operation_counts = await _safe_rows(
"operation_counts",
_RUNTIME_OPERATION_COUNTS_SQL,
)
operation_latest = await _safe_rows(
"operation_latest",
_RUNTIME_OPERATION_LATEST_SQL,
)
auto_repair_counts = await _safe_rows(
"auto_repair_counts",
_RUNTIME_AUTO_REPAIR_COUNTS_SQL,
)
auto_repair_latest = await _safe_rows(
"auto_repair_latest",
_RUNTIME_AUTO_REPAIR_LATEST_SQL,
)
verifier_counts = await _safe_rows(
"verifier_counts",
_RUNTIME_VERIFIER_COUNTS_SQL,
)
verifier_latest = await _safe_rows(
"verifier_latest",
_RUNTIME_VERIFIER_LATEST_SQL,
)
km_counts = await _safe_rows(
"km_counts",
_RUNTIME_KM_COUNTS_SQL,
)
km_latest = await _safe_rows(
"km_latest",
_RUNTIME_KM_LATEST_SQL,
)
telegram_counts = await _safe_rows(
"telegram_counts",
_RUNTIME_TELEGRAM_COUNTS_SQL,
)
telegram_latest = await _safe_rows(
"telegram_latest",
_RUNTIME_TELEGRAM_LATEST_SQL,
)
mcp_gateway_counts = await _safe_rows(
"mcp_gateway_counts",
_RUNTIME_MCP_GATEWAY_COUNTS_SQL,
)
legacy_mcp_counts = await _safe_aux_rows(
legacy_mcp_counts = await _safe_rows(
"legacy_mcp_counts",
_RUNTIME_LEGACY_MCP_COUNTS_SQL,
)
service_log_counts = await _safe_aux_rows(
service_log_counts = await _safe_rows(
"service_log_counts",
_RUNTIME_SERVICE_LOG_COUNTS_SQL,
)
executor_log_counts = await _safe_aux_rows(
executor_log_counts = await _safe_rows(
"executor_log_counts",
_RUNTIME_EXECUTOR_LOG_COUNTS_SQL,
)
timeline_counts = await _safe_aux_rows(
timeline_counts = await _safe_rows(
"timeline_counts",
_RUNTIME_TIMELINE_COUNTS_SQL,
_RUNTIME_TIMELINE_COUNTS_FALLBACK_SQL,
)
playbook_trust_counts = await _safe_aux_rows(
playbook_trust_counts = await _safe_rows(
"playbook_trust_counts",
_RUNTIME_PLAYBOOK_TRUST_COUNTS_SQL,
_RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL,
)
alert_operation_counts = await _safe_aux_rows(
alert_operation_counts = await _safe_rows(
"alert_operation_counts",
_RUNTIME_ALERT_OPERATION_COUNTS_SQL,
)
alertmanager_event_counts = await _safe_aux_rows(
alertmanager_event_counts = await _safe_rows(
"alertmanager_event_counts",
_RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL,
)
grouped_alert_event_counts = await _safe_aux_rows(
grouped_alert_event_counts = await _safe_rows(
"grouped_alert_event_counts",
_RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL,
)
@@ -4097,7 +4163,7 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
return build_runtime_receipt_readback_from_rows(
project_id=project_id,
lookback_hours=params["lookback_hours"],
db_read_status="ok",
db_read_status="partial" if partial_query_failures else "ok",
operation_count_rows=operation_counts,
operation_latest_rows=operation_latest,
auto_repair_count_rows=auto_repair_counts,
@@ -4118,6 +4184,8 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
alertmanager_event_count_rows=alertmanager_event_counts,
grouped_alert_event_count_rows=grouped_alert_event_counts,
log_controlled_writeback_consumer=log_controlled_writeback_consumer,
error_type="partial_query_failures" if partial_query_failures else None,
partial_query_failures=partial_query_failures,
)