Coverage for scripts / upgrade_validation / actions.py: 100.00%

1"""The upgrade-validation actions. 

2 

3Five are new: ``preflight`` (the release preflight plus the base release), 

4``prepare`` (the private workspace), ``deploy`` (the base release, deployed by 

5its own ``gco``), ``sentinels`` and ``upgrade`` (the base release's ``gco 

6upgrade`` to this commit), and ``verify-upgrade``. The rest of the registry 

7reuses the release harness's handlers unchanged. 

8 

9``deploy`` and ``upgrade`` each run one long ``gco`` subprocess inside a 

10*phase*. The phase checkpoints the AWS server time before the command starts, 

11and whatever the command did, even if it failed or timed out, the stacks it 

12left are adopted by run tag against that time before the action returns. A 

13phase interrupted by the harness itself is adopted the same way on resume and 

14is never run again: the run fails, and teardown removes what it left. 

15""" 

16 

17from __future__ import annotations 

18 

19import copy 

20import email.utils 

21import json 

22import shutil 

23import sys 

24from collections.abc import Sequence 

25from datetime import UTC, datetime 

26from pathlib import Path 

27from typing import Any 

28 

29from scripts.live_release_validation.actions import action_preflight 

30from scripts.live_release_validation.cleanup.local_images import ( 

31 prune_local_cdk_asset_images_safely, 

32) 

33from scripts.live_release_validation.constants import _HEALTHY_STACK_STATUSES, _RUN_STACK_TAG 

34from scripts.live_release_validation.context import _run_git 

35from scripts.live_release_validation.inventory import describe_stack 

36from scripts.live_release_validation.models import ActionFailure, RunContext, utc_now 

37from scripts.live_release_validation.ownership.ecr import ( 

38 PRIOR_RELEASE_ECR_IMAGES_KEY, 

39 _checkpointed_ecr_image_targets, 

40 _expected_ecr_images, 

41) 

42from scripts.live_release_validation.ownership.kms import _checkpoint_retained_kms_keys 

43from scripts.live_release_validation.ownership.stacks import ( 

44 _adopt_run_tagged_stacks, 

45 _owned_stack_record, 

46 _reconcile_stack_ownership, 

47) 

48 

49from .models import UpgradeRunSettings 

50from .sentinel import create_sentinel, delete_sentinel, verify_sentinel 

51from .workspace import ( 

52 CommandResult, 

53 Workspace, 

54 build_mirror, 

55 checkout_head, 

56 clone_base, 

57 command_environment, 

58 console_echo, 

59 gco_result_document, 

60 read_cloud_assembly, 

61 reset_workspace, 

62 run_logged, 

63 running_command, 

64 sha256_file, 

65 synthetic_release_tag, 

66 tag_candidate, 

67 tracked_changes, 

68 write_run_context, 

69) 

70 

71# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

72# Generated at (UTC): 2026-09-29T17:48:38Z 

73# Generated from Git commit: 5ac48df61b8643924926baf6528255f8b55604a3 

74# Flowchart(s) generated from this file: 

75# * ``action_upgrade`` -> ``diagrams/code_diagrams/scripts/upgrade_validation/actions.action_upgrade.html`` 

76# (PNG: ``diagrams/code_diagrams/scripts/upgrade_validation/actions.action_upgrade.png``) 

77# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

78# <pyflowchart-code-diagram> END 

79 

80 

81#: ``checkpoint.state`` key for everything this harness records. 

82STATE_KEY = "upgrade_validation" 

83#: Wall-clock cap for each preparation step and for ``gco upgrade --check``. 

84STEP_TIMEOUT_SECONDS = 30 * 60 

85_GIB = float(1024**3) 

86 

87 

88def _settings(ctx: RunContext) -> UpgradeRunSettings: 

89 settings = ctx.settings 

90 if not isinstance(settings, UpgradeRunSettings): 

91 raise TypeError("The upgrade-validation actions need UpgradeRunSettings") 

92 return settings 

93 

94 

95def _state(ctx: RunContext) -> dict[str, Any]: 

96 state = ctx.checkpoint.state.setdefault(STATE_KEY, {}) 

