"""The single authoritative vehicle-status evaluator. Used by the data-quality scanner (detection), the status-recommendation preview endpoint, the apply endpoint, and tests -- so scan-time detection and resolve-time recommendation can never structurally disagree (see docs/fleet-ops-correction/ vehicle-status-decision-table.md for the full decision table and rationale). The evaluator only ever reasons from real, freshly-queried domain facts (an actually active rental, a real service-threshold breach, a real overlapping-booking conflict) -- never from a proxy like "does some other high-severity issue happen to be open". It is therefore also order-independent: resolving, deferring or rejecting an unrelated issue on the same vehicle never changes what this function returns, because it never looks at issue history, only at the vehicle's/bookings' current state. """ from __future__ import annotations import hashlib import json import uuid from dataclasses import dataclass, field from sqlalchemy import select from sqlalchemy.orm import Session from app.models.booking import Booking from app.models.data_quality import DataQualityIssue from app.models.vehicle import Vehicle # Every code below is a stable, localizable identifier -- see # frontend/src/i18n/messageCodes.ts and quality:statusRecommendation.codes.* for the # human-language mapping in all three supported locales. The backend never emits prose. RECOMMENDATION_CODE_ACTIVE_RENTAL = "vehicle.active_rental" RECOMMENDATION_CODE_SERVICE_THRESHOLD = "vehicle.service_threshold_reached" RECOMMENDATION_CODE_BOOKING_CONFLICT = "vehicle.booking_conflict" RECOMMENDATION_CODE_RENTAL_ENDED = "vehicle.rental_ended" RECOMMENDATION_CODE_MANUAL_REVIEW = "vehicle.manual_review_required" RECOMMENDATION_CODE_NO_CONFLICT = "vehicle.no_conflict" @dataclass class VehicleStatusFacts: active_booking_refs: list[str] = field(default_factory=list) overlapping_booking_pairs: list[tuple[str, str]] = field(default_factory=list) service_threshold_reached: bool = False odometer_km: int = 0 next_service_km: int = 0 open_booking_overlap_issue_ref: str | None = None @property def has_active_rental(self) -> bool: return len(self.active_booking_refs) > 0 @property def has_booking_conflict(self) -> bool: return ( len(self.overlapping_booking_pairs) > 0 or self.open_booking_overlap_issue_ref is not None ) def as_dict(self) -> dict: return { "active_booking_refs": self.active_booking_refs, "overlapping_booking_pairs": [list(pair) for pair in self.overlapping_booking_pairs], "service_threshold_reached": self.service_threshold_reached, "odometer_km": self.odometer_km, "next_service_km": self.next_service_km, "open_booking_overlap_issue_ref": self.open_booking_overlap_issue_ref, } @dataclass class VehicleStatusRecommendation: current_status: str recommended_status: str | None recommendation_code: str safe_to_apply: bool manual_review_required: bool facts: VehicleStatusFacts blocking_reasons: list[str] def _overlapping_booking_pairs(bookings: list[Booking]) -> list[tuple[Booking, Booking]]: ordered = sorted(bookings, key=lambda b: b.starts_at) pairs: list[tuple[Booking, Booking]] = [] for i, first in enumerate(ordered): for second in ordered[i + 1 :]: if second.starts_at < first.ends_at and first.starts_at < second.ends_at: pairs.append((first, second)) return pairs def gather_vehicle_status_facts( db: Session, vehicle: Vehicle, *, exclude_issue_id: uuid.UUID | None = None ) -> VehicleStatusFacts: """Real, freshly-queried facts only -- see module docstring. Never cached, never derived from another issue's mere existence (only a *specific* booking_overlap issue's presence is used, as a cross-reference to that issue's own public_ref).""" reserved_or_active = list( db.scalars( select(Booking).where( Booking.vehicle_id == vehicle.id, Booking.status.in_(["reserved", "active"]), ) ).all() ) active_refs = [b.public_ref for b in reserved_or_active if b.status == "active"] overlap_pairs = [ (a.public_ref, b.public_ref) for a, b in _overlapping_booking_pairs(reserved_or_active) ] overlap_issue_query = select(DataQualityIssue.public_ref).where( DataQualityIssue.entity_type == "vehicle", DataQualityIssue.entity_id == vehicle.id, DataQualityIssue.status == "open", DataQualityIssue.rule_type == "booking_overlap", ) if exclude_issue_id is not None: overlap_issue_query = overlap_issue_query.where(DataQualityIssue.id != exclude_issue_id) open_overlap_ref = db.scalar(overlap_issue_query) return VehicleStatusFacts( active_booking_refs=active_refs, overlapping_booking_pairs=overlap_pairs, service_threshold_reached=vehicle.odometer_km >= vehicle.next_service_km, odometer_km=vehicle.odometer_km, next_service_km=vehicle.next_service_km, open_booking_overlap_issue_ref=open_overlap_ref, ) def compute_recommendation_token(vehicle: Vehicle, facts: VehicleStatusFacts) -> str: """A short digest of exactly the facts the recommendation was based on, plus the vehicle's optimistic-lock version. The apply endpoint recomputes this from fresh facts and rejects the request if it doesn't match the token the client last saw -- the frontend must never assume a previously-shown preview is still valid without the server re-checking it (see docs/fleet-ops-correction/current-gap-audit.md ยง8F).""" payload = {"version": vehicle.version, "status": vehicle.operational_status, **facts.as_dict()} digest = hashlib.sha256(json.dumps(payload, sort_keys=True, default=str).encode()).hexdigest() return digest[:16] def evaluate_vehicle_status( vehicle: Vehicle, facts: VehicleStatusFacts ) -> VehicleStatusRecommendation: """Pure decision logic over already-gathered facts -- see docs/fleet-ops-correction/vehicle-status-decision-table.md. Never mutates anything, never queries the database itself (call gather_vehicle_status_facts first), so it is trivial to unit-test every branch in isolation.""" current = vehicle.operational_status blocking_reasons: list[str] = [] if facts.service_threshold_reached: blocking_reasons.append(RECOMMENDATION_CODE_SERVICE_THRESHOLD) if facts.has_booking_conflict: blocking_reasons.append(RECOMMENDATION_CODE_BOOKING_CONFLICT) if current == "maintenance" and RECOMMENDATION_CODE_SERVICE_THRESHOLD not in blocking_reasons: # Already being in maintenance is itself a real blocking fact -- an active # booking never overrides it. This is exactly the forbidden shortcut this # evaluator must never take (maintenance + active booking -> auto "rented"). blocking_reasons.append(RECOMMENDATION_CODE_SERVICE_THRESHOLD) def result( recommended: str | None, code: str, *, safe: bool, manual: bool ) -> VehicleStatusRecommendation: return VehicleStatusRecommendation( current_status=current, recommended_status=recommended, recommendation_code=code, safe_to_apply=safe, manual_review_required=manual, facts=facts, blocking_reasons=blocking_reasons, ) if facts.has_active_rental and not blocking_reasons: if current == "rented": return result(None, RECOMMENDATION_CODE_NO_CONFLICT, safe=False, manual=False) return result("rented", RECOMMENDATION_CODE_ACTIVE_RENTAL, safe=True, manual=False) if facts.has_active_rental and blocking_reasons: # Explicitly forbidden shortcut this evaluator must never take: an active # booking is not proof the vehicle should be "rented" when a real blocking # condition also exists (e.g. maintenance-due, or a genuine booking conflict). # This is a real contradiction in the underlying facts, not something safe to # resolve automatically. return result(None, RECOMMENDATION_CODE_MANUAL_REVIEW, safe=False, manual=True) if facts.service_threshold_reached: if current == "maintenance": return result(None, RECOMMENDATION_CODE_NO_CONFLICT, safe=False, manual=False) return result("maintenance", RECOMMENDATION_CODE_SERVICE_THRESHOLD, safe=True, manual=False) if facts.has_booking_conflict: if current == "blocked": return result(None, RECOMMENDATION_CODE_NO_CONFLICT, safe=False, manual=False) return result("blocked", RECOMMENDATION_CODE_BOOKING_CONFLICT, safe=True, manual=False) # No active rental, no maintenance need, no booking conflict. if current in ("available", "cleaning", "blocked"): return result(None, RECOMMENDATION_CODE_NO_CONFLICT, safe=False, manual=False) if current == "rented": return result("available", RECOMMENDATION_CODE_RENTAL_ENDED, safe=True, manual=False) if current == "maintenance": # No positive fact confirms maintenance is actually finished (no completed # service record is tracked here) -- clearing "maintenance" without such a # fact would be exactly the kind of unsafe shortcut this evaluator forbids. # Releasing a vehicle from maintenance remains an explicit, manual decision. return result(None, RECOMMENDATION_CODE_NO_CONFLICT, safe=False, manual=False) return result(None, RECOMMENDATION_CODE_MANUAL_REVIEW, safe=False, manual=True)