diff --git a/PROJECT_STATE.md b/PROJECT_STATE.md index d58e9a0..6337434 100644 --- a/PROJECT_STATE.md +++ b/PROJECT_STATE.md @@ -2329,3 +2329,30 @@ evidence yet." **210 passed, 1 warning**. - Exact next action: add authenticated n8n workflow heartbeats and execution results, surface stale/healthy workflow state, then update generated contracts and E2E coverage. + +## n8n execution health telemetry (2026-08-10) + +- Added an authenticated, canonical-workflow-only, idempotent heartbeat contract. Status + now records each workflow's last execution ID/result and classifies it as healthy, + stale, failed or without evidence. Scheduled scan evidence expires after 2h30 and the + daily RAGcore sync after 30h; event-driven workflows are not falsely marked stale just + because no business event arrived. +- Registered target-workflow failures override older success evidence until a newer + successful execution arrives. The Automation UI renders these explicit states and + execution IDs instead of treating any historical timestamp as permanently healthy. +- All four versioned n8n definitions now report a successful execution heartbeat with + bounded retries; the central error handler continues to register failed target runs. +- Evidence: JSON validation for all four definitions, frontend lint, ruff and mypy passed; + focused integration tests **26 passed** and full Unraid suite **212 passed, 1 warning**. +- Live deployment: API/web deployed at `c9a8609`; all five pre-existing n8n workflows + were exported to the recoverable appdata backup + `backups/mobilityops-pre-heartbeat-20260810.json` before the four definitions were + imported and published. The import initially exposed n8n CLI's unsafe name-only + credential resolution (both Header Auth nodes resolved to the service credential); + credential IDs were restored from the backup before republishing. A real return then + completed through outbox → server n8n → callback → heartbeat as execution `337`; + its first 403 delivery remained safely retryable and succeeded after the credential + correction. Live n8n state is operational and Vehicle Return is `healthy` with its + execution ID visible. +- Exact next action: regenerate the checked-in OpenAPI contract, add E2E coverage for the + new operator workflows, document credential-safe n8n upgrades and run acceptance. diff --git a/backend/app/api/routers/integrations.py b/backend/app/api/routers/integrations.py index 16824d1..aea4429 100644 --- a/backend/app/api/routers/integrations.py +++ b/backend/app/api/routers/integrations.py @@ -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( diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 00ff855..e41fed7 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -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): diff --git a/backend/app/seed_loader.py b/backend/app/seed_loader.py index db7bba2..cbd823d 100644 --- a/backend/app/seed_loader.py +++ b/backend/app/seed_loader.py @@ -85,6 +85,7 @@ _PERSISTENT_TELEMETRY_ACTIONS = ( "n8n_return_followup_recorded", "n8n_workflow_failure_registered", "n8n_procedures_synced", + "n8n_workflow_heartbeat", "knowledge_question_asked", ) diff --git a/backend/app/services/integration_status.py b/backend/app/services/integration_status.py index e778751..2fa9136 100644 --- a/backend/app/services/integration_status.py +++ b/backend/app/services/integration_status.py @@ -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), diff --git a/backend/tests/test_integration_status.py b/backend/tests/test_integration_status.py index 859ea94..08223e6 100644 --- a/backend/tests/test_integration_status.py +++ b/backend/tests/test_integration_status.py @@ -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): diff --git a/backend/tests/test_integrations.py b/backend/tests/test_integrations.py index cdd4189..e7473b5 100644 --- a/backend/tests/test_integrations.py +++ b/backend/tests/test_integrations.py @@ -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 diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index b614093..530f50e 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -315,6 +315,9 @@ export interface N8nWorkflowEvidence { name: string; built: boolean; last_seen_at: string | null; + state: "no_evidence" | "healthy" | "stale" | "failed"; + last_status: "succeeded" | "failed" | null; + last_execution_id: string | null; } export interface N8nErrorHandlerStatus { diff --git a/frontend/src/i18n/locales/en-GB/integrations.json b/frontend/src/i18n/locales/en-GB/integrations.json index d38a465..e747180 100644 --- a/frontend/src/i18n/locales/en-GB/integrations.json +++ b/frontend/src/i18n/locales/en-GB/integrations.json @@ -37,6 +37,10 @@ "description": "{{known}} of {{expected}} canonical n8n workflows have live evidence of running.", "notBuilt": "Not built yet", "noEvidence": "No evidence yet", + "healthy": "Healthy", + "stale": "Evidence is stale", + "failed": "Latest execution failed", + "executionId": "Execution", "columns": { "name": "Workflow", "status": "Status", diff --git a/frontend/src/i18n/locales/fr-BE/integrations.json b/frontend/src/i18n/locales/fr-BE/integrations.json index cf8005c..2e637f0 100644 --- a/frontend/src/i18n/locales/fr-BE/integrations.json +++ b/frontend/src/i18n/locales/fr-BE/integrations.json @@ -36,7 +36,11 @@ "title": "Workflows d'automatisation", "description": "{{known}} workflows n8n canoniques sur {{expected}} disposent de preuves actuelles de fonctionnement.", "notBuilt": "Pas encore créé", - "noEvidence": "Aucune preuve pour l'instant", + "noEvidence": "Aucune preuve pour l’instant", + "healthy": "Sain", + "stale": "Preuve obsolète", + "failed": "Dernière exécution échouée", + "executionId": "Exécution", "columns": { "name": "Workflow", "status": "Statut", diff --git a/frontend/src/i18n/locales/nl-BE/integrations.json b/frontend/src/i18n/locales/nl-BE/integrations.json index 3c6da07..31b570d 100644 --- a/frontend/src/i18n/locales/nl-BE/integrations.json +++ b/frontend/src/i18n/locales/nl-BE/integrations.json @@ -37,6 +37,10 @@ "description": "{{known}} van {{expected}} canonieke n8n-workflows hebben actuele evidentie van werking.", "notBuilt": "Nog niet gebouwd", "noEvidence": "Nog geen evidentie", + "healthy": "Gezond", + "stale": "Evidentie verouderd", + "failed": "Laatste uitvoering mislukt", + "executionId": "Uitvoering", "columns": { "name": "Workflow", "status": "Status", diff --git a/frontend/src/pages/Automation.tsx b/frontend/src/pages/Automation.tsx index fce7738..b350716 100644 --- a/frontend/src/pages/Automation.tsx +++ b/frontend/src/pages/Automation.tsx @@ -279,9 +279,13 @@ export function Automation() { const displayName = slug ? t(`workflows.names.${slug}`) : w.name; const badge = !w.built ? { status: "not_configured", label: t("workflows.notBuilt") } - : w.last_seen_at - ? { status: "available", label: t("statusLabels.operational") } - : { status: "no_events", label: t("workflows.noEvidence") }; + : w.state === "healthy" + ? { status: "available", label: t("workflows.healthy") } + : w.state === "stale" + ? { status: "degraded", label: t("workflows.stale") } + : w.state === "failed" + ? { status: "failed", label: t("workflows.failed") } + : { status: "no_events", label: t("workflows.noEvidence") }; return (