97 if not isinstance(state, dict): 

98 raise RuntimeError(f"Checkpoint {STATE_KEY} must be an object") 

99 return state 

100 

101 

102def _phases(ctx: RunContext) -> dict[str, dict[str, Any]]: 

103 phases = _state(ctx).setdefault("phases", {}) 

104 if not isinstance(phases, dict) or any(not isinstance(item, dict) for item in phases.values()): 

105 raise RuntimeError(f"Checkpoint {STATE_KEY}.phases is malformed") 

106 return phases 

107 

108 

109def _workspace(ctx: RunContext) -> Workspace: 

110 return Workspace(_settings(ctx).workspace_dir) 

111 

112 

113def _gco_identity(ctx: RunContext) -> dict[str, str]: 

114 """``GCO_*`` settings that point the base ``gco`` at the harness's deployment.""" 

115 config = ctx.config 

116 return { 

117 "GCO_PROJECT_NAME": str(config.project_name), 

118 "GCO_GLOBAL_REGION": str(config.global_region), 

119 "GCO_API_GATEWAY_REGION": str(config.api_gateway_region), 

120 "GCO_MONITORING_REGION": str(config.monitoring_region), 

121 "GCO_DEFAULT_REGION": str(config.default_region), 

122 } 

123 

124 

125def _environment(ctx: RunContext) -> dict[str, str]: 

126 return command_environment(_workspace(ctx), identity=_gco_identity(ctx)) 

127 

128 

129def _server_time(ctx: RunContext) -> datetime: 

130 """AWS's clock, from an STS response, after proving the caller's account.""" 

131 response = ctx.session.client("sts", region_name=ctx.config.global_region).get_caller_identity() 

132 account = str(response.get("Account") or "") 

133 if account != ctx.settings.expected_account: 

134 raise RuntimeError( 

135 f"AWS caller account {account or 'unknown'} does not match expected account " 

136 f"{ctx.settings.expected_account}" 

137 ) 

138 headers = (response.get("ResponseMetadata") or {}).get("HTTPHeaders") or {} 

139 if not headers.get("date"): 

140 raise RuntimeError("STS returned no Date header to start the adoption window from") 

141 moment = email.utils.parsedate_to_datetime(str(headers["date"])) 

142 return (moment if moment.tzinfo is not None else moment.replace(tzinfo=UTC)).astimezone(UTC) 

143 

144 

145def _run_step( 

146 ctx: RunContext, 

147 label: str, 

148 argv: Sequence[str], 

149 *, 

150 cwd: Path, 

151 timeout_seconds: float, 

152 phase: str | None = None, 

153) -> CommandResult: 

154 """Run one workspace command; a phase's command has its process ID checkpointed.""" 

155 workspace = _workspace(ctx) 

156 

157 def record_pid(pid: int) -> None: 

158 if phase is not None: 

159 _phases(ctx)[phase]["pid"] = pid 

160 ctx.persist() 

161 

162 return run_logged( 

163 argv, 

164 cwd=cwd, 

165 env=_environment(ctx), 

166 log_path=workspace.logs / f"{label.replace(' ', '-')}.log", 

167 timeout_seconds=timeout_seconds, 

168 echo=console_echo(label), 

169 on_start=record_pid, 

170 ) 

171 

172 

173# ─── Phases ───────────────────────────────────────────────────────── 

174 

175 

176def _begin_phase(ctx: RunContext, name: str, **details: Any) -> dict[str, Any]: 

177 """Checkpoint a phase's start, and its adoption window, before its command runs.""" 

178 window = _server_time(ctx) 

179 phase = {"started_at": utc_now(), "window_started_at": window.isoformat(), **details} 

180 with ctx.state_lock: 

181 _phases(ctx)[name] = phase 

182 if name == "deploy": 

183 ctx.checkpoint.deployment_attempted = True 

184 ctx.checkpoint.destroyed = False 

185 ctx.persist() 

186 return phase 

187 

188 

189def _replaceable(name: str, phase: dict[str, Any]) -> tuple[str, ...]: 

190 """The stacks a phase may recreate: the upgrade's workload tier, nothing else.""" 

