ceremony.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Stage 3 upgrade ceremony classifiers, gates, and orchestrator (§O2.3–§O2.7).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import re |
| 6 | import shlex |
| 7 | import stat |
| 8 | from argparse import Namespace |
| 9 | from dataclasses import dataclass, field |
| 10 | from enum import Enum |
| 11 | from pathlib import Path |
| 12 | from typing import Any |
| 13 | |
| 14 | from adapters.config import OverseerConfig, load_config |
| 15 | from adapters.errors import ConfigError |
| 16 | from adapters.runner import CommandRunner, quote_arg |
| 17 | from cli.atomic import WriteFailure, atomic_write_text |
| 18 | from cli.config_gen import config_dict_to_yaml, config_to_dict, load_config_from_dict |
| 19 | from cli.context import CliContext |
| 20 | from cli.footprint import ( |
| 21 | MUSE_BRIDGE_DEPLOY_DEST, |
| 22 | MUSE_BRIDGE_WORKFLOW_DEST, |
| 23 | resolve_footprint, |
| 24 | ) |
| 25 | from cli.paths import resolve_config_path, resolve_repo_root |
| 26 | from cli.sanitize import format_config_error, sanitize_text |
| 27 | from cli.version_lock import read_version_lock |
| 28 | |
| 29 | SUPPORTED_FROM = "muse-only" |
| 30 | SUPPORTED_TO = "muse+git-mirror" |
| 31 | BRIDGE_DESTINATIONS = frozenset({MUSE_BRIDGE_WORKFLOW_DEST, MUSE_BRIDGE_DEPLOY_DEST}) |
| 32 | |
| 33 | # Heuristic secret / home-path patterns for G6 (aligned with K7 security tests). |
| 34 | _SECRET_ASSIGN_RE = re.compile( |
| 35 | r"(?i)(?:api[_-]?key|secret|password|token)\s*=\s*['\"][^'\"]{8,}['\"]", |
| 36 | ) |
| 37 | _HOME_PATH_RE = re.compile(r"(?:/Users/|/home/[a-zA-Z])") |
| 38 | _AWS_KEY_RE = re.compile(r"AKIA[0-9A-Z]{16}") |
| 39 | |
| 40 | |
| 41 | class StartState(str, Enum): |
| 42 | """C1 start-state classification (§O2.3).""" |
| 43 | |
| 44 | MUSE_ONLY = "muse-only" |
| 45 | COMPLETE_UPGRADE = "complete-upgrade" |
| 46 | INCOMPLETE_UPGRADE = "incomplete-upgrade" |
| 47 | WRONG_REGIME = "wrong-regime" |
| 48 | MISSING_CONFIG = "missing-config" |
| 49 | |
| 50 | |
| 51 | @dataclass |
| 52 | class GateResult: |
| 53 | """One bridge dry-run gate outcome.""" |
| 54 | |
| 55 | gate_id: str |
| 56 | ok: bool |
| 57 | detail: str |
| 58 | |
| 59 | |
| 60 | @dataclass |
| 61 | class UpgradeReport: |
| 62 | """Machine-readable ceremony report (§O2.5 G7 / C6).""" |
| 63 | |
| 64 | start_state: str = "" |
| 65 | steps_planned: list[str] = field(default_factory=list) |
| 66 | steps_completed: list[str] = field(default_factory=list) |
| 67 | gates: dict[str, dict[str, Any]] = field(default_factory=dict) |
| 68 | ready_for_live_bridge: bool = False |
| 69 | ready_for_footprint: bool = False |
| 70 | dry_run: bool = True |
| 71 | apply: bool = False |
| 72 | live_bridge: bool = False |
| 73 | live_bridge_invoked: bool = False |
| 74 | next_live_step: str = ( |
| 75 | "Run ./scripts/muse-bridge-deploy.sh → push muse-mirror → open PR to main; " |
| 76 | "merge muse-mirror → main remains Tier 3 (never auto)." |
| 77 | ) |
| 78 | hard_stop_c8: str = "Merge muse-mirror → main is Tier 3 — ceremony stops before C8." |
| 79 | errors: list[str] = field(default_factory=list) |
| 80 | warnings: list[str] = field(default_factory=list) |
| 81 | docs_preserved: bool = True |
| 82 | config_written: bool = False |
| 83 | footprint_seeded: bool = False |
| 84 | |
| 85 | def to_payload(self) -> dict[str, Any]: |
| 86 | return { |
| 87 | "start_state": self.start_state, |
| 88 | "steps_planned": list(self.steps_planned), |
| 89 | "steps_completed": list(self.steps_completed), |
| 90 | "gates": dict(self.gates), |
| 91 | "ready_for_live_bridge": self.ready_for_live_bridge, |
| 92 | "ready_for_footprint": self.ready_for_footprint, |
| 93 | "dry_run": self.dry_run, |
| 94 | "apply": self.apply, |
| 95 | "live_bridge": self.live_bridge, |
| 96 | "live_bridge_invoked": self.live_bridge_invoked, |
| 97 | "next_live_step": self.next_live_step, |
| 98 | "hard_stop_c8": self.hard_stop_c8, |
| 99 | "errors": list(self.errors), |
| 100 | "warnings": list(self.warnings), |
| 101 | "docs_preserved": self.docs_preserved, |
| 102 | "config_written": self.config_written, |
| 103 | "footprint_seeded": self.footprint_seeded, |
| 104 | } |
| 105 | |
| 106 | |
| 107 | def required_vcs_complete(config: OverseerConfig) -> bool: |
| 108 | """True when on-disk VCS block satisfies §O2.4.2 minimum for muse+git-mirror.""" |
| 109 | vcs = config.vcs |
| 110 | if vcs.regime != SUPPORTED_TO: |
| 111 | return False |
| 112 | if vcs.canonical != "muse": |
| 113 | return False |
| 114 | if not vcs.git.mirror_branch: |
| 115 | return False |
| 116 | if not vcs.git.remote or not vcs.git.main_branch: |
| 117 | return False |
| 118 | if not vcs.muse.staging_remote: |
| 119 | return False |
| 120 | if not vcs.muse.main_branch: |
| 121 | return False |
| 122 | return True |
| 123 | |
| 124 | |
| 125 | def is_silent_regime_only_patch(before: OverseerConfig | None, after_dict: dict) -> bool: |
| 126 | """Detect forbidden silent edit: regime flip without required mirror fields (§O2.4.1).""" |
| 127 | if before is None: |
| 128 | return False |
| 129 | after_vcs = after_dict.get("vcs") or {} |
| 130 | if after_vcs.get("regime") != SUPPORTED_TO: |
| 131 | return False |
| 132 | git = after_vcs.get("git") or {} |
| 133 | muse = after_vcs.get("muse") or {} |
| 134 | incomplete = ( |
| 135 | not git.get("mirror_branch") |
| 136 | or after_vcs.get("canonical") != "muse" |
| 137 | or not muse.get("staging_remote") |
| 138 | or not muse.get("main_branch") |
| 139 | ) |
| 140 | if not incomplete: |
| 141 | return False |
| 142 | # Regime-only (or regime + partial) relative to a muse-only start. |
| 143 | return before.vcs.regime == SUPPORTED_FROM |
| 144 | |
| 145 | |
| 146 | def docs_preserved(before: OverseerConfig, after: OverseerConfig) -> bool: |
| 147 | """Living-doc paths and repo identity must survive C2 (§O2.4.2).""" |
| 148 | return ( |
| 149 | before.docs.handover == after.docs.handover |
| 150 | and before.docs.roadmap == after.docs.roadmap |
| 151 | and before.docs.coordination == after.docs.coordination |
| 152 | and before.docs.standing_decisions == after.docs.standing_decisions |
| 153 | and before.repo.name == after.repo.name |
| 154 | and before.repo.root_relative_docs == after.repo.root_relative_docs |
| 155 | ) |
| 156 | |
| 157 | |
| 158 | def build_upgraded_config_dict(config: OverseerConfig) -> dict: |
| 159 | """Build muse+git-mirror config preserving docs.*/repo.*/thresholds./freeze (§O2.4.2).""" |
| 160 | data = config_to_dict(config) |
| 161 | prev_git = data["vcs"]["git"] |
| 162 | prev_muse = data["vcs"]["muse"] |
| 163 | data["vcs"] = { |
| 164 | "regime": SUPPORTED_TO, |
| 165 | "canonical": "muse", |
| 166 | "git": { |
| 167 | "remote": prev_git.get("remote") or "origin", |
| 168 | "main_branch": prev_git.get("main_branch") or "main", |
| 169 | "mirror_branch": prev_git.get("mirror_branch") or "muse-mirror", |
| 170 | "feature_branch_pattern": prev_git.get("feature_branch_pattern") |
| 171 | or "feat/{slug}", |
| 172 | }, |
| 173 | "muse": { |
| 174 | "staging_remote": prev_muse.get("staging_remote") or "staging", |
| 175 | "main_branch": prev_muse.get("main_branch") or "main", |
| 176 | "working_dir": prev_muse.get("working_dir"), |
| 177 | }, |
| 178 | } |
| 179 | return data |
| 180 | |
| 181 | |
| 182 | def bridge_footprint_present(repo_root: Path, config: OverseerConfig) -> bool: |
| 183 | """True when both bridge destinations are in lock + on disk (executable script).""" |
| 184 | lock_file = repo_root / ".overseer" / "version.lock" |
| 185 | if not lock_file.is_file(): |
| 186 | return False |
| 187 | try: |
| 188 | lock = read_version_lock(lock_file) |
| 189 | except (OSError, ValueError, KeyError): |
| 190 | return False |
| 191 | locked = {e.path for e in lock.footprint} |
| 192 | if not BRIDGE_DESTINATIONS.issubset(locked): |
| 193 | return False |
| 194 | workflow = repo_root / MUSE_BRIDGE_WORKFLOW_DEST |
| 195 | script = repo_root / MUSE_BRIDGE_DEPLOY_DEST |
| 196 | if not workflow.is_file() or not script.is_file(): |
| 197 | return False |
| 198 | mode = script.stat().st_mode |
| 199 | if not (mode & stat.S_IXUSR): |
| 200 | return False |
| 201 | # Config must resolve bridge destinations for this regime. |
| 202 | dests = {f.destination for f in resolve_footprint(config)} |
| 203 | return BRIDGE_DESTINATIONS.issubset(dests) |
| 204 | |
| 205 | |
| 206 | def classify_start_state(repo_root: Path, config_path: Path) -> StartState: |
| 207 | """C1 classifier (§O2.3).""" |
| 208 | if not config_path.is_file(): |
| 209 | return StartState.MISSING_CONFIG |
| 210 | try: |
| 211 | config = load_config(config_path) |
| 212 | except ConfigError: |
| 213 | return StartState.WRONG_REGIME |
| 214 | |
| 215 | regime = config.vcs.regime |
| 216 | if regime == SUPPORTED_FROM: |
| 217 | return StartState.MUSE_ONLY |
| 218 | if regime == SUPPORTED_TO: |
| 219 | if required_vcs_complete(config) and bridge_footprint_present(repo_root, config): |
| 220 | return StartState.COMPLETE_UPGRADE |
| 221 | return StartState.INCOMPLETE_UPGRADE |
| 222 | return StartState.WRONG_REGIME |
| 223 | |
| 224 | |
| 225 | def check_g3_deploy_script(script: str) -> GateResult: |
| 226 | """G3: S3 refusal (mirror ≠ root) and never export to ``--git-dir .``.""" |
| 227 | if "mirror directory equals repo root" not in script and "MIRROR_ABS == REPO_ABS" not in script: |
| 228 | return GateResult("G3", False, "missing mirror≠root refusal") |
| 229 | # Live export must use absolute mirror path variable, not literal `.` |
| 230 | if re.search(r"""--git-dir\s+['\"]?\.(?:['\"]|\s|$)""", script): |
| 231 | return GateResult("G3", False, "script instructs --git-dir .") |
| 232 | if "--git-dir" in script and "MIRROR_ABS" not in script and "MIRROR_DIR" not in script: |
| 233 | return GateResult("G3", False, "export target not isolated mirror path") |
| 234 | return GateResult("G3", True, "S3 mirror-isolation present") |
| 235 | |
| 236 | |
| 237 | def check_g4_deploy_script(script: str) -> GateResult: |
| 238 | """G4: never push main; publish via mirror_branch only (S8/S13).""" |
| 239 | if "git push origin main" in script: |
| 240 | return GateResult("G4", False, "literal push origin main present") |
| 241 | if 'push "${GIT_REMOTE}" "${MAIN_BRANCH}"' in script: |
| 242 | return GateResult("G4", False, "push of MAIN_BRANCH present") |
| 243 | if 'push "${GIT_REMOTE}" "${MIRROR_BRANCH}"' not in script and "MIRROR_BRANCH" not in script: |
| 244 | return GateResult("G4", False, "mirror_branch push path missing") |
| 245 | return GateResult("G4", True, "mirror-only publish path") |
| 246 | |
| 247 | |
| 248 | def check_g5_deploy_script(script: str) -> GateResult: |
| 249 | """G5: cwd-safe ``muse -C`` + absolute ``--git-dir`` (S7).""" |
| 250 | if 'muse -C "' not in script and "muse -C '${" not in script and 'muse -C "${' not in script: |
| 251 | if "muse -C" not in script: |
| 252 | return GateResult("G5", False, "missing muse -C") |
| 253 | if '--git-dir "${MIRROR_ABS}"' not in script and "--git-dir" not in script: |
| 254 | return GateResult("G5", False, "missing absolute --git-dir mirror path") |
| 255 | return GateResult("G5", True, "muse -C + absolute --git-dir") |
| 256 | |
| 257 | |
| 258 | def check_g6_deploy_script(script: str) -> GateResult: |
| 259 | """G6: no secret-assignment / absolute home paths (S11).""" |
| 260 | if _HOME_PATH_RE.search(script): |
| 261 | return GateResult("G6", False, "absolute operator home path in script") |
| 262 | if _AWS_KEY_RE.search(script) or _SECRET_ASSIGN_RE.search(script): |
| 263 | return GateResult("G6", False, "secret-assignment pattern in script") |
| 264 | return GateResult("G6", True, "no secrets / home paths") |
| 265 | |
| 266 | |
| 267 | def check_g8_git_remote( |
| 268 | repo_root: Path, |
| 269 | remote_name: str, |
| 270 | runner: CommandRunner, |
| 271 | ) -> GateResult: |
| 272 | """G8: local read-only ``git remote get-url`` returns non-empty URL (no network fetch).""" |
| 273 | cmd = f"git remote get-url {quote_arg(remote_name)}" |
| 274 | result = runner.run(cmd, cwd=str(repo_root)) |
| 275 | url = (result.stdout or "").strip() |
| 276 | if result.exit_code != 0 or not url: |
| 277 | return GateResult( |
| 278 | "G8", |
| 279 | False, |
| 280 | f"git remote {remote_name!r} has no usable URL (create repo + remote before C7)", |
| 281 | ) |
| 282 | return GateResult("G8", True, f"remote {remote_name} URL present") |
| 283 | |
| 284 | |
| 285 | def evaluate_bridge_gates( |
| 286 | *, |
| 287 | config: OverseerConfig, |
| 288 | repo_root: Path, |
| 289 | runner: CommandRunner, |
| 290 | ) -> list[GateResult]: |
| 291 | """Evaluate G1–G8 without live export (§O2.5).""" |
| 292 | results: list[GateResult] = [] |
| 293 | |
| 294 | # G1 — config shape |
| 295 | g1_ok = required_vcs_complete(config) |
| 296 | results.append( |
| 297 | GateResult( |
| 298 | "G1", |
| 299 | g1_ok, |
| 300 | "vcs muse+git-mirror complete" if g1_ok else "vcs shape incomplete for muse+git-mirror", |
| 301 | ) |
| 302 | ) |
| 303 | |
| 304 | # G2 — footprint lock + disk + executable |
| 305 | g2_ok = bridge_footprint_present(repo_root, config) |
| 306 | results.append( |
| 307 | GateResult( |
| 308 | "G2", |
| 309 | g2_ok, |
| 310 | "bridge footprint present" if g2_ok else "bridge destinations missing from lock/disk", |
| 311 | ) |
| 312 | ) |
| 313 | |
| 314 | script_path = repo_root / MUSE_BRIDGE_DEPLOY_DEST |
| 315 | script_text = script_path.read_text(encoding="utf-8") if script_path.is_file() else "" |
| 316 | |
| 317 | results.append(check_g3_deploy_script(script_text) if script_text else GateResult("G3", False, "deploy script missing")) |
| 318 | results.append(check_g4_deploy_script(script_text) if script_text else GateResult("G4", False, "deploy script missing")) |
| 319 | results.append(check_g5_deploy_script(script_text) if script_text else GateResult("G5", False, "deploy script missing")) |
| 320 | results.append(check_g6_deploy_script(script_text) if script_text else GateResult("G6", False, "deploy script missing")) |
| 321 | |
| 322 | # G7 — informational always "pass" when report includes next_live_step (orchestrator sets it) |
| 323 | results.append( |
| 324 | GateResult( |
| 325 | "G7", |
| 326 | True, |
| 327 | "next live step is deploy script → muse-mirror PR; merge remains Tier 3", |
| 328 | ) |
| 329 | ) |
| 330 | |
| 331 | remote = config.vcs.git.remote |
| 332 | results.append(check_g8_git_remote(repo_root, remote, runner)) |
| 333 | return results |
| 334 | |
| 335 | |
| 336 | def _path_escape_ok(repo_root: Path, target: Path) -> bool: |
| 337 | try: |
| 338 | target.resolve().relative_to(repo_root.resolve()) |
| 339 | return True |
| 340 | except ValueError: |
| 341 | return False |
| 342 | |
| 343 | |
| 344 | def run_upgrade_regime(args: Namespace, ctx: CliContext) -> tuple[int, UpgradeReport]: |
| 345 | """Orchestrate C0–C5 (and optional C7); hard-stop before C8 (§O2.3 / §O2.7).""" |
| 346 | report = UpgradeReport() |
| 347 | |
| 348 | from_regime = getattr(args, "from_regime", None) or getattr(args, "from", None) |
| 349 | to_regime = getattr(args, "to_regime", None) or getattr(args, "to", None) |
| 350 | if from_regime != SUPPORTED_FROM or to_regime != SUPPORTED_TO: |
| 351 | report.errors.append( |
| 352 | f"refused: only supported pair is --from {SUPPORTED_FROM} --to {SUPPORTED_TO}" |
| 353 | ) |
| 354 | return 4, report |
| 355 | |
| 356 | apply = bool(getattr(args, "apply", False)) |
| 357 | live_bridge = bool(getattr(args, "live_bridge", False)) |
| 358 | force = bool(getattr(args, "force", False)) |
| 359 | yes = bool(getattr(args, "yes", False)) |
| 360 | # Default is dry-run unless --apply (or live with apply path). |
| 361 | dry_run = bool(getattr(args, "dry_run", False)) or not apply |
| 362 | if live_bridge and not apply: |
| 363 | # Live bridge implies apply first unless already complete (handled below). |
| 364 | apply = True |
| 365 | dry_run = False |
| 366 | |
| 367 | report.dry_run = dry_run |
| 368 | report.apply = apply |
| 369 | report.live_bridge = live_bridge |
| 370 | |
| 371 | try: |
| 372 | repo_root = resolve_repo_root(cwd=ctx.cwd, repo_arg=getattr(args, "repo", None), command="upgrade-regime") |
| 373 | except Exception as exc: # noqa: BLE001 — path resolution failures are refuse |
| 374 | report.errors.append(sanitize_text(str(exc), ctx.cwd)) |
| 375 | return 4, report |
| 376 | |
| 377 | config_path = resolve_config_path(repo_root, getattr(args, "config", None)) |
| 378 | if not _path_escape_ok(repo_root, config_path): |
| 379 | report.errors.append("refused: config path outside repo root") |
| 380 | return 4, report |
| 381 | |
| 382 | # C0 prerequisites (non-writing) |
| 383 | report.steps_planned.append("C0") |
| 384 | report.steps_completed.append("C0") |
| 385 | |
| 386 | # C1 start-state |
| 387 | start = classify_start_state(repo_root, config_path) |
| 388 | report.start_state = start.value |
| 389 | report.steps_planned.append("C1") |
| 390 | report.steps_completed.append("C1") |
| 391 | |
| 392 | if start in {StartState.WRONG_REGIME, StartState.MISSING_CONFIG}: |
| 393 | report.errors.append( |
| 394 | "refused: wrong ceremony start state " |
| 395 | f"({start.value}); use greenfield ok init or a later git→mirror freeze" |
| 396 | ) |
| 397 | return 4, report |
| 398 | |
| 399 | if start == StartState.COMPLETE_UPGRADE: |
| 400 | config = load_config(config_path) |
| 401 | gates = evaluate_bridge_gates(config=config, repo_root=repo_root, runner=ctx.runner) |
| 402 | report.gates = {g.gate_id: {"ok": g.ok, "detail": g.detail} for g in gates} |
| 403 | g1_g7 = all(g.ok for g in gates if g.gate_id != "G8") |
| 404 | g8 = next(g for g in gates if g.gate_id == "G8") |
| 405 | report.ready_for_footprint = g1_g7 |
| 406 | report.ready_for_live_bridge = all(g.ok for g in gates) |
| 407 | report.steps_planned.extend(["C5", "G1-G8"]) |
| 408 | report.steps_completed.extend(["C5", "G1-G8"]) |
| 409 | if live_bridge: |
| 410 | if not report.ready_for_live_bridge or not yes: |
| 411 | report.errors.append( |
| 412 | "refused: --live-bridge requires G1–G8 pass and -y/--yes consent (C6)" |
| 413 | ) |
| 414 | return 4, report |
| 415 | return _invoke_live_bridge(repo_root, ctx, report) |
| 416 | report.warnings.append("idempotent success: already muse+git-mirror with bridge footprint") |
| 417 | return 0, report |
| 418 | |
| 419 | # Load pre-upgrade config |
| 420 | try: |
| 421 | before = load_config(config_path) |
| 422 | except ConfigError as exc: |
| 423 | report.errors.append(format_config_error(exc, repo_root)) |
| 424 | return 2, report |
| 425 | |
| 426 | need_c2 = start == StartState.MUSE_ONLY or not required_vcs_complete(before) |
| 427 | report.steps_planned.extend(["C2", "C3", "C4", "C5"] if need_c2 else ["C3", "C4", "C5"]) |
| 428 | |
| 429 | upgraded_dict = build_upgraded_config_dict(before) |
| 430 | if is_silent_regime_only_patch(before, upgraded_dict): |
| 431 | report.errors.append( |
| 432 | "refused: silent regime-only patch detected (complete VCS fields required)" |
| 433 | ) |
| 434 | return 4, report |
| 435 | |
| 436 | try: |
| 437 | upgraded = load_config_from_dict(upgraded_dict, str(config_path)) |
| 438 | except ConfigError as exc: |
| 439 | report.errors.append(format_config_error(exc, repo_root)) |
| 440 | return 2, report |
| 441 | |
| 442 | if not required_vcs_complete(upgraded): |
| 443 | report.errors.append("refused: projected post-upgrade VCS incomplete") |
| 444 | return 4, report |
| 445 | |
| 446 | report.docs_preserved = docs_preserved(before, upgraded) |
| 447 | if not report.docs_preserved: |
| 448 | report.errors.append("refused: docs.*/repo.* would not be preserved") |
| 449 | return 4, report |
| 450 | # Dry-run: plan only through C5/G1–G8; no writes. |
| 451 | if dry_run and not apply: |
| 452 | for step in report.steps_planned: |
| 453 | if step not in report.steps_completed: |
| 454 | report.steps_completed.append(step) |
| 455 | probe_config = upgraded if need_c2 else before |
| 456 | gates = evaluate_bridge_gates(config=probe_config, repo_root=repo_root, runner=ctx.runner) |
| 457 | # Projected G1 always reflects the post-upgrade VCS shape. |
| 458 | gates = [ |
| 459 | GateResult("G1", True, "projected vcs complete") if g.gate_id == "G1" else g |
| 460 | for g in gates |
| 461 | ] |
| 462 | if start == StartState.MUSE_ONLY and not bridge_footprint_present(repo_root, upgraded): |
| 463 | patched: list[GateResult] = [] |
| 464 | for g in gates: |
| 465 | if g.gate_id in {"G2", "G3", "G4", "G5", "G6"} and not g.ok: |
| 466 | patched.append(GateResult(g.gate_id, False, "requires --apply footprint seed")) |
| 467 | else: |
| 468 | patched.append(g) |
| 469 | gates = patched |
| 470 | report.warnings.append("dry-run: no writes; run --apply to perform C2–C4") |
| 471 | report.gates = {g.gate_id: {"ok": g.ok, "detail": g.detail} for g in gates} |
| 472 | report.ready_for_footprint = all( |
| 473 | report.gates[g]["ok"] for g in ("G1", "G2", "G3", "G4", "G5", "G6", "G7") |
| 474 | ) |
| 475 | report.ready_for_live_bridge = all(g["ok"] for g in report.gates.values()) |
| 476 | if live_bridge: |
| 477 | report.errors.append("refused: --live-bridge requires successful --apply path + G1–G8") |
| 478 | return 4, report |
| 479 | return 0, report |
| 480 | |
| 481 | # --- Apply path: C2–C4 writes --- |
| 482 | if need_c2: |
| 483 | try: |
| 484 | atomic_write_text(config_path, config_dict_to_yaml(upgraded_dict)) |
| 485 | report.config_written = True |
| 486 | report.steps_completed.append("C2") |
| 487 | except WriteFailure as exc: |
| 488 | report.errors.append(sanitize_text(str(exc), repo_root)) |
| 489 | return 5, report |
| 490 | else: |
| 491 | report.steps_completed.append("C2-skipped") |
| 492 | |
| 493 | # Reload post-C2 |
| 494 | try: |
| 495 | config = load_config(config_path) |
| 496 | except ConfigError as exc: |
| 497 | report.errors.append(format_config_error(exc, repo_root)) |
| 498 | return 2, report |
| 499 | |
| 500 | if config.vcs.regime != SUPPORTED_TO: |
| 501 | report.errors.append("refused: post-C2 config did not load as muse+git-mirror") |
| 502 | return 4, report |
| 503 | |
| 504 | # C3 footprint re-seed via sync (never --include-preserved) |
| 505 | sync_code = _run_sync_seed(args, ctx, force=force) |
| 506 | if sync_code == 4: |
| 507 | report.errors.append( |
| 508 | "refused: shared-asset conflict on bridge files without --force " |
| 509 | "(living docs never force-promoted on Stage 3 path)" |
| 510 | ) |
| 511 | return 4, report |
| 512 | if sync_code not in (0,): |
| 513 | report.errors.append(f"footprint re-seed failed (exit {sync_code})") |
| 514 | return sync_code if sync_code else 1, report |
| 515 | report.footprint_seeded = True |
| 516 | report.steps_completed.append("C3") |
| 517 | |
| 518 | # C4 footprint gate |
| 519 | if not bridge_footprint_present(repo_root, config): |
| 520 | report.errors.append("refused: C4 footprint gate failed — bridge destinations missing") |
| 521 | return 4, report |
| 522 | report.steps_completed.append("C4") |
| 523 | |
| 524 | # C5 gates |
| 525 | gates = evaluate_bridge_gates(config=config, repo_root=repo_root, runner=ctx.runner) |
| 526 | report.gates = {g.gate_id: {"ok": g.ok, "detail": g.detail} for g in gates} |
| 527 | report.steps_completed.append("C5") |
| 528 | report.ready_for_footprint = all( |
| 529 | report.gates[g]["ok"] for g in ("G1", "G2", "G3", "G4", "G5", "G6", "G7") |
| 530 | ) |
| 531 | report.ready_for_live_bridge = all(g["ok"] for g in report.gates.values()) |
| 532 | |
| 533 | if not report.ready_for_footprint: |
| 534 | report.errors.append("refused: bridge dry-run gates G1–G7 failed") |
| 535 | return 4, report |
| 536 | |
| 537 | if live_bridge: |
| 538 | if not report.ready_for_live_bridge: |
| 539 | report.errors.append( |
| 540 | "refused: not ready for live bridge (G8 or earlier gate failed); " |
| 541 | "config+footprint remain muse+git-mirror for retry" |
| 542 | ) |
| 543 | return 4, report |
| 544 | if not yes: |
| 545 | report.errors.append("refused: --live-bridge requires -y/--yes after gates pass (C6)") |
| 546 | return 4, report |
| 547 | return _invoke_live_bridge(repo_root, ctx, report) |
| 548 | |
| 549 | if yes and not live_bridge: |
| 550 | # --yes alone without gate success / live-bridge is not a C7 consent. |
| 551 | pass |
| 552 | |
| 553 | return 0, report |
| 554 | |
| 555 | |
| 556 | def _run_sync_seed(args: Namespace, ctx: CliContext, *, force: bool) -> int: |
| 557 | """Compose ``ok sync -y`` for C3; ``--force`` only for bridge shared-asset conflicts.""" |
| 558 | from cli.commands.sync import run_sync |
| 559 | |
| 560 | sync_args = Namespace( |
| 561 | repo=getattr(args, "repo", None), |
| 562 | config=getattr(args, "config", None), |
| 563 | dry_run=False, |
| 564 | diff=False, |
| 565 | only=None, |
| 566 | force=force, |
| 567 | yes=True, |
| 568 | include_preserved=False, # never on product Stage 3 path |
| 569 | ) |
| 570 | return run_sync(sync_args, ctx) |
| 571 | |
| 572 | |
| 573 | def _invoke_live_bridge(repo_root: Path, ctx: CliContext, report: UpgradeReport) -> tuple[int, UpgradeReport]: |
| 574 | """C7: invoke vendored deploy script only — never C8 merge.""" |
| 575 | script = repo_root / MUSE_BRIDGE_DEPLOY_DEST |
| 576 | if not script.is_file(): |
| 577 | report.errors.append("refused: deploy script missing for live bridge") |
| 578 | return 4, report |
| 579 | if not _path_escape_ok(repo_root, script): |
| 580 | report.errors.append("refused: deploy script path escapes repo") |
| 581 | return 4, report |
| 582 | msg = "mirror: Stage 3 upgrade-regime first live bridge" |
| 583 | cmd = f"{shlex.quote(str(script))} {shlex.quote(msg)}" |
| 584 | result = ctx.runner.run(cmd, cwd=str(repo_root)) |
| 585 | report.live_bridge_invoked = True |
| 586 | report.steps_completed.append("C7") |
| 587 | if result.exit_code != 0: |
| 588 | report.errors.append( |
| 589 | sanitize_text( |
| 590 | result.stderr or f"live bridge failed (exit {result.exit_code})", |
| 591 | repo_root, |
| 592 | ) |
| 593 | ) |
| 594 | # Default: do not roll back config/footprint (§O2.5). |
| 595 | return result.exit_code if result.exit_code else 1, report |
| 596 | report.warnings.append( |
| 597 | "C7 complete: merge muse-mirror → main remains Tier 3 (C8 hard stop)" |
| 598 | ) |
| 599 | return 0, report |