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 copy 

6import json 

7import os 

8import time 

9import traceback as traceback_module 

10from collections.abc import Callable, Mapping 

11from dataclasses import asdict, dataclass, field, is_dataclass 

12from datetime import UTC, datetime 

13from pathlib import Path 

14from threading import RLock 

15from typing import Any, ClassVar, Literal, cast 

16 

17from .artifact_io import ( 

18 REPORT_FILENAMES as _REPORT_FILENAMES, 

19) 

20from .artifact_io import ( 

21 atomic_write_text, 

22 ensure_private_directory, 

23 ensure_private_run_directory, 

24) 

25from .artifact_io import ( 

26 read_private_text as _read_private_text, 

27) 

28from .inference_contract import ( 

29 INFERENCE_OWNER_LABEL, 

30 MOONCAKE_CONFIG_DIGEST_ANNOTATION, 

31 MOONCAKE_STORE_RECONFIGURED_LOCAL_BUFFER_BYTES, 

32 STRUCTURED_DECISION_PATH, 

33 STRUCTURED_DECISION_QUESTIONS, 

34 TRITON_INPUT_DATA, 

35 TRITON_MODEL_NAME, 

36 TRITON_MODEL_VERSION, 

37 EndpointRole, 

38 Framework, 

39 InferenceRuntimeSpec, 

40 expected_endpoint_count, 

41 inference_deploy_extra_args, 

42 inference_framework_env, 

43 inference_health_path, 

44 inference_identity_fields, 

45 inference_mooncake_block, 

46 inference_request_body, 

47 inference_structured_decision_body, 

48 validate_inference_settings, 

49 write_triton_model_repository, 

50) 

51from .json_utils import loads_without_duplicate_keys 

52 

53__all__ = [ 

54 "INFERENCE_OWNER_LABEL", 

55 "MOONCAKE_CONFIG_DIGEST_ANNOTATION", 

56 "MOONCAKE_STORE_RECONFIGURED_LOCAL_BUFFER_BYTES", 

57 "STRUCTURED_DECISION_PATH", 

58 "STRUCTURED_DECISION_QUESTIONS", 

59 "TRITON_INPUT_DATA", 

60 "TRITON_MODEL_NAME", 

61 "TRITON_MODEL_VERSION", 

62 "ActionFailure", 

63 "ActionResult", 

64 "EndpointRole", 

65 "Framework", 

66 "InferenceRuntimeSpec", 

67 "RunCheckpoint", 

68 "RunContext", 

69 "RunSettings", 

70 "ValidationReport", 

71 "atomic_write_json", 

72 "atomic_write_text", 

73 "ensure_private_directory", 

74 "ensure_private_run_directory", 

75 "utc_now", 

76 "write_triton_model_repository", 

77] 

78 

79SCHEMA_VERSION = 2 

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

81 

82_LIVE_VALIDATION_DISABLE_EFS_BACKUPS_CONTEXT = "gco_live_validation_disable_efs_automatic_backups" 

83 

84 

85def utc_now() -> str: 

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

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

88 

89 

90class ActionFailure(RuntimeError): 

91 """An action failure that keeps the evidence gathered before it failed. 

92 

93 The runner records ``details`` on the failed :class:`ActionResult`, so the 

94 report carries what a passing run would have returned (for example every 

95 example's result) instead of an empty object. The mapping is deep-copied: 

96 callers often raise with a structure that also lives in ``checkpoint.state``. 

97 """ 

98 

99 def __init__(self, message: str, details: Mapping[str, Any]) -> None: 

100 super().__init__(message) 

101 self.details: dict[str, Any] = copy.deepcopy(dict(details)) 

102 

103 

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

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

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

107 return to_jsonable(asdict(value)) 

108 if isinstance(value, datetime): 

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

110 if isinstance(value, Path): 

111 return str(value) 

112 if isinstance(value, dict): 

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

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

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

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

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

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

119 return value 

120 return str(value) 

121 

