from __future__ import annotations from datetime import datetime from sqlalchemy import select from sqlalchemy.orm import Session 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.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 settings = get_settings() _FAILED_DEMO_EVENT_ID = "00000000-0000-4000-8000-000000000020" def _last_reset(db: Session) -> tuple[datetime | None, str | None]: marker = db.scalar( select(AuditEvent) .where(AuditEvent.action == "demo_data_seeded") .order_by(AuditEvent.occurred_at.desc()) ) if marker is None: return None, None metadata = marker.metadata_json or {} return marker.occurred_at, metadata.get("anchor_date") def _scenarios(db: Session) -> list[DemoScenarioOut]: booking = db.scalar(select(Booking).where(Booking.public_ref == "BK-DEMO-RETURN")) duplicate_issue = db.scalar( select(DataQualityIssue).where(DataQualityIssue.public_ref == "DQ-DEMO-DUPLICATE") ) overlap_issue = db.scalar( select(DataQualityIssue).where(DataQualityIssue.public_ref == "DQ-DEMO-OVERLAP") ) failed_run = db.scalar( select(OutboxEvent).where(OutboxEvent.event_id == _FAILED_DEMO_EVENT_ID) ) knowledge_health = get_knowledge_provider().health() # Human copy (title, problem statement, "demonstrates" summary) lives entirely in the # frontend's demo.json (scenarios.items..*) so it's available in all three UI # languages. This service only emits stable identifiers and message codes -- never # display prose -- per the message_code + params architecture used across the app. 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") return [ DemoScenarioOut( id="return-anomaly", estimated_minutes=3, required_roles=["rental_employee", "operations_manager"], start_path=f"/bookings/{booking.public_ref}" if booking else "/bookings", ready=return_ready, blocked_reason_code=( None if return_ready else "bookingNotFound" if booking is None else "bookingAlreadyProcessed" ), ), DemoScenarioOut( id="duplicate-customer", estimated_minutes=3, required_roles=["operations_manager"], start_path=( f"/data-quality/{duplicate_issue.public_ref}" if duplicate_issue else "/data-quality" ), ready=duplicate_ready, blocked_reason_code=( None if duplicate_ready else "duplicateIssueNotFound" if duplicate_issue is None else "issueAlreadyResolved" ), ), DemoScenarioOut( id="booking-overlap", estimated_minutes=2, required_roles=["operations_manager"], start_path=( f"/data-quality/{overlap_issue.public_ref}" if overlap_issue else "/data-quality" ), ready=overlap_ready, blocked_reason_code=( None if overlap_ready else "overlapIssueNotFound" if overlap_issue is None else "issueAlreadyResolved" ), ), DemoScenarioOut( id="automation-retry", estimated_minutes=2, required_roles=["operations_manager"], start_path="/automation", ready=automation_ready, blocked_reason_code=( None if automation_ready else "failedEventNotFound" if failed_run is None else "eventAlreadyRecovered" ), ), DemoScenarioOut( id="knowledge-question", 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", ), ] def _integrations(db: Session) -> list[DemoIntegrationSummaryOut]: n8n = derive_n8n_status(db) knowledge_health = get_knowledge_provider().health() mcp_hub = derive_mcp_hub_status(db) return [ DemoIntegrationSummaryOut( key="n8n", status_code=n8n.state, detail_code="n8nDetail", detail_params={ "succeeded": n8n.succeeded, "failed": n8n.failed, "pending": n8n.pending, }, ), DemoIntegrationSummaryOut( key="ragcore", status_code="operational" if knowledge_health.provider == "ragcore" else "demoMode", detail_code="ragcoreDetail", detail_params={ "count": ( knowledge_health.document_count if knowledge_health.document_count is not None else "unknown" ), "collection": knowledge_health.collection, }, ), DemoIntegrationSummaryOut( key="mcp_hub", # `MCP_HUB_REGISTRATION_ENABLED` on its own proves nothing: registration is # catalog-driven on the Hub's side, so the flag only says Fleet Ops expects # to be called. Only real recorded `mcp_tool_request` calls make this # "operational" -- same evidence rule the integration status page uses. status_code="operational" if mcp_hub.state == "operational" else "notConnected", detail_code=( "mcpDetailEnabled" if mcp_hub.state == "operational" else "mcpDetailNotConnected" ), detail_params={}, ), ] def scenario_integrity_report(db: Session) -> dict: """Server-side scenario-integrity check run after every reset (section 15): confirms each of the 5 named scenarios is actually present and ready, rather than trusting the seed loader silently. Reuses the same readiness derivation the manifest/scenario overview already use, so this can never drift from what a visitor actually sees.""" scenarios = _scenarios(db) not_ready = [ {"id": s.id, "reason_code": s.blocked_reason_code} for s in scenarios if not s.ready ] return {"all_ready": len(not_ready) == 0, "not_ready": not_ready} def build_demo_manifest(db: Session) -> DemoManifestOut: last_reset_at, anchor_date = _last_reset(db) return DemoManifestOut( demo_mode=settings.mobilityops_demo_mode, organization_name=settings.demo_organization_name, timezone=settings.demo_timezone, synthetic_data=True, allow_reset=settings.demo_allow_reset, last_reset_at=last_reset_at, anchor_date=anchor_date, guide_available=True, required_roles=["operations_manager", "rental_employee"], scenarios=_scenarios(db), integrations=_integrations(db), )