191 if name != "upgrade": 

192 return () 

193 return tuple((phase.get("plan") or {}).get("workload_stacks") or ()) 

194 

195 

196def _adopt_phase(ctx: RunContext, name: str) -> dict[str, Any]: 

197 phase = _phases(ctx)[name] 

198 return _adopt_run_tagged_stacks( 

199 ctx, 

200 phase=name, 

201 window_started_at=datetime.fromisoformat(str(phase["window_started_at"])), 

202 replaceable=_replaceable(name, phase), 

203 ) 

204 

205 

206def _settle_phase(ctx: RunContext, name: str) -> None: 

207 """Take ownership of what a phase's command left, whatever its outcome.""" 

208 phase = _phases(ctx)[name] 

209 phase["adoption"] = _adopt_phase(ctx, name) 

210 _reconcile_stack_ownership(ctx) 

211 # Retained resources (the EKS keys, their log groups) are only checkpointed 

212 # while their stack stands; the upgrade destroys the base generation. 

213 phase["owned_kms_keys"] = len(_checkpoint_retained_kms_keys(ctx)) 

214 phase["local_image_prune"] = prune_local_cdk_asset_images_safely() 

215 phase["finished_at"] = utc_now() 

216 ctx.persist() 

217 

218 

219def _recover_interrupted_phases(ctx: RunContext) -> list[str]: 

220 """Adopt what a phase the harness itself was interrupted in left behind. 

221 

222 A harness killed outright (``kill -9``) cannot stop its command, which then 

223 keeps deploying on its own. Nothing is adopted while it still runs. 

224 """ 

225 gco = str(_workspace(ctx).gco) 

226 recovered: list[str] = [] 

227 for name, phase in _phases(ctx).items(): 

228 if phase.get("finished_at"): 

229 continue 

230 pid = phase.get("pid") 

231 command = running_command(pid) if isinstance(pid, int) else None 

232 if command is not None and gco in command: 

233 raise RuntimeError( 

234 f"The {name} phase's command is still running (process {pid}: {command}); " 

235 f"let it finish or stop it with 'kill -TERM -{pid}', then resume" 

236 ) 

237 phase["adoption"] = _adopt_phase(ctx, name) 

238 phase["interrupted"] = True 

239 phase["finished_at"] = utc_now() 

240 ctx.persist() 

241 recovered.append(name) 

242 return recovered 

243 

244 

245def _phase_outcome(phase: dict[str, Any], *, action: str, command: str) -> None: 

246 """Fail the action unless its phase's command finished and succeeded.""" 

247 details = {"phase": copy.deepcopy(phase)} 

248 if phase.get("interrupted"): 

249 raise ActionFailure( 

250 f"{command} was interrupted before it finished; the {action} action never " 

251 "reruns it, so this run can only be torn down", 

252 details, 

253 ) 

254 result = phase.get("command") or {} 

255 if result.get("timed_out"): 

256 raise ActionFailure(f"{command} timed out; see {result.get('log')}", details) 

257 if result.get("exit_code") != 0: 

258 raise ActionFailure( 

259 f"{command} exited with {result.get('exit_code')}; see {result.get('log')}", details 

260 ) 

261 

262 

263# ─── Checks shared by the actions ─────────────────────────────────── 

264 

265 

266def _target_regions(ctx: RunContext) -> dict[str, str]: 

267 targets = ctx.checkpoint.state.get("target_stack_regions") 

268 if not isinstance(targets, dict) or not targets: 

269 raise RuntimeError("Checkpoint lacks target stack Regions") 

270 return {str(name): str(region) for name, region in targets.items()} 

271 

272 

273def _require_prepared_workspace(ctx: RunContext) -> dict[str, Any]: 

274 """The prepared base checkout, still at the base release and unmodified.""" 

275 settings = _settings(ctx) 

276 workspace = _workspace(ctx) 

277 prepared = _state(ctx).get("workspace") 

278 if not isinstance(prepared, dict) or not prepared.get("prepared_at"): 

279 raise RuntimeError("The prepare action has not completed") 

