344 lines
14 KiB
Python
344 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections.abc import Sequence
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any, Literal
|
|
|
|
import httpx
|
|
from sqlalchemy import Row, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import get_settings
|
|
from app.models.audit import AuditEvent
|
|
from app.models.outbox import DEMO_SCENARIO_ERROR_CODE, OutboxEvent
|
|
from app.schemas import (
|
|
McpHubIntegrationStatus,
|
|
N8nErrorHandlerStatus,
|
|
N8nIntegrationStatus,
|
|
N8nWorkflowEvidence,
|
|
)
|
|
|
|
settings = get_settings()
|
|
_hub_health_lock = threading.Lock()
|
|
_hub_health_cached_at = 0.0
|
|
_hub_health_cached_value: bool | None = None
|
|
|
|
# The 4 canonical Fleet Ops n8n workflows (see n8n/workflows/MANIFEST.md). All 4 are
|
|
# built (all with their full node set saved).
|
|
_CANONICAL_WORKFLOWS = (
|
|
"Fleet Ops — Vehicle Return Orchestration",
|
|
"Fleet Ops — Scheduled Data Quality Scan",
|
|
"Fleet Ops — RAGcore Procedure Sync",
|
|
"Fleet Ops — Workflow Error Handler",
|
|
)
|
|
_STALE_AFTER = {
|
|
"Fleet Ops — Scheduled Data Quality Scan": timedelta(hours=2, minutes=30),
|
|
"Fleet Ops — RAGcore Procedure Sync": timedelta(hours=30),
|
|
}
|
|
|
|
|
|
def _latest_rows_per_workflow(db: Session, action: str) -> Sequence[Row[Any]]:
|
|
"""Return the most recent audit row per workflow name for one action.
|
|
|
|
Heartbeats arrive on every scheduled run, so loading *all* rows and picking the
|
|
latest in Python would grow linearly with deployment age. ``DISTINCT ON`` lets
|
|
PostgreSQL return exactly one (latest) row per workflow instead.
|
|
"""
|
|
workflow_name = AuditEvent.after_json["workflow_name"].astext
|
|
return db.execute(
|
|
select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json)
|
|
.where(AuditEvent.action == action)
|
|
.distinct(workflow_name)
|
|
.order_by(workflow_name, AuditEvent.occurred_at.desc())
|
|
).all()
|
|
|
|
|
|
def derive_n8n_status(db: Session) -> N8nIntegrationStatus:
|
|
counts: dict[str, int] = dict(
|
|
db.execute(
|
|
select(OutboxEvent.delivery_status, func.count()).group_by(OutboxEvent.delivery_status)
|
|
).all() # type: ignore[arg-type]
|
|
)
|
|
pending = counts.get("pending", 0)
|
|
delivering = counts.get("delivering", 0)
|
|
failed = counts.get("failed", 0)
|
|
succeeded = counts.get("succeeded", 0)
|
|
|
|
# Prepared demo failures are props, not health signals. They stay visible and
|
|
# counted -- hiding them would be its own kind of lie -- but they are counted
|
|
# *separately*, and only genuinely unexpected failures are allowed to move n8n off
|
|
# "operational". Without this split the demo seed's single staged failure pins the
|
|
# integration to "degraded" forever, which tells a viewer something untrue about
|
|
# the automation.
|
|
demo_scenario_failed = (
|
|
db.scalar(
|
|
select(func.count())
|
|
.select_from(OutboxEvent)
|
|
.where(
|
|
OutboxEvent.delivery_status == "failed",
|
|
OutboxEvent.last_error_code == DEMO_SCENARIO_ERROR_CODE,
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
unexpected_failed = max(failed - demo_scenario_failed, 0)
|
|
|
|
latest_success_at = db.scalar(
|
|
select(func.max(OutboxEvent.updated_at)).where(OutboxEvent.delivery_status == "succeeded")
|
|
)
|
|
# Health talks about real failures only, so the "latest failure" a health reader
|
|
# sees must exclude the staged one too.
|
|
latest_failure_at = db.scalar(
|
|
select(func.max(OutboxEvent.updated_at)).where(
|
|
OutboxEvent.delivery_status == "failed",
|
|
or_(
|
|
OutboxEvent.last_error_code.is_(None),
|
|
OutboxEvent.last_error_code != DEMO_SCENARIO_ERROR_CODE,
|
|
),
|
|
)
|
|
)
|
|
latest_demo_scenario_at = db.scalar(
|
|
select(func.max(OutboxEvent.updated_at)).where(
|
|
OutboxEvent.delivery_status == "failed",
|
|
OutboxEvent.last_error_code == DEMO_SCENARIO_ERROR_CODE,
|
|
)
|
|
)
|
|
|
|
state: Literal["disabled", "unavailable", "degraded", "operational", "no_evidence"]
|
|
if not settings.n8n_dispatch_enabled:
|
|
state = "disabled"
|
|
elif not settings.n8n_webhook_url:
|
|
# Persisted history does not make a currently unconfigured dispatcher green.
|
|
state = "unavailable"
|
|
elif unexpected_failed > 0 and succeeded == 0:
|
|
state = "unavailable"
|
|
elif unexpected_failed > 0:
|
|
state = "degraded"
|
|
# Queued/in-flight work proves only that MobilityOps has work for the dispatcher;
|
|
# it does not prove that n8n has ever accepted a delivery. A green state requires
|
|
# at least one persisted successful round trip.
|
|
elif succeeded > 0:
|
|
state = "operational"
|
|
else:
|
|
state = "no_evidence"
|
|
|
|
# Scheduled scan evidence: only service-triggered runs count as n8n evidence, not
|
|
# runs an operator triggered manually from the Data Quality page.
|
|
latest_scan_at = db.scalar(
|
|
select(func.max(AuditEvent.occurred_at)).where(
|
|
AuditEvent.action == "data_quality_scan_run",
|
|
AuditEvent.actor_type == "service",
|
|
)
|
|
)
|
|
|
|
# RAGcore Procedure Sync evidence: result reports posted by the workflow itself once
|
|
# it finishes uploading procedures to RAGcore (app/api/routers/integrations.py::
|
|
# procedures_sync_result), the same "the workflow's own callback is the evidence"
|
|
# pattern the scheduled scan and error handler already use below.
|
|
latest_procedure_sync_row = db.execute(
|
|
select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json)
|
|
.where(AuditEvent.action == "n8n_procedures_synced")
|
|
.order_by(AuditEvent.occurred_at.desc())
|
|
.limit(1)
|
|
).first()
|
|
latest_procedure_sync_at = (
|
|
latest_procedure_sync_row[0] if latest_procedure_sync_row is not None else None
|
|
)
|
|
|
|
# Error handler evidence: registrations posted by the "Fleet Ops — Workflow Error
|
|
# Handler" n8n workflow itself, which also doubles as proof that workflow is wired
|
|
# up and firing correctly.
|
|
total_failures_registered = (
|
|
db.scalar(
|
|
select(func.count(AuditEvent.id)).where(
|
|
AuditEvent.action == "n8n_workflow_failure_registered"
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
latest_failure_row = db.execute(
|
|
select(AuditEvent.occurred_at, AuditEvent.after_json)
|
|
.where(AuditEvent.action == "n8n_workflow_failure_registered")
|
|
.order_by(AuditEvent.occurred_at.desc())
|
|
.limit(1)
|
|
).first()
|
|
latest_handler_failure_at = latest_failure_row[0] if latest_failure_row else None
|
|
latest_handler_failure_workflow = (
|
|
(latest_failure_row[1] or {}).get("workflow_name") if latest_failure_row else None
|
|
)
|
|
|
|
legacy_evidence_by_workflow = {
|
|
"Fleet Ops — Vehicle Return Orchestration": latest_success_at,
|
|
"Fleet Ops — Scheduled Data Quality Scan": latest_scan_at,
|
|
"Fleet Ops — RAGcore Procedure Sync": latest_procedure_sync_at,
|
|
"Fleet Ops — Workflow Error Handler": latest_handler_failure_at,
|
|
}
|
|
heartbeat_by_workflow: dict[str, tuple[datetime, str, str | None]] = {}
|
|
heartbeat_rows = _latest_rows_per_workflow(db, "n8n_workflow_heartbeat")
|
|
for occurred_at, after, metadata in heartbeat_rows:
|
|
workflow_name = (after or {}).get("workflow_name")
|
|
if workflow_name in _CANONICAL_WORKFLOWS and workflow_name not in heartbeat_by_workflow:
|
|
heartbeat_by_workflow[workflow_name] = (
|
|
occurred_at,
|
|
(after or {}).get("status", "succeeded"),
|
|
(metadata or {}).get("execution_id"),
|
|
)
|
|
|
|
failure_by_workflow: dict[str, tuple[datetime, str | None]] = {}
|
|
failure_rows = _latest_rows_per_workflow(db, "n8n_workflow_failure_registered")
|
|
for occurred_at, after, metadata in failure_rows:
|
|
workflow_name = (after or {}).get("workflow_name")
|
|
if workflow_name in _CANONICAL_WORKFLOWS and workflow_name not in failure_by_workflow:
|
|
failure_by_workflow[workflow_name] = (
|
|
occurred_at,
|
|
(metadata or {}).get("execution_id"),
|
|
)
|
|
|
|
now = datetime.now(UTC)
|
|
workflows: list[N8nWorkflowEvidence] = []
|
|
for name in _CANONICAL_WORKFLOWS:
|
|
legacy_seen = legacy_evidence_by_workflow[name]
|
|
heartbeat = heartbeat_by_workflow.get(name)
|
|
failure_signal = failure_by_workflow.get(name)
|
|
seen_at = heartbeat[0] if heartbeat else legacy_seen
|
|
last_status: Literal["succeeded", "failed"] | None = (
|
|
"succeeded" if seen_at is not None else None
|
|
)
|
|
execution_id = heartbeat[2] if heartbeat else None
|
|
if heartbeat and heartbeat[1] == "failed":
|
|
last_status = "failed"
|
|
if failure_signal and (seen_at is None or failure_signal[0] > seen_at):
|
|
seen_at = failure_signal[0]
|
|
last_status = "failed"
|
|
execution_id = failure_signal[1]
|
|
if name == "Fleet Ops — RAGcore Procedure Sync" and latest_procedure_sync_row:
|
|
sync_at, sync_result, sync_metadata = latest_procedure_sync_row
|
|
synced = (sync_result or {}).get("synced", 0)
|
|
failed_syncs = (sync_result or {}).get("failed", 0)
|
|
# The result callback is authoritative for corpus delivery. A generic
|
|
# succeeded heartbeat cannot turn a zero/partial upload green.
|
|
sync_result_failed = (
|
|
not isinstance(synced, int)
|
|
or not isinstance(failed_syncs, int)
|
|
or synced <= 0
|
|
or failed_syncs > 0
|
|
)
|
|
if sync_result_failed:
|
|
last_status = "failed"
|
|
if seen_at is None or sync_at > seen_at:
|
|
seen_at = sync_at
|
|
if sync_result_failed:
|
|
execution_id = (sync_metadata or {}).get("execution_id")
|
|
workflow_state: Literal["no_evidence", "healthy", "stale", "failed"]
|
|
if seen_at is None:
|
|
workflow_state = "no_evidence"
|
|
elif last_status == "failed":
|
|
workflow_state = "failed"
|
|
elif name in _STALE_AFTER and now - seen_at > _STALE_AFTER[name]:
|
|
workflow_state = "stale"
|
|
else:
|
|
workflow_state = "healthy"
|
|
workflows.append(
|
|
N8nWorkflowEvidence(
|
|
name=name,
|
|
built=True,
|
|
last_seen_at=seen_at,
|
|
state=workflow_state,
|
|
last_status=last_status,
|
|
last_execution_id=execution_id,
|
|
)
|
|
)
|
|
|
|
if state in {"operational", "no_evidence"} and any(
|
|
workflow.state in {"failed", "stale"} for workflow in workflows
|
|
):
|
|
state = "degraded"
|
|
|
|
return N8nIntegrationStatus(
|
|
configured=bool(settings.n8n_webhook_url),
|
|
dispatch_enabled=settings.n8n_dispatch_enabled,
|
|
state=state,
|
|
pending=pending,
|
|
delivering=delivering,
|
|
failed=failed,
|
|
unexpected_failed=unexpected_failed,
|
|
demo_scenario_failed=demo_scenario_failed,
|
|
succeeded=succeeded,
|
|
latest_success_at=latest_success_at,
|
|
latest_failure_at=latest_failure_at,
|
|
latest_demo_scenario_at=latest_demo_scenario_at,
|
|
expected_workflow_count=len(_CANONICAL_WORKFLOWS),
|
|
known_workflow_count=sum(1 for w in workflows if w.last_seen_at is not None),
|
|
workflows=workflows,
|
|
error_handler=N8nErrorHandlerStatus(
|
|
total_failures_registered=total_failures_registered,
|
|
latest_failure_at=latest_handler_failure_at,
|
|
latest_failure_workflow=latest_handler_failure_workflow,
|
|
),
|
|
)
|
|
|
|
|
|
def derive_mcp_hub_status(db: Session) -> McpHubIntegrationStatus:
|
|
"""Evidence-based MCP Hub status: real tool-call audit history, not just the
|
|
`MCP_HUB_REGISTRATION_ENABLED` flag flipped on. Every `mcp_tool_request` call
|
|
already writes an `AuditEvent` (see `app/api/routers/mcp_integrations.py`)."""
|
|
total_calls = (
|
|
db.scalar(select(func.count(AuditEvent.id)).where(AuditEvent.action == "mcp_tool_request"))
|
|
or 0
|
|
)
|
|
latest_call_row = db.execute(
|
|
select(AuditEvent.occurred_at, AuditEvent.actor_label, AuditEvent.metadata_json)
|
|
.where(AuditEvent.action == "mcp_tool_request")
|
|
.order_by(AuditEvent.occurred_at.desc())
|
|
.limit(1)
|
|
).first()
|
|
last_called_at = latest_call_row[0] if latest_call_row else None
|
|
last_client = latest_call_row[1] if latest_call_row else None
|
|
last_tool = (latest_call_row[2] or {}).get("tool") if latest_call_row else None
|
|
|
|
hub_reachable = _check_hub_reachable()
|
|
|
|
state: Literal["not_configured", "no_evidence", "operational"]
|
|
if not settings.mcp_hub_registration_enabled:
|
|
state = "not_configured"
|
|
elif total_calls > 0 and hub_reachable is not False:
|
|
state = "operational"
|
|
else:
|
|
# Historical tool calls remain useful telemetry, but cannot support a current
|
|
# operational claim when the configured Hub health endpoint is unreachable.
|
|
# ``hub_reachable`` stays available separately so consumers can distinguish
|
|
# this from a provider that simply has no call evidence yet.
|
|
state = "no_evidence"
|
|
|
|
return McpHubIntegrationStatus(
|
|
registration_enabled=settings.mcp_hub_registration_enabled,
|
|
state=state,
|
|
total_calls=total_calls,
|
|
last_tool=last_tool,
|
|
last_client=last_client,
|
|
last_called_at=last_called_at,
|
|
hub_reachable=hub_reachable,
|
|
)
|
|
|
|
|
|
def _check_hub_reachable() -> bool | None:
|
|
"""Real Hub-side health signal (MCP Hub's own registration is catalog-driven on
|
|
its side, so this is the only thing Fleet Ops itself can honestly check).
|
|
`None` means not configured / not checked, never a guess."""
|
|
if not settings.mcp_hub_base_url:
|
|
return None
|
|
global _hub_health_cached_at, _hub_health_cached_value
|
|
now = time.monotonic()
|
|
with _hub_health_lock:
|
|
if now - _hub_health_cached_at < settings.mcp_hub_health_cache_seconds:
|
|
return _hub_health_cached_value
|
|
try:
|
|
response = httpx.get(f"{settings.mcp_hub_base_url.rstrip('/')}/health", timeout=1.5)
|
|
_hub_health_cached_value = response.status_code == 200
|
|
except httpx.HTTPError:
|
|
_hub_health_cached_value = False
|
|
_hub_health_cached_at = time.monotonic()
|
|
return _hub_health_cached_value
|