diff --git a/backend/app/api/routers/integrations.py b/backend/app/api/routers/integrations.py index b3a421a..86b8271 100644 --- a/backend/app/api/routers/integrations.py +++ b/backend/app/api/routers/integrations.py @@ -13,7 +13,7 @@ from app.core.config import get_settings from app.core.errors import AppError from app.models.audit import AuditEvent from app.models.outbox import OutboxEvent -from app.schemas import ScanResultOut +from app.schemas import ScanResultOut, WorkflowErrorReportIn, WorkflowErrorReportResult from app.services.audit import record_audit_event from app.services.data_quality import run_scan @@ -89,3 +89,60 @@ def scheduled_scan( result = run_scan(db, actor_label="n8n scheduled scan", actor_type="service") return ScanResultOut(created=result.created) + + +@router.post("/workflow-error", response_model=WorkflowErrorReportResult) +def workflow_error( + body: WorkflowErrorReportIn, + service_token: str = Header(..., alias="X-Service-Token"), + db: Session = Depends(get_db), +) -> WorkflowErrorReportResult: + """Receives a bounded, secret-free failure report from the central n8n "Fleet Ops -- + Workflow Error Handler" workflow, which is attached as the Error Workflow on every + other Fleet Ops n8n workflow. Idempotent on execution_id: n8n may redeliver the same + error report (e.g. after a timed-out response), so this must not double-record.""" + if service_token != settings.n8n_callback_token: + raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401) + + already_recorded = ( + db.scalar( + select(AuditEvent.id).where( + AuditEvent.action == "n8n_workflow_failure_registered", + AuditEvent.metadata_json["execution_id"].astext == body.execution_id, + ) + ) + is not None + ) + if not already_recorded: + correlation_id: uuid.UUID | None = None + if body.correlation_id: + try: + correlation_id = uuid.UUID(body.correlation_id) + except ValueError: + correlation_id = None + record_audit_event( + db, + actor_type="service", + actor_label="n8n error handler", + action="n8n_workflow_failure_registered", + entity_type="automation", + correlation_id=correlation_id, + after={ + "workflow_id": body.workflow_id, + "workflow_name": body.workflow_name, + "error_category": body.error_category, + "error_summary": body.error_summary, + "trigger_context": body.trigger_context, + "attempt": body.attempt, + "retry_action": body.retry_action, + "failed_at": body.failed_at.isoformat(), + }, + metadata={"execution_id": body.execution_id}, + ) + db.commit() + + return WorkflowErrorReportResult( + status="already_registered" if already_recorded else "registered", + execution_id=body.execution_id, + occurred_at=datetime.now(UTC), + ) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 5c2ef51..c95f887 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -143,6 +143,27 @@ class ScanResultOut(BaseModel): created: dict[str, int] +class WorkflowErrorReportIn(BaseModel): + workflow_id: str = Field(max_length=120) + workflow_name: str = Field(max_length=200) + execution_id: str = Field(max_length=120) + failed_at: datetime + error_category: Literal[ + "timeout", "authError", "connectionError", "httpError", "validationError", "unknown" + ] + error_summary: str = Field(max_length=500) + trigger_context: str | None = Field(default=None, max_length=200) + correlation_id: str | None = None + attempt: int = Field(default=1, ge=1, le=1000) + retry_action: str | None = Field(default=None, max_length=200) + + +class WorkflowErrorReportResult(BaseModel): + status: Literal["registered", "already_registered"] + execution_id: str + occurred_at: datetime + + class ProvideFieldsRequest(BaseModel): fields: dict[str, str] diff --git a/backend/tests/test_integrations.py b/backend/tests/test_integrations.py index 1699bca..58c8bdc 100644 --- a/backend/tests/test_integrations.py +++ b/backend/tests/test_integrations.py @@ -122,3 +122,72 @@ def test_scheduled_scan_is_idempotent_across_repeated_triggers(client): assert first.status_code == 200 assert second.status_code == 200 assert second.json()["created"] == {} + + +def _workflow_error_body(execution_id: str, **overrides): + body = { + "workflow_id": "mobilityops-return-processing", + "workflow_name": "Fleet Ops — Vehicle Return Orchestration", + "execution_id": execution_id, + "failed_at": "2026-08-04T10:15:00Z", + "error_category": "httpError", + "error_summary": "Callback request failed with status 500", + "trigger_context": "webhook", + "correlation_id": None, + "attempt": 1, + "retry_action": "n8n will retry automatically", + } + body.update(overrides) + return body + + +def test_workflow_error_rejects_wrong_service_token(client): + response = client.post( + "/api/v1/integrations/n8n/workflow-error", + json=_workflow_error_body(str(uuid.uuid4())), + headers={"X-Service-Token": "wrong-token"}, + ) + assert response.status_code == 401 + + +def test_workflow_error_rejects_unknown_category(client): + settings = get_settings() + response = client.post( + "/api/v1/integrations/n8n/workflow-error", + json=_workflow_error_body(str(uuid.uuid4()), error_category="somethingElse"), + headers={"X-Service-Token": settings.n8n_callback_token}, + ) + assert response.status_code == 422 + + +def test_workflow_error_registers_and_is_idempotent_by_execution_id(client, ops_client): + settings = get_settings() + headers = {"X-Service-Token": settings.n8n_callback_token} + execution_id = str(uuid.uuid4()) + body = _workflow_error_body(execution_id) + + first = client.post("/api/v1/integrations/n8n/workflow-error", json=body, headers=headers) + second = client.post("/api/v1/integrations/n8n/workflow-error", json=body, headers=headers) + + assert first.status_code == 200 + assert first.json()["status"] == "registered" + assert second.status_code == 200 + assert second.json()["status"] == "already_registered" + + audit_events = ops_client.get( + "/api/v1/audit", params={"action": "n8n_workflow_failure_registered"} + ).json() + matching = [e for e in audit_events if e["metadata"]["execution_id"] == execution_id] + assert len(matching) == 1 + assert matching[0]["after"]["error_category"] == "httpError" + assert matching[0]["after"]["retry_action"] == "n8n will retry automatically" + + +def test_workflow_error_bounds_summary_length(client): + settings = get_settings() + response = client.post( + "/api/v1/integrations/n8n/workflow-error", + json=_workflow_error_body(str(uuid.uuid4()), error_summary="x" * 501), + headers={"X-Service-Token": settings.n8n_callback_token}, + ) + assert response.status_code == 422