280 if not workspace.owned_by(settings.run_id) or not workspace.gco.is_file(): 

281 raise RuntimeError(f"The prepared workspace {workspace.root} is missing or incomplete") 

282 head = checkout_head(workspace) 

283 if head != settings.base_commit: 

284 raise RuntimeError(f"The base checkout is at {head}, not {settings.base_ref}") 

285 changes = tracked_changes(workspace) 

286 if changes != ["cdk.json"]: 

287 raise RuntimeError( 

288 "The base checkout's tracked files changed beyond the run's cdk.json: " 

289 + ", ".join(changes or ["none (cdk.json was reverted)"]) 

290 ) 

291 if sha256_file(workspace.clone / "cdk.json") != prepared["cdk_json"]["sha256"]: 

292 raise RuntimeError("The base checkout's cdk.json changed after the run wrote it") 

293 return prepared 

294 

295 

296def _require_mirror_repositories_in_baseline( 

297 ctx: RunContext, baseline: dict[str, Any] 

298) -> dict[str, Any]: 

299 """Every repository either release's image mirror copies into must predate the run. 

300 

301 The mirror runs inside the base ``gco`` and, after the upgrade, inside the 

302 candidate's, with no creation callback into this harness: a repository it 

303 created would have no ownership record, so final-inventory could neither 

304 accept nor remove it. Copying into a repository the baseline already holds 

305 is covered the way a release run covers it: a tag the baseline lacked is 

306 a retained image delta. 

307 """ 

308 present = { 

309 (str(region), str(repository.get("name") or "")) 

310 for region, repositories in (baseline.get("ecr_repositories") or {}).items() 

311 for repository in repositories 

312 } 

313 wanted = sorted( 

314 { 

315 (str(item["region"]), str(item["repository"])) 

316 for item in _checkpointed_ecr_image_targets(ctx) 

317 if any(source.get("kind") == "configured-mirror" for source in item["sources"]) 

318 } 

319 ) 

320 missing = [f"{region}:{name}" for region, name in wanted if (region, name) not in present] 

321 if missing: 

322 raise RuntimeError( 

323 "The image mirror would create ECR repositories outside this run's ownership " 

324 f"records: {', '.join(missing)}. Seed them with 'gco images mirror --region " 

325 "<region>' first (a release validation run leaves them in place), then rerun" 

326 ) 

327 return {"repositories": [f"{region}:{name}" for region, name in wanted]} 

328 

329 

330def _stack_generations(ctx: RunContext) -> dict[str, str]: 

331 """The stack ID this run owns under each target name now.""" 

332 return { 

333 name: str((_owned_stack_record(ctx, region, name) or {}).get("stack_id") or "") 

334 for name, region in sorted(_target_regions(ctx).items()) 

335 } 

336 

337 

338def _require_healthy_targets(ctx: RunContext) -> dict[str, dict[str, Any]]: 

339 """Every target stands, under the ID this run owns, in a completed state.""" 

340 stacks: dict[str, dict[str, Any]] = {} 

341 unhealthy: list[str] = [] 

342 for name, region in sorted(_target_regions(ctx).items()): 

343 record = _owned_stack_record(ctx, region, name) 

344 live = describe_stack(ctx.session, region, name) 

345 status = str((live or {}).get("status") or "absent") 

346 stacks[name] = { 

347 "region": region, 

348 "stack_id": (record or {}).get("stack_id"), 

349 "status": status, 

350 } 

351 if ( 

352 record is None 

353 or live is None 

354 or live.get("stack_id") != record.get("stack_id") 

355 or status not in _HEALTHY_STACK_STATUSES 

356 ): 

357 unhealthy.append(f"{name} ({status})") 

358 if unhealthy: 

359 raise ActionFailure( 

360 "Target stacks are not deployed and owned by this run: " + ", ".join(unhealthy), 

361 {"stacks": stacks}, 

362 ) 

363 return stacks 

364 

365 

366# ─── preflight ────────────────────────────────────────────────────── 

367 

368 

369def _verify_base_release(ctx: RunContext) -> dict[str, Any]: 

370 """The base tag still names the pinned commit, and deploys this checkout's topology.""" 

