Coverage for scripts / live_release_validation / models.py: 100.00%

1"""Serializable run, checkpoint, action, and report models.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7import time 

8import traceback as traceback_module 

9from collections.abc import Callable 

10from dataclasses import asdict, dataclass, field, is_dataclass 

11from datetime import UTC, datetime 

12from pathlib import Path 

13from threading import RLock 

14from typing import Any, Literal, cast 

15 

16from .artifact_io import ( 

17 REPORT_FILENAMES as _REPORT_FILENAMES, 

18) 

19from .artifact_io import ( 

20 atomic_write_text, 

21 ensure_private_directory, 

22 ensure_private_run_directory, 

23) 

24from .artifact_io import ( 

25 read_private_text as _read_private_text, 

26) 

27from .inference_contract import ( 

28 INFERENCE_OWNER_LABEL, 

29 Framework, 

30 InferenceRuntimeSpec, 

31 inference_deploy_extra_args, 

32 inference_framework_env, 

33 inference_identity_fields, 

34 inference_request_body, 

35 validate_inference_settings, 

36) 

37from .json_utils import loads_without_duplicate_keys 

38 

39__all__ = [ 

40 "INFERENCE_OWNER_LABEL", 

41 "ActionResult", 

42 "Framework", 

43 "InferenceRuntimeSpec", 

44 "RunCheckpoint", 

45 "RunContext", 

46 "RunSettings", 

47 "ValidationReport", 

48 "atomic_write_json", 

49 "atomic_write_text", 

50 "ensure_private_directory", 

51 "ensure_private_run_directory", 

52 "utc_now", 

53] 

54 

55SCHEMA_VERSION = 2 

56ActionStatus = Literal["passed", "failed", "skipped"] 

57 

58_LIVE_VALIDATION_DISABLE_EFS_BACKUPS_CONTEXT = "gco_live_validation_disable_efs_automatic_backups" 

59 

60 

61def utc_now() -> str: 

62 """Return an RFC 3339-compatible UTC timestamp.""" 

63 return datetime.now(UTC).isoformat() 

64 

65 

66def to_jsonable(value: Any) -> Any: 

67 """Convert report values to stable JSON-compatible primitives.""" 

68 if is_dataclass(value) and not isinstance(value, type): 

69 return to_jsonable(asdict(value)) 

70 if isinstance(value, datetime): 

71 return value.astimezone(UTC).isoformat() 

72 if isinstance(value, Path): 

73 return str(value) 

74 if isinstance(value, dict): 

75 return {str(key): to_jsonable(item) for key, item in value.items()} 

76 if isinstance(value, (list, tuple)): 

77 return [to_jsonable(item) for item in value] 

78 if isinstance(value, (set, frozenset)): 

79 return sorted(to_jsonable(item) for item in value) 

80 if value is None or isinstance(value, (str, int, float, bool)): 

81 return value 

82 return str(value) 

83 

84 

85def atomic_write_json(path: Path, value: Any) -> None: 

86 """Atomically persist owner-only JSON inside a validated private directory.""" 

87 content = json.dumps(to_jsonable(value), indent=2, sort_keys=True) + "\n" 

88 atomic_write_text(path, content) 

89 

90 

91@dataclass(frozen=True) 

92class RunSettings: 

93 """Immutable operator inputs for one live validation run.""" 

94 

95 run_id: str 

96 repo_root: Path 

97 report_dir: Path 

98 checkpoint_path: Path 

99 expected_account: str 

100 expected_sha: str 

101 expected_branch: str 

102 profile: str 

103 requested_actions: tuple[str, ...] 

104 protected_stack_names: tuple[str, ...] = ("CDKToolkit", "GCOGitHubOIDCStack") 

105 max_workers: int = 4 

106 job_timeout_seconds: int = 1800 

107 queue_timeout_seconds: int = 900 

108 poll_interval_seconds: int = 10 

109 destroy_attempts: int = 3 

110 destroy_retry_delay_seconds: int = 30 

111 confirm_kms_key_deletion: bool = False 

112 resume: bool = False 

113 #: Free-space floor (GiB) preflight enforces on the checkout, the report 

114 #: directory, and the home volume before ``deploy`` builds container 

115 #: images. A host that runs out of space mid-deploy fails the image 

116 #: build, then fails to persist the checkpoint, which also aborts the 

117 #: guaranteed cleanup — so the floor is checked before anything is 

118 #: created. ``0`` disables the check. 

119 min_free_disk_gib: int = 20 

120 #: Off-by-default schedulers force-enabled for this run's deploy (threaded 

121 #: to CDK as the ``helm_enabled_overrides`` context; see the ``schedulers`` 

122 #: action). Part of the resume identity: a resumed run must deploy and 

123 #: validate the same chart set it started with. 

124 optional_schedulers: tuple[str, ...] = () 

125 

126 # First-class inference action contract. ``inference_enabled`` is explicit 

127 # because sibling harnesses reuse RunSettings with their own ``all`` action. 

128 inference_enabled: bool = False 

129 selected_region: str = "" 

130 inference_runtimes: tuple[InferenceRuntimeSpec, ...] = () 

131 request_prompt: str = "Reply with a short deterministic validation response." 

132 request_max_tokens: int = 8 

133 namespace: str = "gco-inference" 

134 health_path: str = "/health" 

135 gpu_count: int = 0 

136 baseline_replicas: int = 1 

137 autoscale_initial_replicas: int = 1 

138 hpa_min_replicas: int = 2 

139 hpa_max_replicas: int = 2 

140 hpa_cpu_target: int = 70 

141 endpoint_count: int = 4 

142 proxy_tls_cpu_request: str = "100m" 

143 proxy_tls_cpu_target: int = 70 

144 command_timeout_seconds: int = 300 

145 readiness_timeout_seconds: int = 1800 

146 hpa_timeout_seconds: int = 900 

147 deletion_timeout_seconds: int = 900 

148 monitor_interval_seconds: int = 15 

149 hpa_stability_intervals: int = 2 

150 consent: bool = False 

151 

152 def __post_init__(self) -> None: 

153 """Normalize output paths and validate the selected action contracts.""" 

154 report_dir = Path(os.path.abspath(os.fspath(self.report_dir))) 

155 checkpoint_path = Path(os.path.abspath(os.fspath(self.checkpoint_path))) 

156 object.__setattr__(self, "report_dir", report_dir) 

157 object.__setattr__(self, "checkpoint_path", checkpoint_path) 

158 if checkpoint_path.parent != report_dir: 

159 raise ValueError("Checkpoint must be a direct child of the report directory") 

160 if checkpoint_path.name in _REPORT_FILENAMES: 

161 raise ValueError( 

162 f"Checkpoint filename is reserved for a validation report: {checkpoint_path.name}" 

163 ) 

164 if self.inference_enabled: 

165 validate_inference_settings(self) 

166 

167 def request_body(self, runtime: InferenceRuntimeSpec) -> dict[str, Any]: 

168 """Return the deterministic body for one runtime in the matrix.""" 

169 return inference_request_body(self, runtime) 

170 

171 @staticmethod 

172 def framework_env(runtime: InferenceRuntimeSpec) -> dict[str, str]: 

173 return inference_framework_env(runtime) 

174 

175 @staticmethod 

176 def deploy_extra_args(runtime: InferenceRuntimeSpec) -> tuple[str, ...]: 

177 return inference_deploy_extra_args(runtime) 

178 

179 @property 

180 def kubeconfig_path(self) -> Path: 

181 """Return the isolated kubeconfig path without touching the filesystem.""" 

182 return self.report_dir / "kubeconfig" 

183 

184 def _inference_identity_fields(self) -> dict[str, Any]: 

185 return inference_identity_fields(self) 

186 

187 def extra_cdk_context(self) -> dict[str, str]: 

188 """Extra ``--context`` pairs every CDK invocation of this run must carry.""" 

189 context = {_LIVE_VALIDATION_DISABLE_EFS_BACKUPS_CONTEXT: "true"} 

190 if self.optional_schedulers: 

191 context["helm_enabled_overrides"] = ",".join(self.optional_schedulers) 

192 return context 

193 

194 def identity(self) -> dict[str, Any]: 

195 """Return fields that must remain identical across resume attempts.""" 

196 identity = { 

197 "run_id": self.run_id, 

198 "repo_root": str(self.repo_root.resolve()), 

199 "expected_account": self.expected_account, 

200 "expected_sha": self.expected_sha, 

201 "expected_branch": self.expected_branch, 

202 "profile": self.profile, 

203 "requested_actions": list(self.requested_actions), 

204 "protected_stack_names": list(self.protected_stack_names), 

205 "confirm_kms_key_deletion": self.confirm_kms_key_deletion, 

206 "optional_schedulers": list(self.optional_schedulers), 

207 "extra_cdk_context": self.extra_cdk_context(), 

208 } 

209 if self.inference_enabled: 

210 identity["inference"] = self._inference_identity_fields() 

211 return identity 

212 

213 

214@dataclass 

215class ActionResult: 

216 """One action's durable report entry.""" 

