M4: implement n8n automation

Outbox dispatcher (background thread, FOR UPDATE SKIP LOCKED claim, exponential backoff, no transaction held during HTTP I/O). n8n callback endpoint with shared-secret auth and idempotency by event ID. Automation nav + UI with manual retry. 49 backend tests passing, ruff clean. Fixed a crash-on-redelivery bug in seeded outbox payloads and made the dispatcher defensive against malformed payloads. Verified the full live round trip against a real n8n instance: return -> outbox -> dispatcher -> n8n workflow -> callback -> succeeded, including the S5 failed-retry demo scenario.

backend/app/api/routers/integrations.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+from typing import Any
+
+from fastapi import APIRouter, Depends, Header
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db
+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.services.audit import record_audit_event
+
+router = APIRouter(prefix="/api/v1/integrations/n8n", tags=["integrations"])
+settings = get_settings()
+
+
+@router.post("/return-callback")
+def return_callback(
+    body: dict[str, Any],
+    idempotency_key: str = Header(..., alias="Idempotency-Key"),
+    service_token: str = Header(..., alias="X-Service-Token"),
+    db: Session = Depends(get_db),
+) -> dict:
+    if service_token != settings.n8n_callback_token:
+        raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401)
+
+    try:
+        event_id = uuid.UUID(idempotency_key)
+    except ValueError as exc:
+        raise AppError(
+            "INVALID_IDEMPOTENCY_KEY", "Idempotency-Key must be the event's UUID.", status_code=422
+        ) from exc
+
+    event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id))
+    if event is None:
+        raise AppError("EVENT_NOT_FOUND", "No outbox event matches this event ID.", status_code=404)
+
+    # Idempotent by event ID: n8n or our own dispatcher may redeliver the same event
+    # (e.g. a lost response after a timeout), so this callback must not double-record.
+    already_recorded = (
+        db.scalar(
+            select(AuditEvent.id).where(
+                AuditEvent.action == "n8n_return_followup_recorded",
+                AuditEvent.metadata_json["event_id"].astext == str(event_id),
+            )
+        )
+        is not None
+    )
+    if not already_recorded:
+        record_audit_event(
+            db,
+            actor_type="service",
+            actor_label="n8n",
+            action="n8n_return_followup_recorded",
+            entity_type="booking",
+            correlation_id=uuid.UUID(body.get("correlation_id"))
+            if body.get("correlation_id")
+            else None,
+            after={"follow_up": body.get("follow_up"), "summary": body.get("summary")},
+            metadata={"event_id": str(event_id)},
+        )
+        db.commit()
+
+    return {
+        "status": "recorded",
+        "event_id": str(event_id),
+        "occurred_at": datetime.now(UTC).isoformat(),
+    }

backend/app/api/routers/workflows.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db, require_operations_manager
+from app.core.errors import AppError
+from app.models.outbox import OutboxEvent
+from app.schemas import AutomationRunOut, CurrentUser
+from app.services.audit import record_audit_event
+
+router = APIRouter(prefix="/api/v1/workflows", tags=["workflows"])
+
+
+def _to_out(event: OutboxEvent) -> AutomationRunOut:
+    return AutomationRunOut(
+        event_id=str(event.event_id),
+        event_type=event.event_type,
+        aggregate_ref=event.payload_json.get("aggregate_ref", ""),
+        status=event.delivery_status,
+        attempts=event.attempts,
+        last_error=event.last_error,
+        occurred_at=event.occurred_at,
+    )
+
+
+@router.get("", response_model=list[AutomationRunOut])
+def list_workflows(
+    status: str | None = Query(default=None),
+    db: Session = Depends(get_db),
+    _user: CurrentUser = Depends(require_operations_manager),
+) -> list[AutomationRunOut]:
+    stmt = select(OutboxEvent).order_by(OutboxEvent.occurred_at.desc())
+    if status:
+        stmt = stmt.where(OutboxEvent.delivery_status == status)
+    events = db.scalars(stmt).all()
+    return [_to_out(e) for e in events]
+
+
+@router.post("/{event_id}/retry", response_model=AutomationRunOut)
+def retry_workflow(
+    event_id: str,
+    db: Session = Depends(get_db),
+    user: CurrentUser = Depends(require_operations_manager),
+) -> AutomationRunOut:
+    try:
+        parsed_id = uuid.UUID(event_id)
+    except ValueError as exc:
+        raise AppError("INVALID_EVENT_ID", "event_id must be a UUID.", status_code=422) from exc
+
+    event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == parsed_id))
+    if event is None:
+        raise AppError("EVENT_NOT_FOUND", "Workflow event not found.", status_code=404)
+    if event.delivery_status != "failed":
+        raise AppError(
+            "NOT_RETRYABLE",
+            f"Event is '{event.delivery_status}', not 'failed'; "
+            "only failed deliveries can be retried.",
+            status_code=409,
+        )
+
+    event.delivery_status = "pending"
+    event.next_attempt_at = None
+    record_audit_event(
+        db,
+        actor_type="user",
+        actor_label=user.display_name,
+        action="workflow_retry",
+        entity_type="outbox_event",
+        metadata={"event_id": event_id, "previous_attempts": event.attempts},
+    )
+    db.commit()
+    return _to_out(event)

