n8n: add Fleet Ops endpoint to receive workflow error reports

New POST /api/v1/integrations/n8n/workflow-error, service-token
authenticated, for the central "Fleet Ops — Workflow Error Handler"
n8n workflow to report a bounded, secret-free failure (workflow id/
name, execution id, safe error category, trigger context, correlation
id, attempt, retry action). Idempotent on execution_id via the same
audit-event precheck pattern used by /return-callback, so a
redelivered error report is not registered twice.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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),
+    )

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]
 

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