217 

218 name: str 

219 description: str 

220 status: ActionStatus 

221 started_at: str 

222 ended_at: str 

223 duration_seconds: float 

224 details: dict[str, Any] = field(default_factory=dict) 

225 error: str | None = None 

226 traceback: str | None = None 

227 

228 @classmethod 

229 def passed( 

230 cls, 

231 *, 

232 name: str, 

233 description: str, 

234 started_at: str, 

235 started_monotonic: float, 

236 ended_monotonic: float, 

237 details: dict[str, Any] | None = None, 

238 ) -> ActionResult: 

239 return cls( 

240 name=name, 

241 description=description, 

242 status="passed", 

243 started_at=started_at, 

244 ended_at=utc_now(), 

245 duration_seconds=round(ended_monotonic - started_monotonic, 3), 

246 details=details or {}, 

247 ) 

248 

249 @classmethod 

250 def failed( 

251 cls, 

252 *, 

253 name: str, 

254 description: str, 

255 started_at: str, 

256 started_monotonic: float, 

257 ended_monotonic: float, 

258 error: BaseException, 

259 details: dict[str, Any] | None = None, 

260 ) -> ActionResult: 

261 return cls( 

262 name=name, 

263 description=description, 

264 status="failed", 

265 started_at=started_at, 

266 ended_at=utc_now(), 

267 duration_seconds=round(ended_monotonic - started_monotonic, 3), 

268 details=details or {}, 

269 error=f"{type(error).__name__}: {error}", 

270 traceback="".join( 

271 traceback_module.format_exception(type(error), error, error.__traceback__) 

272 ), 

273 ) 