122 

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

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

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

126 atomic_write_text(path, content) 

127 

128 

129@dataclass(frozen=True) 

130class RunSettings: 

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

132 

133 #: Whether this harness may record and honor run-tag adoption authority 

134 #: for stacks (``ownership/stacks.py``). Only the upgrade harness, whose 

135 #: stacks are deployed by a ``gco`` subprocess rather than change sets the 

136 #: harness prepares, opts in; a class attribute, so no flag can enable it. 

137 allows_run_tag_adoption: ClassVar[bool] = False 

138 

139 run_id: str 

140 repo_root: Path 

141 report_dir: Path 

142 checkpoint_path: Path 

143 expected_account: str 

144 expected_sha: str 

145 expected_branch: str 

146 profile: str 

147 requested_actions: tuple[str, ...] 

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

149 max_workers: int = 4 

150 job_timeout_seconds: int = 1800 

151 queue_timeout_seconds: int = 900 

152 poll_interval_seconds: int = 10 

153 destroy_attempts: int = 3 

154 destroy_retry_delay_seconds: int = 30 

155 confirm_kms_key_deletion: bool = False 

156 resume: bool = False 

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

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

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

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

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

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

163 min_free_disk_gib: int = 20 

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

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

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

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

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

169 #: Canonical JSON of the ``eks_capabilities_overrides`` CDK context this 

170 #: run deploys with (``--eks-capabilities``; see the ``eks-capabilities`` 

171 #: action), or ``""`` for none. Part of the resume identity (through 

172 #: ``extra_cdk_context``) like the scheduler overrides. 

173 eks_capabilities_overrides_json: str = "" 

174 #: Canonical JSON of the ``tracing_overrides`` CDK context (deep-merged 

175 #: over cdk.json ``tracing``), or ``""`` for none. The release entry point 

176 #: always sets ``{"sample_ratio":1.0}`` so the ``tracing`` action finds a 

177 #: span for every request it drives; part of the resume identity. 

178 tracing_overrides_json: str = "" 

179 

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

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

182 inference_enabled: bool = False 

183 selected_region: str = "" 

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

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

186 request_max_tokens: int = 8 

187 namespace: str = "gco-inference" 

188 health_path: str = "/health" 

189 gpu_count: int = 0 

190 baseline_replicas: int = 1 

191 autoscale_initial_replicas: int = 1 

192 hpa_min_replicas: int = 2 

193 hpa_max_replicas: int = 2 

194 hpa_cpu_target: int = 70 

195 #: Scenarios in the matrix. ``None`` derives it from the runtimes (every 

196 #: role of every runtime); an explicit value must match, so a checkpoint 

197 #: or caller that still assumes the old four-scenario matrix fails loudly. 

198 endpoint_count: int | None = None 

199 proxy_tls_cpu_request: str = "100m" 

200 proxy_tls_cpu_target: int = 70 

201 command_timeout_seconds: int = 300 

202 readiness_timeout_seconds: int = 1800 

203 hpa_timeout_seconds: int = 900 

204 deletion_timeout_seconds: int = 900 

205 monitor_interval_seconds: int = 15 

206 hpa_stability_intervals: int = 2 

207 consent: bool = False 

208 

209 def __post_init__(self) -> None: 

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

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

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

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

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

215 if checkpoint_path.parent != report_dir: 

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

217 if checkpoint_path.name in _REPORT_FILENAMES: 

218 raise ValueError( 

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

220 ) 

221 if self.inference_enabled: 

222 if self.endpoint_count is None: 

223 object.__setattr__( 

224 self, "endpoint_count", expected_endpoint_count(self.inference_runtimes) 

225 ) 

226 validate_inference_settings(self) 

227 

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

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

230 return inference_request_body(self, runtime) 

231 

232 @staticmethod 

233 def structured_decision_body(runtime: InferenceRuntimeSpec) -> dict[str, Any]: 

234 """Return the ``/v1/systemone`` body the vLLM baseline scenario sends.""" 

