|
| 1 | +"""Executable onboarding and quickstart flow for IX-Style.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from dataclasses import dataclass, field |
| 6 | +from pathlib import Path |
| 7 | +from typing import Any |
| 8 | + |
| 9 | +from ix_style.messages import EvidenceBundleBuilder |
| 10 | + |
| 11 | +from .invariants import InvariantChecker |
| 12 | +from .reports import JsonArtifactIO, ReviewArtifactBuilder |
| 13 | +from .repository_audit import RepositorySelfAuditor |
| 14 | +from .runner import ScenarioRunner |
| 15 | +from .sample_scenarios import ( |
| 16 | + build_nav_spoof_transition_scenario, |
| 17 | + build_power_fault_clamp_scenario, |
| 18 | + build_recovery_deferred_scenario, |
| 19 | +) |
| 20 | + |
| 21 | + |
| 22 | +@dataclass(slots=True, frozen=True) |
| 23 | +class QuickstartScenarioSummary: |
| 24 | + """Summary of one quickstart scenario execution.""" |
| 25 | + |
| 26 | + name: str |
| 27 | + scenario_id: str |
| 28 | + scenario_passed: bool |
| 29 | + invariants_passed: bool |
| 30 | + bundle_valid: bool |
| 31 | + final_outcome: str |
| 32 | + dominant_safety_posture: str |
| 33 | + operator_headline: str |
| 34 | + exported_review_dir: str |
| 35 | + |
| 36 | + def as_dict(self) -> dict[str, Any]: |
| 37 | + """Return a JSON-friendly dictionary.""" |
| 38 | + return { |
| 39 | + "name": self.name, |
| 40 | + "scenario_id": self.scenario_id, |
| 41 | + "scenario_passed": self.scenario_passed, |
| 42 | + "invariants_passed": self.invariants_passed, |
| 43 | + "bundle_valid": self.bundle_valid, |
| 44 | + "final_outcome": self.final_outcome, |
| 45 | + "dominant_safety_posture": self.dominant_safety_posture, |
| 46 | + "operator_headline": self.operator_headline, |
| 47 | + "exported_review_dir": self.exported_review_dir, |
| 48 | + } |
| 49 | + |
| 50 | + |
| 51 | +@dataclass(slots=True, frozen=True) |
| 52 | +class QuickstartRunSummary: |
| 53 | + """Aggregated quickstart execution summary.""" |
| 54 | + |
| 55 | + overall_passed: bool |
| 56 | + audit_passed: bool |
| 57 | + audit_report: dict[str, Any] |
| 58 | + scenario_results: tuple[QuickstartScenarioSummary, ...] |
| 59 | + exported_root: str |
| 60 | + |
| 61 | + def as_dict(self) -> dict[str, Any]: |
| 62 | + """Return a JSON-friendly dictionary.""" |
| 63 | + return { |
| 64 | + "overall_passed": self.overall_passed, |
| 65 | + "audit_passed": self.audit_passed, |
| 66 | + "audit_report": self.audit_report, |
| 67 | + "scenario_results": [item.as_dict() for item in self.scenario_results], |
| 68 | + "exported_root": self.exported_root, |
| 69 | + } |
| 70 | + |
| 71 | + |
| 72 | +@dataclass(slots=True) |
| 73 | +class QuickstartRunner: |
| 74 | + """Runs the baseline onboarding flow end-to-end.""" |
| 75 | + |
| 76 | + self_auditor: RepositorySelfAuditor = field(default_factory=RepositorySelfAuditor) |
| 77 | + scenario_runner: ScenarioRunner = field(default_factory=ScenarioRunner) |
| 78 | + invariant_checker: InvariantChecker = field(default_factory=InvariantChecker) |
| 79 | + review_builder: ReviewArtifactBuilder = field(default_factory=ReviewArtifactBuilder) |
| 80 | + artifact_io: JsonArtifactIO = field(default_factory=JsonArtifactIO) |
| 81 | + bundle_builder: EvidenceBundleBuilder = field(default_factory=EvidenceBundleBuilder) |
| 82 | + |
| 83 | + def run(self, output_root: str | Path) -> QuickstartRunSummary: |
| 84 | + """Run the baseline onboarding flow and export review artifacts.""" |
| 85 | + output_dir = Path(output_root) |
| 86 | + output_dir.mkdir(parents=True, exist_ok=True) |
| 87 | + |
| 88 | + audit_report = self.self_auditor.run() |
| 89 | + scenario_results: list[QuickstartScenarioSummary] = [] |
| 90 | + |
| 91 | + scenarios = ( |
| 92 | + ("power_fault_clamp", build_power_fault_clamp_scenario()), |
| 93 | + ("nav_spoof_transition", build_nav_spoof_transition_scenario()), |
| 94 | + ("recovery_deferred", build_recovery_deferred_scenario()), |
| 95 | + ) |
| 96 | + |
| 97 | + for name, scenario in scenarios: |
| 98 | + result = self.scenario_runner.run(scenario) |
| 99 | + invariant_report = self.invariant_checker.evaluate(result) |
| 100 | + review_package = self.review_builder.build(result) |
| 101 | + |
| 102 | + export_dir = output_dir / name |
| 103 | + self.artifact_io.export_package(review_package, export_dir) |
| 104 | + |
| 105 | + bundle_errors = self.bundle_builder.validate(review_package.evidence_bundle) |
| 106 | + scenario_results.append( |
| 107 | + QuickstartScenarioSummary( |
| 108 | + name=name, |
| 109 | + scenario_id=scenario.scenario_id, |
| 110 | + scenario_passed=result.passed, |
| 111 | + invariants_passed=invariant_report.passed, |
| 112 | + bundle_valid=not bundle_errors, |
| 113 | + final_outcome=str( |
| 114 | + review_package.decision_receipt["final_outcome"] |
| 115 | + ), |
| 116 | + dominant_safety_posture=str( |
| 117 | + review_package.mission_health_snapshot["dominant_safety_posture"] |
| 118 | + ), |
| 119 | + operator_headline=str( |
| 120 | + review_package.operator_safety_summary["headline"] |
| 121 | + ), |
| 122 | + exported_review_dir=str(export_dir), |
| 123 | + ) |
| 124 | + ) |
| 125 | + |
| 126 | + overall_passed = audit_report.passed and all( |
| 127 | + item.scenario_passed and item.invariants_passed and item.bundle_valid |
| 128 | + for item in scenario_results |
| 129 | + ) |
| 130 | + |
| 131 | + return QuickstartRunSummary( |
| 132 | + overall_passed=overall_passed, |
| 133 | + audit_passed=audit_report.passed, |
| 134 | + audit_report=audit_report.as_dict(), |
| 135 | + scenario_results=tuple(scenario_results), |
| 136 | + exported_root=str(output_dir), |
| 137 | + ) |
0 commit comments