backend/app/core/config.py

@@ -15,6 +15,11 @@ class Settings(BaseSettings):
     ragcore_workspace: str = "mobilityops"
     ragcore_collection: str = "internal-procedures"
     n8n_webhook_url: str = "http://n8n:5678/webhook/mobilityops-return"
+    n8n_callback_token: str = "replace-me-n8n-callback-token"
+    n8n_dispatch_enabled: bool = True
+    n8n_dispatch_interval_seconds: float = 3.0
+    n8n_http_timeout_seconds: float = 5.0
+    n8n_max_attempts: int = 5
     app_secret: str = "replace-in-production"
     session_cookie_name: str = "mobilityops_session"
     session_ttl_seconds: int = 60 * 60 * 8

backend/app/main.py

@@ -1,15 +1,35 @@
 import uuid
+from contextlib import asynccontextmanager
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
-from app.api.routers import audit, bookings, dashboard, data_quality, demo, vehicles
+from app.api.routers import (
+    audit,
+    bookings,
+    dashboard,
+    data_quality,
+    demo,
+    integrations,
+    vehicles,
+    workflows,
+)
 from app.core.config import get_settings
 from app.core.errors import AppError, error_body
+from app.services.dispatcher import start_background_dispatcher, stop_background_dispatcher
 
 settings = get_settings()
