feat(agents): read live autonomous runtime receipts
This commit is contained in:
Your Name
@@ -59,7 +59,7 @@ from src.services.ai_agent_automation_inventory_snapshot import (
|
||||
load_latest_ai_agent_automation_inventory_snapshot,
|
||||
)
|
||||
from src.services.ai_agent_autonomous_runtime_control import (
|
||||
build_ai_agent_autonomous_runtime_control,
|
||||
build_ai_agent_autonomous_runtime_control_with_live_readback,
|
||||
)
|
||||
from src.services.ai_agent_candidate_operation_dry_run_evidence import (
|
||||
load_latest_ai_agent_candidate_operation_dry_run_evidence,
|
||||
@@ -841,7 +841,7 @@ async def get_automation_inventory_snapshot() -> dict[str, Any]:
|
||||
async def get_agent_autonomous_runtime_control() -> dict[str, Any]:
|
||||
"""回傳目前有效 AI Agent 自主化控制層。"""
|
||||
try:
|
||||
return await asyncio.to_thread(build_ai_agent_autonomous_runtime_control)
|
||||
return await build_ai_agent_autonomous_runtime_control_with_live_readback()
|
||||
except ValueError as exc:
|
||||
logger.error("ai_agent_autonomous_runtime_control_invalid", error=str(exc))
|
||||
raise HTTPException(
|
||||
|
||||
@@ -9,10 +9,14 @@ KM, and Telegram receipts are present.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable, Mapping
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from src.core.config import settings
|
||||
from src.core.logging import get_logger
|
||||
from sqlalchemy import text
|
||||
from src.db.base import get_db_context
|
||||
from src.services.report_generation_service import (
|
||||
DAILY_REPORT_HOUR_TAIPEI,
|
||||
MONTHLY_REPORT_DAY_TAIPEI,
|
||||
@@ -25,6 +29,18 @@ _SCHEMA_VERSION = "ai_agent_autonomous_runtime_control_v1"
|
||||
_RUNTIME_AUTHORITY = "current_owner_directive_controlled_ai_automation"
|
||||
_DEPLOY_READBACK_MARKER = "p2_416_d1n_autonomous_runtime_control_prod_readback_v2"
|
||||
_DEPLOY_ATTEMPT_NOTE = "cd_3673_retry_after_host_pressure_gate_fix"
|
||||
_LIVE_READBACK_SCHEMA_VERSION = "ai_agent_autonomous_runtime_receipt_readback_v1"
|
||||
_DEFAULT_PROJECT_ID = "awoooi"
|
||||
_DEFAULT_LOOKBACK_HOURS = 24
|
||||
_EXECUTOR_OPERATION_TYPES = (
|
||||
"ansible_candidate_matched",
|
||||
"ansible_check_mode_executed",
|
||||
"ansible_apply_executed",
|
||||
"ansible_rollback_executed",
|
||||
"ansible_execution_skipped",
|
||||
)
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
def _allowed_risk_levels() -> list[str]:
|
||||
@@ -32,6 +48,340 @@ def _allowed_risk_levels() -> list[str]:
|
||||
return sorted({item.strip().lower() for item in raw.split(",") if item.strip()})
|
||||
|
||||
|
||||
def _utc_iso(value: Any) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, datetime):
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(tzinfo=timezone.utc)
|
||||
return value.astimezone(timezone.utc).isoformat()
|
||||
return str(value)
|
||||
|
||||
|
||||
def _row_mapping(row: Mapping[str, Any] | Any) -> dict[str, Any]:
|
||||
if isinstance(row, Mapping):
|
||||
return dict(row)
|
||||
mapping = getattr(row, "_mapping", None)
|
||||
if mapping is not None:
|
||||
return dict(mapping)
|
||||
return dict(row)
|
||||
|
||||
|
||||
def _int_value(value: Any) -> int:
|
||||
try:
|
||||
return int(value or 0)
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
def _sanitize_latest_rows(
|
||||
rows: Iterable[Mapping[str, Any] | Any],
|
||||
*,
|
||||
allowed_keys: tuple[str, ...],
|
||||
time_keys: tuple[str, ...] = ("created_at", "collected_at", "queued_at", "sent_at"),
|
||||
limit: int = 5,
|
||||
) -> list[dict[str, Any]]:
|
||||
clean_rows: list[dict[str, Any]] = []
|
||||
for row in rows:
|
||||
item = _row_mapping(row)
|
||||
clean: dict[str, Any] = {}
|
||||
for key in allowed_keys:
|
||||
if key not in item:
|
||||
continue
|
||||
value = item.get(key)
|
||||
clean[key] = _utc_iso(value) if key in time_keys else value
|
||||
clean_rows.append(clean)
|
||||
if len(clean_rows) >= limit:
|
||||
break
|
||||
return clean_rows
|
||||
|
||||
|
||||
def _operation_counts(
|
||||
rows: Iterable[Mapping[str, Any] | Any],
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
counts = {
|
||||
operation_type: {
|
||||
"total": 0,
|
||||
"recent": 0,
|
||||
"by_status": {},
|
||||
}
|
||||
for operation_type in _EXECUTOR_OPERATION_TYPES
|
||||
}
|
||||
for row in rows:
|
||||
item = _row_mapping(row)
|
||||
operation_type = str(item.get("operation_type") or "unknown")
|
||||
status = str(item.get("status") or "unknown")
|
||||
bucket = counts.setdefault(
|
||||
operation_type,
|
||||
{
|
||||
"total": 0,
|
||||
"recent": 0,
|
||||
"by_status": {},
|
||||
},
|
||||
)
|
||||
total = _int_value(item.get("total"))
|
||||
recent = _int_value(item.get("recent"))
|
||||
bucket["total"] += total
|
||||
bucket["recent"] += recent
|
||||
bucket["by_status"][status] = bucket["by_status"].get(status, 0) + total
|
||||
return counts
|
||||
|
||||
|
||||
def _status_counts(
|
||||
rows: Iterable[Mapping[str, Any] | Any],
|
||||
*,
|
||||
status_key: str,
|
||||
) -> dict[str, Any]:
|
||||
by_status: dict[str, int] = {}
|
||||
total = 0
|
||||
recent = 0
|
||||
for row in rows:
|
||||
item = _row_mapping(row)
|
||||
status = str(item.get(status_key) or "unknown")
|
||||
row_total = _int_value(item.get("total"))
|
||||
by_status[status] = by_status.get(status, 0) + row_total
|
||||
total += row_total
|
||||
recent += _int_value(item.get("recent"))
|
||||
return {
|
||||
"total": total,
|
||||
"recent": recent,
|
||||
"by_status": by_status,
|
||||
}
|
||||
|
||||
|
||||
def _latest_flow_closure(
|
||||
*,
|
||||
operation_latest_rows: Iterable[Mapping[str, Any] | Any],
|
||||
verifier_latest_rows: Iterable[Mapping[str, Any] | Any],
|
||||
km_latest_rows: Iterable[Mapping[str, Any] | Any],
|
||||
telegram_latest_rows: Iterable[Mapping[str, Any] | Any],
|
||||
) -> dict[str, Any]:
|
||||
operation_rows = [_row_mapping(row) for row in operation_latest_rows]
|
||||
verifier_rows = [_row_mapping(row) for row in verifier_latest_rows]
|
||||
km_rows = [_row_mapping(row) for row in km_latest_rows]
|
||||
telegram_rows = [_row_mapping(row) for row in telegram_latest_rows]
|
||||
latest_apply = next(
|
||||
(
|
||||
row
|
||||
for row in operation_rows
|
||||
if str(row.get("operation_type") or "") == "ansible_apply_executed"
|
||||
),
|
||||
None,
|
||||
)
|
||||
if latest_apply is None:
|
||||
return {
|
||||
"apply_op_id": None,
|
||||
"incident_id": None,
|
||||
"has_post_apply_verifier": False,
|
||||
"has_km_writeback": False,
|
||||
"has_telegram_receipt": False,
|
||||
"closed": False,
|
||||
"missing": [
|
||||
"ansible_apply_executed",
|
||||
"post_apply_verifier",
|
||||
"km_writeback",
|
||||
"telegram_receipt",
|
||||
],
|
||||
}
|
||||
|
||||
apply_op_id = str(latest_apply.get("op_id") or "")
|
||||
incident_id = str(latest_apply.get("incident_id") or "")
|
||||
km_path_type = f"ansible_apply_receipt:{apply_op_id[:8]}" if apply_op_id else ""
|
||||
has_verifier = any(
|
||||
str(row.get("apply_op_id") or "") == apply_op_id
|
||||
for row in verifier_rows
|
||||
)
|
||||
has_km = any(
|
||||
str(row.get("path_type") or "") == km_path_type
|
||||
or (
|
||||
incident_id
|
||||
and str(row.get("related_incident_id") or "") == incident_id
|
||||
)
|
||||
for row in km_rows
|
||||
)
|
||||
has_telegram = any(
|
||||
str(row.get("send_status") or "") == "sent"
|
||||
and str(row.get("action") or "") == "controlled_apply_result"
|
||||
and (
|
||||
not incident_id
|
||||
or str(row.get("incident_id") or "") == incident_id
|
||||
)
|
||||
for row in telegram_rows
|
||||
)
|
||||
missing = [
|
||||
name
|
||||
for name, present in (
|
||||
("post_apply_verifier", has_verifier),
|
||||
("km_writeback", has_km),
|
||||
("telegram_receipt", has_telegram),
|
||||
)
|
||||
if not present
|
||||
]
|
||||
return {
|
||||
"apply_op_id": apply_op_id or None,
|
||||
"incident_id": incident_id or None,
|
||||
"has_post_apply_verifier": has_verifier,
|
||||
"has_km_writeback": has_km,
|
||||
"has_telegram_receipt": has_telegram,
|
||||
"closed": not missing,
|
||||
"missing": missing,
|
||||
}
|
||||
|
||||
|
||||
def build_runtime_receipt_readback_from_rows(
|
||||
*,
|
||||
project_id: str = _DEFAULT_PROJECT_ID,
|
||||
lookback_hours: int = _DEFAULT_LOOKBACK_HOURS,
|
||||
db_read_status: str = "ok",
|
||||
operation_count_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
operation_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
verifier_count_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
verifier_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
km_count_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
km_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
telegram_count_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
telegram_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
|
||||
error_type: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Build the live executor receipt readback from already-fetched rows."""
|
||||
|
||||
operation_latest = list(operation_latest_rows)
|
||||
verifier_latest = list(verifier_latest_rows)
|
||||
km_latest = list(km_latest_rows)
|
||||
telegram_latest = list(telegram_latest_rows)
|
||||
operation_summary = _operation_counts(operation_count_rows)
|
||||
verifier_summary = _status_counts(
|
||||
verifier_count_rows,
|
||||
status_key="verification_result",
|
||||
)
|
||||
km_summary = _status_counts(km_count_rows, status_key="status")
|
||||
telegram_summary = _status_counts(telegram_count_rows, status_key="send_status")
|
||||
latest_closure = _latest_flow_closure(
|
||||
operation_latest_rows=operation_latest,
|
||||
verifier_latest_rows=verifier_latest,
|
||||
km_latest_rows=km_latest,
|
||||
telegram_latest_rows=telegram_latest,
|
||||
)
|
||||
apply_summary = operation_summary.get("ansible_apply_executed") or {}
|
||||
readback = {
|
||||
"schema_version": _LIVE_READBACK_SCHEMA_VERSION,
|
||||
"project_id": project_id,
|
||||
"lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)),
|
||||
"db_read_status": db_read_status,
|
||||
"writes_on_read": False,
|
||||
"ansible_operations": {
|
||||
"counts": operation_summary,
|
||||
"latest": _sanitize_latest_rows(
|
||||
operation_latest,
|
||||
allowed_keys=(
|
||||
"op_id",
|
||||
"parent_op_id",
|
||||
"operation_type",
|
||||
"status",
|
||||
"actor",
|
||||
"incident_id",
|
||||
"catalog_id",
|
||||
"playbook_path",
|
||||
"execution_mode",
|
||||
"returncode",
|
||||
"duration_ms",
|
||||
"created_at",
|
||||
),
|
||||
),
|
||||
},
|
||||
"ansible_apply_executed": {
|
||||
"total": _int_value(apply_summary.get("total")),
|
||||
"recent": _int_value(apply_summary.get("recent")),
|
||||
"by_status": apply_summary.get("by_status") or {},
|
||||
},
|
||||
"post_apply_verifier": {
|
||||
**verifier_summary,
|
||||
"latest": _sanitize_latest_rows(
|
||||
verifier_latest,
|
||||
allowed_keys=(
|
||||
"id",
|
||||
"incident_id",
|
||||
"matched_playbook_id",
|
||||
"verification_result",
|
||||
"apply_op_id",
|
||||
"catalog_id",
|
||||
"playbook_path",
|
||||
"returncode",
|
||||
"collected_at",
|
||||
),
|
||||
),
|
||||
},
|
||||
"km_writeback": {
|
||||
**km_summary,
|
||||
"latest": _sanitize_latest_rows(
|
||||
km_latest,
|
||||
allowed_keys=(
|
||||
"id",
|
||||
"title",
|
||||
"related_incident_id",
|
||||
"related_playbook_id",
|
||||
"path_type",
|
||||
"status",
|
||||
"created_by",
|
||||
"created_at",
|
||||
),
|
||||
),
|
||||
},
|
||||
"telegram_receipt": {
|
||||
**telegram_summary,
|
||||
"latest": _sanitize_latest_rows(
|
||||
telegram_latest,
|
||||
allowed_keys=(
|
||||
"message_id",
|
||||
"run_id",
|
||||
"message_type",
|
||||
"send_status",
|
||||
"provider_message_id",
|
||||
"incident_id",
|
||||
"action",
|
||||
"queued_at",
|
||||
"sent_at",
|
||||
),
|
||||
),
|
||||
},
|
||||
"latest_flow_closure": latest_closure,
|
||||
}
|
||||
if error_type:
|
||||
readback["error"] = {
|
||||
"type": error_type,
|
||||
"message": "runtime receipt DB read failed; see API logs",
|
||||
}
|
||||
return readback
|
||||
|
||||
|
||||
def _attach_runtime_receipt_readback(
|
||||
payload: dict[str, Any],
|
||||
readback: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
payload["runtime_receipt_readback"] = readback
|
||||
rollups = payload.setdefault("rollups", {})
|
||||
rollups.update({
|
||||
"live_ansible_apply_executed_count": _int_value(
|
||||
readback.get("ansible_apply_executed", {}).get("total")
|
||||
),
|
||||
"live_post_apply_verifier_count": _int_value(
|
||||
readback.get("post_apply_verifier", {}).get("total")
|
||||
),
|
||||
"live_km_writeback_count": _int_value(
|
||||
readback.get("km_writeback", {}).get("total")
|
||||
),
|
||||
"live_telegram_receipt_count": _int_value(
|
||||
readback.get("telegram_receipt", {}).get("total")
|
||||
),
|
||||
"live_executor_latest_flow_closed_count": (
|
||||
1
|
||||
if (readback.get("latest_flow_closure") or {}).get("closed") is True
|
||||
else 0
|
||||
),
|
||||
})
|
||||
return payload
|
||||
|
||||
|
||||
def build_ai_agent_autonomous_runtime_control() -> dict[str, Any]:
|
||||
"""Build the current AI Agent autonomy control-plane readback."""
|
||||
|
||||
@@ -240,10 +590,257 @@ def build_ai_agent_autonomous_runtime_control() -> dict[str, Any]:
|
||||
"legacy_policy_overridden_count": len(legacy_overrides),
|
||||
},
|
||||
}
|
||||
_attach_runtime_receipt_readback(
|
||||
payload,
|
||||
build_runtime_receipt_readback_from_rows(
|
||||
project_id=_DEFAULT_PROJECT_ID,
|
||||
db_read_status="not_queried",
|
||||
),
|
||||
)
|
||||
_validate_payload(payload)
|
||||
return payload
|
||||
|
||||
|
||||
async def load_ai_agent_autonomous_runtime_receipt_readback(
|
||||
*,
|
||||
project_id: str = _DEFAULT_PROJECT_ID,
|
||||
lookback_hours: int = _DEFAULT_LOOKBACK_HOURS,
|
||||
limit: int = 20,
|
||||
) -> dict[str, Any]:
|
||||
"""Read live executor receipts without sending messages or mutating runtime state."""
|
||||
|
||||
params = {
|
||||
"project_id": project_id,
|
||||
"lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)),
|
||||
"limit": max(1, int(limit or 20)),
|
||||
}
|
||||
try:
|
||||
async with get_db_context(project_id) as db:
|
||||
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
|
||||
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()
|
||||
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()
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"ai_agent_autonomous_runtime_receipt_readback_failed",
|
||||
project_id=project_id,
|
||||
error_type=type(exc).__name__,
|
||||
)
|
||||
return build_runtime_receipt_readback_from_rows(
|
||||
project_id=project_id,
|
||||
lookback_hours=params["lookback_hours"],
|
||||
db_read_status="unavailable",
|
||||
error_type=type(exc).__name__,
|
||||
)
|
||||
|
||||
return build_runtime_receipt_readback_from_rows(
|
||||
project_id=project_id,
|
||||
lookback_hours=params["lookback_hours"],
|
||||
db_read_status="ok",
|
||||
operation_count_rows=operation_counts,
|
||||
operation_latest_rows=operation_latest,
|
||||
verifier_count_rows=verifier_counts,
|
||||
verifier_latest_rows=verifier_latest,
|
||||
km_count_rows=km_counts,
|
||||
km_latest_rows=km_latest,
|
||||
telegram_count_rows=telegram_counts,
|
||||
telegram_latest_rows=telegram_latest,
|
||||
)
|
||||
|
||||
|
||||
async def build_ai_agent_autonomous_runtime_control_with_live_readback(
|
||||
*,
|
||||
project_id: str = _DEFAULT_PROJECT_ID,
|
||||
lookback_hours: int = _DEFAULT_LOOKBACK_HOURS,
|
||||
) -> dict[str, Any]:
|
||||
"""Build the control plane and attach live DB receipt readback."""
|
||||
|
||||
payload = build_ai_agent_autonomous_runtime_control()
|
||||
readback = await load_ai_agent_autonomous_runtime_receipt_readback(
|
||||
project_id=project_id,
|
||||
lookback_hours=lookback_hours,
|
||||
)
|
||||
_attach_runtime_receipt_readback(payload, readback)
|
||||
_validate_payload(payload)
|
||||
return payload
|
||||
|
||||
|
||||
_RUNTIME_OPERATION_COUNTS_SQL = """
|
||||
SELECT
|
||||
operation_type,
|
||||
status,
|
||||
count(*) AS total,
|
||||
count(*) FILTER (
|
||||
WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
|
||||
) AS recent
|
||||
FROM automation_operation_log
|
||||
WHERE operation_type IN (
|
||||
'ansible_candidate_matched',
|
||||
'ansible_check_mode_executed',
|
||||
'ansible_apply_executed',
|
||||
'ansible_rollback_executed',
|
||||
'ansible_execution_skipped'
|
||||
)
|
||||
GROUP BY operation_type, status
|
||||
ORDER BY operation_type, status
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_OPERATION_LATEST_SQL = """
|
||||
SELECT
|
||||
op_id::text AS op_id,
|
||||
parent_op_id::text AS parent_op_id,
|
||||
operation_type,
|
||||
status,
|
||||
actor,
|
||||
coalesce(incident_id::text, input ->> 'incident_id') AS incident_id,
|
||||
input ->> 'catalog_id' AS catalog_id,
|
||||
coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path,
|
||||
input ->> 'execution_mode' AS execution_mode,
|
||||
coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode,
|
||||
duration_ms,
|
||||
created_at
|
||||
FROM automation_operation_log
|
||||
WHERE operation_type IN (
|
||||
'ansible_candidate_matched',
|
||||
'ansible_check_mode_executed',
|
||||
'ansible_apply_executed',
|
||||
'ansible_rollback_executed',
|
||||
'ansible_execution_skipped'
|
||||
)
|
||||
ORDER BY created_at DESC
|
||||
LIMIT :limit
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_VERIFIER_COUNTS_SQL = """
|
||||
SELECT
|
||||
coalesce(verification_result, 'missing') AS verification_result,
|
||||
count(*) AS total,
|
||||
count(*) FILTER (
|
||||
WHERE collected_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
|
||||
) AS recent
|
||||
FROM incident_evidence
|
||||
WHERE post_execution_state ->> 'apply_op_id' IS NOT NULL
|
||||
GROUP BY coalesce(verification_result, 'missing')
|
||||
ORDER BY verification_result
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_VERIFIER_LATEST_SQL = """
|
||||
SELECT
|
||||
id,
|
||||
incident_id,
|
||||
matched_playbook_id,
|
||||
coalesce(verification_result, 'missing') AS verification_result,
|
||||
post_execution_state ->> 'apply_op_id' AS apply_op_id,
|
||||
post_execution_state ->> 'catalog_id' AS catalog_id,
|
||||
post_execution_state ->> 'playbook_path' AS playbook_path,
|
||||
post_execution_state ->> 'returncode' AS returncode,
|
||||
collected_at
|
||||
FROM incident_evidence
|
||||
WHERE post_execution_state ->> 'apply_op_id' IS NOT NULL
|
||||
ORDER BY collected_at DESC
|
||||
LIMIT :limit
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_KM_COUNTS_SQL = """
|
||||
SELECT
|
||||
status,
|
||||
count(*) AS total,
|
||||
count(*) FILTER (
|
||||
WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
|
||||
) AS recent
|
||||
FROM knowledge_entries
|
||||
WHERE project_id = :project_id
|
||||
AND (
|
||||
path_type LIKE 'ansible_apply_receipt:%'
|
||||
OR tags::text LIKE '%ansible_controlled_apply%'
|
||||
)
|
||||
GROUP BY status
|
||||
ORDER BY status
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_KM_LATEST_SQL = """
|
||||
SELECT
|
||||
id,
|
||||
title,
|
||||
related_incident_id,
|
||||
related_playbook_id,
|
||||
path_type,
|
||||
status,
|
||||
created_by,
|
||||
created_at
|
||||
FROM knowledge_entries
|
||||
WHERE project_id = :project_id
|
||||
AND (
|
||||
path_type LIKE 'ansible_apply_receipt:%'
|
||||
OR tags::text LIKE '%ansible_controlled_apply%'
|
||||
)
|
||||
ORDER BY created_at DESC
|
||||
LIMIT :limit
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_TELEGRAM_COUNTS_SQL = """
|
||||
SELECT
|
||||
send_status,
|
||||
count(*) AS total,
|
||||
count(*) FILTER (
|
||||
WHERE queued_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
|
||||
) AS recent
|
||||
FROM awooop_outbound_message
|
||||
WHERE project_id = :project_id
|
||||
AND channel_type = 'telegram'
|
||||
AND source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result'
|
||||
GROUP BY send_status
|
||||
ORDER BY send_status
|
||||
"""
|
||||
|
||||
|
||||
_RUNTIME_TELEGRAM_LATEST_SQL = """
|
||||
SELECT
|
||||
message_id::text AS message_id,
|
||||
run_id::text AS run_id,
|
||||
message_type,
|
||||
send_status,
|
||||
provider_message_id,
|
||||
source_envelope #>> '{callback_reply,incident_id}' AS incident_id,
|
||||
source_envelope #>> '{callback_reply,action}' AS action,
|
||||
queued_at,
|
||||
sent_at
|
||||
FROM awooop_outbound_message
|
||||
WHERE project_id = :project_id
|
||||
AND channel_type = 'telegram'
|
||||
AND source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result'
|
||||
ORDER BY queued_at DESC
|
||||
LIMIT :limit
|
||||
"""
|
||||
|
||||
|
||||
def _validate_payload(payload: dict[str, Any]) -> None:
|
||||
if payload.get("schema_version") != _SCHEMA_VERSION:
|
||||
raise ValueError(f"schema_version must be {_SCHEMA_VERSION}")
|
||||
|
||||
Reference in New Issue
Block a user