Jens/MobilityOps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

M14: add n8n execution health telemetry

Browse Source
This commit is contained in:
NuklearRabbit
2026-08-10 03:41:06 +02:00
parent 218599af7d
commit 58fb515337
17 changed files with 368 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routers/integrations.py
+54
View File
@@ -15,6 +15,8 @@ from app.core.errors import AppError
from app.models.audit import AuditEvent
from app.models.outbox import OutboxEvent
from app.schemas import (
N8nHeartbeatIn,
N8nHeartbeatResult,
ProcedureDocumentOut,
ProcedureListOut,
ProcedureSyncResultIn,
@@ -30,6 +32,58 @@ from app.services.knowledge.procedures import iter_procedure_documents
router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"])
settings = get_settings()
_CANONICAL_WORKFLOW_NAMES = frozenset(
{
"Fleet Ops — Vehicle Return Orchestration",
"Fleet Ops — Scheduled Data Quality Scan",
"Fleet Ops — RAGcore Procedure Sync",
"Fleet Ops — Workflow Error Handler",
}
)
@router.post("/heartbeat", response_model=N8nHeartbeatResult)
def workflow_heartbeat(
body: N8nHeartbeatIn,
service_token: str = Header(..., alias="X-Service-Token"),
db: Session = Depends(get_db),
) -> N8nHeartbeatResult:
"""Authenticated, idempotent execution evidence from a canonical n8n workflow."""
if service_token != settings.n8n_callback_token:
raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401)
if body.workflow_name not in _CANONICAL_WORKFLOW_NAMES:
raise AppError("UNKNOWN_WORKFLOW", "Unknown Fleet Ops workflow.", status_code=422)
already_recorded = (
db.scalar(
select(AuditEvent.id).where(
AuditEvent.action == "n8n_workflow_heartbeat",
AuditEvent.metadata_json["execution_id"].astext == body.execution_id,
AuditEvent.after_json["status"].astext == body.status,
)
)
is not None
)
if not already_recorded:
record_audit_event(
db,
actor_type="service",
actor_label="n8n workflow heartbeat",
action="n8n_workflow_heartbeat",
entity_type="automation",
after={
"workflow_id": body.workflow_id,
"workflow_name": body.workflow_name,
"status": body.status,
},
metadata={"execution_id": body.execution_id},
)
db.commit()
return N8nHeartbeatResult(
status="already_registered" if already_recorded else "registered",
execution_id=body.execution_id,
occurred_at=datetime.now(UTC),
)
@router.post("/return-callback")
def return_callback(
backend/app/schemas.py
+16
View File
@@ -363,6 +363,22 @@ class N8nWorkflowEvidence(BaseModel):
name: str
built: bool
last_seen_at: datetime | None
state: Literal["no_evidence", "healthy", "stale", "failed"] = "no_evidence"
last_status: Literal["succeeded", "failed"] | None = None
last_execution_id: str | None = None
class N8nHeartbeatIn(BaseModel):
workflow_id: str = Field(min_length=1, max_length=120)
workflow_name: str = Field(min_length=1, max_length=200)
execution_id: str = Field(min_length=1, max_length=120)
status: Literal["succeeded", "failed"]
class N8nHeartbeatResult(BaseModel):
status: Literal["registered", "already_registered"]
execution_id: str
occurred_at: datetime
class N8nErrorHandlerStatus(BaseModel):
backend/app/seed_loader.py
+1
View File
@@ -85,6 +85,7 @@ _PERSISTENT_TELEMETRY_ACTIONS = (
"n8n_return_followup_recorded",
"n8n_workflow_failure_registered",
"n8n_procedures_synced",
"n8n_workflow_heartbeat",
"knowledge_question_asked",
)
backend/app/services/integration_status.py
+75 -8
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from typing import Literal
import httpx
@@ -26,6 +27,10 @@ _CANONICAL_WORKFLOWS = (
"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 derive_n8n_status(db: Session) -> N8nIntegrationStatus:
@@ -129,20 +134,82 @@ def derive_n8n_status(db: Session) -> N8nIntegrationStatus:
(latest_failure_row[1] or {}).get("workflow_name") if latest_failure_row else None
)
evidence_by_workflow = {
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,
}
workflows = [
N8nWorkflowEvidence(
name=name,
built=True,
last_seen_at=evidence_by_workflow[name],
heartbeat_by_workflow: dict[str, tuple[datetime, str, str | None]] = {}
heartbeat_rows = db.execute(
select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json)
.where(AuditEvent.action == "n8n_workflow_heartbeat")
.order_by(AuditEvent.occurred_at.desc())
).all()
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 = db.execute(
select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json)
.where(AuditEvent.action == "n8n_workflow_failure_registered")
.order_by(AuditEvent.occurred_at.desc())
).all()
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
)
for name in _CANONICAL_WORKFLOWS
]
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]
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),
backend/tests/test_integration_status.py
+1
View File
@@ -124,6 +124,7 @@ def test_integration_status_lists_all_four_canonical_workflows(ops_client):
# their own first real signal -- there is no run evidence yet either.
assert ragcore_sync["built"] is True
assert ragcore_sync["last_seen_at"] is None
assert ragcore_sync["state"] == "no_evidence"
def test_integration_status_scheduled_scan_evidence_only_counts_service_runs(client, ops_client):
backend/tests/test_integrations.py
+37
View File
@@ -264,3 +264,40 @@ def test_procedures_sync_result_registers_and_is_idempotent(client, ops_client):
status = ops_client.get("/api/v1/integrations/status").json()["n8n"]
ragcore_sync = next(w for w in status["workflows"] if "RAGcore" in w["name"])
assert ragcore_sync["last_seen_at"] is not None
def test_workflow_heartbeat_is_idempotent_and_drives_live_status(client, ops_client):
settings = get_settings()
execution_id = str(uuid.uuid4())
body = {
"workflow_id": "mobilityops-scheduled-quality-scan",
"workflow_name": "Fleet Ops — Scheduled Data Quality Scan",
"execution_id": execution_id,
"status": "succeeded",
}
headers = {"X-Service-Token": settings.n8n_callback_token}
first = client.post("/api/v1/integrations/n8n/heartbeat", json=body, headers=headers)
second = client.post("/api/v1/integrations/n8n/heartbeat", json=body, headers=headers)
assert first.status_code == 200
assert first.json()["status"] == "registered"
assert second.json()["status"] == "already_registered"
status = ops_client.get("/api/v1/integrations/status").json()["n8n"]
workflow = next(w for w in status["workflows"] if w["name"] == body["workflow_name"])
assert workflow["state"] == "healthy"
assert workflow["last_status"] == "succeeded"
assert workflow["last_execution_id"] == execution_id
def test_workflow_heartbeat_rejects_unknown_workflow(client):
response = client.post(
"/api/v1/integrations/n8n/heartbeat",
json={
"workflow_id": "unknown",
"workflow_name": "Unknown workflow",
"execution_id": "test-unknown-001",
"status": "succeeded",
},
headers={"X-Service-Token": get_settings().n8n_callback_token},
)
assert response.status_code == 422