-app = FastAPI(title="MobilityOps API", version="0.1.0")
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    start_background_dispatcher()
+    yield
+    stop_background_dispatcher()
+
+
+app = FastAPI(title="MobilityOps API", version="0.1.0", lifespan=lifespan)
 
 app.add_middleware(
     CORSMiddleware,
@@ -62,3 +82,5 @@ app.include_router(vehicles.router)
 app.include_router(bookings.router)
 app.include_router(audit.router)
 app.include_router(data_quality.router)
+app.include_router(workflows.router)
+app.include_router(integrations.router)

backend/app/seed_loader.py

@@ -235,16 +235,37 @@ def load_seed(db: Session) -> SeedResult:
     db.execute(insert(DataQualityIssue), dq_rows)
     counts["data_quality_issues"] = len(dq_rows)
 
+    vehicle_ref_by_booking_ref = {
+        row["public_ref"]: row["vehicle_ref"] for row in _read_csv("bookings.csv")
+    }
+
     outbox_rows = []
     for row in _read_csv("workflow_runs.csv"):
         booking_id = booking_id_by_ref.get(row["aggregate_ref"])
+        # Build the same schema-complete envelope the live return workflow (M2) produces,
+        # so a seeded/historical event is redeliverable (e.g. via manual retry) without the
+        # dispatcher crashing on a missing key. See PROJECT_STATE.md M4 notes.
         outbox_rows.append(
             {
                 "event_id": uuid.UUID(row["event_id"]),
                 "event_type": row["event_type"],
                 "aggregate_type": "booking",
                 "aggregate_id": booking_id or uuid.uuid4(),
-                "payload_json": {"aggregate_ref": row["aggregate_ref"]},
+                "payload_json": {
+                    "correlation_id": str(uuid.uuid4()),
+                    "aggregate": {
+                        "type": "booking",
+                        "id": str(booking_id or uuid.uuid4()),
+                        "public_ref": row["aggregate_ref"],
+                    },
+                    "data": {
+                        "vehicle_ref": vehicle_ref_by_booking_ref.get(row["aggregate_ref"], ""),
+                        "inspection_ref": "",
+                        "resulting_vehicle_status": "cleaning",
+                        "attention_reasons": [],
+                    },
+                    "aggregate_ref": row["aggregate_ref"],
+                },
                 "occurred_at": _parse_dt(row["occurred_at"]),
                 "delivery_status": row["status"],
                 "attempts": int(row["attempts"]),

backend/app/services/dispatcher.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+import logging
+import threading
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import httpx
+from sqlalchemy import select
+
+from app.core.config import get_settings
+from app.core.db import SessionLocal
+from app.models.outbox import OutboxEvent
+
+logger = logging.getLogger("mobilityops.dispatcher")
+settings = get_settings()
+
+_stop_event = threading.Event()
+
+
+def _backoff_seconds(attempts: int) -> int:
+    return min(2**attempts, 60)
+
+
+def _claim_due_events(batch_size: int = 5) -> list[uuid.UUID]:
+    """Claim a batch of due events with a short-lived transaction (no network I/O held open)."""
+    db = SessionLocal()
+    try:
+        now = datetime.now(UTC)
+        rows = db.scalars(
+            select(OutboxEvent)
+            .where(
+                OutboxEvent.delivery_status == "pending",
+                (OutboxEvent.next_attempt_at.is_(None)) | (OutboxEvent.next_attempt_at <= now),
+            )
+            .order_by(OutboxEvent.occurred_at.asc())
+            .limit(batch_size)
+            .with_for_update(skip_locked=True)
+        ).all()
+        claimed_ids = [row.event_id for row in rows]
+        for row in rows:
+            row.delivery_status = "delivering"
+        db.commit()
+        return claimed_ids
+    finally:
+        db.close()
+
+
+def _deliver_one(event_id: uuid.UUID) -> None:
+    db = SessionLocal()
+    try:
+        event = db.get(OutboxEvent, event_id)
+        if event is None:
+            return
+        # Reconstruct the wire envelope from contracts/events.schema.json: only the fields
+        # the schema declares (additionalProperties: false), sourced from real columns where
+        # possible. `payload_json` also carries an internal `aggregate_ref` convenience field
+        # for our own dashboard/audit reads, which must not be forwarded to n8n.
+        try:
+            wire_event = {
+                "event_id": str(event.event_id),
+                "event_type": event.event_type,
+                "occurred_at": event.occurred_at.isoformat(),
+                "correlation_id": event.payload_json["correlation_id"],
+                "aggregate": event.payload_json["aggregate"],
+                "data": event.payload_json["data"],
+            }
+            payload_error: str | None = None
+        except KeyError as exc:
+            # A malformed payload must still resolve the claimed "delivering" row to a
+            # terminal-or-retryable state below, rather than leaving it stuck forever.
+            wire_event = None
+            payload_error = f"Malformed outbox payload, missing key {exc}"
+        attempts = event.attempts
+    finally:
+        db.close()
+
+    if wire_event is None:
+        success, error, body = False, payload_error, None
+    else:
+        try:
+            response = httpx.post(
+                settings.n8n_webhook_url,
+                json=wire_event,
+                timeout=settings.n8n_http_timeout_seconds,
+            )
+            response.raise_for_status()
+            body = response.json()
+            success = bool(body.get("ok", True))
+            error = None if success else f"n8n reported failure: {body}"
+        except httpx.HTTPError as exc:
+            success = False
+            error = f"{type(exc).__name__}: {exc}"
+            body = None
+
+    db = SessionLocal()
+    try:
+        event = db.get(OutboxEvent, event_id)
+        if event is None:
+            return
+        event.attempts = attempts + 1
+        if success:
+            event.delivery_status = "succeeded"
+            event.last_error = None
+            event.next_attempt_at = None
+            event.external_run_id = str((body or {}).get("event_id", event_id))
+        else:
+            event.last_error = (error or "delivery failed")[:2000]
+            if event.attempts >= settings.n8n_max_attempts:
+                event.delivery_status = "failed"
+                event.next_attempt_at = None
+            else:
+                event.delivery_status = "pending"
+                event.next_attempt_at = datetime.now(UTC) + timedelta(
+                    seconds=_backoff_seconds(event.attempts)
+                )
+        db.commit()
+    finally:
+        db.close()
+
+
+def run_dispatch_cycle() -> int:
+    """Run one claim+deliver cycle. Returns the number of events processed."""
+    claimed = _claim_due_events()
+    for event_id in claimed:
+        _deliver_one(event_id)
+    return len(claimed)
+
+
+def _loop() -> None:
+    while not _stop_event.is_set():
+        try:
+            run_dispatch_cycle()
+        except Exception:  # noqa: BLE001
+            logger.exception("Outbox dispatch cycle failed")
+        _stop_event.wait(settings.n8n_dispatch_interval_seconds)
+
+
+def start_background_dispatcher() -> None:
+    if not settings.n8n_dispatch_enabled:
+        return
+    _stop_event.clear()
+    thread = threading.Thread(target=_loop, name="outbox-dispatcher", daemon=True)
+    thread.start()
+
+
+def stop_background_dispatcher() -> None:
+    _stop_event.set()