M16: isolate acceptance and harden readiness

backend/app/main.py

@@ -4,6 +4,7 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
+from sqlalchemy import text
 
 from app.api.routers import (
     audit,
@@ -77,6 +78,28 @@ def health() -> dict[str, str]:
     return {"status": "ok", "service": "mobilityops-api"}
 
 
+@app.get("/health/live")
+def liveness() -> dict[str, str]:
+    """Process liveness only; external dependencies deliberately do not affect it."""
+    return {"status": "ok", "service": "mobilityops-api"}
+
+
+@app.get("/health/ready")
+def readiness() -> JSONResponse:
+    """Traffic readiness: the API is useful only while its canonical database responds."""
+    try:
+        with SessionLocal() as db:
+            db.execute(text("SELECT 1"))
+    except Exception:  # noqa: BLE001 -- readiness must convert infrastructure errors to 503
+        return JSONResponse(
+            status_code=503,
+            content={"status": "not_ready", "service": "mobilityops-api", "database": "down"},
+        )
+    return JSONResponse(
+        content={"status": "ready", "service": "mobilityops-api", "database": "up"}
+    )
+
+
 @app.get("/api/v1/system/status")
 def system_status() -> dict[str, object]:
     return {

backend/app/services/data_quality.py

@@ -54,15 +54,9 @@ def _has_open_issue(db: Session, rule_type: str, entity_type: str, entity_id: uu
     )
 
 
-def _next_public_ref(db: Session, prefix: str) -> str:
-    existing = db.execute(select(DataQualityIssue.public_ref)).scalars().all()
-    numbers = [
-        int(ref.rsplit("-", 1)[-1])
-        for ref in existing
-        if ref.startswith(f"{prefix}-") and ref.rsplit("-", 1)[-1].isdigit()
-    ]
-    next_number = (max(numbers) + 1) if numbers else 1
-    return f"{prefix}-{next_number:04d}"
+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(
@@ -109,7 +103,7 @@ def _open_issue(
         evidence["previous_decision"] = previous.status
 
     issue = DataQualityIssue(
-        public_ref=_next_public_ref(db, "DQ-SCAN"),
+        public_ref=_new_scan_ref("DQ-SCAN"),
         rule_type=rule_type,
         entity_type=entity_type,
         entity_id=entity_id,

backend/app/services/returns.py

@@ -18,12 +18,14 @@ from app.models.vehicle import Vehicle
 from app.schemas import CurrentUser, RegisterReturnRequest
 from app.services.audit import record_audit_event
 
-REF_PREFIX = "INSP"
 
+def _new_inspection_ref() -> str:
+    """Generate a collision-resistant public reference without reading mutable counts.
 
-def _next_public_ref(db: Session) -> str:
-    existing = db.execute(select(Inspection.public_ref)).scalars().all()
-    return f"{REF_PREFIX}-{len(existing) + 1:04d}"
+    Return commands for different bookings can commit concurrently. A count-based
+    reference made those independent transactions race for the same unique value.
+    """
+    return f"INSP-{uuid.uuid4().hex[:10].upper()}"
 
 
 def _derive_vehicle_status_with_reason(
@@ -211,7 +213,7 @@ def register_vehicle_return(
     evaluation = evaluate_return(db, booking, vehicle, body, now=now)
 
     inspection = Inspection(
-        public_ref=_next_public_ref(db),
+        public_ref=_new_inspection_ref(),
         booking_id=booking.id,
         vehicle_id=vehicle.id,
         type="return",

backend/tests/test_data_quality.py

@@ -148,9 +148,7 @@ def test_merge_customers_s2_scenario_rewires_and_audits(ops_client):
     issue = ops_client.get("/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE").json()
     assert issue["status"] == "resolved"
 
-    audit_events = ops_client.get(
-        "/api/v1/audit", params={"action": "customer_merged"}
-    ).json()
+    audit_events = ops_client.get("/api/v1/audit", params={"action": "customer_merged"}).json()
     assert len(audit_events) >= 1
 
     # Already-resolved issue cannot be merged again.
@@ -431,9 +429,7 @@ def test_manual_scan_records_audit_event(ops_client):
     scan = ops_client.post("/api/v1/data-quality/scan")
     assert scan.status_code == 200
 
-    events = ops_client.get(
-        "/api/v1/audit", params={"action": "data_quality_scan_run"}
-    ).json()
+    events = ops_client.get("/api/v1/audit", params={"action": "data_quality_scan_run"}).json()
     assert len(events) >= 1
     assert "created" in events[0]["metadata"]
 
@@ -444,9 +440,7 @@ def _reset_demo(ops_client) -> None:
     # in before making any further authenticated call with the same client.
     response = ops_client.post("/api/v1/demo/reset")
     assert response.status_code == 200, response.text
-    login_response = ops_client.post(
-        "/api/v1/demo/login", json={"role": "operations_manager"}
-    )
+    login_response = ops_client.post("/api/v1/demo/login", json={"role": "operations_manager"})
     assert login_response.status_code == 200, login_response.text
 
 
@@ -548,3 +542,11 @@ def test_rejected_issue_recurrence_links_to_prior_decision(ops_client):
     )
     assert match is not None, "expected a new issue linked back to the rejected one"
     assert match["evidence"]["previous_decision"] == "rejected"
+
+
+def test_scan_public_refs_are_collision_resistant() -> None:
+    from app.services.data_quality import _new_scan_ref
+
+    refs = {_new_scan_ref("DQ-SCAN") for _ in range(1000)}
+    assert len(refs) == 1000
+    assert all(ref.startswith("DQ-SCAN-") and len(ref) == 18 for ref in refs)

backend/tests/test_health.py

@@ -7,3 +7,35 @@ def test_health() -> None:
     response = TestClient(app).get("/health")
     assert response.status_code == 200
     assert response.json() == {"status": "ok", "service": "mobilityops-api"}
+
+
+def test_liveness_is_process_only() -> None:
+    response = TestClient(app).get("/health/live")
+    assert response.status_code == 200
+    assert response.json()["status"] == "ok"
+
+
+def test_readiness_checks_the_canonical_database() -> None:
+    response = TestClient(app).get("/health/ready")
+    assert response.status_code == 200
+    assert response.json() == {
+        "status": "ready",
+        "service": "mobilityops-api",
+        "database": "up",
+    }
+
+
+def test_readiness_degrades_when_database_is_unavailable(monkeypatch) -> None:
+    import app.main as main_module
+
+    class BrokenSession:
+        def __enter__(self):
+            raise ConnectionError("database unavailable")
+
+        def __exit__(self, *_args):
+            return False
+
+    monkeypatch.setattr(main_module, "SessionLocal", BrokenSession)
+    response = TestClient(app).get("/health/ready")
+    assert response.status_code == 503
+    assert response.json()["database"] == "down"

backend/tests/test_return.py

@@ -305,3 +305,11 @@ def test_concurrent_returns_only_one_succeeds():
 
     assert results.count(201) == 1
     assert results.count(409) == 2
+
+
+def test_return_public_refs_are_collision_resistant() -> None:
+    from app.services.returns import _new_inspection_ref
+
+    refs = {_new_inspection_ref() for _ in range(1000)}
+    assert len(refs) == 1000
+    assert all(ref.startswith("INSP-") and len(ref) == 15 for ref in refs)