371 settings = _settings(ctx) 

372 root = settings.repo_root 

373 commit = _run_git( 

374 root, 

375 "rev-parse", 

376 "--verify", 

377 "--quiet", 

378 f"refs/tags/{settings.base_ref}^{{commit}}", 

379 check=False, 

380 ) 

381 if commit != settings.base_commit: 

382 raise RuntimeError( 

383 f"{settings.base_ref} names {commit or 'no commit'} in this repository, not " 

384 f"{settings.base_commit}" 

385 ) 

386 if _run_git(root, "merge-base", settings.base_commit, settings.expected_sha) != commit: 

387 raise RuntimeError( 

388 f"{settings.expected_sha} does not descend from {settings.base_ref}; the upgrade " 

389 "path this harness validates starts from a release in the checkout's history" 

390 ) 

391 try: 

392 base_config = json.loads(_run_git(root, "show", f"{commit}:cdk.json")) 

393 except ValueError as exc: 

394 raise RuntimeError(f"{settings.base_ref}'s cdk.json is not valid JSON: {exc}") from exc 

395 base_context = base_config.get("context") if isinstance(base_config, dict) else None 

396 if not isinstance(base_context, dict): 

397 raise RuntimeError(f"{settings.base_ref}'s cdk.json has no context object") 

398 differing = [ 

399 key 

400 for key in ("project_name", "deployment_regions") 

401 if base_context.get(key) != ctx.cdk_context.get(key) 

402 ] 

403 if differing: 

404 raise RuntimeError( 

405 f"{settings.base_ref} deploys a different {' and '.join(differing)} than this " 

406 "checkout; the teardown and inventory would not cover what it deploys" 

407 ) 

408 return { 

409 "ref": settings.base_ref, 

410 "commit": commit, 

411 "project_name": base_context["project_name"], 

412 "deployment_regions": base_context["deployment_regions"], 

413 # Checked against the baseline before the deploy phase begins. 

414 "image_mirror": bool((base_context.get("volcano_image_mirror") or {}).get("enabled")), 

415 } 

416 

417 

418def _verify_candidate_stack_tags(ctx: RunContext) -> dict[str, Any]: 

419 """This checkout's app applies cdk.json ``context.tags`` to every target stack. 

420 

421 The upgrade's redeploy passes no ``--tags``: the run tag reaches the stacks 

422 it recreates only through cdk.json, and a stack without it could never be 

423 adopted or destroyed by the run. The release preflight has just 

424 synthesized this checkout to list its stacks, so its assembly shows 

425 whether the app turns cdk.json's tags into stack tags. 

426 """ 

427 expected = ctx.cdk_context.get("tags") 

428 if not isinstance(expected, dict) or not expected: 

429 raise RuntimeError( 

430 "cdk.json context.tags is empty, so nothing proves this checkout's app tags its " 

431 "stacks from cdk.json; the run tag the upgrade's redeploy needs travels that way" 

432 ) 

433 assembly = read_cloud_assembly(_settings(ctx).repo_root) 

434 untagged = sorted( 

435 name 

436 for name in _target_regions(ctx) 

437 if any( 

438 (assembly.get(name) or {}).get("tags", {}).get(str(key)) != str(value) 

439 for key, value in expected.items() 

440 ) 

441 ) 

442 if untagged: 

443 raise RuntimeError( 

444 "This checkout's app does not apply cdk.json context.tags to " 

445 + ", ".join(untagged) 

446 + ", so the stacks the upgrade recreates would lack the run tag" 

447 ) 

448 return {"cdk_json_tags": sorted(str(key) for key in expected), "stacks": len(assembly)} 

449 

450 

451def _required_tools() -> dict[str, str]: 

452 tools = {name: shutil.which(name) for name in ("git", "node", "npm")} 

453 missing = sorted(name for name, path in tools.items() if not path) 

454 if missing: 

455 raise RuntimeError( 

456 "Preparing the base release needs " + ", ".join(missing) + " on PATH " 

457 "(git for the private clone, node and npm for its CDK CLI)" 

458 ) 

459 return {name: str(path) for name, path in tools.items()} 

