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

Merge remote-tracking branch 'origin/main' into codex/p0-006-ai-log-triage-20260709

Browse Source
This commit is contained in:
ogt
2026-07-10 00:15:51 +08:00
parent 3f6c62dd03 a1ec18b9ba
commit ae0c6d291d
4 changed files with 303 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -53,6 +53,7 @@ _RUNTIME_RECEIPT_QUERY_BUDGET_SECONDS = 2.0
_RUNTIME_RECEIPT_SINGLE_QUERY_TIMEOUT_SECONDS = 0.45
_RUNTIME_RECEIPT_STATEMENT_TIMEOUT_MS = 400
_LOG_CONTROLLED_WRITEBACK_CONSUMER_TIMEOUT_SECONDS = 4.0
_LOG_CONTROLLED_WRITEBACK_CONSUMER_READBACK_CACHE_TTL_SECONDS = 30.0
_RUNTIME_RECEIPT_READBACK_CACHE_TTL_SECONDS = 20.0
_AUXILIARY_RUNTIME_RECEIPT_QUERY_NAMES = {
"alert_operation_counts",
@@ -60,6 +61,7 @@ _AUXILIARY_RUNTIME_RECEIPT_QUERY_NAMES = {
"grouped_alert_event_counts",
"db_context_exit",
"legacy_mcp_counts",
"playbook_trust_counts",
}
_CONSUMER_RECEIPT_FALLBACK_ERROR_TYPES = {
"RuntimeReceiptDbContextTimeout",
@@ -102,6 +104,10 @@ _runtime_receipt_readback_cache: dict[
tuple[str, int, int],
tuple[float, dict[str, Any]],
] = {}
_log_controlled_writeback_consumer_readback_cache: dict[
str,
tuple[float, dict[str, Any]],
] = {}
_runtime_receipt_readback_lock: asyncio.Lock | None = None
@@ -213,6 +219,7 @@ def _runtime_receipt_readback_cache_store(
def _clear_runtime_receipt_readback_cache() -> None:
_runtime_receipt_readback_cache.clear()
_log_controlled_writeback_consumer_readback_cache.clear()
def _sanitize_latest_rows(
@@ -471,18 +478,33 @@ async def _load_log_controlled_writeback_consumer_readback(
"""Attach LOG consumer bindings without writing KM/RAG/PlayBook/MCP targets."""
try:
return await asyncio.wait_for(
readback = await asyncio.wait_for(
load_latest_ai_agent_log_controlled_writeback_consumer_readback(
project_id=project_id,
),
timeout=_LOG_CONTROLLED_WRITEBACK_CONSUMER_TIMEOUT_SECONDS,
)
_log_controlled_writeback_consumer_readback_cache_store(
project_id=project_id,
readback=readback,
)
return readback
except Exception as exc: # pragma: no cover - keeps runtime control API visible
logger.warning(
"log_controlled_writeback_consumer_readback_failed",
project_id=project_id,
error_type=type(exc).__name__,
)
cached_readback = _log_controlled_writeback_consumer_readback_cache_get(
project_id=project_id,
)
if cached_readback is not None:
cached_readback["cache_fallback_active"] = True
cached_readback["record_quality"] = "cached_live_consumer_readback"
readback_info = cached_readback.get("readback")
if isinstance(readback_info, dict):
readback_info["cache_fallback_error_type"] = type(exc).__name__
return cached_readback
return _fallback_log_controlled_writeback_consumer_readback(
error_type=type(exc).__name__,
)
@@ -602,6 +624,35 @@ def _log_controlled_writeback_consumer_ready(
)
def _log_controlled_writeback_consumer_readback_cache_get(
*,
project_id: str,
) -> dict[str, Any] | None:
cached = _log_controlled_writeback_consumer_readback_cache.get(project_id)
if cached is None:
return None
stored_at, readback = cached
if (
time.monotonic() - stored_at
> _LOG_CONTROLLED_WRITEBACK_CONSUMER_READBACK_CACHE_TTL_SECONDS
):
_log_controlled_writeback_consumer_readback_cache.pop(project_id, None)
return None
return copy.deepcopy(readback)
def _log_controlled_writeback_consumer_readback_cache_store(
*,
project_id: str,
readback: Mapping[str, Any],
) -> None:
if _log_controlled_writeback_consumer_ready(readback):
_log_controlled_writeback_consumer_readback_cache[project_id] = (
time.monotonic(),
copy.deepcopy(dict(readback)),
)
def _consumer_receipt_fallback_active(
*,
db_read_status: str,

119
apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
View File

@@ -19,6 +19,7 @@ from typing import Any
from sqlalchemy import text
from src.core.config import settings
from src.core.logging import get_logger
from src.db.base import get_db_context
from src.services.ai_agent_log_controlled_writeback_consumer_readback import (
@@ -36,6 +37,10 @@ from src.services.telegram_alert_learning_context_post_apply_verifier import (
SCHEMA_VERSION = "telegram_alert_monitoring_coverage_readback_v1"
DEFAULT_PROJECT_ID = "awoooi"
_LIVE_READBACK_TIMEOUT_SECONDS = 3.0
_RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS = 0.75
_RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25
_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS = 650
_MONITORING_INVENTORY = "monitoring-alerting-observability-inventory.snapshot.json"
_MONITORING_POST_INCIDENT_READBACK = "monitoring-post-incident-readback-plan.snapshot.json"
_ALERT_LOG_EVENT_TYPES = (
@@ -382,6 +387,10 @@ def build_telegram_alert_monitoring_coverage_readback(
"all_verified_ai_alert_context_receipts_reusable_by_ai_agent": (
effective_verifier_ready
),
"km_rag_mcp_playbook_ai_agent_context_ready": effective_verifier_ready,
"km_rag_mcp_playbook_ai_agent_context_source": str(
ai_loop_context["source"]
),
"all_monitoring_inventory_surfaces_have_accepted_live_receipt": (
monitoring_live_gap_count == 0
),
@@ -453,6 +462,10 @@ def build_telegram_alert_monitoring_coverage_readback(
"ai_loop_ai_agent_context_receipt_count": _int(
ai_loop_context["ai_agent_context_receipt_count"]
),
"km_rag_mcp_playbook_ai_agent_context_ready": effective_verifier_ready,
"km_rag_mcp_playbook_ai_agent_context_source": str(
ai_loop_context["source"]
),
"log_controlled_consumer_readback_ready": bool(
consumer_rollups.get("controlled_consumer_readback_ready")
),
@@ -675,6 +688,9 @@ async def _load_runtime_log_readback(*, project_id: str) -> dict[str, Any]:
timeout=_LIVE_READBACK_TIMEOUT_SECONDS,
)
except Exception as exc: # pragma: no cover - live DB pressure
direct_payload = await _load_runtime_log_readback_direct(project_id=project_id)
if direct_payload is not None:
return direct_payload
logger.warning(
"telegram_alert_monitoring_runtime_log_readback_unavailable",
project_id=project_id,
@@ -732,16 +748,117 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]:
)
event_rows = [_dict(row) for row in event_result.mappings().all()]
outbound_row = _dict(outbound_result.mappings().first() or {})
return _runtime_log_readback_from_rows(
event_rows=event_rows,
outbound_row=outbound_row,
readback_source="db_session",
)
async def _load_runtime_log_readback_direct(
*,
project_id: str,
) -> dict[str, Any] | None:
"""Read alert log rollups through a short-lived connection when the pool is busy."""
db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
event_types_sql = ", ".join(f"'{event_type}'" for event_type in _ALERT_LOG_EVENT_TYPES)
try:
import asyncpg
conn = await asyncio.wait_for(
asyncpg.connect(db_url),
timeout=_RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS,
)
try:
await asyncio.wait_for(
conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
)
await asyncio.wait_for(
conn.execute(
"SET statement_timeout = "
f"'{int(_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
),
timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
)
event_rows = await asyncio.wait_for(
conn.fetch(f"""
SELECT
event_type,
COUNT(*) AS count,
MAX(created_at) AS latest_at
FROM alert_operation_log
WHERE created_at >= NOW() - INTERVAL '7 days'
AND event_type IN ({event_types_sql})
GROUP BY event_type
"""),
timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
)
outbound_row = await asyncio.wait_for(
conn.fetchrow("""
SELECT
COUNT(*) AS outbound_telegram_total,
COUNT(*) FILTER (WHERE send_status = 'sent') AS outbound_sent_total,
COUNT(*) FILTER (
WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb)
? 'ai_automation_alert_card'
) AS outbound_ai_alert_card_total,
COUNT(*) FILTER (
WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb)
? 'source_refs'
) AS outbound_source_refs_total,
MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at
FROM awooop_outbound_message
WHERE project_id = $1
AND channel_type = 'telegram'
AND queued_at >= NOW() - INTERVAL '7 days'
""", project_id),
timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
)
return _runtime_log_readback_from_rows(
event_rows=[dict(row) for row in event_rows],
outbound_row=dict(outbound_row) if outbound_row is not None else {},
readback_source="direct_connection",
)
finally:
try:
await asyncio.wait_for(
conn.close(),
timeout=_RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS,
)
except Exception as close_exc: # pragma: no cover - live cleanup
logger.warning(
"telegram_alert_monitoring_runtime_log_direct_close_failed",
project_id=project_id,
error_type=type(close_exc).__name__,
)
except Exception as exc: # pragma: no cover - live DB pressure
logger.warning(
"telegram_alert_monitoring_runtime_log_direct_readback_failed",
project_id=project_id,
error_type=type(exc).__name__,
)
return None
def _runtime_log_readback_from_rows(
*,
event_rows: list[dict[str, Any]],
outbound_row: dict[str, Any],
readback_source: str,
) -> dict[str, Any]:
event_counts = {str(row["event_type"]): _int(row["count"]) for row in event_rows}
latest_event_at = max(
[str(row.get("latest_at")) for row in event_rows if row.get("latest_at")],
default=None,
)
outbound_row = _dict(outbound_result.mappings().first() or {})
total = sum(event_counts.values())
return {
"status": "ok",
"summary": {
"readback_source": readback_source,
"alert_operation_log_total_7d": total,
"telegram_sent_event_count_7d": event_counts.get("TELEGRAM_SENT", 0),
"telegram_result_sent_event_count_7d": event_counts.get(