from __future__ import annotations from typing import Literal import httpx from sqlalchemy import func, select from sqlalchemy.orm import Session from app.core.config import get_settings from app.models.audit import AuditEvent from app.models.outbox import OutboxEvent from app.schemas import ( McpHubIntegrationStatus, N8nErrorHandlerStatus, N8nIntegrationStatus, N8nWorkflowEvidence, ) settings = get_settings() # The 4 canonical Fleet Ops n8n workflows (see n8n/workflows/MANIFEST.md). All 4 are # built (workflow 3, RAGcore Procedure Sync, has all 6 nodes saved); workflow 3 is not # yet published/active, so it will show no *run* evidence until it is. _CANONICAL_WORKFLOWS = ( "Fleet Ops — Vehicle Return Orchestration", "Fleet Ops — Scheduled Data Quality Scan", "Fleet Ops — RAGcore Procedure Sync", "Fleet Ops — Workflow Error Handler", ) 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) latest_success_at = db.scalar( select(func.max(OutboxEvent.updated_at)).where(OutboxEvent.delivery_status == "succeeded") ) latest_failure_at = db.scalar( select(func.max(OutboxEvent.updated_at)).where(OutboxEvent.delivery_status == "failed") ) state: Literal["disabled", "unavailable", "degraded", "operational", "no_evidence"] if not settings.n8n_dispatch_enabled: state = "disabled" elif failed > 0 and succeeded == 0: state = "unavailable" elif failed > 0: state = "degraded" elif succeeded > 0 or pending > 0 or delivering > 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", ) ) # 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 ) 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": None, "Fleet Ops — Workflow Error Handler": latest_handler_failure_at, } workflows = [ N8nWorkflowEvidence( name=name, built=True, last_seen_at=evidence_by_workflow[name], ) for name in _CANONICAL_WORKFLOWS ] return N8nIntegrationStatus( configured=bool(settings.n8n_webhook_url), dispatch_enabled=settings.n8n_dispatch_enabled, state=state, pending=pending, delivering=delivering, failed=failed, succeeded=succeeded, latest_success_at=latest_success_at, latest_failure_at=latest_failure_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 state: Literal["not_configured", "no_evidence", "operational"] if not settings.mcp_hub_registration_enabled: state = "not_configured" elif total_calls > 0: state = "operational" else: state = "no_evidence" hub_reachable = _check_hub_reachable() 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 try: response = httpx.get(f"{settings.mcp_hub_base_url.rstrip('/')}/health", timeout=1.5) return response.status_code == 200 except httpx.HTTPError: return False