460 

461 

462def _check_workspace_disk(settings: UpgradeRunSettings) -> float | str: 

463 """The workspace volume must clear the same free-space floor as the checkout.""" 

464 if settings.min_free_disk_gib <= 0: 

465 return "not-required" 

466 probe = settings.workspace_dir 

467 while not probe.exists() and probe.parent != probe: 

468 probe = probe.parent 

469 free_gib = shutil.disk_usage(probe).free / _GIB 

470 if free_gib < settings.min_free_disk_gib: 

471 raise RuntimeError( 

472 f"The workspace volume ({probe}) has {free_gib:.1f} GiB free, below the " 

473 f"{settings.min_free_disk_gib} GiB floor; the base clone, its venv, and its " 

474 "image builds need the room" 

475 ) 

476 return round(free_gib, 2) 

477 

478 

479def action_upgrade_preflight(ctx: RunContext) -> dict[str, Any]: 

480 """Run the release preflight, then pin the base release, the tools, and the workspace.""" 

481 settings = _settings(ctx) 

482 recovered: list[str] = [] 

483 if ctx.checkpoint.deployment_attempted: 

484 # A phase the harness was interrupted in left stacks it has not adopted 

485 # yet; the release preflight's ownership reconcile would refuse them. 

486 _server_time(ctx) 

487 recovered = _recover_interrupted_phases(ctx) 

488 details = action_preflight(ctx) 

489 base = _verify_base_release(ctx) 

490 candidate_tags = _verify_candidate_stack_tags(ctx) 

491 state = _state(ctx) 

492 if state.get("base") not in (None, base): 

493 raise RuntimeError("The base release changed since the checkpoint was created") 

494 state["base"] = base 

495 ctx.persist() 

496 preparing = "prepare" in set(ctx.report.selected_actions) 

497 return { 

498 **details, 

499 "base_release": base, 

500 "candidate_stack_tags": candidate_tags, 

501 "synthetic_tag": synthetic_release_tag(settings.base_ref), 

502 "tools": _required_tools() if preparing else "not-required", 

503 "workspace": str(settings.workspace_dir), 

504 "workspace_free_disk_gib": _check_workspace_disk(settings) if preparing else None, 

505 "recovered_phases": recovered, 

506 } 

507 

508 

509# ─── prepare ──────────────────────────────────────────────────────── 

510 

511 

512def _prepare_commands(workspace: Workspace) -> tuple[tuple[str, list[str], Path], ...]: 

513 clone = workspace.clone 

514 return ( 

515 ("prepare venv", [sys.executable, "-m", "venv", str(workspace.venv)], workspace.root), 

516 ( 

517 "prepare pip", 

518 [ 

519 str(workspace.python), 

520 "-m", 

521 "pip", 

522 "install", 

523 "--quiet", 

524 "--no-input", 

525 "--constraint", 

526 str(clone / "requirements-lock.txt"), 

527 "--editable", 

528 f"{clone}[cdk]", 

529 ], 

530 clone, 

531 ), 

532 ("prepare npm", ["npm", "ci", "--ignore-scripts", "--no-audit", "--no-fund"], clone), 

533 ("prepare synth", [str(workspace.gco), "stacks", "synth"], clone), 

534 ) 

535 

536 

537def action_prepare(ctx: RunContext) -> dict[str, Any]: 

538 """Clone the base release privately, install its gco, and synthesize its app.""" 

539 settings = _settings(ctx) 

540 state = _state(ctx) 

541 if isinstance(state.get("workspace"), dict) and state["workspace"].get("prepared_at"): 

542 return {"reused": True, **_require_prepared_workspace(ctx)} 

543 workspace = _workspace(ctx) 

544 reset_workspace(workspace, settings.run_id) 

545 mirror = build_mirror( 

546 workspace, 

547 source=settings.repo_root, 

548 branch=settings.expected_branch, 

549 candidate_sha=settings.expected_sha, 

550 base_ref=settings.base_ref, 

551 base_commit=settings.base_commit, 

552 ) 

553 clone = clone_base(workspace, base_ref=settings.base_ref, base_commit=settings.base_commit) 

