M54: harden operations and demo resilience

backend/app/api/routers/bookings.py

@@ -30,6 +30,7 @@ from app.schemas import (
     ReturnPreviewResult,
 )
 from app.services.audit import record_audit_event
+from app.services.data_quality import open_odometer_regression_issue
 from app.services.returns import preview_vehicle_return, register_vehicle_return
 
 router = APIRouter(prefix="/api/v1/bookings", tags=["bookings"])
@@ -146,8 +147,14 @@ def create_booking(
 ) -> BookingOut:
     if body.ends_at <= body.starts_at:
         raise HTTPException(status_code=422, detail="Booking end must be after its start")
-    customer = db.scalar(select(Customer).where(Customer.public_ref == body.customer_ref))
-    if customer is None or customer.merged_into_customer_id is not None:
+    customer = db.scalar(
+        select(Customer).where(Customer.public_ref == body.customer_ref).with_for_update()
+    )
+    if (
+        customer is None
+        or customer.merged_into_customer_id is not None
+        or customer.anonymized_at is not None
+    ):
         raise HTTPException(status_code=422, detail="Customer is unavailable for booking")
     # Serialise booking creation per vehicle. The overlap check must run after
     # acquiring this lock, otherwise two concurrent requests can both pass it.
@@ -179,7 +186,9 @@ def create_booking(
         status="reserved",
         start_odometer_km=None,
         end_odometer_km=None,
-        requirements_complete=body.requirements_complete,
+        # Requirements are intentionally confirmed in a separate, audited action.
+        # Never allow booking creation to bypass that operational checkpoint.
+        requirements_complete=False,
     )
     db.add(booking)
     db.flush()