235 return inference_structured_decision_body(runtime) 

236 

237 @staticmethod 

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

239 return inference_framework_env(runtime) 

240 

241 @staticmethod 

242 def deploy_extra_args( 

243 runtime: InferenceRuntimeSpec, *, endpoint_name: str = "<endpoint>" 

244 ) -> tuple[str, ...]: 

245 return inference_deploy_extra_args(runtime, endpoint_name=endpoint_name) 

246 

247 def runtime_health_path(self, runtime: InferenceRuntimeSpec) -> str: 

248 """The readiness path the deploy probes for ``runtime``.""" 

249 return inference_health_path(self, runtime) 

250 

251 @staticmethod 

252 def mooncake_block(role: EndpointRole, *, reconfigured: bool = False) -> dict[str, Any] | None: 

253 """The ``spec.mooncake`` block a scenario persists, or ``None``.""" 

254 return inference_mooncake_block(role, reconfigured=reconfigured) 

255 

256 @property 

257 def kubeconfig_path(self) -> Path: 

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

259 return self.report_dir / "kubeconfig" 

260 

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

262 return inference_identity_fields(self) 

263 

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

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

266 context = {_LIVE_VALIDATION_DISABLE_EFS_BACKUPS_CONTEXT: "true"} 

267 if self.optional_schedulers: 

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

269 if self.eks_capabilities_overrides_json: 

270 context["eks_capabilities_overrides"] = self.eks_capabilities_overrides_json 

271 if self.tracing_overrides_json: 

272 context["tracing_overrides"] = self.tracing_overrides_json 

273 return context 

274 

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

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

277 identity = { 

278 "run_id": self.run_id, 

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

280 "expected_account": self.expected_account, 

281 "expected_sha": self.expected_sha, 

282 "expected_branch": self.expected_branch, 

283 "profile": self.profile, 

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

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

286 "confirm_kms_key_deletion": self.confirm_kms_key_deletion, 

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

288 "extra_cdk_context": self.extra_cdk_context(), 

289 } 

290 if self.inference_enabled: 

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

292 return identity 

293 

294 

295@dataclass 

296class ActionResult: 

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

298 

299 name: str 

300 description: str 

301 status: ActionStatus 

302 started_at: str 

303 ended_at: str 

304 duration_seconds: float 

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

306 error: str | None = None 

307 traceback: str | None = None 

308 

309 @classmethod 

310 def passed( 

311 cls, 

312 *, 

313 name: str, 

314 description: str, 

315 started_at: str, 

316 started_monotonic: float, 

317 ended_monotonic: float, 

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

319 ) -> ActionResult: 

320 return cls( 

321 name=name, 

322 description=description, 

323 status="passed", 

324 started_at=started_at, 

325 ended_at=utc_now(), 

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

327 details=details or {}, 

328 ) 

329 

330 @classmethod 

331 def failed( 

332 cls, 

333 *, 

334 name: str, 

335 description: str, 

336 started_at: str, 

337 started_monotonic: float, 

338 ended_monotonic: float, 

339 error: BaseException, 

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

341 ) -> ActionResult: 

342 return cls( 

343 name=name, 

344 description=description, 

345 status="failed", 

346 started_at=started_at, 

347 ended_at=utc_now(), 

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

349 details=details or {}, 

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

351 traceback="".join( 

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

353 ), 

354 ) 

355 

356 @classmethod 

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

358 return cls( 

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

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

361 status=value["status"], 

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

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

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

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

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

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

368 ) 

369 

370 

371@dataclass 

372class RunCheckpoint: 

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

374 

375 identity: dict[str, Any] 

376 created_at: str = field(default_factory=utc_now) 

377 updated_at: str = field(default_factory=utc_now) 

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

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

380 deployment_attempted: bool = False 

381 destroyed: bool = False 

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

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

384 schema_version: int = SCHEMA_VERSION 

385 

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

387 self.updated_at = utc_now() 