274 

275 @classmethod 

276 def from_dict(cls, value: dict[str, Any]) -> ActionResult: 

277 return cls( 

278 name=str(value["name"]), 

279 description=str(value.get("description", "")), 

280 status=value["status"], 

281 started_at=str(value.get("started_at", "")), 

282 ended_at=str(value.get("ended_at", "")), 

283 duration_seconds=float(value.get("duration_seconds", 0.0)), 

284 details=dict(value.get("details") or {}), 

285 error=value.get("error"), 

286 traceback=value.get("traceback"), 

287 ) 

288 

289 

290@dataclass 

291class RunCheckpoint: 

292 """Crash-safe state used to resume and prove resource ownership.""" 

293 

294 identity: dict[str, Any] 

295 created_at: str = field(default_factory=utc_now) 

296 updated_at: str = field(default_factory=utc_now) 

297 completed_actions: list[str] = field(default_factory=list) 

298 action_results: dict[str, ActionResult] = field(default_factory=dict) 

299 deployment_attempted: bool = False 

300 destroyed: bool = False 

301 baseline: dict[str, Any] | None = None 

302 state: dict[str, Any] = field(default_factory=dict) 

303 schema_version: int = SCHEMA_VERSION 

304 

305 def to_dict(self) -> dict[str, Any]: 

306 self.updated_at = utc_now() 

307 serialized = to_jsonable(self) 

308 if not isinstance(serialized, dict): 

309 raise TypeError("RunCheckpoint did not serialize to an object") 

310 return serialized 

311 

312 @classmethod 

313 def from_path(cls, path: Path) -> RunCheckpoint: 

314 try: 

315 checkpoint_text = _read_private_text(path) 

316 except (OSError, UnicodeError) as exc: 

317 raise ValueError(f"Unable to read checkpoint {path}: {exc}") from exc 

318 try: 

319 raw = loads_without_duplicate_keys(checkpoint_text) 

320 except ValueError as exc: 

321 raise ValueError(f"Unable to read checkpoint {path}: {exc}") from exc 

322 if not isinstance(raw, dict) or raw.get("schema_version") != SCHEMA_VERSION: 

323 raise ValueError(f"Checkpoint {path} does not use supported schema {SCHEMA_VERSION}") 

324 results_raw = raw.get("action_results") or {} 

325 if not isinstance(results_raw, dict): 

326 raise ValueError(f"Checkpoint {path} has invalid action_results") 