@@ -225,6 +234,15 @@ def checkout_booking(
     if active_conflict is not None:
         raise HTTPException(status_code=409, detail="Vehicle already has an active booking")
 
+    correlation_id = uuid.uuid4()
+    before_booking = {
+        "status": booking.status,
+        "start_odometer_km": booking.start_odometer_km,
+    }
+    before_vehicle = {
+        "operational_status": vehicle.operational_status,
+        "odometer_km": vehicle.odometer_km,
+    }
     attention_reasons: list[str] = []
     if body.start_odometer_km < vehicle.odometer_km:
         attention_reasons.append("odometer_regression")
@@ -250,6 +268,18 @@ def checkout_booking(
         completed_by=user.display_name,
     )
     db.add(inspection)
+    if "odometer_regression" in attention_reasons:
+        open_odometer_regression_issue(
+            db,
+            vehicle=vehicle,
+            reading_ref=inspection.public_ref,
+            reading_km=body.start_odometer_km,
+            canonical_km=vehicle.odometer_km,
+            source_type="checkout",
+            related_refs=[booking.public_ref, inspection.public_ref],
+            actor_label=user.display_name,
+            correlation_id=correlation_id,
+        )
     if attention_reasons:
         booking.status = "blocked"
         vehicle.operational_status = (
@@ -269,13 +299,31 @@ def checkout_booking(
         action="booking_checkout_recorded",
         entity_type="booking",
         entity_id=booking.id,
+        correlation_id=correlation_id,
+        before=before_booking,
         after={
             "inspection_ref": inspection.public_ref,
             "booking_status": booking.status,
+            "start_odometer_km": booking.start_odometer_km,
             "vehicle_status": vehicle.operational_status,
             "attention_reasons": attention_reasons,
         },
     )
+    record_audit_event(
+        db,
+        actor_type="user",
+        actor_label=user.display_name,
+        action="vehicle_status_changed",
+        entity_type="vehicle",
+        entity_id=vehicle.id,
+        correlation_id=correlation_id,
+        before=before_vehicle,
+        after={
+            "operational_status": vehicle.operational_status,
+            "odometer_km": vehicle.odometer_km,
+        },
+        metadata={"booking_ref": booking.public_ref, "inspection_ref": inspection.public_ref},
+    )
     db.commit()
     return CheckoutBookingResult(
         booking_ref=booking.public_ref,

backend/app/api/routers/customers.py

@@ -21,6 +21,7 @@ def search_customers(
         select(Customer)
         .where(
             Customer.merged_into_customer_id.is_(None),
+            Customer.anonymized_at.is_(None),
             or_(
                 Customer.public_ref.ilike(term),
                 Customer.first_name.ilike(term),

backend/app/api/routers/data_quality.py

@@ -11,6 +11,7 @@ from app.models.booking import Booking
 from app.models.customer import Customer
 from app.models.data_quality import DataQualityIssue
 from app.models.inspection import Inspection
+from app.models.maintenance import MaintenanceRecord
 from app.models.user import User
 from app.models.vehicle import Vehicle
 from app.schemas import (
@@ -224,6 +225,8 @@ _PREFIX_TO_TYPE = {
     "MO-": "vehicle",
     "BK-": "booking",
     "INSP-": "inspection",
+    "MAINT-": "maintenance",
+    "MNT-": "maintenance",
 }
 
 
@@ -292,6 +295,17 @@ def _snapshot(entity_type: str, ref: str, db: Session) -> dict | None:
             "completed_at": inspection.completed_at.isoformat(),
             "booking_ref": booking.public_ref if booking else None,
         }
+    if entity_type == "maintenance":
+        record = db.scalar(select(MaintenanceRecord).where(MaintenanceRecord.public_ref == ref))
+        if record is None:
+            return None
+        return {
+            "entity_type": "maintenance",
+            "public_ref": record.public_ref,
+            "odometer_km": record.odometer_km,
+            "occurred_at": record.occurred_at.isoformat(),
+            "category": record.category,
+        }
     return None
 
 

backend/app/api/routers/demo.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import threading
 import time
 import uuid
+from datetime import UTC, datetime
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 from sqlalchemy import func, select
@@ -10,7 +11,9 @@ from sqlalchemy.orm import Session
 
 from app.api.deps import get_current_user, get_db, require_operations_manager
 from app.core.config import get_settings
+from app.core.db import begin_exclusive_demo_reset, end_exclusive_demo_reset, engine
 from app.core.security import SessionPayload, create_session_token, read_session_token
+from app.models.audit import AuditEvent
 from app.models.user import User
 from app.schemas import CurrentUser, DemoLoginRequest, DemoManifestOut
 from app.seed_loader import reset_and_seed
@@ -21,7 +24,6 @@ from app.services.sessions import revoke_session
 router = APIRouter(prefix="/api/v1/demo", tags=["demo"])
 settings = get_settings()
 _reset_guard = threading.Lock()
-_last_reset_monotonic = 0.0
 _RESET_ADVISORY_LOCK_ID = 706_533_149
 
 
@@ -110,7 +112,6 @@ def demo_reset(
     db: Session = Depends(get_db),
     user: CurrentUser = Depends(require_operations_manager),
 ) -> dict:
-    global _last_reset_monotonic
     if not settings.mobilityops_demo_mode:
         # Outside demo mode the reset endpoint must not exist at all: it wipes
         # operational data and replaces it with synthetic records.
@@ -122,19 +123,35 @@ def demo_reset(
         )
     if not _reset_guard.acquire(blocking=False):
         raise HTTPException(status_code=409, detail="A demo reset is already running.")
+    replica_lock_connection = None
     try:
-        elapsed = time.monotonic() - _last_reset_monotonic
-        if _last_reset_monotonic and elapsed < settings.demo_reset_cooldown_seconds:
+        # A session-level lock on its own connection rejects another replica immediately;
+        # the main DB session can then safely end its auth read transaction and wait on
+        # the normal shared/exclusive data barrier without releasing this replica guard.
+        replica_lock_connection = engine.connect()
+        locked = replica_lock_connection.scalar(
+            select(func.pg_try_advisory_lock(_RESET_ADVISORY_LOCK_ID))
+        )
+        if not locked:
+            raise HTTPException(status_code=409, detail="A demo reset is already running.")
+        begin_exclusive_demo_reset(db)
+        # The audit timestamp is shared by every replica. A process-local monotonic
+        # timestamp cannot protect a multi-replica deployment.
+        last_reset_at = db.scalar(
+            select(AuditEvent.occurred_at)
+            .where(AuditEvent.action == "demo_reset")
+            .order_by(AuditEvent.occurred_at.desc())
+            .limit(1)
+        )
+        elapsed = (datetime.now(UTC) - last_reset_at).total_seconds() if last_reset_at else None
+        if elapsed is not None and elapsed < settings.demo_reset_cooldown_seconds:
             retry_after = max(1, int(settings.demo_reset_cooldown_seconds - elapsed + 0.999))
             raise HTTPException(
                 status_code=429,
                 detail=f"Demo reset is cooling down. Retry in {retry_after} seconds.",
                 headers={"Retry-After": str(retry_after)},
             )
-        locked = db.scalar(select(func.pg_try_advisory_xact_lock(_RESET_ADVISORY_LOCK_ID)))
-        if not locked:
-            raise HTTPException(status_code=409, detail="A demo reset is already running.")
-        result = reset_and_seed(db, preserve_integration_telemetry=True)
+        result = reset_and_seed(db, preserve_integration_telemetry=True, commit=False)
         integrity = scenario_integrity_report(db)
         record_audit_event(
             db,
@@ -149,8 +166,15 @@ def demo_reset(
             },
         )
         db.commit()
-        _last_reset_monotonic = time.monotonic()
+        end_exclusive_demo_reset(db)
     finally:
+        if replica_lock_connection is not None:
+            try:
+                replica_lock_connection.scalar(
+                    select(func.pg_advisory_unlock(_RESET_ADVISORY_LOCK_ID))
+                )
+            finally:
+                replica_lock_connection.close()
         _reset_guard.release()
     response.delete_cookie(settings.session_cookie_name)
     return {

backend/app/api/routers/integrations.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
+import hashlib
 import hmac
 import uuid
 from datetime import UTC, datetime
 from pathlib import Path
 
 from fastapi import APIRouter, Depends, Header
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.api.deps import get_db
@@ -51,6 +52,13 @@ def _require_service_token(service_token: str) -> None:
         raise AppError("UNAUTHORIZED_SERVICE", "Invalid service token.", status_code=401)
 
 
+def _lock_idempotency_key(db: Session, namespace: str, key: str) -> None:
+    """Serialize callback check+insert by a stable, transaction-scoped key."""
+    digest = hashlib.sha256(f"{namespace}:{key}".encode()).digest()
+    lock_id = int.from_bytes(digest[:8], byteorder="big", signed=True)
+    db.scalar(select(func.pg_advisory_xact_lock(lock_id)))
+
+
 @router.post("/heartbeat", response_model=N8nHeartbeatResult)
 def workflow_heartbeat(
     body: N8nHeartbeatIn,
@@ -61,6 +69,7 @@ def workflow_heartbeat(
     _require_service_token(service_token)
     if body.workflow_name not in _CANONICAL_WORKFLOW_NAMES:
         raise AppError("UNKNOWN_WORKFLOW", "Unknown Fleet Ops workflow.", status_code=422)
+    _lock_idempotency_key(db, "n8n_workflow_heartbeat", f"{body.execution_id}:{body.status}")
     already_recorded = (
         db.scalar(
             select(AuditEvent.id).where(
@@ -109,9 +118,29 @@ def return_callback(
             "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))
+    event = db.scalar(select(OutboxEvent).where(OutboxEvent.event_id == event_id).with_for_update())
     if event is None:
         raise AppError("EVENT_NOT_FOUND", "No outbox event matches this event ID.", status_code=404)
+    if body.event_id != event_id:
+        raise AppError(
+            "CALLBACK_EVENT_MISMATCH",
+            "Callback event_id does not match Idempotency-Key.",
+            status_code=409,
+        )
+    try:
+        expected_correlation_id = uuid.UUID(str(event.payload_json["correlation_id"]))
+    except (KeyError, TypeError, ValueError) as exc:
+        raise AppError(
+            "INVALID_EVENT_CORRELATION",
+            "The stored outbox event has no valid correlation ID.",
+            status_code=409,
+        ) from exc
+    if body.correlation_id != expected_correlation_id:
+        raise AppError(
+            "CALLBACK_CORRELATION_MISMATCH",
+            "Callback correlation_id does not match the outbox event.",
+            status_code=409,
+        )
 
     # 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.
@@ -131,7 +160,7 @@ def return_callback(
             actor_label="n8n",
             action="n8n_return_followup_recorded",
             entity_type="booking",
-            correlation_id=body.correlation_id,
+            correlation_id=expected_correlation_id,
             after={"follow_up": body.follow_up, "summary": body.summary},
             metadata={"event_id": str(event_id)},
         )
@@ -170,6 +199,7 @@ def workflow_error(
     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."""
     _require_service_token(service_token)
+    _lock_idempotency_key(db, "n8n_workflow_failure", body.execution_id)
 
     already_recorded = (
         db.scalar(
@@ -181,19 +211,13 @@ def workflow_error(
         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,
+            correlation_id=body.correlation_id,
             after={
                 "workflow_id": body.workflow_id,
                 "workflow_name": body.workflow_name,
@@ -248,6 +272,7 @@ def procedures_sync_result(
     RAGcore Procedure Sync" workflow once it finishes uploading procedures to RAGcore.
     Idempotent on execution_id, matching the workflow-error and return-callback pattern."""
     _require_service_token(service_token)
+    _lock_idempotency_key(db, "n8n_procedure_sync", body.execution_id)
 
     already_recorded = (
         db.scalar(

backend/app/api/routers/knowledge.py

@@ -53,7 +53,9 @@ def ask_question(
         client_ip = (
             forwarded.split(",")[-1].strip()
             if forwarded
-            else request.client.host if request.client else "unknown"
+            else request.client.host
+            if request.client
+            else "unknown"
         )
         token = request.cookies.get(settings.session_cookie_name, "")
         session_key = hashlib.sha256(token.encode("utf-8")).hexdigest()

backend/app/api/routers/vehicles.py

@@ -27,6 +27,7 @@ from app.schemas import (
     VehiclePageOut,
 )
 from app.services.audit import record_audit_event
+from app.services.data_quality import open_odometer_regression_issue
 
 router = APIRouter(prefix="/api/v1/vehicles", tags=["vehicles"])
 
@@ -143,7 +144,7 @@ def list_vehicles(
 def get_vehicle(
     public_ref: str,
     db: Session = Depends(get_db),
-    _user: CurrentUser = Depends(get_current_user),
+    user: CurrentUser = Depends(get_current_user),
 ) -> VehicleDetailOut:
     vehicle = db.scalar(select(Vehicle).where(Vehicle.public_ref == public_ref))
     if vehicle is None:
@@ -163,11 +164,21 @@ def get_vehicle(
         .where(MaintenanceRecord.vehicle_id == vehicle.id)
         .order_by(MaintenanceRecord.occurred_at.desc())
     ).all()
-    issues = db.scalars(
-        select(DataQualityIssue)
-        .where(DataQualityIssue.entity_type == "vehicle", DataQualityIssue.entity_id == vehicle.id)
-        .order_by(DataQualityIssue.detected_at.desc())
-    ).all()
+    # Detailed data-quality evidence is an operations-manager surface. Employees still
+    # get the operational vehicle record they need, but never receive hidden evidence in
+    # the payload merely because the frontend omits the Quality tab.
+    issues = (
+        db.scalars(
+            select(DataQualityIssue)
+            .where(
+                DataQualityIssue.entity_type == "vehicle",
+                DataQualityIssue.entity_id == vehicle.id,
+            )
+            .order_by(DataQualityIssue.detected_at.desc())
+        ).all()
+        if user.role == "operations_manager"
+        else []
+    )
 
     booking_by_id = {b.id: b.public_ref for b in bookings}
     next_booking = next(
@@ -276,6 +287,12 @@ def create_maintenance_record(
     vehicle = db.scalar(select(Vehicle).where(Vehicle.public_ref == public_ref).with_for_update())
     if vehicle is None:
         raise HTTPException(status_code=404, detail="Vehicle not found")
+    correlation_id = uuid.uuid4()
+    before_vehicle = {
+        "operational_status": vehicle.operational_status,
+        "odometer_km": vehicle.odometer_km,
+        "next_service_km": vehicle.next_service_km,
+    }
     record = MaintenanceRecord(
         public_ref=f"MAINT-{uuid.uuid4().hex[:8].upper()}",
         vehicle_id=vehicle.id,
@@ -285,6 +302,17 @@ def create_maintenance_record(
         summary=body.summary.strip(),
     )
     db.add(record)
+    open_odometer_regression_issue(
+        db,
+        vehicle=vehicle,
+        reading_ref=record.public_ref,
+        reading_km=body.odometer_km,
+        canonical_km=vehicle.odometer_km,
+        source_type="maintenance",
+        related_refs=[record.public_ref],
+        actor_label=user.display_name,
+        correlation_id=correlation_id,
+    )
     vehicle.odometer_km = max(vehicle.odometer_km, body.odometer_km)
     if body.next_service_km is not None:
         if body.next_service_km < vehicle.odometer_km:
@@ -301,7 +329,14 @@ def create_maintenance_record(
         action="maintenance_record_created",
         entity_type="vehicle",
         entity_id=vehicle.id,
-        after={"maintenance_ref": record.public_ref, "status": vehicle.operational_status},
+        correlation_id=correlation_id,
+        before=before_vehicle,
+        after={
+            "maintenance_ref": record.public_ref,
+            "operational_status": vehicle.operational_status,
+            "odometer_km": vehicle.odometer_km,
+            "next_service_km": vehicle.next_service_km,
+        },
     )
     db.commit()
     return MaintenanceOut(

backend/app/core/config.py

@@ -33,7 +33,11 @@ class Settings(BaseSettings):
     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
+    # The synchronous n8n workflow performs two bounded, retried callbacks before it
+    # acknowledges an event. Keep this above that complete workflow budget, while the
+    # delivery lease remains the wider crash-recovery boundary (enforced by the contract
+    # check in scripts/check-contracts.py).
+    n8n_http_timeout_seconds: float = 15.0
     n8n_max_attempts: int = 5
     n8n_delivery_lease_seconds: float = 120.0
     app_secret: str = "replace-in-production"

backend/app/core/db.py

@@ -1,6 +1,6 @@
 from collections.abc import Generator
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, event, func, select
 from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
 
 from app.core.config import get_settings
@@ -10,6 +10,37 @@ settings = get_settings()
 engine = create_engine(settings.database_url, pool_pre_ping=True, future=True)
 SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
 
+# Every SQLAlchemy transaction participates in a shared database-wide barrier. Normal
+# reads/writes coexist; the short demo reset takes the exclusive form so it can never
+# interleave deletes/inserts with an API request, scanner, callback, or dispatcher cycle.
+DEMO_DATA_BARRIER_LOCK_ID = 5_344_725_149_212_793_901
+_EXCLUSIVE_RESET_INFO_KEY = "mobilityops_demo_reset_exclusive"
+
+
+@event.listens_for(Session, "after_begin")
+def _acquire_demo_data_barrier(session: Session, _transaction, connection) -> None:
+    if connection.dialect.name != "postgresql":
+        return
+    lock = (
+        func.pg_advisory_xact_lock(DEMO_DATA_BARRIER_LOCK_ID)
+        if session.info.get(_EXCLUSIVE_RESET_INFO_KEY)
+        else func.pg_advisory_xact_lock_shared(DEMO_DATA_BARRIER_LOCK_ID)
+    )
+    connection.execute(select(lock))
+
+
+def begin_exclusive_demo_reset(db: Session) -> None:
+    """Make the session's next transaction the exclusive side of the reset barrier."""
+    if db.in_transaction():
+        # Auth normally already read the user under a shared barrier. End that read-only
+        # transaction before requesting exclusive; in-place lock upgrades can deadlock.
+        db.rollback()
+    db.info[_EXCLUSIVE_RESET_INFO_KEY] = True
+
+
+def end_exclusive_demo_reset(db: Session) -> None:
+    db.info.pop(_EXCLUSIVE_RESET_INFO_KEY, None)
+
 
 class Base(DeclarativeBase):
     pass

backend/app/schemas.py

@@ -4,9 +4,13 @@ import uuid
 from datetime import datetime
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, StringConstraints
 
 Role = Literal["operations_manager", "rental_employee"]
+NonBlankOperationalReason = Annotated[
+    str,
+    StringConstraints(strip_whitespace=True, min_length=3, max_length=500),
+]
 
 
 class DemoLoginRequest(BaseModel):
@@ -93,17 +97,16 @@ class CreateBookingRequest(BaseModel):
     vehicle_ref: str = Field(min_length=3, max_length=20)
     starts_at: datetime
     ends_at: datetime
-    requirements_complete: bool = False
 
 
 class CompleteBookingRequirementsRequest(BaseModel):
-    confirmation: str = Field(min_length=3, max_length=500)
+    confirmation: NonBlankOperationalReason
 
 
 class RescheduleBookingRequest(BaseModel):
     starts_at: datetime
     ends_at: datetime
-    reason: str = Field(min_length=3, max_length=500)
+    reason: NonBlankOperationalReason
 
 
 class CustomerOptionOut(BaseModel):
@@ -122,7 +125,7 @@ class AvailableVehicleOut(BaseModel):
 
 
 class CancelBookingRequest(BaseModel):
-    reason: str = Field(min_length=3, max_length=500)
+    reason: NonBlankOperationalReason
 
 
 class CheckoutBookingRequest(BaseModel):
@@ -225,7 +228,7 @@ class CreateMaintenanceRequest(BaseModel):
 
 
 class ReleaseVehicleRequest(BaseModel):
-    reason: str = Field(min_length=3, max_length=500)
+    reason: NonBlankOperationalReason
 
 
 class DataQualityIssueOut(BaseModel):
@@ -295,7 +298,7 @@ class WorkflowErrorReportIn(BaseModel):
     ]
     error_summary: str = Field(max_length=500)
     trigger_context: str | None = Field(default=None, max_length=200)
-    correlation_id: str | None = None
+    correlation_id: uuid.UUID | None = None
     attempt: int = Field(default=1, ge=1, le=1000)
     retry_action: str | None = Field(default=None, max_length=200)
 
@@ -433,7 +436,8 @@ class ReturnCallbackIn(BaseModel):
 
     model_config = ConfigDict(extra="ignore")
 
-    correlation_id: uuid.UUID | None = None
+    event_id: uuid.UUID
+    correlation_id: uuid.UUID
     follow_up: str | None = Field(default=None, max_length=200)
     summary: str | None = Field(default=None, max_length=2000)
 

backend/app/seed_loader.py

@@ -11,6 +11,7 @@ from sqlalchemy import delete, insert, update
 from sqlalchemy.orm import Session
 
 from app.core.config import get_settings
+from app.core.db import begin_exclusive_demo_reset, end_exclusive_demo_reset
 from app.models.audit import AuditEvent
 from app.models.booking import Booking
 from app.models.customer import Customer
@@ -201,7 +202,11 @@ def load_seed(db: Session) -> SeedResult:
     counts["bookings"] = len(booking_rows)
 
     inspection_rows = []
-    for row in _read_csv("inspections.csv"):
+    inspection_source_rows = _read_csv("inspections.csv")
+    return_inspection_row_by_booking_ref = {
+        row["booking_ref"]: row for row in inspection_source_rows if row["type"] == "return"
+    }
+    for row in inspection_source_rows:
         inspection_rows.append(
             {
                 "id": uuid.uuid4(),
@@ -259,16 +264,18 @@ def load_seed(db: Session) -> SeedResult:
         }
 
     def _odometer_regression_signal(later_ref: str, earlier_ref: str) -> list[dict]:
-        later = booking_row_by_ref[later_ref]
-        earlier = booking_row_by_ref[earlier_ref]
+        later = return_inspection_row_by_booking_ref[later_ref]
+        earlier = return_inspection_row_by_booking_ref[earlier_ref]
         return [
             {
                 "code": "odometer.regression",
+                "source_type": "return",
                 "params": {
-                    "later_ref": later_ref,
-                    "later_km": later["end_odometer_km"],
-                    "earlier_ref": earlier_ref,
-                    "earlier_km": earlier["end_odometer_km"],
+                    "later_ref": later["public_ref"],
+                    "later_km": int(later["odometer_km"]),
+                    "earlier_ref": earlier["public_ref"],
+                    "earlier_km": int(earlier["odometer_km"]),
+                    "booking_ref": later_ref,
                 },
             }
         ]
@@ -354,6 +361,27 @@ def load_seed(db: Session) -> SeedResult:
         entity_type, entity_id = resolve_entity(row["entity_ref"])
         related_ref = row.get("related_ref") or ""
         related_refs = related_ref.split("|") if related_ref else []
+        signals = _seed_signals(row["public_ref"], row["entity_ref"], related_refs)
+        evidence_extra: dict[str, object] = {}
+        if row["rule_type"] == "odometer_regression" and signals:
+            params = signals[0].get("params", {})
+            later_ref = params.get("later_ref")
+            earlier_ref = params.get("earlier_ref")
+            booking_ref = params.get("booking_ref")
+            if (
+                isinstance(earlier_ref, str)
+                and isinstance(later_ref, str)
+                and isinstance(booking_ref, str)
+            ):
+                # The authored CSV prose predates structured references. Preserve the
+                # two real source bookings so the detail page can show evidence, while
+                # only the later (regressing) reading is eligible for correction.
+                related_refs = [earlier_ref, booking_ref, later_ref]
+                evidence_extra = {
+                    "source_type": signals[0].get("source_type", "return"),
+                    "source_types": [signals[0].get("source_type", "return")],
+                    "correctable_booking_refs": [booking_ref],
+                }
         severity_due_delta = {
             "high": timedelta(hours=4),
             "medium": timedelta(days=1),
@@ -372,7 +400,8 @@ def load_seed(db: Session) -> SeedResult:
                     "summary": row["evidence"],
                     "entity_ref": row["entity_ref"],
                     "related_refs": related_refs,
-                    "signals": _seed_signals(row["public_ref"], row["entity_ref"], related_refs),
+                    "signals": signals,
+                    **evidence_extra,
                 },
                 "proposed_action_json": {},
                 "detected_at": now,
@@ -449,12 +478,23 @@ def load_seed(db: Session) -> SeedResult:
     return SeedResult(counts=counts, anchor_date=today, seeded_at=seeded_at)
 
 
-def reset_and_seed(db: Session, *, preserve_integration_telemetry: bool = False) -> SeedResult:
+def reset_and_seed(
+    db: Session,
+    *,
+    preserve_integration_telemetry: bool = False,
+    commit: bool = True,
+) -> SeedResult:
     from app.services.data_quality import run_scan
 
+    if not db.info.get("mobilityops_demo_reset_exclusive"):
+        begin_exclusive_demo_reset(db)
     clear_all(db, preserve_integration_telemetry=preserve_integration_telemetry)
     result = load_seed(db)
-    db.commit()
-    scan = run_scan(db)
+    scan = run_scan(db, commit=False)
     result.counts["data_quality_issues"] += sum(scan.created.values())
+    if commit:
+        db.commit()
+        end_exclusive_demo_reset(db)
+    else:
+        db.flush()
     return result

backend/app/services/data_quality.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import uuid
 from dataclasses import dataclass, field
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime
 
 from sqlalchemy import func, select, update
 from sqlalchemy.orm import Session
@@ -11,10 +11,22 @@ from app.core.errors import AppError
 from app.models.booking import Booking
 from app.models.customer import Customer
 from app.models.data_quality import DataQualityIssue
+from app.models.inspection import Inspection
+from app.models.maintenance import MaintenanceRecord
 from app.models.vehicle import Vehicle
-from app.schemas import CurrentUser, ResolveOdometerRegressionRequest
+from app.schemas import CurrentUser
 from app.services.audit import record_audit_event
+from app.services.data_quality_common import has_open_issue as _has_open_issue
+from app.services.data_quality_common import issue_due_at
+from app.services.data_quality_common import load_open_issue as _load_open_issue
+from app.services.data_quality_common import new_scan_ref as _new_scan_ref
 from app.services.data_quality_duplicate_scan import scan_duplicate_customers
+from app.services.data_quality_odometer import (
+    open_odometer_regression_issue as open_odometer_regression_issue,
+)
+from app.services.data_quality_odometer import (
+    resolve_odometer_regression as resolve_odometer_regression,
+)
 from app.services.vehicle_status import (
     RECOMMENDATION_CODE_NO_CONFLICT,
     VehicleStatusRecommendation,
@@ -28,15 +40,6 @@ REQUIRED_VEHICLE_FIELDS = ("registration_number", "make", "model", "location")
 DATA_QUALITY_SCAN_LOCK_ID = 6_138_493_717_091_029_491
 
 
-def issue_due_at(detected_at: datetime, severity: str) -> datetime:
-    """Return the local operational SLA deadline for a newly detected issue."""
-    return detected_at + {
-        "high": timedelta(hours=4),
-        "medium": timedelta(days=1),
-        "low": timedelta(days=3),
-    }.get(severity, timedelta(days=1))
-
-
 @dataclass
 class ScanResult:
     created: dict[str, int] = field(default_factory=dict)
@@ -45,25 +48,6 @@ class ScanResult:
         self.created[rule_type] = self.created.get(rule_type, 0) + 1
 
 
-def _has_open_issue(db: Session, rule_type: str, entity_type: str, entity_id: uuid.UUID) -> bool:
-    return (
-        db.scalar(
-            select(DataQualityIssue.id).where(
-                DataQualityIssue.rule_type == rule_type,
-                DataQualityIssue.entity_type == entity_type,
-                DataQualityIssue.entity_id == entity_id,
-                DataQualityIssue.status == "open",
-            )
-        )
-        is not None
-    )
-
-
-def _new_scan_ref(prefix: str) -> str:
-    """Generate a stable human-readable prefix with a concurrent-safe suffix."""
-    return f"{prefix}-{uuid.uuid4().hex[:10].upper()}"
-
-
 def _open_issue(
     db: Session,
     scan: ScanResult,
@@ -76,6 +60,7 @@ def _open_issue(
     entity_ref: str,
     related_refs: list[str],
     signals: list[dict] | None = None,
+    evidence_extra: dict | None = None,
 ) -> None:
     if _has_open_issue(db, rule_type, entity_type, entity_id):
         return
@@ -103,6 +88,7 @@ def _open_issue(
         "related_refs": related_refs,
         "signals": signals or [],
     }
+    evidence.update(evidence_extra or {})
     if previous is not None:
         evidence["reopened_from"] = previous.public_ref
         evidence["previous_decision"] = previous.status
@@ -235,57 +221,110 @@ def _scan_vehicle_status_conflicts(db: Session, scan: ScanResult) -> None:
         )
 
 
-def _scan_odometer_regressions(db: Session, scan: ScanResult) -> None:
-    # The seed dataset's vehicle.odometer_km is generated independently of booking
-    # history, so comparing every historical booking against it produces near-universal
-    # false positives. Instead check the booking sequence's own internal consistency:
-    # each vehicle's completed bookings should show a non-decreasing odometer reading.
-    vehicles = {v.id: v for v in db.scalars(select(Vehicle)).all()}
-    bookings_by_vehicle: dict[uuid.UUID, list[Booking]] = {}
-    for booking in db.scalars(
+def _scan_odometer_regressions(
+    db: Session,
+    scan: ScanResult,
+    *,
+    actor_label: str | None,
+    actor_type: str,
+) -> None:
+    # Compare the chronological history with itself rather than every historical reading
+    # to today's canonical value. That detects imported checkout/return/maintenance
+    # regressions without flagging every legitimate older reading.
+    # Runtime checkout/return/maintenance mutations take the vehicle lock before they
+    # inspect or append DQ-03 evidence. Taking the same lock here makes scan-vs-command
+    # check/merge atomic and prevents a partial-unique race for the open issue.
+    vehicles = {
+        v.id: v for v in db.scalars(select(Vehicle).order_by(Vehicle.id).with_for_update()).all()
+    }
+    readings_by_vehicle: dict[uuid.UUID, list[tuple[datetime, str, int, str, str | None]]] = {}
+    inspections = db.scalars(select(Inspection)).all()
+    return_inspection_booking_ids = {
+        inspection.booking_id for inspection in inspections if inspection.type == "return"
+    }
+    returned_bookings = db.scalars(
         select(Booking).where(Booking.status == "returned", Booking.end_odometer_km.is_not(None))
-    ).all():
-        bookings_by_vehicle.setdefault(booking.vehicle_id, []).append(booking)
+    ).all()
+    booking_ref_by_id = {booking.id: booking.public_ref for booking in returned_bookings}
+    for booking in returned_bookings:
+        # A real return inspection owns the actual reading timestamp. Using the booking's
+        # planned ends_at as a duplicate second reading can make an early return appear to
+        # go backwards after a newer real inspection. Keep booking data only as the legacy
+        # import fallback when no return inspection exists.
+        if booking.id in return_inspection_booking_ids:
+            continue
+        assert booking.end_odometer_km is not None
+        readings_by_vehicle.setdefault(booking.vehicle_id, []).append(
+            (
+                booking.ends_at,
+                booking.public_ref,
+                booking.end_odometer_km,
+                "booking",
+                booking.public_ref,
+            )
+        )
+    for inspection in inspections:
+        readings_by_vehicle.setdefault(inspection.vehicle_id, []).append(
+            (
+                inspection.completed_at,
+                inspection.public_ref,
+                inspection.odometer_km,
+                inspection.type,
+                (
+                    booking_ref_by_id.get(inspection.booking_id)
+                    if inspection.type == "return"
+                    else None
+                ),
+            )
+        )
+    for record in db.scalars(select(MaintenanceRecord)).all():
+        readings_by_vehicle.setdefault(record.vehicle_id, []).append(
+            (record.occurred_at, record.public_ref, record.odometer_km, "maintenance", None)
+        )
 
-    for vehicle_id, bookings in bookings_by_vehicle.items():
-        bookings.sort(key=lambda b: b.ends_at)
-        for earlier, later in zip(bookings, bookings[1:], strict=False):
-            # The query above filters end_odometer_km IS NOT NULL, so both are ints here.
-            assert earlier.end_odometer_km is not None
-            assert later.end_odometer_km is not None
-            if later.end_odometer_km < earlier.end_odometer_km:
+    for vehicle_id, readings in readings_by_vehicle.items():
+        readings.sort(key=lambda reading: (reading[0], reading[1]))
+        highest = readings[0] if readings else None
+        for later in readings[1:]:
+            assert highest is not None
+            if later[2] < highest[2]:
                 vehicle = vehicles[vehicle_id]
-                _open_issue(
+                had_open_issue = _has_open_issue(
                     db,
-                    scan,
-                    rule_type="odometer_regression",
-                    entity_type="vehicle",
-                    entity_id=vehicle_id,
-                    severity="medium",
-                    summary=(
-                        f"Booking {later.public_ref} recorded {later.end_odometer_km} km, "
-                        f"below the {earlier.end_odometer_km} km recorded by earlier "
-                        f"booking {earlier.public_ref}."
-                    ),
-                    entity_ref=vehicle.public_ref,
-                    related_refs=[earlier.public_ref, later.public_ref],
-                    signals=[
-                        {
-                            "code": "odometer.regression",
-                            "params": {
-                                "later_ref": later.public_ref,
-                                "later_km": later.end_odometer_km,
-                                "earlier_ref": earlier.public_ref,
-                                "earlier_km": earlier.end_odometer_km,
-                            },
-                        }
-                    ],
+                    "odometer_regression",
+                    "vehicle",
+                    vehicle_id,
                 )
-                break
+                issue = open_odometer_regression_issue(
+                    db,
+                    vehicle=vehicle,
+                    reading_ref=later[1],
+                    reading_km=later[2],
+                    canonical_km=highest[2],
+                    canonical_ref=highest[1],
+                    source_type=later[3],
+                    related_refs=list(
+                        dict.fromkeys(
+                            [highest[1], later[1], *([later[4]] if later[4] is not None else [])]
+                        )
+                    ),
+                    correctable_booking_refs=[later[4]] if later[4] is not None else [],
+                    public_ref=_new_scan_ref("DQ-SCAN"),
+                    actor_label=actor_label,
+                    actor_type=actor_type,
+                )
+                if issue is not None and not had_open_issue:
+                    scan.bump("odometer_regression")
+            if later[2] > highest[2]:
+                highest = later
 
 
 def run_scan(
-    db: Session, *, actor_label: str | None = None, actor_type: str = "user"
+    db: Session,
+    *,
+    actor_label: str | None = None,
+    actor_type: str = "user",
+    commit: bool = True,
 ) -> ScanResult:
     # The check-then-insert work below spans several rules. Serialise whole scans at the
     # database boundary so API and n8n triggers cannot both observe an empty condition.
@@ -293,7 +332,12 @@ def run_scan(
     scan = ScanResult()
     scan_duplicate_customers(db, scan, _open_issue)
     _scan_missing_required_fields(db, scan)
-    _scan_odometer_regressions(db, scan)
+    _scan_odometer_regressions(
+        db,
+        scan,
+        actor_label=actor_label,
+        actor_type=actor_type,
+    )
     _scan_booking_overlaps(db, scan)
     _scan_vehicle_status_conflicts(db, scan)
     if actor_label is not None:
@@ -305,28 +349,13 @@ def run_scan(
             entity_type="system",
             metadata={"created": scan.created},
         )
-    db.commit()
+    if commit:
+        db.commit()
+    else:
+        db.flush()
     return scan
 
 
-def _load_open_issue(
-    db: Session, public_ref: str, *, lock: bool = True
-) -> DataQualityIssue:
-    statement = select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref)
-    if lock:
-        statement = statement.with_for_update()
-    issue = db.scalar(statement)
-    if issue is None:
-        raise AppError("ISSUE_NOT_FOUND", "Data quality issue not found.", status_code=404)
-    if issue.status != "open":
-        raise AppError(
-            "ISSUE_NOT_OPEN",
-            f"Issue is '{issue.status}', not 'open'.",
-            status_code=409,
-        )
-    return issue
-
-
 def defer_issue(db: Session, public_ref: str, actor: CurrentUser) -> DataQualityIssue:
     issue = _load_open_issue(db, public_ref)
     issue.status = "deferred"
@@ -456,115 +485,6 @@ def provide_missing_fields(
     return issue
 
 
-def resolve_odometer_regression(
-    db: Session, public_ref: str, body: ResolveOdometerRegressionRequest, actor: CurrentUser
-) -> DataQualityIssue:
-    issue = _load_open_issue(db, public_ref)
-    if issue.rule_type != "odometer_regression":
-        raise AppError(
-            "NOT_AN_ODOMETER_ISSUE",
-            "This issue is not an odometer_regression issue.",
-            status_code=409,
-        )
-    # Lock order is booking -> vehicle everywhere (checkout, return, reschedule); taking
-    # the vehicle lock first here would be a deadlock waiting to happen under concurrency.
-    booking: Booking | None = None
-    if body.decision != "retain_canonical":
-        related_refs = issue.evidence_json.get("related_refs", [])
-        if body.booking_ref not in related_refs:
-            raise AppError(
-                "INVALID_BOOKING_REFERENCE",
-                "booking_ref must be one of this issue's related bookings.",
-                status_code=422,
-            )
-        if body.corrected_odometer_km is None:
-            raise AppError(
-                "CORRECTED_VALUE_REQUIRED",
-                "corrected_odometer_km is required when correcting a reading.",
-                status_code=422,
-            )
-        booking = db.scalar(
-            select(Booking).where(Booking.public_ref == body.booking_ref).with_for_update()
-        )
-        if booking is None:
-            raise AppError(
-                "BOOKING_NOT_FOUND", "The booking to correct was not found.", status_code=404
-            )
-    vehicle = db.scalar(select(Vehicle).where(Vehicle.id == issue.entity_id).with_for_update())
-    if vehicle is None:
-        raise AppError(
-            "VEHICLE_NOT_FOUND", "The vehicle for this issue was not found.", status_code=404
-        )
-
-    correlation_id = uuid.uuid4()
-
-    if body.decision == "retain_canonical":
-        record_audit_event(
-            db,
-            actor_type="user",
-            actor_label=actor.display_name,
-            action="data_quality_odometer_retained",
-            entity_type="vehicle",
-            entity_id=vehicle.id,
-            correlation_id=correlation_id,
-            metadata={"issue_ref": issue.public_ref, "canonical_odometer_km": vehicle.odometer_km},
-        )
-    else:
-        assert booking is not None and body.corrected_odometer_km is not None
-        # Never silently lower the canonical odometer: a correction must be at or above
-        # the current canonical value, otherwise it would just create a new regression.
-        if body.corrected_odometer_km < vehicle.odometer_km:
-            raise AppError(
-                "CORRECTION_BELOW_CANONICAL",
-                (
-                    f"Corrected value {body.corrected_odometer_km} km is still below the "
-                    f"canonical {vehicle.odometer_km} km; it would not resolve the regression."
-                ),
-                status_code=422,
-            )
-
-        before = {
-            "booking_end_odometer_km": booking.end_odometer_km,
-            "vehicle_odometer_km": vehicle.odometer_km,
-        }
-        booking.end_odometer_km = body.corrected_odometer_km
-        vehicle.odometer_km = body.corrected_odometer_km
-        vehicle.version += 1
-        record_audit_event(
-            db,
-            actor_type="user",
-            actor_label=actor.display_name,
-            action="data_quality_odometer_corrected",
-            entity_type="vehicle",
-            entity_id=vehicle.id,
-            correlation_id=correlation_id,
-            before=before,
-            after={
-                "booking_end_odometer_km": booking.end_odometer_km,
-                "vehicle_odometer_km": vehicle.odometer_km,
-            },
-            metadata={"issue_ref": issue.public_ref, "booking_ref": booking.public_ref},
-        )
-
-    issue.status = "resolved"
-    issue.resolved_at = datetime.now(UTC)
-    issue.resolved_by = actor.display_name
-    record_audit_event(
-        db,
-        actor_type="user",
-        actor_label=actor.display_name,
-        action="data_quality_issue_resolved",
-        entity_type="data_quality_issue",
-        entity_id=issue.id,
-        correlation_id=correlation_id,
-        before={"status": "open"},
-        after={"status": "resolved"},
-        metadata={"decision": body.decision, "note": body.note},
-    )
-    db.commit()
-    return issue
-
-
 def resolve_booking_overlap(
     db: Session, public_ref: str, booking_ref: str, note: str | None, actor: CurrentUser
 ) -> DataQualityIssue:

backend/app/services/data_quality_common.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timedelta
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.core.errors import AppError
+from app.models.data_quality import DataQualityIssue
+
+
+def issue_due_at(detected_at: datetime, severity: str) -> datetime:
+    """Return the local operational SLA deadline for a newly detected issue."""
+    return detected_at + {
+        "high": timedelta(hours=4),
+        "medium": timedelta(days=1),
+        "low": timedelta(days=3),
+    }.get(severity, timedelta(days=1))
+
+
+def has_open_issue(
+    db: Session,
+    rule_type: str,
+    entity_type: str,
+    entity_id: uuid.UUID,
+) -> bool:
+    return (
+        db.scalar(
+            select(DataQualityIssue.id).where(
+                DataQualityIssue.rule_type == rule_type,
+                DataQualityIssue.entity_type == entity_type,
+                DataQualityIssue.entity_id == entity_id,
+                DataQualityIssue.status == "open",
+            )
+        )
+        is not None
+    )
+
+
+def new_scan_ref(prefix: str) -> str:
+    """Generate a stable human-readable prefix with a concurrent-safe suffix."""
+    return f"{prefix}-{uuid.uuid4().hex[:10].upper()}"
+
+
+def load_open_issue(db: Session, public_ref: str, *, lock: bool = True) -> DataQualityIssue:
+    statement = select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref)
+    if lock:
+        statement = statement.with_for_update().execution_options(populate_existing=True)
+    issue = db.scalar(statement)
+    if issue is None:
+        raise AppError("ISSUE_NOT_FOUND", "Data quality issue not found.", status_code=404)
+    if issue.status != "open":
+        raise AppError(
+            "ISSUE_NOT_OPEN",
+            f"Issue is '{issue.status}', not 'open'.",
+            status_code=409,
+        )
+    return issue

backend/app/services/data_quality_duplicate_scan.py

@@ -90,9 +90,7 @@ def scan_duplicate_customers[ScanType: ScanAccumulator](
         ratio = SequenceMatcher(None, name_a, name_b).ratio()
         if ratio >= 0.5:
             score += round(ratio * 30)
-            signals.append(
-                {"code": "duplicate.similar_name", "params": {"score": round(ratio, 2)}}
-            )
+            signals.append({"code": "duplicate.similar_name", "params": {"score": round(ratio, 2)}})
             summary_parts.append("similar name")
 
         if score >= DUPLICATE_THRESHOLD:

backend/app/services/data_quality_odometer.py

@@ -0,0 +1,507 @@
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.core.errors import AppError
+from app.models.booking import Booking
+from app.models.data_quality import DataQualityIssue
+from app.models.inspection import Inspection
+from app.models.vehicle import Vehicle
+from app.schemas import CurrentUser, ResolveOdometerRegressionRequest
+from app.services.audit import record_audit_event
+from app.services.data_quality_common import issue_due_at, load_open_issue, new_scan_ref
+
+
+def _odometer_fingerprint(
+    *,
+    source_type: str,
+    later_ref: str,
+    later_km: int,
+) -> dict[str, str | int]:
+    """Stable identity for one reviewed regression, independent of issue IDs."""
+    return {
+        "source_type": source_type,
+        "later_ref": later_ref,
+        "later_km": later_km,
+    }
+
+
+def _was_odometer_regression_retained(
+    db: Session,
+    *,
+    vehicle_id: uuid.UUID,
+    fingerprint: dict[str, str | int],
+) -> bool:
+    reviewed = db.scalars(
+        select(DataQualityIssue).where(
+            DataQualityIssue.rule_type == "odometer_regression",
+            DataQualityIssue.entity_type == "vehicle",
+            DataQualityIssue.entity_id == vehicle_id,
+            DataQualityIssue.status == "resolved",
+        )
+    ).all()
+    return any(
+        issue.evidence_json.get("resolution_decision") == "retain_canonical"
+        and fingerprint in issue.evidence_json.get("retained_odometer_fingerprints", [])
+        for issue in reviewed
+    )
+
+
+def open_odometer_regression_issue(
+    db: Session,
+    *,
+    vehicle: Vehicle,
+    reading_ref: str,
+    reading_km: int,
+    canonical_km: int,
+    source_type: str,
+    related_refs: list[str],
+    correctable_booking_refs: list[str] | None = None,
+    canonical_ref: str | None = None,
+    detected_at: datetime | None = None,
+    public_ref: str | None = None,
+    actor_label: str | None = None,
+    actor_type: str = "user",
+    correlation_id: uuid.UUID | None = None,
+) -> DataQualityIssue | None:
+    """Open one explainable DQ-03 issue while the caller holds the vehicle lock."""
+    if reading_km >= canonical_km:
+        return None
+    earlier_ref = canonical_ref or vehicle.public_ref
+    fingerprint = _odometer_fingerprint(
+        source_type=source_type,
+        later_ref=reading_ref,
+        later_km=reading_km,
+    )
+    # A manager's explicit "retain canonical" decision acknowledges this exact source
+    # fact. Reopening it on every scheduled scan would create churn; only changed or new
+    # evidence (and therefore a different fingerprint) is actionable again.
+    if _was_odometer_regression_retained(
+        db,
+        vehicle_id=vehicle.id,
+        fingerprint=fingerprint,
+    ):
+        return None
+    existing = db.scalar(
+        select(DataQualityIssue)
+        .where(
+            DataQualityIssue.rule_type == "odometer_regression",
+            DataQualityIssue.entity_type == "vehicle",
+            DataQualityIssue.entity_id == vehicle.id,
+            DataQualityIssue.status == "open",
+        )
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    if existing is not None:
+        before_related_refs = list(existing.evidence_json.get("related_refs", []))
+        before_correctable_refs = list(existing.evidence_json.get("correctable_booking_refs", []))
+        new_signal = {
+            "code": "odometer.regression",
+            "source_type": source_type,
+            "params": {
+                "later_ref": reading_ref,
+                "later_km": reading_km,
+                "earlier_ref": earlier_ref,
+                "earlier_km": canonical_km,
+            },
+        }
+        existing_signals = list(existing.evidence_json.get("signals", []))
+        signal_was_new = not any(
+            isinstance(signal, dict)
+            and isinstance(signal.get("params"), dict)
+            and _odometer_fingerprint(
+                source_type=str(
+                    signal.get("source_type", existing.evidence_json.get("source_type"))
+                ),
+                later_ref=str(signal["params"].get("later_ref")),
+                later_km=signal["params"].get("later_km"),
+            )
+            == fingerprint
+            for signal in existing_signals
+            if isinstance(signal, dict)
+            and isinstance(signal.get("params"), dict)
+            and isinstance(signal["params"].get("later_km"), int)
+        )
+        if signal_was_new:
+            existing_signals.append(new_signal)
+        merged_related_refs = list(
+            dict.fromkeys([*before_related_refs, *(related_refs if signal_was_new else [])])
+        )
+        merged_correctable_refs = list(
+            dict.fromkeys([*before_correctable_refs, *(correctable_booking_refs or [])])
+        )
+        if (
+            not signal_was_new
+            and merged_related_refs == before_related_refs
+            and merged_correctable_refs == before_correctable_refs
+        ):
+            return existing
+        source_types = list(existing.evidence_json.get("source_types", []))
+        previous_source_type = existing.evidence_json.get("source_type")
+        if (
+            isinstance(previous_source_type, str)
+            and previous_source_type != "multiple"
+            and previous_source_type not in source_types
+        ):
+            source_types.append(previous_source_type)
+        if source_type not in source_types:
+            source_types.append(source_type)
+        existing.evidence_json = {
+            **existing.evidence_json,
+            "summary": (
+                f"{source_type.title()} {reading_ref} recorded {reading_km} km, below "
+                f"the canonical {canonical_km} km for {vehicle.public_ref}."
+            ),
+            "related_refs": merged_related_refs,
+            "correctable_booking_refs": merged_correctable_refs,
+            "source_type": source_type if len(source_types) == 1 else "multiple",
+            "source_types": source_types,
+            "signals": existing_signals,
+        }
+        if actor_label is not None and (
+            merged_related_refs != before_related_refs
+            or merged_correctable_refs != before_correctable_refs
+            or signal_was_new
+        ):
+            record_audit_event(
+                db,
+                actor_type=actor_type,
+                actor_label=actor_label,
+                action="data_quality_issue_evidence_updated",
+                entity_type="data_quality_issue",
+                entity_id=existing.id,
+                correlation_id=correlation_id,
+                before={
+                    "related_refs": before_related_refs,
+                    "correctable_booking_refs": before_correctable_refs,
+                },
+                after={
+                    "related_refs": merged_related_refs,
+                    "correctable_booking_refs": merged_correctable_refs,
+                },
+                metadata={"vehicle_ref": vehicle.public_ref, "reading_ref": reading_ref},
+            )
+        return existing
+    now = detected_at or datetime.now(UTC)
+    previous = db.scalar(
+        select(DataQualityIssue)
+        .where(
+            DataQualityIssue.rule_type == "odometer_regression",
+            DataQualityIssue.entity_type == "vehicle",
+            DataQualityIssue.entity_id == vehicle.id,
+            DataQualityIssue.status != "open",
+        )
+        .order_by(DataQualityIssue.detected_at.desc())
+    )
+    evidence = {
+        "summary": (
+            f"{source_type.title()} {reading_ref} recorded {reading_km} km, below "
+            f"the canonical {canonical_km} km for {vehicle.public_ref}."
+        ),
+        "entity_ref": vehicle.public_ref,
+        "related_refs": related_refs,
+        "correctable_booking_refs": correctable_booking_refs or [],
+        "source_type": source_type,
+        "source_types": [source_type],
+        "signals": [
+            {
+                "code": "odometer.regression",
+                "source_type": source_type,
+                "params": {
+                    "later_ref": reading_ref,
+                    "later_km": reading_km,
+                    "earlier_ref": earlier_ref,
+                    "earlier_km": canonical_km,
+                },
+            }
+        ],
+    }
+    if previous is not None:
+        evidence["reopened_from"] = previous.public_ref
+        evidence["previous_decision"] = previous.status
+    issue = DataQualityIssue(
+        public_ref=public_ref or new_scan_ref("DQ-ODO"),
+        rule_type="odometer_regression",
+        entity_type="vehicle",
+        entity_id=vehicle.id,
+        severity="medium",
+        status="open",
+        evidence_json=evidence,
+        proposed_action_json={},
+        detected_at=now,
+        due_at=issue_due_at(now, "medium"),
+    )
+    db.add(issue)
+    db.flush()
+    if actor_label is not None:
+        record_audit_event(
+            db,
+            actor_type=actor_type,
+            actor_label=actor_label,
+            action="data_quality_issue_created",
+            entity_type="data_quality_issue",
+            entity_id=issue.id,
+            correlation_id=correlation_id,
+            after={"status": "open", "rule_type": "odometer_regression"},
+            metadata={"vehicle_ref": vehicle.public_ref, "reading_ref": reading_ref},
+        )
+    return issue
+
+
+def resolve_odometer_regression(
+    db: Session, public_ref: str, body: ResolveOdometerRegressionRequest, actor: CurrentUser
+) -> DataQualityIssue:
+    # Read the routing data without a row lock first. The canonical mutation order is
+    # booking -> vehicle -> issue -> inspection everywhere, matching checkout/return.
+    # Locking the issue before the booking creates a resolver-vs-return deadlock.
+    issue_snapshot = load_open_issue(db, public_ref, lock=False)
+    if issue_snapshot.rule_type != "odometer_regression":
+        raise AppError(
+            "NOT_AN_ODOMETER_ISSUE",
+            "This issue is not an odometer_regression issue.",
+            status_code=409,
+        )
+
+    def correctable_refs(issue: DataQualityIssue) -> list[str]:
+        if "correctable_booking_refs" in issue.evidence_json:
+            refs = issue.evidence_json.get("correctable_booking_refs", [])
+        else:
+            # Backward compatibility for issues created before source-aware evidence.
+            refs = [
+                ref
+                for ref in issue.evidence_json.get("related_refs", [])
+                if isinstance(ref, str) and ref.startswith("BK-")
+            ]
+        return [ref for ref in refs if isinstance(ref, str)]
+
+    booking: Booking | None = None
+    if body.decision != "retain_canonical":
+        if body.booking_ref not in correctable_refs(issue_snapshot):
+            raise AppError(
+                "INVALID_BOOKING_REFERENCE",
+                "booking_ref must be one of this issue's related bookings.",
+                status_code=422,
+            )
+        if body.corrected_odometer_km is None:
+            raise AppError(
+                "CORRECTED_VALUE_REQUIRED",
+                "corrected_odometer_km is required when correcting a reading.",
+                status_code=422,
+            )
+        booking = db.scalar(
+            select(Booking).where(Booking.public_ref == body.booking_ref).with_for_update()
+        )
+        if booking is None:
+            raise AppError(
+                "BOOKING_NOT_FOUND", "The booking to correct was not found.", status_code=404
+            )
+    vehicle = db.scalar(
+        select(Vehicle).where(Vehicle.id == issue_snapshot.entity_id).with_for_update()
+    )
+    if vehicle is None:
+        raise AppError(
+            "VEHICLE_NOT_FOUND", "The vehicle for this issue was not found.", status_code=404
+        )
+
+    # A return or scan may have appended evidence while we waited for the domain locks.
+    # Lock and refresh the issue only now, then revalidate every decision against that
+    # current evidence instead of resolving a stale snapshot.
+    issue = load_open_issue(db, public_ref)
+    if issue.rule_type != "odometer_regression" or issue.entity_id != vehicle.id:
+        raise AppError(
+            "ISSUE_CHANGED",
+            "The issue changed while the correction was being prepared. Review it again.",
+            status_code=409,
+        )
+    if booking is not None:
+        if booking.vehicle_id != vehicle.id or booking.public_ref not in correctable_refs(issue):
+            raise AppError(
+                "INVALID_BOOKING_REFERENCE",
+                "booking_ref must be one of this issue's related bookings.",
+                status_code=422,
+            )
+
+    correlation_id = uuid.uuid4()
+
+    if body.decision == "retain_canonical":
+        retained_fingerprints: list[dict[str, str | int]] = []
+        fallback_source_type = str(issue.evidence_json.get("source_type", "unknown"))
+        for signal in issue.evidence_json.get("signals", []):
+            if not isinstance(signal, dict) or signal.get("code") != "odometer.regression":
+                continue
+            params = signal.get("params")
+            if not isinstance(params, dict):
+                continue
+            source_type = signal.get("source_type", fallback_source_type)
+            later_ref = params.get("later_ref")
+            later_km = params.get("later_km")
+            earlier_ref = params.get("earlier_ref")
+            earlier_km = params.get("earlier_km")
+            if (
+                isinstance(source_type, str)
+                and isinstance(later_ref, str)
+                and isinstance(later_km, int)
+                and isinstance(earlier_ref, str)
+                and isinstance(earlier_km, int)
+            ):
+                retained_fingerprints.append(
+                    _odometer_fingerprint(
+                        source_type=source_type,
+                        later_ref=later_ref,
+                        later_km=later_km,
+                    )
+                )
+        issue.evidence_json = {
+            **issue.evidence_json,
+            "resolution_decision": "retain_canonical",
+            "retained_odometer_fingerprints": retained_fingerprints,
+        }
+        record_audit_event(
+            db,
+            actor_type="user",
+            actor_label=actor.display_name,
+            action="data_quality_odometer_retained",
+            entity_type="vehicle",
+            entity_id=vehicle.id,
+            correlation_id=correlation_id,
+            metadata={
+                "issue_ref": issue.public_ref,
+                "canonical_odometer_km": vehicle.odometer_km,
+                "retained_fingerprint_count": len(retained_fingerprints),
+            },
+        )
+    else:
+        assert booking is not None and body.corrected_odometer_km is not None
+        # Never silently lower the canonical odometer: a correction must be at or above
+        # the current canonical value, otherwise it would just create a new regression.
+        if body.corrected_odometer_km < vehicle.odometer_km:
+            raise AppError(
+                "CORRECTION_BELOW_CANONICAL",
+                (
+                    f"Corrected value {body.corrected_odometer_km} km is still below the "
+                    f"canonical {vehicle.odometer_km} km; it would not resolve the regression."
+                ),
+                status_code=422,
+            )
+
+        # A returned booking has two persisted representations of the same reading.
+        # Correct every return inspection for the selected booking, including historical
+        # seed rows whose old issue evidence did not yet cite the inspection explicitly.
+        related_inspections = db.scalars(
+            select(Inspection)
+            .where(Inspection.booking_id == booking.id, Inspection.type == "return")
+            .with_for_update()
+        ).all()
+        before = {
+            "booking_end_odometer_km": booking.end_odometer_km,
+            "vehicle_odometer_km": vehicle.odometer_km,
+            "inspection_odometer_km": {
+                inspection.public_ref: inspection.odometer_km for inspection in related_inspections
+            },
+        }
+        booking.end_odometer_km = body.corrected_odometer_km
+        for inspection in related_inspections:
+            inspection.odometer_km = body.corrected_odometer_km
+        vehicle.odometer_km = body.corrected_odometer_km
+        vehicle.version += 1
+        record_audit_event(
+            db,
+            actor_type="user",
+            actor_label=actor.display_name,
+            action="data_quality_odometer_corrected",
+            entity_type="vehicle",
+            entity_id=vehicle.id,
+            correlation_id=correlation_id,
+            before=before,
+            after={
+                "booking_end_odometer_km": booking.end_odometer_km,
+                "vehicle_odometer_km": vehicle.odometer_km,
+                "inspection_odometer_km": {
+                    inspection.public_ref: inspection.odometer_km
+                    for inspection in related_inspections
+                },
+            },
+            metadata={"issue_ref": issue.public_ref, "booking_ref": booking.public_ref},
+        )
+
+    has_remaining_evidence = False
+    if booking is not None:
+        target_refs = {booking.public_ref, *(item.public_ref for item in related_inspections)}
+        before_related = [
+            ref for ref in issue.evidence_json.get("related_refs", []) if isinstance(ref, str)
+        ]
+        before_correctable = correctable_refs(issue)
+        before_signals = [
+            signal for signal in issue.evidence_json.get("signals", []) if isinstance(signal, dict)
+        ]
+        remaining_signals = [
+            signal
+            for signal in before_signals
+            if not (
+                isinstance(signal.get("params"), dict)
+                and signal["params"].get("later_ref") in target_refs
+            )
+        ]
+        remaining_related = [ref for ref in before_related if ref not in target_refs]
+        remaining_correctable = [ref for ref in before_correctable if ref != booking.public_ref]
+        has_remaining_evidence = bool(remaining_signals or remaining_correctable)
+        issue.evidence_json = {
+            **issue.evidence_json,
+            "summary": (
+                "Additional odometer regression evidence remains for review."
+                if has_remaining_evidence
+                else issue.evidence_json.get("summary", "Odometer reading corrected.")
+            ),
+            "related_refs": remaining_related,
+            "correctable_booking_refs": remaining_correctable,
+            "signals": remaining_signals,
+        }
+        if has_remaining_evidence:
+            record_audit_event(
+                db,
+                actor_type="user",
+                actor_label=actor.display_name,
+                action="data_quality_issue_partially_resolved",
+                entity_type="data_quality_issue",
+                entity_id=issue.id,
+                correlation_id=correlation_id,
+                before={
+                    "related_refs": before_related,
+                    "correctable_booking_refs": before_correctable,
+                    "signal_count": len(before_signals),
+                },
+                after={
+                    "related_refs": remaining_related,
+                    "correctable_booking_refs": remaining_correctable,
+                    "signal_count": len(remaining_signals),
+                },
+                metadata={
+                    "decision": body.decision,
+                    "booking_ref": booking.public_ref,
+                    "note": body.note,
+                },
+            )
+
+    if not has_remaining_evidence:
+        issue.status = "resolved"
+        issue.resolved_at = datetime.now(UTC)
+        issue.resolved_by = actor.display_name
+        record_audit_event(
+            db,
+            actor_type="user",
+            actor_label=actor.display_name,
+            action="data_quality_issue_resolved",
+            entity_type="data_quality_issue",
+            entity_id=issue.id,
+            correlation_id=correlation_id,
+            before={"status": "open"},
+            after={"status": "resolved"},
+            metadata={"decision": body.decision, "note": body.note},
+        )
+    db.commit()
+    return issue

backend/app/services/demo_manifest.py

@@ -9,16 +9,28 @@ from app.core.config import get_settings
 from app.models.audit import AuditEvent
 from app.models.booking import Booking
 from app.models.data_quality import DataQualityIssue
-from app.models.outbox import OutboxEvent
+from app.models.outbox import OutboxEvent, is_demo_scenario_failure
 from app.schemas import DemoIntegrationSummaryOut, DemoManifestOut, DemoScenarioOut
 from app.services.integration_status import derive_mcp_hub_status, derive_n8n_status
-from app.services.knowledge import get_knowledge_provider
+from app.services.knowledge import KnowledgeHealth, get_knowledge_provider
 
 settings = get_settings()
 
 _FAILED_DEMO_EVENT_ID = "00000000-0000-4000-8000-000000000020"
 
 
+def _knowledge_scenario_ready(health: KnowledgeHealth) -> bool:
+    """Require a reachable provider and independently verified indexed documents.
+
+    ``source_document_count`` describes local Markdown files, while a sync report only
+    describes an upload attempt. Neither proves that the active provider can retrieve a
+    corpus, so an unavailable verification count remains honestly not ready.
+    """
+    return bool(
+        health.available and health.document_count is not None and health.document_count > 0
+    )
+
+
 def _last_reset(db: Session) -> tuple[datetime | None, str | None]:
     marker = db.scalar(
         select(AuditEvent)
@@ -49,7 +61,8 @@ def _scenarios(db: Session) -> list[DemoScenarioOut]:
     return_ready = bool(booking and booking.status == "active" and booking.end_odometer_km is None)
     duplicate_ready = bool(duplicate_issue and duplicate_issue.status == "open")
     overlap_ready = bool(overlap_issue and overlap_issue.status == "open")
-    automation_ready = bool(failed_run and failed_run.delivery_status == "failed")
+    automation_ready = bool(failed_run and is_demo_scenario_failure(failed_run))
+    knowledge_ready = _knowledge_scenario_ready(knowledge_health)
 
     return [
         DemoScenarioOut(
@@ -119,8 +132,8 @@ def _scenarios(db: Session) -> list[DemoScenarioOut]:
             estimated_minutes=2,
             required_roles=["rental_employee", "operations_manager"],
             start_path="/knowledge",
-            ready=knowledge_health.available,
-            blocked_reason_code=None if knowledge_health.available else "knowledgeUnavailable",
+            ready=knowledge_ready,
+            blocked_reason_code=None if knowledge_ready else "knowledgeUnavailable",
         ),
     ]
 
@@ -145,9 +158,11 @@ def _integrations(db: Session) -> list[DemoIntegrationSummaryOut]:
             key="ragcore",
             status_code=(
                 "operational"
-                if knowledge_health.provider == "ragcore" and knowledge_health.available
+                if knowledge_health.provider == "ragcore"
+                and _knowledge_scenario_ready(knowledge_health)
                 else "demoMode"
-                if knowledge_health.provider == "demo" and knowledge_health.available
+                if knowledge_health.provider == "demo"
+                and _knowledge_scenario_ready(knowledge_health)
                 else "unavailable"
             ),
             detail_code=(
@@ -169,9 +184,7 @@ def _integrations(db: Session) -> list[DemoIntegrationSummaryOut]:
             # "operational" -- same evidence rule the integration status page uses.
             status_code="operational" if mcp_hub.state == "operational" else "notConnected",
             detail_code=(
-                "mcpDetailOperational"
-                if mcp_hub.state == "operational"
-                else "mcpDetailPrepared"
+                "mcpDetailOperational" if mcp_hub.state == "operational" else "mcpDetailPrepared"
             ),
             detail_params={},
         ),

backend/app/services/dispatcher.py

@@ -77,6 +77,14 @@ def _claim_due_events(batch_size: int = 5) -> list[uuid.UUID]:
         for row in rows:
             row.delivery_status = "delivering"
             row.next_attempt_at = lease_deadline
+            # The token is stored inside the internal payload (the wire envelope below
+            # explicitly selects only contract fields). It lets the outcome transaction
+            # prove that this is still the same lease after network I/O. A stale worker
+            # must never overwrite a later reclaim/retry or an idempotent callback.
+            row.payload_json = {
+                **row.payload_json,
+                "_delivery_claim_token": str(uuid.uuid4()),
+            }
         db.commit()
         return claimed_ids
     finally:
@@ -109,6 +117,9 @@ def _deliver_one(event_id: uuid.UUID) -> None:
             wire_event = None
             payload_error = f"Malformed outbox payload, missing key {exc}"
         attempts = event.attempts
+        claim_token = event.payload_json.get("_delivery_claim_token")
+        if event.delivery_status != "delivering" or not isinstance(claim_token, str):
+            return
     finally:
         db.close()
 
@@ -130,9 +141,33 @@ def _deliver_one(event_id: uuid.UUID) -> None:
             except ValueError:
                 body = None
             if isinstance(body, dict):
-                success = bool(body.get("ok", True))
-                error = None if success else f"n8n reported failure: {body}"
-                error_code = None if success else "remoteReportedFailure"
+                acknowledged = body.get("ok") is True
+                response_event_id = body.get("event_id")
+                event_id_matches = response_event_id == str(event_id)
+                result = body.get("result")
+                execution_id = result.get("execution_id") if isinstance(result, dict) else None
+                execution_id_valid = isinstance(execution_id, str) and bool(execution_id.strip())
+                success = acknowledged and event_id_matches and execution_id_valid
+                if success:
+                    error = None
+                    error_code = None
+                elif not acknowledged:
+                    error = (
+                        "n8n response did not explicitly acknowledge the event with ok=true: "
+                        f"{body}"
+                    )
+                    error_code = (
+                        "remoteReportedFailure" if body.get("ok") is False else "malformedResponse"
+                    )
+                elif not event_id_matches:
+                    error = (
+                        "n8n acknowledged a different event ID "
+                        f"(expected {event_id}, received {response_event_id!r})"
+                    )
+                    error_code = "mismatchedEventId"
+                else:
+                    error = "n8n response omitted a valid result.execution_id"
+                    error_code = "malformedResponse"
             else:
                 # A 2xx status with a non-object (or unparsable) body means the workflow
                 # itself errored before its "Respond to Webhook" node ran -- n8n's default
@@ -152,16 +187,35 @@ def _deliver_one(event_id: uuid.UUID) -> None:
 
     db = SessionLocal()
     try:
-        event = db.get(OutboxEvent, event_id)
+        event = db.scalar(
+            select(OutboxEvent).where(OutboxEvent.event_id == event_id).with_for_update()
+        )
         if event is None:
             return
+        if (
+            event.delivery_status != "delivering"
+            or event.attempts != attempts
+            or event.payload_json.get("_delivery_claim_token") != claim_token
+        ):
+            logger.info(
+                "Ignoring stale delivery outcome for event %s because lease ownership changed",
+                event_id,
+            )
+            return
         event.attempts = attempts + 1
+        event.payload_json = {
+            key: value
+            for key, value in event.payload_json.items()
+            if key != "_delivery_claim_token"
+        }
         if success:
             event.delivery_status = "succeeded"
             event.last_error = None
             event.last_error_code = None
             event.next_attempt_at = None
-            event.external_run_id = str((body or {}).get("event_id", event_id))
+            result = (body or {}).get("result")
+            execution_id = result.get("execution_id") if isinstance(result, dict) else None
+            event.external_run_id = execution_id if isinstance(execution_id, str) else None
         else:
             event.last_error = (error or "delivery failed")[:2000]
             event.last_error_code = error_code or "unknownError"

backend/app/services/integration_status.py

@@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta
 from typing import Any, Literal
 
 import httpx
-from sqlalchemy import Row, func, select
+from sqlalchemy import Row, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core.config import get_settings
@@ -93,7 +93,10 @@ def derive_n8n_status(db: Session) -> N8nIntegrationStatus:
     latest_failure_at = db.scalar(
         select(func.max(OutboxEvent.updated_at)).where(
             OutboxEvent.delivery_status == "failed",
-            OutboxEvent.last_error_code != DEMO_SCENARIO_ERROR_CODE,
+            or_(
+                OutboxEvent.last_error_code.is_(None),
+                OutboxEvent.last_error_code != DEMO_SCENARIO_ERROR_CODE,
+            ),
         )
     )
     latest_demo_scenario_at = db.scalar(
@@ -106,11 +109,17 @@ def derive_n8n_status(db: Session) -> N8nIntegrationStatus:
     state: Literal["disabled", "unavailable", "degraded", "operational", "no_evidence"]
     if not settings.n8n_dispatch_enabled:
         state = "disabled"
+    elif not settings.n8n_webhook_url:
+        # Persisted history does not make a currently unconfigured dispatcher green.
+        state = "unavailable"
     elif unexpected_failed > 0 and succeeded == 0:
         state = "unavailable"
     elif unexpected_failed > 0:
         state = "degraded"
-    elif succeeded > 0 or pending > 0 or delivering > 0:
+    # Queued/in-flight work proves only that MobilityOps has work for the dispatcher;
+    # it does not prove that n8n has ever accepted a delivery.  A green state requires
+    # at least one persisted successful round trip.
+    elif succeeded > 0:
         state = "operational"
     else:
         state = "no_evidence"
@@ -128,8 +137,14 @@ def derive_n8n_status(db: Session) -> N8nIntegrationStatus:
     # it finishes uploading procedures to RAGcore (app/api/routers/integrations.py::
     # procedures_sync_result), the same "the workflow's own callback is the evidence"
     # pattern the scheduled scan and error handler already use below.
-    latest_procedure_sync_at = db.scalar(
-        select(func.max(AuditEvent.occurred_at)).where(AuditEvent.action == "n8n_procedures_synced")
+    latest_procedure_sync_row = db.execute(
+        select(AuditEvent.occurred_at, AuditEvent.after_json, AuditEvent.metadata_json)
+        .where(AuditEvent.action == "n8n_procedures_synced")
+        .order_by(AuditEvent.occurred_at.desc())
+        .limit(1)
+    ).first()
+    latest_procedure_sync_at = (
+        latest_procedure_sync_row[0] if latest_procedure_sync_row is not None else None
     )
 
     # Error handler evidence: registrations posted by the "Fleet Ops — Workflow Error
@@ -198,6 +213,24 @@ def derive_n8n_status(db: Session) -> N8nIntegrationStatus:
             seen_at = failure_signal[0]
             last_status = "failed"
             execution_id = failure_signal[1]
+        if name == "Fleet Ops — RAGcore Procedure Sync" and latest_procedure_sync_row:
+            sync_at, sync_result, sync_metadata = latest_procedure_sync_row
+            synced = (sync_result or {}).get("synced", 0)
+            failed_syncs = (sync_result or {}).get("failed", 0)
+            # The result callback is authoritative for corpus delivery. A generic
+            # succeeded heartbeat cannot turn a zero/partial upload green.
+            sync_result_failed = (
+                not isinstance(synced, int)
+                or not isinstance(failed_syncs, int)
+                or synced <= 0
+                or failed_syncs > 0
+            )
+            if sync_result_failed:
+                last_status = "failed"
+            if seen_at is None or sync_at > seen_at:
+                seen_at = sync_at
+            if sync_result_failed:
+                execution_id = (sync_metadata or {}).get("execution_id")
         workflow_state: Literal["no_evidence", "healthy", "stale", "failed"]
         if seen_at is None:
             workflow_state = "no_evidence"
@@ -265,16 +298,20 @@ def derive_mcp_hub_status(db: Session) -> McpHubIntegrationStatus:
     last_client = latest_call_row[1] if latest_call_row else None
     last_tool = (latest_call_row[2] or {}).get("tool") if latest_call_row else None
 
+    hub_reachable = _check_hub_reachable()
+
     state: Literal["not_configured", "no_evidence", "operational"]
     if not settings.mcp_hub_registration_enabled:
         state = "not_configured"
-    elif total_calls > 0:
+    elif total_calls > 0 and hub_reachable is not False:
         state = "operational"
     else:
+        # Historical tool calls remain useful telemetry, but cannot support a current
+        # operational claim when the configured Hub health endpoint is unreachable.
+        # ``hub_reachable`` stays available separately so consumers can distinguish
+        # this from a provider that simply has no call evidence yet.
         state = "no_evidence"
 
-    hub_reachable = _check_hub_reachable()
-
     return McpHubIntegrationStatus(
         registration_enabled=settings.mcp_hub_registration_enabled,
         state=state,

backend/app/services/knowledge/ragcore.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import hashlib
+import re
+import uuid
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from threading import Lock
@@ -13,6 +16,10 @@ from app.services.knowledge import EvidenceState, GroundedAnswer, KnowledgeHealt
 from app.services.knowledge.procedures import ProcedureDocument, iter_procedure_documents
 
 _GROUNDED_ANSWERABILITY = {"answerable", "partially_answerable"}
+_ANSWERABILITY_STATES = _GROUNDED_ANSWERABILITY | {
+    "not_answerable",
+    "conflicting_evidence",
+}
 
 # Mirrors DemoKnowledgeProvider's own extractive template in spirit: a real cited
 # excerpt wrapped in a fixed sentence, never a generated summary. Used only as a
@@ -47,6 +54,11 @@ _DOMAIN_CONCEPTS: dict[str, tuple[str, ...]] = {
     "availability": ("available", "availability", "beschikbaar", "disponible", "disponibilité"),
     "technical": ("technical", "warning", "technisch", "waarschuwing", "technique", "alerte"),
 }
+_ANSWERABLE_INTENT_CONCEPTS = frozenset(_DOMAIN_CONCEPTS) - {"vehicle", "customer"}
+
+
+def _normalize_evidence_text(value: str) -> str:
+    return " ".join(re.findall(r"\w+", value.casefold()))
 
 
 def _question_concepts(question: str) -> set[str]:
@@ -93,7 +105,9 @@ def _rank_sources_for_concepts(sources: list[SourceCard], concepts: set[str]) ->
     )
 
 
-def _retrieval_score(result: dict) -> float | None:
+def _retrieval_score(result: object) -> float | None:
+    if not isinstance(result, dict):
+        return None
     scores = result.get("scores")
     if not isinstance(scores, dict):
         return None
@@ -124,23 +138,81 @@ class RAGcoreKnowledgeProvider:
     excerpt RAGcore's own search actually found, wrapped in the same fixed citation
     template `DemoKnowledgeProvider` uses -- never a fabricated summary.
 
-    Known gap, not fixable from this side: RAGcore's ingest pipeline currently tags every
-    chunk's `language` payload field as `"en"` regardless of actual document language (the
-    `/v1/uploads` contract has no per-file language field for a caller to set correctly).
-    Filtering search/answer requests by requested UI language would therefore silently
-    exclude genuinely-relevant nl-BE/fr-BE content, so this adapter deliberately does not
-    filter by language -- retrieval relies on the embedding model's cross-lingual matching.
+    Retrieval is scoped to the stable RAGcore ``source_id`` values owned by Fleet Ops for
+    the requested UI language. Returned internal document/version UUIDs are deliberately
+    not treated as Fleet Ops identifiers: every citation must instead prove the complete
+    managed-source chain (source id, URI, locator, checksum and extractive local text).
     """
 
     name = "ragcore"
 
     def __init__(self) -> None:
         self._settings = get_settings()
+        self._managed_documents_by_source_id: dict[str, ProcedureDocument] = {}
+        for document in iter_procedure_documents(Path(self._settings.knowledge_dir)):
+            self._managed_documents_by_source_id[document.source_id] = document
         self._verification_cache: dict[str, tuple[float, int]] = {}
         self._verification_lock = Lock()
         self._answers_circuit_lock = Lock()
         self._answers_circuit_open_until = 0.0
 
+    def _managed_source(self, citation: object, language: str) -> SourceCard | None:
+        if not isinstance(citation, dict):
+            return None
+        source_id = citation.get("source_id")
+        if not isinstance(source_id, str):
+            return None
+        document = self._managed_documents_by_source_id.get(str(source_id))
+        if document is None or document.language != language:
+            return None
+        if citation.get("locator") != f"{document.document_id}.md":
+            return None
+        if citation.get("source_uri") != f"ragcore://source/{document.source_id}":
+            return None
+        for field_name in ("id", "document_id", "document_version_id"):
+            value = citation.get(field_name)
+            if not isinstance(value, str):
+                return None
+            try:
+                parsed = uuid.UUID(value)
+            except (ValueError, TypeError, AttributeError):
+                return None
+            if parsed.int == 0:
+                return None
+        if not isinstance(citation.get("title"), str):
+            return None
+        section = citation.get("section")
+        if section is not None and not isinstance(section, str):
+            return None
+        excerpt = citation.get("excerpt")
+        if not isinstance(excerpt, str) or not excerpt.strip():
+            return None
+        expected_hash = hashlib.sha256(excerpt.encode("utf-8")).hexdigest()
+        if citation.get("excerpt_sha256") != expected_hash:
+            return None
+        # RAGcore owns chunking, but the cited excerpt must still be extractive evidence
+        # from the authoritative local procedure. Token normalization tolerates Markdown
+        # punctuation/whitespace while rejecting provider text that was never uploaded.
+        normalized_excerpt = _normalize_evidence_text(excerpt)
+        if not normalized_excerpt or normalized_excerpt not in _normalize_evidence_text(
+            document.content
+        ):
+            return None
+        return SourceCard(
+            document_id=document.document_id,
+            title=document.title,
+            version=document.version,
+            section=section or "",
+            excerpt=excerpt,
+        )
+
+    def _managed_source_ids(self, language: str) -> list[str]:
+        return [
+            document.source_id
+            for document in self._managed_documents_by_source_id.values()
+            if document.language == language
+        ]
+
     def _answers_circuit_is_open(self) -> bool:
         with self._answers_circuit_lock:
             return monotonic() < self._answers_circuit_open_until
@@ -284,7 +356,7 @@ class RAGcoreKnowledgeProvider:
         if self._answers_circuit_is_open():
             KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", "circuit_open").inc()
         else:
-            answered = self._ask_via_answers(question, correlation_id)
+            answered = self._ask_via_answers(question, correlation_id, language)
             if answered is not None:
                 return answered
         # /v1/answers itself is unavailable (non-2xx or unreachable) -- fall back to
@@ -295,7 +367,9 @@ class RAGcoreKnowledgeProvider:
         # generation step.
         return self._ask_via_search_fallback(question, correlation_id, language)
 
-    def _ask_via_answers(self, question: str, correlation_id: str) -> GroundedAnswer | None:
+    def _ask_via_answers(
+        self, question: str, correlation_id: str, language: str
+    ) -> GroundedAnswer | None:
         """Returns None (not a GroundedAnswer) when /v1/answers itself is unavailable,
         so the caller can fall back to search -- as opposed to a real 200 response
         classifying the question as insufficiently answerable, which is a genuine,
@@ -307,6 +381,7 @@ class RAGcoreKnowledgeProvider:
                     json={
                         "query": question,
                         "requested_space_ids": [self._settings.ragcore_space_id],
+                        "filters": {"source_ids": self._managed_source_ids(language)},
                     },
                 )
                 if response.status_code != 200:
@@ -320,25 +395,73 @@ class RAGcoreKnowledgeProvider:
             return None
 
         try:
-            citations = {c["id"]: c for c in body.get("citations", [])}
-            sources = [
-                SourceCard(
-                    document_id=str(citation["document_id"]),
-                    title=citation["title"],
-                    version=str(citation["document_version_id"]),
-                    section=citation.get("section") or "",
-                    excerpt=citation["excerpt"],
-                )
-                for citation in citations.values()
+            if not isinstance(body, dict):
+                raise TypeError("answer response must be an object")
+            # Validate the provider's complete AnswerResponse contract before trusting
+            # generated prose. These IDs and claim bindings are the evidence that RAGcore
+            # ran its deterministic claim/citation validator; one unrelated but otherwise
+            # valid citation must never make arbitrary answer text appear grounded.
+            uuid.UUID(str(body["answer_id"]))
+            uuid.UUID(str(body["retrieval_run_id"]))
+            raw_citations = body.get("citations", [])
+            if not isinstance(raw_citations, list):
+                raise TypeError("citations must be a list")
+            citation_ids: set[uuid.UUID] = set()
+            for citation in raw_citations:
+                if not isinstance(citation, dict):
+                    raise TypeError("citation must be an object")
+                citation_ids.add(uuid.UUID(str(citation["id"])))
+            raw_claims = body["claims"]
+            if not isinstance(raw_claims, list):
+                raise TypeError("claims must be a list")
+            claims_are_bound = bool(raw_claims)
+            answer_text = body.get("answer")
+            for claim in raw_claims:
+                if not isinstance(claim, dict):
+                    raise TypeError("claim must be an object")
+                claim_text = claim.get("text")
+                claim_citation_ids = claim.get("citation_ids")
+                if (
+                    not isinstance(claim_text, str)
+                    or not claim_text.strip()
+                    or not isinstance(answer_text, str)
+                    or claim_text.strip() not in answer_text
+                    or not isinstance(claim_citation_ids, list)
+                    or not claim_citation_ids
+                ):
+                    claims_are_bound = False
+                    continue
+                try:
+                    bound_ids = {uuid.UUID(str(item)) for item in claim_citation_ids}
+                except (TypeError, ValueError):
+                    claims_are_bound = False
+                    continue
+                if not bound_ids.issubset(citation_ids):
+                    claims_are_bound = False
+            mapped_sources = [
+                source
+                for citation in raw_citations
+                if (source := self._managed_source(citation, language)) is not None
             ]
-            sources = _deduplicate_sources(sources)
+            citations_are_managed = len(mapped_sources) == len(raw_citations)
+            sources = _deduplicate_sources(mapped_sources)
             answerability = body.get("answerability", "not_answerable")
-            is_grounded = answerability in _GROUNDED_ANSWERABILITY and sources
+            if answerability not in _ANSWERABILITY_STATES:
+                raise TypeError("answerability is invalid")
+            if not isinstance(answer_text, str):
+                raise TypeError("answer must be a string")
+            is_grounded = (
+                answerability in _GROUNDED_ANSWERABILITY
+                and bool(sources)
+                and citations_are_managed
+                and claims_are_bound
+                and bool(answer_text.strip())
+            )
             evidence_state: EvidenceState = "grounded" if is_grounded else "insufficient"
             KNOWLEDGE_PROVIDER_REQUESTS.labels("answers", evidence_state).inc()
             self._close_answers_circuit()
             return GroundedAnswer(
-                answer=body.get("answer", "") if evidence_state == "grounded" else "",
+                answer=answer_text if evidence_state == "grounded" else "",
                 evidence_state=evidence_state,
                 sources=sources if evidence_state == "grounded" else [],
                 provider=self.name,
@@ -366,6 +489,7 @@ class RAGcoreKnowledgeProvider:
                     json={
                         "query": question,
                         "requested_space_ids": [self._settings.ragcore_space_id],
+                        "filters": {"source_ids": self._managed_source_ids(language)},
                         "max_results": 5,
                     },
                 )
@@ -378,24 +502,24 @@ class RAGcoreKnowledgeProvider:
             return unavailable
 
         try:
+            if not isinstance(body, dict):
+                raise TypeError("search response must be an object")
             results = body.get("results", [])
             if not isinstance(results, list):
                 raise TypeError("results must be a list")
             sources = [
-                SourceCard(
-                    document_id=str(result["citation"]["document_id"]),
-                    title=result["citation"]["title"],
-                    version=str(result["citation"]["document_version_id"]),
-                    section=result["citation"].get("section") or "",
-                    excerpt=result["citation"]["excerpt"],
-                )
+                source
                 for result in results
+                if isinstance(result, dict)
+                and (source := self._managed_source(result.get("citation"), language)) is not None
             ]
         except (TypeError, KeyError, ValueError):
             KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "malformed").inc()
             return unavailable
 
         for result in results:
+            if not isinstance(result, dict):
+                continue
             score = _retrieval_score(result)
             if score is not None:
                 KNOWLEDGE_RETRIEVAL_SCORE.observe(score)
@@ -403,15 +527,11 @@ class RAGcoreKnowledgeProvider:
         all_sources = _deduplicate_sources(sources)
         concepts = _question_concepts(question)
         qualified_sources = [
-            SourceCard(
-                document_id=str(result["citation"]["document_id"]),
-                title=result["citation"]["title"],
-                version=str(result["citation"]["document_version_id"]),
-                section=result["citation"].get("section") or "",
-                excerpt=result["citation"]["excerpt"],
-            )
+            source
             for result in results
-            if (_retrieval_score(result) or 0.0) >= self._settings.ragcore_min_search_score
+            if isinstance(result, dict)
+            and (_retrieval_score(result) or 0.0) >= self._settings.ragcore_min_search_score
+            and (source := self._managed_source(result.get("citation"), language)) is not None
         ]
         sources = _rank_sources_for_concepts(_deduplicate_sources(qualified_sources), concepts)
         if not all_sources:
@@ -424,15 +544,17 @@ class RAGcoreKnowledgeProvider:
                 correlation_id=correlation_id,
             )
 
-        damage_evidence = any(
-            term
-            in (
-                f"{source.document_id} {source.title} {source.section} {source.excerpt}"
-            ).casefold()
+        required_concepts = concepts & _ANSWERABLE_INTENT_CONCEPTS
+        evidence_text = " ".join(
+            f"{source.document_id} {source.title} {source.section} {source.excerpt}"
             for source in sources
-            for term in _DOMAIN_CONCEPTS["damage"]
-        )
-        if not concepts or not sources or ("damage" in concepts and not damage_evidence):
+        ).casefold()
+        covered_concepts = {
+            concept
+            for concept in required_concepts
+            if any(term in evidence_text for term in _DOMAIN_CONCEPTS[concept])
+        }
+        if not required_concepts or not sources or covered_concepts != required_concepts:
             KNOWLEDGE_PROVIDER_REQUESTS.labels("search", "insufficient").inc()
             return GroundedAnswer(
                 answer="",

backend/app/services/returns.py

@@ -4,7 +4,7 @@ import hashlib
 import json
 import uuid
 from dataclasses import dataclass
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime
 
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
@@ -12,13 +12,13 @@ from sqlalchemy.orm import Session
 
 from app.core.errors import AppError
 from app.models.booking import Booking
-from app.models.data_quality import DataQualityIssue
 from app.models.idempotency import IdempotencyRecord
 from app.models.inspection import Inspection
 from app.models.outbox import OutboxEvent
 from app.models.vehicle import Vehicle
 from app.schemas import CurrentUser, RegisterReturnRequest
 from app.services.audit import record_audit_event
+from app.services.data_quality import open_odometer_regression_issue
 
 
 def _new_inspection_ref() -> str:
@@ -261,28 +261,21 @@ def register_vehicle_return(
 
     quality_issue_ref: str | None = None
     if evaluation.odometer_regression:
-        issue = DataQualityIssue(
-            public_ref=f"DQ-RET-{str(inspection.public_ref).split('-')[-1]}",
-            rule_type="odometer_regression",
-            entity_type="vehicle",
-            entity_id=vehicle.id,
-            severity="medium",
-            status="open",
-            evidence_json={
-                "summary": (
-                    f"Return submitted {body.end_odometer_km} km, below canonical "
-                    f"{evaluation.canonical_odometer_km} km."
-                ),
-                "entity_ref": vehicle.public_ref,
-                "related_refs": [booking.public_ref, inspection.public_ref],
-            },
-            proposed_action_json={},
+        issue = open_odometer_regression_issue(
+            db,
+            vehicle=vehicle,
+            reading_ref=inspection.public_ref,
+            reading_km=body.end_odometer_km,
+            canonical_km=evaluation.canonical_odometer_km,
+            source_type="return",
+            related_refs=[booking.public_ref, inspection.public_ref],
+            correctable_booking_refs=[booking.public_ref],
             detected_at=now,
-            due_at=now + timedelta(days=1),
+            public_ref=f"DQ-RET-{str(inspection.public_ref).split('-')[-1]}",
+            actor_label=actor.display_name,
+            correlation_id=correlation_id,
         )
-        db.add(issue)
-        db.flush()
-        quality_issue_ref = issue.public_ref
+        quality_issue_ref = issue.public_ref if issue is not None else None
 
     resulting_status = evaluation.resulting_vehicle_status
     vehicle.odometer_km = evaluation.resulting_odometer_km