388 serialized = to_jsonable(self) 

389 if not isinstance(serialized, dict): 

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

391 return serialized 

392 

393 @classmethod 

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

395 try: 

396 checkpoint_text = _read_private_text(path) 

397 except (OSError, UnicodeError) as exc: 

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

399 try: 

400 raw = loads_without_duplicate_keys(checkpoint_text) 

401 except ValueError as exc: 

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

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

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

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

406 if not isinstance(results_raw, dict): 

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

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

409 if not isinstance(state_raw, dict): 

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

411 return cls( 

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

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

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

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

416 action_results={ 

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

418 for name, value in results_raw.items() 

419 if isinstance(value, dict) 

420 }, 

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

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

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

424 state=dict(state_raw), 

425 schema_version=SCHEMA_VERSION, 

426 ) 

427 

428 

429@dataclass 

430class ValidationReport: 

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

432 

433 run_id: str 

434 identity: dict[str, Any] 

435 selected_actions: list[str] 

436 started_at: str 

437 ended_at: str | None = None 

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

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

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

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

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

443 fatal_error: str | None = None 

444 schema_version: int = SCHEMA_VERSION 

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

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

447 title: str = "GCO Live Release Validation" 

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

449 

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

451 serialized = to_jsonable(self) 

452 if not isinstance(serialized, dict): 

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

454 return serialized 

455 

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

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

458 ensure_private_directory(directory) 

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

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

461 atomic_write_json(json_path, self.to_dict()) 

462 atomic_write_text(markdown_path, self.to_markdown()) 

463 return json_path, markdown_path 

464 

465 def to_markdown(self) -> str: 

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

467 identity = self.identity 

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

469 lines = [ 

470 f"# {self.title}", 

471 "", 

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

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

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

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

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

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

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

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

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

481 "", 

482 "## Actions", 

483 "", 

484 "| Action | Status | Duration | Error |", 

485 "|---|---:|---:|---|", 

486 ] 

487 for result in self.action_results: 

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

489 lines.append( 

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

491 ) 

492 if not self.action_results: 

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

494 

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

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

497 lines.append("```") 

498 

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

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

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

502 lines.append("```") 

503 

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

505 if self.fatal_error or failures: 

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

507 if self.fatal_error: 

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

509 for result in failures: 

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

511 lines.extend( 

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

513 ) 

514 

515 lines.append("") 

516 return "\n".join(lines) 

517 

518 

519@dataclass 

520class RunContext: 

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

522 

523 settings: RunSettings 

524 checkpoint: RunCheckpoint 

525 report: ValidationReport 

526 cdk_context: dict[str, Any] 

527 deployment_regions: tuple[str, ...] 

528 config: Any 

529 session: Any 

530 stack_manager: Any 

531 aws_client: Any 

532 job_manager: Any 

533 persist_callback: Callable[[RunCheckpoint], None] 

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

535 

536 def persist(self) -> None: 

537 with self.state_lock: 

538 self.persist_callback(self.checkpoint) 

539 

540 def register_job( 

541 self, 

542 *, 

543 name: str, 

544 namespace: str, 

545 region: str, 

546 path: str, 

547 run_label: str, 

548 transport_region: str | None, 

549 ) -> dict[str, Any]: 

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

551 

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

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