327 state_raw = raw.get("state", {}) 

328 if not isinstance(state_raw, dict): 

329 raise ValueError(f"Checkpoint {path} state must be an object") 

330 return cls( 

331 identity=dict(raw.get("identity") or {}), 

332 created_at=str(raw.get("created_at") or utc_now()), 

333 updated_at=str(raw.get("updated_at") or utc_now()), 

334 completed_actions=[str(item) for item in raw.get("completed_actions") or []], 

335 action_results={ 

336 str(name): ActionResult.from_dict(value) 

337 for name, value in results_raw.items() 

338 if isinstance(value, dict) 

339 }, 

340 deployment_attempted=bool(raw.get("deployment_attempted", False)), 

341 destroyed=bool(raw.get("destroyed", False)), 

342 baseline=dict(raw["baseline"]) if isinstance(raw.get("baseline"), dict) else None, 

343 state=dict(state_raw), 

344 schema_version=SCHEMA_VERSION, 

345 ) 

346 

347 

348@dataclass 

349class ValidationReport: 

350 """Local JSON/Markdown evidence for one run; contains account-specific identifiers.""" 

351 

352 run_id: str 

353 identity: dict[str, Any] 

354 selected_actions: list[str] 

355 started_at: str 

356 ended_at: str | None = None 

357 status: Literal["running", "passed", "partial", "failed", "interrupted"] = "running" 

358 action_results: list[ActionResult] = field(default_factory=list) 

359 cleanup: dict[str, Any] = field(default_factory=dict) 

360 baseline: dict[str, Any] | None = None 

361 final_inventory: dict[str, Any] | None = None 

362 fatal_error: str | None = None 

363 schema_version: int = SCHEMA_VERSION 

364 #: Markdown heading and report-filename stem; sibling harnesses override 

365 #: both (e.g. "GCO Example Job Validation" / "example-job-validation"). 

366 title: str = "GCO Live Release Validation" 

367 report_stem: str = "live-release-validation" 

368 

369 def to_dict(self) -> dict[str, Any]: 

370 serialized = to_jsonable(self) 

371 if not isinstance(serialized, dict): 

372 raise TypeError("ValidationReport did not serialize to an object") 

373 return serialized 

374 

375 def write(self, directory: Path) -> tuple[Path, Path]: 

376 """Write both report formats and return their paths.""" 

377 ensure_private_directory(directory) 

378 json_path = directory / f"{self.report_stem}.json" 

379 markdown_path = directory / f"{self.report_stem}.md" 

380 atomic_write_json(json_path, self.to_dict()) 

381 atomic_write_text(markdown_path, self.to_markdown()) 

382 return json_path, markdown_path 

383 

384 def to_markdown(self) -> str: 

385 """Render a compact human-reviewable report.""" 

386 identity = self.identity 

387 selected_scope = ", ".join(f"`{name}`" for name in self.selected_actions) or "_none_" 

388 lines = [ 

389 f"# {self.title}", 

390 "", 

391 f"- **Run:** `{self.run_id}`", 

392 f"- **Status:** **{self.status.upper()}**", 

393 f"- **Account:** `{identity.get('expected_account', 'unknown')}`", 

394 f"- **Commit:** `{identity.get('expected_sha', 'unknown')}`", 

395 f"- **Branch:** `{identity.get('expected_branch', 'unknown')}`", 

396 f"- **Profile:** `{identity.get('profile', 'unknown')}`", 

397 f"- **Selected action scope:** {selected_scope}", 

398 f"- **Started:** `{self.started_at}`", 

399 f"- **Ended:** `{self.ended_at or 'in progress'}`", 

400 "", 

401 "## Actions", 

402 "", 

403 "| Action | Status | Duration | Error |", 

404 "|---|---:|---:|---|", 

405 ] 

406 for result in self.action_results: 

407 error = (result.error or "").replace("|", "\\|").replace("\n", " ") 

408 lines.append( 

409 f"| `{result.name}` | {result.status} | {result.duration_seconds:.3f}s | {error} |" 

410 ) 

411 if not self.action_results: 

412 lines.append("| _none_ | skipped | 0s | |") 

413 

414 lines.extend(["", "## Cleanup", "", "```json"]) 

415 lines.append(json.dumps(to_jsonable(self.cleanup), indent=2, sort_keys=True)) 

416 lines.append("```") 

417 

418 final_summary = (self.final_inventory or {}).get("summary", {}) 

