NuklearRabbit
6f77a30dce
The demo seed plants exactly one failed delivery (BK-H-0020) to demonstrate retry and audit. Because derive_n8n_status() counted any failure, every fresh reset pinned the n8n integration to "degraded" -- the demo showed a warning about a prop, which tells a viewer something untrue about the automation. The seeded failure now carries its own error code, demoScenarioTimeout, rather than the generic connectionError a real timeout produces. No column and no migration: last_error_code already existed, is already surfaced to the UI and is already localizable. - integration status splits failed into unexpected_failed and demo_scenario_failed; only unexpected failures may move the state. A staged failure alone leaves n8n operational. - latest_failure_at is a health signal and now ignores the staged failure; latest_demo_scenario_at reports it separately. - /api/v1/workflows exposes is_demo_scenario. The Automation page labels the run as a prepared demo scenario, explains that it is a simulated temporary failure that does not affect automation health, and offers a distinct "retry demo scenario" action. Translated in nl-BE, en-GB and fr-BE. - the carve-out stays narrow: a real failure still degrades n8n, and a genuine later failure of the same event overwrites the demo code with the real one. - the retry itself is unchanged and real: the event goes back on the outbox and the dispatcher delivers it to n8n like any other, so 19+1 becomes 20+0 only on an actual round trip. The audit records which kind of failure was retried. Tests that assert on the seeded scenario now reseed first, since earlier test files legitimately mutate the outbox and the suite shares one database. Verified locally against a real PostgreSQL 16: 181 passed, ruff clean, mypy clean (50 files), tsc clean, frontend build clean. Not deployed and not browser-verified.
362 lines
14 KiB
Python
362 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from difflib import SequenceMatcher
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import delete, insert, update
|
|
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.customer import Customer
|
|
from app.models.data_quality import DataQualityIssue
|
|
from app.models.idempotency import IdempotencyRecord
|
|
from app.models.inspection import Inspection
|
|
from app.models.maintenance import MaintenanceRecord
|
|
from app.models.outbox import DEMO_SCENARIO_ERROR_CODE, OutboxEvent
|
|
from app.models.user import User
|
|
from app.models.vehicle import Vehicle
|
|
from app.services.audit import record_audit_event
|
|
|
|
settings = get_settings()
|
|
|
|
DEMO_USERS = [
|
|
{
|
|
"public_ref": "USR-OPS",
|
|
"display_name": "Amelie De Ridder",
|
|
"role": "operations_manager",
|
|
},
|
|
{
|
|
"public_ref": "USR-EMP",
|
|
"display_name": "Karim Boujaddaine",
|
|
"role": "rental_employee",
|
|
},
|
|
]
|
|
|
|
# seed/generate_seed.py authored the committed CSVs relative to this fixed date
|
|
# (`--anchor 2026-08-01`, matching Settings.demo_today). Every reset shifts every
|
|
# seeded date by (today - SEED_AUTHORED_ANCHOR) so "today" / "near-future" / "overlaps
|
|
# right now" scenarios stay true to the actual reset moment instead of decaying as real
|
|
# time passes between resets -- a fixed anchor with no shift goes stale within days.
|
|
SEED_AUTHORED_ANCHOR = date(2026, 8, 1)
|
|
|
|
|
|
def _seed_anchor_shift(today: date) -> timedelta:
|
|
return today - SEED_AUTHORED_ANCHOR
|
|
|
|
|
|
def _parse_dt(value: str) -> datetime:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
|
|
|
|
def _parse_bool(value: str) -> bool:
|
|
return value.strip().lower() == "true"
|
|
|
|
|
|
def _parse_optional_int(value: str) -> int | None:
|
|
value = value.strip()
|
|
return int(value) if value else None
|
|
|
|
|
|
@dataclass
|
|
class SeedResult:
|
|
counts: dict[str, int]
|
|
anchor_date: date
|
|
seeded_at: datetime
|
|
|
|
|
|
def _seed_dir() -> Path:
|
|
return Path(settings.seed_dir)
|
|
|
|
|
|
def _read_csv(name: str) -> list[dict[str, str]]:
|
|
path = _seed_dir() / name
|
|
with path.open(newline="", encoding="utf-8") as handle:
|
|
return list(csv.DictReader(handle))
|
|
|
|
|
|
def clear_all(db: Session) -> None:
|
|
for model in (
|
|
AuditEvent,
|
|
OutboxEvent,
|
|
IdempotencyRecord,
|
|
DataQualityIssue,
|
|
Inspection,
|
|
MaintenanceRecord,
|
|
Booking,
|
|
Vehicle,
|
|
Customer,
|
|
User,
|
|
):
|
|
db.execute(delete(model))
|
|
|
|
|
|
def load_seed(db: Session) -> SeedResult:
|
|
counts: dict[str, int] = {}
|
|
today = datetime.now(UTC).date()
|
|
shift = _seed_anchor_shift(today)
|
|
|
|
user_rows = [
|
|
{"id": uuid.uuid4(), **user, "active": True} for user in DEMO_USERS
|
|
]
|
|
db.execute(insert(User), user_rows)
|
|
counts["users"] = len(user_rows)
|
|
|
|
customer_id_by_ref: dict[str, uuid.UUID] = {}
|
|
customer_rows = []
|
|
customer_row_by_ref: dict[str, dict] = {}
|
|
for row in _read_csv("customers.csv"):
|
|
cid = uuid.uuid4()
|
|
customer_id_by_ref[row["public_ref"]] = cid
|
|
customer_row = {
|
|
"id": cid,
|
|
"public_ref": row["public_ref"],
|
|
"first_name": row["first_name"],
|
|
"last_name": row["last_name"],
|
|
"email": row["email"] or None,
|
|
"phone": row["phone"] or None,
|
|
"postal_code": row["postal_code"] or None,
|
|
"city": row["city"] or None,
|
|
}
|
|
customer_rows.append(customer_row)
|
|
customer_row_by_ref[row["public_ref"]] = customer_row
|
|
db.execute(insert(Customer), customer_rows)
|
|
counts["customers"] = len(customer_rows)
|
|
# Second pass for merged_into (self-referencing FK) since target must exist first.
|
|
for row in _read_csv("customers.csv"):
|
|
merged_ref = row.get("merged_into") or ""
|
|
if merged_ref:
|
|
db.execute(
|
|
update(Customer)
|
|
.where(Customer.id == customer_id_by_ref[row["public_ref"]])
|
|
.values(merged_into_customer_id=customer_id_by_ref[merged_ref])
|
|
)
|
|
|
|
vehicle_id_by_ref: dict[str, uuid.UUID] = {}
|
|
vehicle_rows = []
|
|
for row in _read_csv("vehicles.csv"):
|
|
vid = uuid.uuid4()
|
|
vehicle_id_by_ref[row["public_ref"]] = vid
|
|
vehicle_rows.append(
|
|
{
|
|
"id": vid,
|
|
"public_ref": row["public_ref"],
|
|
"make": row["make"],
|
|
"model": row["model"],
|
|
"model_year": int(row["model_year"]),
|
|
"registration_number": row["registration_number"],
|
|
"location": row["location"],
|
|
"operational_status": row["operational_status"],
|
|
"odometer_km": int(row["odometer_km"]),
|
|
"next_service_km": int(row["next_service_km"]),
|
|
"active": _parse_bool(row["active"]),
|
|
"version": 1,
|
|
}
|
|
)
|
|
db.execute(insert(Vehicle), vehicle_rows)
|
|
counts["vehicles"] = len(vehicle_rows)
|
|
|
|
booking_id_by_ref: dict[str, uuid.UUID] = {}
|
|
booking_rows = []
|
|
for row in _read_csv("bookings.csv"):
|
|
bid = uuid.uuid4()
|
|
booking_id_by_ref[row["public_ref"]] = bid
|
|
booking_rows.append(
|
|
{
|
|
"id": bid,
|
|
"public_ref": row["public_ref"],
|
|
"customer_id": customer_id_by_ref[row["customer_ref"]],
|
|
"vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
|
|
"starts_at": _parse_dt(row["starts_at"]) + shift,
|
|
"ends_at": _parse_dt(row["ends_at"]) + shift,
|
|
"status": row["status"],
|
|
"start_odometer_km": _parse_optional_int(row["start_odometer_km"]),
|
|
"end_odometer_km": _parse_optional_int(row["end_odometer_km"]),
|
|
"requirements_complete": _parse_bool(row["requirements_complete"]),
|
|
}
|
|
)
|
|
db.execute(insert(Booking), booking_rows)
|
|
counts["bookings"] = len(booking_rows)
|
|
|
|
inspection_rows = []
|
|
for row in _read_csv("inspections.csv"):
|
|
inspection_rows.append(
|
|
{
|
|
"id": uuid.uuid4(),
|
|
"public_ref": row["public_ref"],
|
|
"booking_id": booking_id_by_ref[row["booking_ref"]],
|
|
"vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
|
|
"type": row["type"],
|
|
"fuel_level_percent": int(row["fuel_level_percent"]),
|
|
"cleanliness_ok": _parse_bool(row["cleanliness_ok"]),
|
|
"damage_reported": _parse_bool(row["damage_reported"]),
|
|
"technical_warning": _parse_bool(row["technical_warning"]),
|
|
"odometer_km": int(row["odometer_km"]),
|
|
"completed_at": _parse_dt(row["completed_at"]) + shift,
|
|
"completed_by": None,
|
|
}
|
|
)
|
|
db.execute(insert(Inspection), inspection_rows)
|
|
counts["inspections"] = len(inspection_rows)
|
|
|
|
maintenance_rows = []
|
|
for row in _read_csv("maintenance.csv"):
|
|
maintenance_rows.append(
|
|
{
|
|
"id": uuid.uuid4(),
|
|
"public_ref": row["public_ref"],
|
|
"vehicle_id": vehicle_id_by_ref[row["vehicle_ref"]],
|
|
"occurred_at": _parse_dt(row["occurred_at"]) + shift,
|
|
"odometer_km": int(row["odometer_km"]),
|
|
"category": row["category"],
|
|
"summary": row["summary"],
|
|
}
|
|
)
|
|
db.execute(insert(MaintenanceRecord), maintenance_rows)
|
|
counts["maintenance"] = len(maintenance_rows)
|
|
|
|
def resolve_entity(entity_ref: str) -> tuple[str, uuid.UUID]:
|
|
if entity_ref.startswith("CUS-"):
|
|
return "customer", customer_id_by_ref[entity_ref]
|
|
return "vehicle", vehicle_id_by_ref[entity_ref]
|
|
|
|
def _seed_signals(public_ref: str, entity_ref: str, related_refs: list[str]) -> list[dict]:
|
|
# The four named DQ-DEMO-* rows anchor the guided demo's scripted scenarios, so
|
|
# they carry real, accurate structured signals (not just a legacy English
|
|
# sentence) -- the frontend renders these as the primary, localized evidence;
|
|
# see docs/fleet-ops-correction/current-gap-audit.md §6.
|
|
if public_ref == "DQ-DEMO-DUPLICATE":
|
|
a = customer_row_by_ref[entity_ref]
|
|
b = customer_row_by_ref[related_refs[0]]
|
|
name_a = f"{a['first_name']} {a['last_name']}".strip().lower()
|
|
name_b = f"{b['first_name']} {b['last_name']}".strip().lower()
|
|
ratio = SequenceMatcher(None, name_a, name_b).ratio()
|
|
return [
|
|
{"code": "duplicate.exact_email"},
|
|
{"code": "duplicate.exact_phone"},
|
|
{"code": "duplicate.same_postal_code"},
|
|
{"code": "duplicate.similar_name", "params": {"score": round(ratio, 2)}},
|
|
]
|
|
if public_ref == "DQ-DEMO-OVERLAP":
|
|
return [{"code": "overlap.reserved_bookings", "params": {"refs": related_refs}}]
|
|
if public_ref == "DQ-DEMO-STATUS":
|
|
return [{"code": "vehicle.booking_conflict"}]
|
|
if public_ref == "DQ-DEMO-ATTENTION":
|
|
return [
|
|
{
|
|
"code": "attention.upcoming_booking_missing_inspection",
|
|
"params": {"booking_ref": related_refs[0] if related_refs else ""},
|
|
}
|
|
]
|
|
return []
|
|
|
|
dq_rows = []
|
|
now = datetime.now(UTC)
|
|
for row in _read_csv("data_quality_issues.csv"):
|
|
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 []
|
|
dq_rows.append(
|
|
{
|
|
"id": uuid.uuid4(),
|
|
"public_ref": row["public_ref"],
|
|
"rule_type": row["rule_type"],
|
|
"entity_type": entity_type,
|
|
"entity_id": entity_id,
|
|
"severity": row["severity"],
|
|
"status": row["status"],
|
|
"evidence_json": {
|
|
"summary": row["evidence"],
|
|
"entity_ref": row["entity_ref"],
|
|
"related_refs": related_refs,
|
|
"signals": _seed_signals(row["public_ref"], row["entity_ref"], related_refs),
|
|
},
|
|
"proposed_action_json": {},
|
|
"detected_at": now,
|
|
"resolved_at": now if row["status"] == "resolved" else None,
|
|
"resolved_by": "USR-OPS" if row["status"] == "resolved" else None,
|
|
}
|
|
)
|
|
db.execute(insert(DataQualityIssue), dq_rows)
|
|
counts["data_quality_issues"] = len(dq_rows)
|
|
|
|
vehicle_ref_by_booking_ref = {
|
|
row["public_ref"]: row["vehicle_ref"] for row in _read_csv("bookings.csv")
|
|
}
|
|
|
|
outbox_rows = []
|
|
for row in _read_csv("workflow_runs.csv"):
|
|
booking_id = booking_id_by_ref.get(row["aggregate_ref"])
|
|
# Build the same schema-complete envelope the live return workflow (M2) produces,
|
|
# so a seeded/historical event is redeliverable (e.g. via manual retry) without the
|
|
# dispatcher crashing on a missing key. See PROJECT_STATE.md M4 notes.
|
|
outbox_rows.append(
|
|
{
|
|
"event_id": uuid.UUID(row["event_id"]),
|
|
"event_type": row["event_type"],
|
|
"aggregate_type": "booking",
|
|
"aggregate_id": booking_id or uuid.uuid4(),
|
|
"payload_json": {
|
|
"correlation_id": str(uuid.uuid4()),
|
|
"aggregate": {
|
|
"type": "booking",
|
|
"id": str(booking_id or uuid.uuid4()),
|
|
"public_ref": row["aggregate_ref"],
|
|
},
|
|
"data": {
|
|
"vehicle_ref": vehicle_ref_by_booking_ref.get(row["aggregate_ref"], ""),
|
|
"inspection_ref": "",
|
|
"resulting_vehicle_status": "cleaning",
|
|
"attention_reasons": [],
|
|
},
|
|
"aggregate_ref": row["aggregate_ref"],
|
|
},
|
|
"occurred_at": _parse_dt(row["occurred_at"]) + shift,
|
|
"delivery_status": row["status"],
|
|
"attempts": int(row["attempts"]),
|
|
"next_attempt_at": None,
|
|
"last_error": row["last_error"] or None,
|
|
# The seed dataset's one synthetic failure (BK-H-0020) models a
|
|
# connection-timeout-style delivery failure -- see workflow_runs.csv.
|
|
# It is coded as a *prepared demo scenario*, not as a real
|
|
# connectionError, so integration health never degrades because of a
|
|
# prop and a viewer is told plainly that this failure is staged.
|
|
"last_error_code": DEMO_SCENARIO_ERROR_CODE if row["last_error"] else None,
|
|
"external_run_id": None,
|
|
}
|
|
)
|
|
db.execute(insert(OutboxEvent), outbox_rows)
|
|
counts["workflow_runs"] = len(outbox_rows)
|
|
|
|
seeded_at = datetime.now(UTC)
|
|
record_audit_event(
|
|
db,
|
|
actor_type="system",
|
|
actor_label="seed loader",
|
|
action="demo_data_seeded",
|
|
entity_type="system",
|
|
metadata={
|
|
"anchor_date": today.isoformat(),
|
|
"seed_authored_anchor": SEED_AUTHORED_ANCHOR.isoformat(),
|
|
"counts": counts,
|
|
},
|
|
)
|
|
|
|
return SeedResult(counts=counts, anchor_date=today, seeded_at=seeded_at)
|
|
|
|
|
|
def reset_and_seed(db: Session) -> SeedResult:
|
|
from app.services.data_quality import run_scan
|
|
|
|
clear_all(db)
|
|
result = load_seed(db)
|
|
db.commit()
|
|
scan = run_scan(db)
|
|
result.counts["data_quality_issues"] += sum(scan.created.values())
|
|
return result
|