554 """ 

555 with self.state_lock: 

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

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

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

559 ): 

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

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

562 matches = [ 

563 item 

564 for item in jobs 

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

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

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

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

569 ] 

570 if len(matches) > 1: 

571 raise RuntimeError( 

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

573 ) 

574 expected = { 

575 "name": name, 

576 "namespace": namespace, 

577 "region": region, 

578 "path": path, 

579 "run_label": run_label, 

580 "transport_region": transport_region, 

581 } 

582 if matches: 

583 record = matches[0] 

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

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

586 raise RuntimeError( 

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

588 ) 

589 record.setdefault( 

590 "submission_state", 

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

592 ) 

593 else: 

594 record = { 

595 **expected, 

596 "uid": None, 

597 "deleted": False, 

598 "submission_state": "registered", 

599 } 

600 jobs.append(record) 

601 self.persist_callback(self.checkpoint) 

602 return record 

603 

604 def prepare_job_submission( 

605 self, 

606 record: dict[str, Any], 

607 *, 

608 envelope: dict[str, Any], 

609 resumable: bool, 

610 ) -> None: 

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

612 canonical = to_jsonable(envelope) 

613 if not isinstance(canonical, dict): 

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

615 with self.state_lock: 

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

617 previous = record.get("submission_envelope") 

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

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

620 previous_resumable = record.get("submission_resumable") 

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

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

623 if state == "registered": 

624 record["submission_state"] = "prepared" 

625 elif state not in { 

626 "prepared", 

627 "submitting", 

628 "submitted", 

629 "appeared", 

630 "deleted", 

631 "blocked", 

632 "not_submitted", 

633 }: 

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

635 record["submission_envelope"] = canonical 

636 record["submission_resumable"] = resumable 

637 self.persist_callback(self.checkpoint) 

638 

639 def begin_job_submission( 

640 self, 

641 record: dict[str, Any], 

642 *, 

643 reconciliation_timeout_seconds: int, 

644 ) -> None: 

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

646 with self.state_lock: 

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

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

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

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

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

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

653 now = time.time() 

654 record["submission_state"] = "submitting" 

655 record["submission_started_at"] = now 

656 record["submission_reconcile_deadline"] = now + reconciliation_timeout_seconds 

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

658 self.persist_callback(self.checkpoint) 

659 

660 def finish_job_submission( 

661 self, 

662 record: dict[str, Any], 

663 submission: dict[str, Any], 

664 *, 

665 appearance_timeout_seconds: int, 

666 ) -> None: 

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

668 with self.state_lock: 

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

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

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

672 acknowledged_at = time.time() 

673 if state != "appeared": 

674 record["submission_state"] = "submitted" 

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

676 record["submission_acknowledged_at"] = acknowledged_at 

677 record["appearance_deadline"] = acknowledged_at + appearance_timeout_seconds 

678 self.persist_callback(self.checkpoint) 

679 

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

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

682 with self.state_lock: 

683 record["submission_state"] = "blocked" 

684 record["submission_blocked_reason"] = reason 

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

686 self.persist_callback(self.checkpoint) 

687 

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

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

690 with self.state_lock: 

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

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

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

694 record["submission_state"] = "not_submitted" 

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

696 self.persist_callback(self.checkpoint) 

697 

698 def mark_central_job_cancelled_before_claim( 

699 self, 

700 record: dict[str, Any], 

701 *, 

702 job_id: str, 

703 ) -> None: 

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

705 if not job_id: 

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

707 with self.state_lock: 

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

709 previous_job_id = record.get("central_cancelled_before_claim_job_id") 

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

711 return 

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

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

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

715 raise RuntimeError( 

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

717 ) 

718 record["submission_state"] = "not_submitted" 

719 record["central_cancelled_before_claim_job_id"] = job_id 

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

721 self.persist_callback(self.checkpoint) 

722 

723 def mark_central_job_not_created_by_worker( 

724 self, 

725 record: dict[str, Any], 

726 *, 

727 job_id: str, 

728 ) -> None: 

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

730 if not job_id: 

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

732 with self.state_lock: 

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

734 previous_job_id = record.get("central_worker_not_created_job_id") 

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

736 return 

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

738 raise RuntimeError( 

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

740 ) 

741 central_identity = ( 

742 record.get("k8s_job_name"), 

743 record.get("k8s_job_namespace"), 

744 record.get("k8s_job_uid"), 

745 ) 

746 if ( 

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

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

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

750 ): 

751 raise RuntimeError( 

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

753 ) 

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

755 raise RuntimeError( 

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

757 ) 

758 record["submission_state"] = "not_submitted"