419 lines.extend(["", "## Final inventory", "", "```json"]) 

420 lines.append(json.dumps(to_jsonable(final_summary), indent=2, sort_keys=True)) 

421 lines.append("```") 

422 

423 failures = [result for result in self.action_results if result.status == "failed"] 

424 if self.fatal_error or failures: 

425 lines.extend(["", "## Failures", ""]) 

426 if self.fatal_error: 

427 lines.extend(["```text", self.fatal_error, "```", ""]) 

428 for result in failures: 

429 lines.append(f"### `{result.name}`") 

430 lines.extend( 

431 ["", "```text", result.traceback or result.error or "unknown", "```", ""] 

432 ) 

433 

434 lines.append("") 

435 return "\n".join(lines) 

436 

437 

438@dataclass 

439class RunContext: 

440 """Mutable dependencies and durable state shared by action handlers.""" 

441 

442 settings: RunSettings 

443 checkpoint: RunCheckpoint 

444 report: ValidationReport 

445 cdk_context: dict[str, Any] 

446 deployment_regions: tuple[str, ...] 

447 config: Any 

448 session: Any 

449 stack_manager: Any 

450 aws_client: Any 

451 job_manager: Any 

452 persist_callback: Callable[[RunCheckpoint], None] 

453 state_lock: RLock = field(default_factory=RLock, repr=False) 

454 

455 def persist(self) -> None: 

456 with self.state_lock: 

457 self.persist_callback(self.checkpoint) 

458 

459 def register_job( 

460 self, 

461 *, 

462 name: str, 

463 namespace: str, 

464 region: str, 

465 path: str, 

466 run_label: str, 

467 transport_region: str | None, 

468 ) -> dict[str, Any]: 

469 """Checkpoint one deterministic Job before submitting it. 

470 

471 The record is not destructive authority until an exact Kubernetes UID 

472 has been observed together with the expected run/path labels. 

473 """ 

474 with self.state_lock: 

475 raw_jobs = self.checkpoint.state.setdefault("jobs", []) 

476 if not isinstance(raw_jobs, list) or any( 

477 not isinstance(item, dict) for item in raw_jobs 

478 ): 

479 raise RuntimeError("Checkpoint jobs must be a list of objects") 

480 jobs = cast(list[dict[str, Any]], raw_jobs) 

481 matches = [ 

482 item 

483 for item in jobs 

484 if item.get("name") == name 

485 and item.get("namespace") == namespace 

486 and item.get("region") == region 

487 and item.get("path") == path 

488 ] 

489 if len(matches) > 1: 

490 raise RuntimeError( 

491 f"Checkpoint contains duplicate Job records for {region}:{namespace}/{name}" 

492 ) 

493 expected = { 

494 "name": name, 

495 "namespace": namespace, 

496 "region": region, 

497 "path": path, 

498 "run_label": run_label, 

499 "transport_region": transport_region, 

500 } 

501 if matches: 

502 record = matches[0] 

503 for key, value in expected.items(): 

504 if record.get(key) != value: 

505 raise RuntimeError( 

506 f"Checkpoint Job identity changed for {region}:{namespace}/{name}: {key}" 

507 ) 

508 record.setdefault( 

509 "submission_state", 

510 "appeared" if record.get("uid") else "registered", 

511 ) 

512 else: 

513 record = { 

514 **expected, 

515 "uid": None, 

516 "deleted": False, 

517 "submission_state": "registered", 

518 } 

519 jobs.append(record) 

520 self.persist_callback(self.checkpoint) 

521 return record 

522 

523 def prepare_job_submission( 

524 self, 

525 record: dict[str, Any], 

526 *, 

527 envelope: dict[str, Any], 

528 resumable: bool, 

529 ) -> None: 

530 """Persist one immutable canonical envelope before any submission attempt.""" 

531 canonical = to_jsonable(envelope) 

532 if not isinstance(canonical, dict): 

533 raise RuntimeError("A Job submission envelope must be a JSON object") 

534 with self.state_lock: 

535 state = str(record.get("submission_state") or "registered") 

536 previous = record.get("submission_envelope") 

537 if previous is not None and previous != canonical: 

538 raise RuntimeError("Checkpointed Job submission envelope changed") 

539 previous_resumable = record.get("submission_resumable") 

540 if previous_resumable is not None and bool(previous_resumable) != resumable: 

541 raise RuntimeError("Checkpointed Job resumability contract changed") 