554 cdk_json = write_run_context( 

555 workspace.clone, 

556 run_tag_key=_RUN_STACK_TAG, 

557 run_id=settings.run_id, 

558 context=settings.base_cdk_context(), 

559 ) 

560 steps: list[dict[str, Any]] = [] 

561 for label, argv, cwd in _prepare_commands(workspace): 

562 result = _run_step(ctx, label, argv, cwd=cwd, timeout_seconds=STEP_TIMEOUT_SECONDS) 

563 steps.append({"step": label, **result.to_dict()}) 

564 if not result.ok: 

565 raise ActionFailure( 

566 f"Preparing {settings.base_ref} failed at '{label}' " 

567 f"(exit {result.exit_code}); see {result.log_path}", 

568 {"steps": steps}, 

569 ) 

570 

571 assembly = read_cloud_assembly(workspace.clone) 

572 evidence: dict[str, Any] = {"steps": steps, "assembly": assembly} 

573 targets = sorted(_target_regions(ctx)) 

574 if sorted(assembly) != targets: 

575 raise ActionFailure( 

576 f"{settings.base_ref} synthesizes {sorted(assembly)}, not this checkout's " 

577 f"{targets}; a stack renamed or added between the releases would outlive teardown", 

578 evidence, 

579 ) 

580 untagged = sorted( 

581 name 

582 for name, stack in assembly.items() 

583 if stack["tags"].get(_RUN_STACK_TAG) != settings.run_id 

584 ) 

585 if untagged: 

586 raise ActionFailure( 

587 f"{settings.base_ref} would deploy {', '.join(untagged)} without the run tag, so " 

588 "the run could not prove it owns them", 

589 evidence, 

590 ) 

591 changes = tracked_changes(workspace) 

592 if changes != ["cdk.json"]: 

593 raise ActionFailure( 

594 "Preparing the base checkout changed tracked files beyond cdk.json, which " 

595 "gco upgrade refuses: " + ", ".join(changes), 

596 evidence, 

597 ) 

598 prior_images = _expected_ecr_images(ctx, targets, root=workspace.clone) 

599 record = { 

600 "root": str(workspace.root), 

601 "mirror": mirror, 

602 "clone": clone, 

603 "cdk_json": cdk_json, 

604 "synthetic_tag": synthetic_release_tag(settings.base_ref), 

605 "stacks": targets, 

606 "steps": steps, 

607 "prepared_at": utc_now(), 

608 } 

609 with ctx.state_lock: 

610 ctx.checkpoint.state[PRIOR_RELEASE_ECR_IMAGES_KEY] = prior_images 

611 state["workspace"] = record 

612 ctx.persist() 

613 return {**record, "prior_release_ecr_images": prior_images} 

614 

615 

616# ─── deploy ───────────────────────────────────────────────────────── 

617 

618 

619def action_base_deploy(ctx: RunContext) -> dict[str, Any]: 

620 """Deploy the base release with its own gco, then adopt its stacks by run tag.""" 

621 if ctx.checkpoint.baseline is None: 

622 raise RuntimeError("A protected-resource baseline is required before deployment") 

623 settings = _settings(ctx) 

624 workspace = _workspace(ctx) 

625 phase = _phases(ctx).get("deploy") 

626 if phase is None: 

627 _require_prepared_workspace(ctx) 

628 mirror = _require_mirror_repositories_in_baseline(ctx, ctx.checkpoint.baseline) 

629 phase = _begin_phase(ctx, "deploy", image_mirror=mirror) 

630 try: 

631 # No --tag: the run tag comes from cdk.json, exactly as it must for 

632 # the upgrade's redeploy, so adopting these stacks also proves it. 

633 result = _run_step( 

634 ctx, 

635 "base deploy", 

636 [str(workspace.gco), "stacks", "deploy-all", "--yes"], 

637 cwd=workspace.clone, 

638 timeout_seconds=settings.deploy_timeout_seconds, 

639 phase="deploy", 

640 ) 

641 phase["command"] = result.to_dict() 

642 finally: 

643 _settle_phase(ctx, "deploy") 

