test(devtools): publish rebuild safety differential lane

devtools/command_catalog.py

@@ -29,6 +29,7 @@
     "lab probe pipeline",
     "lab probe turso",
     "lab projections",
+    "lab run",
     "lab smoke",
     "lab schema audit",
     "lab schema commit",
@@ -2149,6 +2150,18 @@ class CatalogBypassSite:
         use_when="Assert memory budgets around a concrete query or archive-facing command.",
         examples=("devtools bench memory --max-rss-mb 1536 -- polylogue --plain analyze",),
     ),
+    CommandSpec(
+        "lab run",
+        "verification lab",
+        "Run a named archive verification scenario.",
+        "devtools.lab_scenario",
+        entrypoint="run_main",
+        use_when="Run a scenario such as rebuild-safety through the direct lab command path.",
+        examples=(
+            "devtools lab run rebuild-safety",
+            "devtools lab run rebuild-safety --report-dir .cache/rebuild-safety-report --json",
+        ),
+    ),
     CommandSpec(
         "lab smoke",
         "verification lab",

devtools/lab_scenario.py

@@ -7,12 +7,20 @@
 import subprocess
 import sys
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Protocol, TextIO
 
 from devtools import repo_root as _get_root
 from devtools.cli_boundary import invoke_polylogue_cli
+from devtools.rebuild_safety_scenario import (
+    REBUILD_DIFFERENTIAL_SCENARIO_NAME,
+    REBUILD_SAFETY_SCENARIO_NAME,
+    RebuildComparisonResult,
+    run_rebuild_differential,
+    run_rebuild_safety,
+)
 from devtools.storage_correctness_scenario import (
     STORAGE_CORRECTNESS_SCENARIO_NAME,
     run_storage_correctness,
@@ -25,7 +33,12 @@
 from polylogue.core.outcomes import OutcomeStatus
 from polylogue.scenarios import AssertionSpec, ExecutionSpec, polylogue_execution
 
-_SCENARIO_NAMES = ("archive-smoke", "reader-visual-smoke", STORAGE_CORRECTNESS_SCENARIO_NAME)
+_SCENARIO_NAMES = (
+    "archive-smoke",
+    "reader-visual-smoke",
+    STORAGE_CORRECTNESS_SCENARIO_NAME,
+    REBUILD_SAFETY_SCENARIO_NAME,
+)
 _ARCHIVE_SMOKE_TIER = 0
 
 
@@ -113,6 +126,77 @@ def failed_stages(self) -> tuple[str, ...]:
         return tuple(name for name, status in self.stage_statuses().items() if status is OutcomeStatus.ERROR)
 
 
+class RebuildSafetyResult:
+    """Direct result wrapper for the derived-tier rebuild lab lane."""
+
+    def __init__(self, *, report_dir: Path | None) -> None:
+        self.report_dir = report_dir
+        self.safety, self.safety_error = self._run("rebuild-safety", run_rebuild_safety)
+        self.differential, self.differential_error = self._run("rebuild-differential", run_rebuild_differential)
+        self._write_report()
+
+    @staticmethod
+    def _run(
+        name: str, runner: Callable[[], RebuildComparisonResult]
+    ) -> tuple[RebuildComparisonResult | None, str | None]:
+        try:
+            return runner(), None
+        except Exception as exc:
+            return None, f"{name} failed: {type(exc).__name__}: {exc}"
+
+    @staticmethod
+    def _report(value: RebuildComparisonResult | None, error: str | None) -> str:
+        if error is not None:
+            return error
+        assert value is not None
+        return value.format_report()
+
+    def _write_report(self) -> None:
+        if self.report_dir is None:
+            return
+        self.report_dir.mkdir(parents=True, exist_ok=True)
+        (self.report_dir / "rebuild-safety.txt").write_text(
+            f"{self._report(self.safety, self.safety_error)}\n\n"
+            f"{self._report(self.differential, self.differential_error)}\n",
+            encoding="utf-8",
+        )
+
+    @property
+    def scenario_name(self) -> str:
+        return REBUILD_SAFETY_SCENARIO_NAME
+
+    @property
+    def all_passed(self) -> bool:
+        return (
+            self.safety_error is None
+            and self.differential_error is None
+            and self.safety is not None
+            and self.differential is not None
+            and self.safety.all_passed
+            and self.differential.all_passed
+        )
+
+    def stage_statuses(self) -> dict[str, OutcomeStatus]:
+        return {
+            REBUILD_SAFETY_SCENARIO_NAME: OutcomeStatus.OK
+            if self.safety_error is None and self.safety is not None and self.safety.all_passed
+            else OutcomeStatus.ERROR,
+            REBUILD_DIFFERENTIAL_SCENARIO_NAME: OutcomeStatus.OK
+            if self.differential_error is None and self.differential is not None and self.differential.all_passed
+            else OutcomeStatus.ERROR,
+        }
+
+    def failed_stages(self) -> tuple[str, ...]:
+        return tuple(name for name, status in self.stage_statuses().items() if status is OutcomeStatus.ERROR)
+
+    def extra_payload(self) -> dict[str, object]:
+        payload: dict[str, object] = {
+            "safety_report": self._report(self.safety, self.safety_error),
+            "differential_report": self._report(self.differential, self.differential_error),
+        }
+        return payload
+
+
 def get_archive_smoke_checks() -> tuple[ArchiveSmokeCheck, ...]:
     """Return direct CLI checks for the archive-smoke lab smoke."""
     return _ARCHIVE_SMOKE_CHECKS
@@ -171,6 +255,11 @@ def list_scenarios(*, as_json: bool) -> int:
             "artifact_count": len(reader_visual_artifact_payloads()),
         },
         storage_correctness_scenario_entry(),
+        {
+            "name": REBUILD_SAFETY_SCENARIO_NAME,
+            "kind": "derived-tier-differential",
+            "checks": [REBUILD_SAFETY_SCENARIO_NAME, REBUILD_DIFFERENTIAL_SCENARIO_NAME],
+        },
     ]
     payload = {"scenarios": scenarios}
     if as_json:
@@ -184,6 +273,9 @@ def list_scenarios(*, as_json: bool) -> int:
         if name == "storage-correctness":
             print(f"{name:<20s}  checks: {entry['check_count']}")
             continue
+        if name == REBUILD_SAFETY_SCENARIO_NAME:
+            print(f"{name:<20s}  checks: rebuild safety + differential")
+            continue
         print(f"{name:<20s}  tier-0 checks: {entry['tier_0_check_count']}")
     return 0
 
@@ -379,6 +471,8 @@ def main(argv: list[str] | None = None) -> int:
     result: _ScenarioResult
     if args.scenario == "storage-correctness":
         result = run_storage_correctness(report_dir=args.report_dir)
+    elif args.scenario == REBUILD_SAFETY_SCENARIO_NAME:
+        result = RebuildSafetyResult(report_dir=args.report_dir)
     else:
         result = run_archive_smoke(
             live=bool(args.live),
@@ -395,5 +489,10 @@ def main(argv: list[str] | None = None) -> int:
     return 0 if result.all_passed else 1
 
 
+def run_main(argv: list[str] | None = None) -> int:
+    """Run a named lab scenario through the advertised ``devtools lab run`` route."""
+    return main(["run", *(argv or [])])
+
+
 if __name__ == "__main__":
     raise SystemExit(main())