542 if state == "registered": 

543 record["submission_state"] = "prepared" 

544 elif state not in { 

545 "prepared", 

546 "submitting", 

547 "submitted", 

548 "appeared", 

549 "deleted", 

550 "blocked", 

551 "not_submitted", 

552 }: 

553 raise RuntimeError(f"Cannot prepare Job submission from state {state!r}") 

554 record["submission_envelope"] = canonical 

555 record["submission_resumable"] = resumable 

556 self.persist_callback(self.checkpoint) 

557 

558 def begin_job_submission( 

559 self, 

560 record: dict[str, Any], 

561 *, 

562 reconciliation_timeout_seconds: int, 

563 ) -> None: 

564 """Persist the ambiguous check/use boundary immediately before submission.""" 

565 with self.state_lock: 

566 state = str(record.get("submission_state") or "registered") 

567 resumable = bool(record.get("submission_resumable", False)) 

568 if state != "prepared" and not (state == "submitting" and resumable): 

569 raise RuntimeError(f"Cannot begin Job submission from state {state!r}") 

570 if not isinstance(record.get("submission_envelope"), dict): 

571 raise RuntimeError("Cannot submit a Job without a checkpointed envelope") 

572 now = time.time() 

573 record["submission_state"] = "submitting" 

574 record["submission_started_at"] = now 

575 record["submission_reconcile_deadline"] = now + reconciliation_timeout_seconds 

576 record["submission_attempts"] = int(record.get("submission_attempts") or 0) + 1 

577 self.persist_callback(self.checkpoint) 

578 

579 def finish_job_submission( 

580 self, 

581 record: dict[str, Any], 

582 submission: dict[str, Any], 

583 *, 

584 appearance_timeout_seconds: int, 

585 ) -> None: 

586 """Persist an acknowledgement and begin a fresh bounded appearance window.""" 

587 with self.state_lock: 

588 state = str(record.get("submission_state") or "") 

589 if state not in {"submitting", "submitted", "appeared"}: 

590 raise RuntimeError(f"Cannot finish Job submission from state {state!r}") 

591 acknowledged_at = time.time() 

592 if state != "appeared": 

593 record["submission_state"] = "submitted" 

594 record["submission"] = to_jsonable(submission) 

595 record["submission_acknowledged_at"] = acknowledged_at 

596 record["appearance_deadline"] = acknowledged_at + appearance_timeout_seconds 

597 self.persist_callback(self.checkpoint) 

598 

599 def block_job_submission(self, record: dict[str, Any], reason: str) -> None: 

600 """Fail closed when a non-idempotent submission cannot be reconciled.""" 

601 with self.state_lock: 

602 record["submission_state"] = "blocked" 

603 record["submission_blocked_reason"] = reason 

604 record["submission_blocked_at"] = time.time() 

605 self.persist_callback(self.checkpoint) 

606 

607 def mark_job_not_submitted(self, record: dict[str, Any]) -> None: 

608 """Record authoritative absence only before a side effect could escape.""" 

609 with self.state_lock: 

610 state = str(record.get("submission_state") or "registered") 

611 if state not in {"registered", "prepared", "not_submitted"}: 

612 raise RuntimeError(f"Cannot mark Job not submitted from state {state!r}") 

613 record["submission_state"] = "not_submitted" 

614 record["not_submitted_at"] = time.time() 

615 self.persist_callback(self.checkpoint) 

616 

617 def mark_central_job_cancelled_before_claim( 

618 self, 

619 record: dict[str, Any], 

620 *, 

621 job_id: str, 

622 ) -> None: 

623 """Record terminal queue proof that a central Job never reached a worker.""" 

624 if not job_id: 

625 raise RuntimeError("Central cancellation proof requires a queue Job ID") 

626 with self.state_lock: 

627 state = str(record.get("submission_state") or "registered") 

628 previous_job_id = record.get("central_cancelled_before_claim_job_id") 

629 if state == "not_submitted" and previous_job_id == job_id: 

630 return 

631 if state not in {"submitting", "submitted"}: 

632 raise RuntimeError(f"Cannot apply central cancellation proof from state {state!r}") 

633 if record.get("path") != "dynamodb" or record.get("uid"): 

634 raise RuntimeError( 

635 "Central cancellation proof cannot replace immutable Kubernetes UID evidence" 

636 ) 

637 record["submission_state"] = "not_submitted" 