644 else: 

645 _reconcile_stack_ownership(ctx) 

646 _phase_outcome(phase, action="deploy", command=f"{settings.base_ref}'s gco stacks deploy-all") 

647 return { 

648 "base_release": _state(ctx).get("base"), 

649 "phase": copy.deepcopy(phase), 

650 "stacks": _require_healthy_targets(ctx), 

651 } 

652 

653 

654# ─── sentinels ────────────────────────────────────────────────────── 

655 

656 

657def action_sentinels(ctx: RunContext) -> dict[str, Any]: 

658 """Write a job template through the base release's API for the upgrade to preserve.""" 

659 template = create_sentinel(ctx, timeout_seconds=_settings(ctx).api_ready_timeout_seconds) 

660 state = _state(ctx) 

661 state["sentinel"] = template 

662 ctx.persist() 

663 return {"template": copy.deepcopy(template)} 

664 

665 

666# ─── upgrade ──────────────────────────────────────────────────────── 

667 

668 

669def _validated_plan(ctx: RunContext, check: CommandResult, tag: str) -> dict[str, Any]: 

670 """``gco upgrade --check`` must describe exactly the upgrade this run validates.""" 

671 document = gco_result_document(check.stdout) or {} 

672 plan = document.get("plan") 

673 details = {"check": check.to_dict(), "document": document} 

674 if not check.ok or not isinstance(plan, dict): 

675 raise ActionFailure("gco upgrade --check did not return a plan", details) 

676 control = [str(name) for name in plan.get("control_plane_stacks") or []] 

677 workload = [str(name) for name in plan.get("workload_stacks") or []] 

678 problems: list[str] = [] 

679 if document.get("up_to_date") or plan.get("already_at_target"): 

680 problems.append("the base checkout is already at the target") 

681 if plan.get("target") != tag: 

682 problems.append(f"the target is {plan.get('target')!r}, not {tag}") 

683 if (plan.get("install") or {}).get("cli_editable_from_checkout") is not True: 

684 problems.append("the base gco is not the editable install of its checkout") 

685 if ( 

686 not workload 

687 or set(control) & set(workload) 

688 or {*control, *workload} != set(_target_regions(ctx)) 

689 ): 

690 problems.append("its control-plane and workload stacks do not partition the targets") 

691 if problems: 

692 raise ActionFailure( 

693 "gco upgrade --check does not describe this run's upgrade: " + "; ".join(problems), 

694 details, 

695 ) 

696 return {"target": tag, "control_plane_stacks": control, "workload_stacks": workload} 

697 

698 

699def action_upgrade(ctx: RunContext) -> dict[str, Any]: 

700 """Run the base release's gco upgrade to this commit and adopt the recreated stacks.""" 

701 settings = _settings(ctx) 

702 workspace = _workspace(ctx) 

703 phase = _phases(ctx).get("upgrade") 

704 if phase is None: 

705 prepared = _require_prepared_workspace(ctx) 

706 tag = tag_candidate( 

707 workspace, tag=str(prepared["synthetic_tag"]), candidate_sha=settings.expected_sha 

708 ) 

709 check = _run_step( 

710 ctx, 

711 "upgrade check", 

712 [str(workspace.gco), "-o", "json", "upgrade", "--check", "--ref", tag["tag"]], 

713 cwd=workspace.clone, 

714 timeout_seconds=STEP_TIMEOUT_SECONDS, 

715 ) 

716 plan = _validated_plan(ctx, check, tag["tag"]) 

717 # Checkpoint the base generation's retained resources while it stands. 

718 _reconcile_stack_ownership(ctx) 

719 _checkpoint_retained_kms_keys(ctx) 

720 phase = _begin_phase( 

721 ctx, 

722 "upgrade", 

723 tag=tag, 

724 check=check.to_dict(), 

725 plan=plan, 

726 before=_stack_generations(ctx), 

727 ) 

728 try: 

729 result = _run_step( 

730 ctx, 

731 "upgrade", 

732 [ 

733 str(workspace.gco), 

734 "-o", 

735 "json", 

736 "upgrade", 

737 "--yes", 

738 "--ref",