638 record["central_cancelled_before_claim_job_id"] = job_id 

639 record["central_cancelled_before_claim_at"] = time.time() 

640 self.persist_callback(self.checkpoint) 

641 

642 def mark_central_job_not_created_by_worker( 

643 self, 

644 record: dict[str, Any], 

645 *, 

646 job_id: str, 

647 ) -> None: 

648 """Record explicit worker proof that Kubernetes mutation never began.""" 

649 if not job_id: 

650 raise RuntimeError("Central worker no-workload proof requires a queue Job ID") 

651 with self.state_lock: 

652 state = str(record.get("submission_state") or "registered") 

653 previous_job_id = record.get("central_worker_not_created_job_id") 

654 if state == "not_submitted" and previous_job_id == job_id: 

655 return 

656 if state not in {"submitting", "submitted"}: 

657 raise RuntimeError( 

658 f"Cannot apply central worker no-workload proof from state {state!r}" 

659 ) 

660 central_identity = ( 

661 record.get("k8s_job_name"), 

662 record.get("k8s_job_namespace"), 

663 record.get("k8s_job_uid"), 

664 ) 

665 if ( 

666 record.get("path") != "dynamodb" 

667 or record.get("uid") is not None 

668 or any(value is not None for value in central_identity) 

669 ): 

670 raise RuntimeError( 

671 "Central worker no-workload proof cannot replace Kubernetes identity evidence" 

672 ) 

673 if record.get("central_cancelled_before_claim_job_id") is not None: 

674 raise RuntimeError( 

675 "Central worker no-workload proof conflicts with cancellation proof" 

676 ) 

677 record["submission_state"] = "not_submitted" 

678 record["central_worker_not_created_job_id"] = job_id 

679 record["central_worker_not_created_at"] = time.time() 

680 self.persist_callback(self.checkpoint) 

681 

682 def bind_central_job_identity( 

683 self, 

684 record: dict[str, Any], 

685 *, 

686 job_id: str, 

687 name: str, 

688 namespace: str, 

689 uid: str, 

690 appearance_timeout_seconds: int, 

691 ) -> bool: 

692 """Bind a requested central-queue record to its immutable Kubernetes Job. 

693 

694 The requested name and namespace remain canonical submission/replay 

695 identity. The worker-persisted identity is a separate destructive 

696 authority and starts one fresh workload-appearance window when first 

697 observed. 

698 """ 

699 if record.get("path") != "dynamodb": 

700 raise RuntimeError("Central Kubernetes identity requires a DynamoDB workload record") 

701 if not all((job_id, name, namespace, uid)): 

702 raise RuntimeError("Central Kubernetes identity fields must all be non-empty") 

703 if appearance_timeout_seconds <= 0: 

704 raise RuntimeError("Central workload appearance timeout must be positive") 

705 

706 immutable = { 

707 "central_queue_job_id": job_id, 

708 "k8s_job_name": name, 

709 "k8s_job_namespace": namespace, 

710 "k8s_job_uid": uid, 

711 } 

712 with self.state_lock: 

713 present = {key: record.get(key) for key in immutable} 

714 populated = [value is not None for value in present.values()] 

715 if any(populated): 

716 if not all(populated): 

717 raise RuntimeError("Checkpoint contains a partial central Kubernetes identity") 

718 for key, value in immutable.items(): 

719 if present[key] != value: 

720 raise RuntimeError( 

721 f"Central Kubernetes identity changed for {job_id}: {key}" 

722 ) 

723 if record.get("uid") != uid: 

724 raise RuntimeError( 

725 "Central Kubernetes UID disagrees with checkpoint ownership authority" 

726 ) 

727 return False 

728 

729 previous_uid = record.get("uid") 

730 if previous_uid is not None and previous_uid != uid: 

731 raise RuntimeError( 

732 f"Kubernetes Job UID changed from {previous_uid!r} to {uid!r}; " 

733 "refusing central ownership" 

734 ) 

735 bound_at = time.time() 

736 was_deleted = bool(record.get("deleted")) 

737 record.update(immutable) 

738 record["uid"] = uid 

739 record["central_identity_bound_at"] = bound_at 

740 record["appearance_deadline"] = bound_at + appearance_timeout_seconds 

741 if was_deleted: 

742 record["requested_identity_deletion_superseded_at"] = bound_at 

743 record.pop("deleted_at", None) 

744 record["deleted"] = False 

745 record["submission_state"] = "appeared"