Coverage for scripts / live_release_validation / checks / inference.py: 100.00%
503 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Owned, sequential managed-inference endpoint lifecycle and evidence checks."""
3from __future__ import annotations
5import json
6import os
7import secrets
8import subprocess
9import time
10from pathlib import Path
11from typing import Any, cast
13from scripts.example_job_validation.kube import KubectlRunner
15from ..models import (
16 TRITON_MODEL_VERSION,
17 RunContext,
18 RunSettings,
19)
20from .inference_commands import (
21 OWNER_LABEL,
22 EndpointPlan,
23 _last_json_document,
24 build_configure_store_command,
25 build_delete_command,
26 build_deploy_command,
27 build_endpoint_plans,
28 build_health_command,
29 build_invoke_command,
30 build_model_delete_command,
31 build_model_list_command,
32 build_model_upload_command,
33 build_model_uri_command,
34 build_models_command,
35 build_structured_decision_command,
36 extract_generated_text,
37 extract_structured_decision,
38)
39from .inference_common import (
40 InferenceCommandFailure as _CommandFailure,
41)
42from .inference_common import (
43 ManagedInferenceValidationError,
44)
45from .inference_inventory import KUBERNETES_INVENTORY_KINDS, InferenceInventoryMixin
46from .inference_runtime import InferenceRuntimeMixin
47from .inference_scenarios import InferenceScenarioMixin
49__all__ = [
50 "KUBERNETES_INVENTORY_KINDS",
51 "OWNER_LABEL",
52 "EndpointPlan",
53 "ManagedInferenceLifecycle",
54 "ManagedInferenceValidationError",
55 "build_configure_store_command",
56 "build_delete_command",
57 "build_deploy_command",
58 "build_endpoint_plans",
59 "build_health_command",
60 "build_invoke_command",
61 "build_model_delete_command",
62 "build_model_list_command",
63 "build_model_upload_command",
64 "build_model_uri_command",
65 "build_models_command",
66 "build_structured_decision_command",
67 "extract_generated_text",
68 "extract_structured_decision",
69 "initialize_run_state",
70]
72_STATE_KEY = "inference_validation"
73_MAX_PRIVATE_OUTPUT = 64 * 1024
76def initialize_run_state(
77 ctx: RunContext,
78 settings: RunSettings,
79) -> tuple[tuple[EndpointPlan, ...], dict[str, Any]]:
80 """Persist a random owner nonce and immutable plan before any create check."""
81 raw_state = ctx.checkpoint.state.get(_STATE_KEY)
82 if raw_state is None:
83 owner_nonce = secrets.token_hex(32)
84 plans = build_endpoint_plans(settings, owner_nonce)
85 expected_plan = [plan.private_dict(owner_nonce) for plan in plans]
86 state: dict[str, Any] = {
87 "contract_version": 3,
88 "owner_nonce": owner_nonce,
89 "phase": "planned",
90 "plan": expected_plan,
91 "endpoints": [
92 {
93 **plan.private_dict(owner_nonce),
94 "phase": "planned",
95 "incarnation": 1,
96 "closed_incarnations": [],
97 "cleanup_phase": "not-started",
98 "validation_complete": False,
99 "absence_proven": False,
100 "owned": False,
101 "commands": [],
102 "cleanup_attempts": [],
103 "failures": [],
104 }
105 for plan in plans
106 ],
107 }
108 ctx.checkpoint.state[_STATE_KEY] = state
109 ctx.persist()
110 return plans, state
112 if not isinstance(raw_state, dict):
113 raise ManagedInferenceValidationError("inference checkpoint state is invalid")
114 state = cast(dict[str, Any], raw_state)
115 raw_owner_nonce = state.get("owner_nonce")
116 if (
117 state.get("contract_version") != 3
118 or not isinstance(raw_owner_nonce, str)
119 or len(raw_owner_nonce) != 64
120 or any(character not in "0123456789abcdef" for character in raw_owner_nonce)
121 ):
122 raise ManagedInferenceValidationError("inference checkpoint owner nonce is invalid")
123 owner_nonce = raw_owner_nonce
124 plans = build_endpoint_plans(settings, owner_nonce)
125 expected_plan = [plan.private_dict(owner_nonce) for plan in plans]
126 if state.get("plan") != expected_plan:
127 raise ManagedInferenceValidationError("inference checkpoint plan changed")
128 records = state.get("endpoints")
129 if not isinstance(records, list) or len(records) != len(plans):
130 raise ManagedInferenceValidationError("inference endpoint checkpoint is invalid")
131 for plan, raw_record in zip(plans, records, strict=True):
132 if not isinstance(raw_record, dict):
133 raise ManagedInferenceValidationError("inference endpoint checkpoint is invalid")
134 expected = plan.private_dict(owner_nonce)
135 if any(raw_record.get(key) != value for key, value in expected.items()):
136 raise ManagedInferenceValidationError("inference checkpoint endpoint identity changed")
137 if (
138 not isinstance(raw_record.get("incarnation"), int)
139 or raw_record["incarnation"] < 1
140 or not isinstance(raw_record.get("closed_incarnations"), list)
141 ):
142 raise ManagedInferenceValidationError("inference checkpoint incarnation state changed")
143 return plans, state
146class ManagedInferenceLifecycle(
147 InferenceInventoryMixin, InferenceRuntimeMixin, InferenceScenarioMixin
148):
149 """Run and clean the runtime matrix with durable ownership.
151 Six strictly sequential scenarios: vLLM baseline, CPU-HPA and Mooncake
152 store (TCP, cold tier, reconfigured mid-leg), SGLang baseline and CPU-HPA,
153 and Triton baseline served from a model repository the run uploads to the
154 registry and removes afterwards.
155 """
157 def __init__(
158 self,
159 *,
160 ctx: RunContext,
161 settings: RunSettings,
162 plans: tuple[EndpointPlan, ...],
163 state: dict[str, Any],
164 kubectl: KubectlRunner,
165 kubeconfig_path: Path,
166 ) -> None:
167 self.ctx = ctx
168 self.settings = settings
169 self.plans = plans
170 self.state = state
171 self.kubectl = kubectl
172 self.kubeconfig_path = kubeconfig_path
173 owner_nonce = state.get("owner_nonce")
174 if not isinstance(owner_nonce, str) or not owner_nonce:
175 raise ManagedInferenceValidationError("inference owner nonce is missing")
176 self.owner_nonce = owner_nonce
177 records = state.get("endpoints")
178 if not isinstance(records, list) or any(not isinstance(item, dict) for item in records):
179 raise ManagedInferenceValidationError("managed inference endpoint state is invalid")
180 self.records = cast(list[dict[str, Any]], records)
181 self._table: Any | None = None
183 def _persist(self) -> None:
184 self.ctx.persist()
186 def _set_phase(
187 self,
188 record: dict[str, Any],
189 phase: str,
190 **values: Any,
191 ) -> None:
192 record["phase"] = phase
193 record.update(values)
194 self.state["phase"] = f"endpoint-{record['ordinal']}:{phase}"
195 self._persist()
197 def _record_failure(
198 self,
199 record: dict[str, Any],
200 stage: str,
201 exc: BaseException,
202 ) -> None:
203 failures = record.setdefault("failures", [])
204 if not isinstance(failures, list):
205 failures = []
206 record["failures"] = failures
207 failures.append({"stage": stage, "error": f"{type(exc).__name__}: {exc}"})
208 record["last_failed_phase"] = stage
209 self._persist()
211 @property
212 def table(self) -> Any:
213 if self._table is None:
214 try:
215 resource = self.ctx.session.resource(
216 "dynamodb", region_name=self.ctx.config.global_region
217 )
218 self._table = resource.Table(f"{self.ctx.config.project_name}-inference-endpoints")
219 except Exception as exc:
220 self.state["ddb_setup_error"] = f"{type(exc).__name__}: {exc}"
221 self._persist()
222 raise ManagedInferenceValidationError(
223 "managed inference state store could not be opened"
224 ) from None
225 return self._table
227 def _strong_get(self, record: dict[str, Any]) -> dict[str, Any] | None:
228 try:
229 response = self.table.get_item(
230 Key={"endpoint_name": record["name"]},
231 ConsistentRead=True,
232 )
233 except Exception as exc:
234 self._record_failure(record, "ddb-strong-read", exc)
235 raise ManagedInferenceValidationError(
236 "managed inference strong state read failed; inspect the private checkpoint"
237 ) from None
238 item = response.get("Item") if isinstance(response, dict) else None
239 if item is None:
240 return None
241 if not isinstance(item, dict):
242 raise ManagedInferenceValidationError("managed inference state record is malformed")
243 return cast(dict[str, Any], item)
245 def _is_owned(self, item: dict[str, Any]) -> bool:
246 labels = item.get("labels")
247 return isinstance(labels, dict) and labels.get(OWNER_LABEL) == self.owner_nonce
249 def _verify_item_contract(
250 self,
251 plan: EndpointPlan,
252 item: dict[str, Any],
253 record: dict[str, Any],
254 ) -> None:
255 lifecycle_id = item.get("lifecycle_id")
256 if not isinstance(lifecycle_id, str) or not lifecycle_id:
257 raise ManagedInferenceValidationError(
258 "inference stored endpoint has no immutable lifecycle identity"
259 )
260 closed = record.get("closed_incarnations")
261 closed_ids = (
262 {
263 entry.get("lifecycle_id")
264 for entry in closed
265 if isinstance(entry, dict) and isinstance(entry.get("lifecycle_id"), str)
266 }
267 if isinstance(closed, list)
268 else set()
269 )
270 if lifecycle_id in closed_ids:
271 raise ManagedInferenceValidationError(
272 "inference endpoint reverted to a closed lifecycle incarnation"
273 )
274 observed_lifecycle = record.get("lifecycle_id")
275 if observed_lifecycle is None:
276 record["lifecycle_id"] = lifecycle_id
277 self._persist()
278 elif observed_lifecycle != lifecycle_id:
279 raise ManagedInferenceValidationError(
280 "inference endpoint incarnation changed; refusing replacement ownership"
281 )
282 spec = item.get("spec")
283 if not isinstance(spec, dict):
284 raise ManagedInferenceValidationError("managed inference stored spec is malformed")
285 runtime = plan.runtime
286 expected_base: dict[str, Any] = {
287 "image": runtime.image,
288 "port": runtime.port,
289 "replicas": plan.replicas,
290 "gpu_count": self.settings.gpu_count,
291 "health_check_path": self.settings.runtime_health_path(runtime),
292 }
293 # The two launcher contracts are persisted explicitly; Triton has no
294 # ``--framework`` and is recognised from its image.
295 expected_base["framework"] = (
296 runtime.framework if runtime.framework in ("vllm", "sglang") else None
297 )
298 # Every launcher receives its arguments on argv, so a stored record
299 # without exactly those args is not this run's endpoint.
300 expected_base["args"] = list(
301 self.settings.deploy_extra_args(runtime, endpoint_name=plan.name)
302 )
303 # The scenario's Mooncake block (store mode with the cold tier, plus
304 # the buffer size once ``configure-store`` has run) or none at all.
305 expected_base["mooncake"] = self.settings.mooncake_block(
306 plan.role, reconfigured=record.get("store_reconfigured") is True
307 )
308 for key, value in expected_base.items():
309 if spec.get(key) != value:
310 raise ManagedInferenceValidationError(
311 "managed inference stored endpoint contract does not match this run"
312 )
313 # A runtime that takes no environment persists none (``None`` or an
314 # empty map); the launchers persist exactly ``MODEL``.
315 expected_env = self.settings.framework_env(runtime)
316 if (spec.get("env") or {}) != expected_env:
317 raise ManagedInferenceValidationError(
318 "managed inference stored endpoint contract does not match this run"
319 )
320 if runtime.uses_model_registry:
321 model_source = record.get("model_source")
322 if not isinstance(model_source, str) or spec.get("model_source") != model_source:
323 raise ManagedInferenceValidationError(
324 "managed inference stored model source does not match this run's upload"
325 )
326 elif spec.get("model_source") is not None:
327 raise ManagedInferenceValidationError(
328 "managed inference stored endpoint unexpectedly serves from a model source"
329 )
330 if item.get("target_regions") != [self.settings.selected_region]:
331 raise ManagedInferenceValidationError(
332 "managed inference stored target region does not match this run"
333 )
334 if item.get("namespace") != self.settings.namespace:
335 raise ManagedInferenceValidationError(
336 "managed inference stored namespace does not match this run"
337 )
338 autoscaling = spec.get("autoscaling")
339 if plan.autoscaling:
340 expected_autoscaling = {
341 "enabled": True,
342 "min_replicas": self.settings.hpa_min_replicas,
343 "max_replicas": self.settings.hpa_max_replicas,
344 "metrics": [{"type": "cpu", "target": self.settings.hpa_cpu_target}],
345 }
346 if autoscaling != expected_autoscaling:
347 raise ManagedInferenceValidationError(
348 "managed inference stored HPA contract does not match this run"
349 )
350 elif autoscaling is not None:
351 raise ManagedInferenceValidationError(
352 "managed inference baseline unexpectedly has autoscaling configured"
353 )
355 @staticmethod
356 def _truncated(value: str) -> str:
357 return value[-_MAX_PRIVATE_OUTPUT:]
359 def _set_invoke_journal_outcome(
360 self,
361 record: dict[str, Any],
362 status: str,
363 **values: Any,
364 ) -> None:
365 """Update the durable non-replay journal in the same checkpoint write."""
366 journal = record.get("invoke_journal")
367 if not isinstance(journal, dict):
368 raise ManagedInferenceValidationError("managed inference invoke journal is invalid")
369 journal["status"] = status
370 journal.update(values)
372 def _run_command(
373 self,
374 record: dict[str, Any],
375 stage: str,
376 command: list[str],
377 *,
378 deadline: float | None = None,
379 ) -> str:
380 environment = dict(os.environ)
381 environment["KUBECONFIG"] = str(self.kubeconfig_path)
382 command_timeout = float(self.settings.command_timeout_seconds)
383 if deadline is not None:
384 remaining = deadline - time.monotonic()
385 if remaining <= 0:
386 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
387 commands.append({"stage": stage, "argv": command, "deadline_exhausted": True})
388 self._persist()
389 raise _CommandFailure(f"{stage} phase deadline exhausted")
390 command_timeout = min(command_timeout, remaining)
391 try:
392 result = subprocess.run(
393 command,
394 cwd=self.settings.repo_root,
395 capture_output=True,
396 text=True,
397 timeout=command_timeout,
398 env=environment,
399 shell=False,
400 check=False,
401 )
402 except subprocess.TimeoutExpired as exc:
403 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
404 commands.append(
405 {
406 "stage": stage,
407 "argv": command,
408 "timed_out": True,
409 "timeout_seconds": command_timeout,
410 "stdout": self._truncated(str(exc.stdout or "")),
411 "stderr": self._truncated(str(exc.stderr or "")),
412 }
413 )
414 if stage == "invoke":
415 self._set_invoke_journal_outcome(
416 record,
417 "ambiguous",
418 reason="timeout",
419 stdout=self._truncated(str(exc.stdout or "")),
420 )
421 self._persist()
422 raise _CommandFailure(f"{stage} timed out") from None
423 except (OSError, UnicodeError) as exc:
424 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
425 commands.append(
426 {
427 "stage": stage,
428 "argv": command,
429 "launch_error": f"{type(exc).__name__}: {exc}",
430 }
431 )
432 if stage == "invoke":
433 self._set_invoke_journal_outcome(
434 record,
435 "failed",
436 reason="launch-error",
437 )
438 self._persist()
439 raise _CommandFailure(f"{stage} could not start") from None
441 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
442 commands.append(
443 {
444 "stage": stage,
445 "argv": command,
446 "returncode": result.returncode,
447 "stdout": self._truncated(result.stdout),
448 "stderr": self._truncated(result.stderr),
449 }
450 )
451 if stage == "invoke":
452 self._set_invoke_journal_outcome(
453 record,
454 "succeeded" if result.returncode == 0 else "failed",
455 returncode=result.returncode,
456 stdout=self._truncated(result.stdout),
457 stderr=self._truncated(result.stderr),
458 )
459 self._persist()
460 if result.returncode != 0:
461 raise _CommandFailure(f"{stage} exited nonzero")
462 return result.stdout
464 def ensure_owned_endpoint(
465 self,
466 plan: EndpointPlan,
467 record: dict[str, Any],
468 ) -> None:
469 """Adopt only this run's marker, or prove collision-free absence then create."""
470 item = self._strong_get(record)
471 if item is not None:
472 if not self._is_owned(item):
473 raise ManagedInferenceValidationError(
474 "managed inference endpoint collision detected; refusing ownership"
475 )
476 self._verify_item_contract(plan, item, record)
477 record["owned"] = True
478 self._set_phase(record, "ownership-confirmed")
479 return
481 inventory = self.kubernetes_inventory(
482 record,
483 deadline=time.monotonic() + self.settings.readiness_timeout_seconds,
484 )
485 if any(inventory.values()):
486 raise ManagedInferenceValidationError(
487 "managed inference Kubernetes name collision detected; refusing creation"
488 )
489 self._set_phase(record, "collision-checked")
490 model_source = self.ensure_model_repository(plan, record)
491 self._set_phase(record, "deploy-started")
492 try:
493 self._run_command(
494 record,
495 "deploy",
496 build_deploy_command(
497 self.settings, plan, self.owner_nonce, model_source=model_source
498 ),
499 )
500 except _CommandFailure as exc:
501 self._record_failure(record, "deploy", exc)
502 item = self._strong_get(record)
503 if item is None or not self._is_owned(item):
504 raise ManagedInferenceValidationError(
505 "managed inference deploy failed; inspect the private checkpoint"
506 ) from None
507 self._wait_for_owned_record(plan, record)
509 def _wait_for_healthy_backend(
510 self,
511 plan: EndpointPlan,
512 record: dict[str, Any],
513 ) -> dict[str, Any]:
514 attempts = record.setdefault("backend_probe_attempts", [])
515 if not isinstance(attempts, list):
516 raise ManagedInferenceValidationError("managed backend probe history is invalid")
517 deadline = time.monotonic() + self.settings.readiness_timeout_seconds
518 heartbeat_at = time.monotonic()
519 while True:
520 if time.monotonic() >= deadline:
521 raise self._timeout_error(
522 plan,
523 record,
524 "managed health did not converge before timeout",
525 reason="health-timeout",
526 )
527 heartbeat_at = self.keep_cluster_tunnel_alive(record, heartbeat_at, deadline=deadline)
528 item = self._strong_get(record)
529 if item is None or not self._is_owned(item):
530 raise ManagedInferenceValidationError("managed ownership changed during health")
531 self._verify_item_contract(plan, item, record)
532 statuses = item.get("region_status")
533 regional = (
534 statuses.get(self.settings.selected_region) if isinstance(statuses, dict) else None
535 )
536 if (
537 item.get("desired_state") != "running"
538 or not isinstance(regional, dict)
539 or regional.get("state") != "running"
540 ):
541 raise ManagedInferenceValidationError(
542 "managed endpoint stopped running during health"
543 )
544 attempt: dict[str, Any] = {
545 "attempt": len(attempts) + 1,
546 "started_at_monotonic": time.monotonic(),
547 "classification": "started",
548 }
549 attempts.append(attempt)
551 def finish(classification: str, **values: Any) -> None:
552 attempt.update( # noqa: B023 - helper is called synchronously in this iteration
553 ended_at_monotonic=time.monotonic(), classification=classification, **values
554 )
555 self._persist()
557 self._persist()
558 try:
559 output = self._run_command(
560 record,
561 "health",
562 build_health_command(self.settings, plan),
563 deadline=deadline,
564 )
565 except _CommandFailure as exc:
566 finish("command-failed", error=str(exc))
567 raise
568 try:
569 health = _last_json_document(output)
570 except ManagedInferenceValidationError as exc:
571 finish("malformed-output", error=str(exc))
572 raise
573 status = health.get("status") if isinstance(health, dict) else None
574 http_status = health.get("http_status") if isinstance(health, dict) else None
575 if (
576 not isinstance(status, str)
577 or isinstance(http_status, bool)
578 or not isinstance(http_status, int)
579 ):
580 finish("malformed-contract")
581 raise ManagedInferenceValidationError(
582 "managed inference health probe returned a malformed contract"
583 )
584 attempt.update(
585 {
586 "status": status,
587 "http_status": http_status,
588 "path": health.get("path"),
589 "latency_ms": health.get("latency_ms"),
590 "body_summary": self._truncated(
591 json.dumps(health.get("body"), sort_keys=True, default=str)
592 ),
593 }
594 )
595 if status == "healthy" and 200 <= http_status < 300:
596 finish("healthy")
597 return cast(dict[str, Any], health)
598 retryable = status == "unhealthy" and (
599 http_status in {404, 429} or 500 <= http_status < 600
600 )
601 if not retryable:
602 finish("terminal-contract")
603 raise ManagedInferenceValidationError(
604 "managed inference health probe returned a terminal status or contract"
605 )
606 remaining = deadline - time.monotonic()
607 if remaining <= 0:
608 finish("deadline-exhausted")
609 raise ManagedInferenceValidationError(
610 "managed inference health endpoint did not converge before timeout"
611 )
612 finish("retryable-unhealthy")
613 time.sleep(min(float(self.settings.poll_interval_seconds), remaining))
615 def verify_backend_probes(
616 self,
617 plan: EndpointPlan,
618 record: dict[str, Any],
619 ) -> None:
620 """Require converged authenticated health and exact model discovery."""
621 try:
622 health = self._wait_for_healthy_backend(plan, record)
623 evidence: dict[str, Any] = {
624 "health": {
625 "healthy": True,
626 "http_status": health["http_status"],
627 "path": self.settings.runtime_health_path(plan.runtime),
628 }
629 }
631 runtime = plan.runtime
632 model_output = self._run_command(
633 record,
634 "model-info",
635 build_models_command(self.settings, plan),
636 )
637 model_info = _last_json_document(model_output)
638 if runtime.framework == "vllm":
639 data = model_info.get("data") if isinstance(model_info, dict) else None
640 model_ids = (
641 [
642 item.get("id")
643 for item in data
644 if isinstance(item, dict) and isinstance(item.get("id"), str)
645 ]
646 if isinstance(data, list)
647 else []
648 )
649 if runtime.model_id not in model_ids:
650 raise ManagedInferenceValidationError(
651 "managed vLLM model inventory omitted the configured model"
652 )
653 evidence["model_info"] = {
654 "path": runtime.model_info_path,
655 "configured_model_present": True,
656 "model_revision_pinned": True,
657 "model_count": len(model_ids),
658 }
659 elif runtime.framework == "triton":
660 # ``/v2/models/stats`` is Triton's inventory: every loaded
661 # model by name and version. The generated identity model must
662 # be loaded at its single version.
663 stats = model_info.get("model_stats") if isinstance(model_info, dict) else None
664 loaded = (
665 {
666 (entry.get("name"), str(entry.get("version")))
667 for entry in stats
668 if isinstance(entry, dict)
669 }
670 if isinstance(stats, list)
671 else set()
672 )
673 if (runtime.model_id, TRITON_MODEL_VERSION) not in loaded:
674 raise ManagedInferenceValidationError(
675 "managed Triton model statistics omitted the generated model"
676 )
677 evidence["model_info"] = {
678 "path": runtime.model_info_path,
679 "configured_model_present": True,
680 "model_version": TRITON_MODEL_VERSION,
681 "model_count": len(loaded),
682 }
683 else:
684 # ``gco inference models --framework sglang`` projects the
685 # server's resolved launcher arguments (``/server_info``), so
686 # the running process itself reports which model path and
687 # which revision it was started with.
688 if (
689 not isinstance(model_info, dict)
690 or model_info.get("model_path") != runtime.model_id
691 or model_info.get("revision") != runtime.model_revision
692 ):
693 raise ManagedInferenceValidationError(
694 "managed SGLang /server_info did not report the exact model path "
695 "and revision"
696 )
697 evidence["model_info"] = {
698 "path": runtime.model_info_path,
699 "configured_model_present": True,
700 "configured_revision_present": True,
701 }
702 except (ManagedInferenceValidationError, _CommandFailure) as exc:
703 self._record_failure(record, "backend-probes", exc)
704 raise ManagedInferenceValidationError(
705 "managed inference health/model probe failed its contract"
706 ) from None
708 record["backend_probe_evidence"] = evidence
709 self._set_phase(record, "backend-probes-verified")
711 def invoke(self, plan: EndpointPlan, record: dict[str, Any]) -> None:
712 """Invoke once, or recover a durable successful outcome without replay."""
713 command = build_invoke_command(self.settings, plan)
714 journal = record.get("invoke_journal")
715 if journal is None:
716 journal = {
717 "status": "started",
718 "framework": plan.runtime.framework,
719 "request_path": plan.runtime.request_path,
720 "argv": command,
721 }
722 record["invoke_journal"] = journal
723 self._persist()
724 output: str | None = None
725 elif not isinstance(journal, dict):
726 raise ManagedInferenceValidationError("managed inference invoke journal is invalid")
727 else:
728 if (
729 journal.get("framework") != plan.runtime.framework
730 or journal.get("request_path") != plan.runtime.request_path
731 or journal.get("argv") != command
732 ):
733 raise ManagedInferenceValidationError(
734 "managed inference invoke journal identity changed"
735 )
736 status = journal.get("status")
737 if status == "succeeded" and isinstance(journal.get("stdout"), str):
738 output = journal["stdout"]
739 elif status in {"started", "ambiguous", "failed"}:
740 raise ManagedInferenceValidationError(
741 "managed inference invocation has a non-replayable persisted outcome"
742 )
743 else:
744 raise ManagedInferenceValidationError(
745 "managed inference invoke journal status is invalid"
746 )
747 try:
748 if output is None:
749 output = self._run_command(record, "invoke", command)
750 generated_text = extract_generated_text(output, plan.runtime.framework)
751 except (ManagedInferenceValidationError, _CommandFailure) as exc:
752 self._record_failure(record, "invoke", exc)
753 raise ManagedInferenceValidationError(
754 "managed inference invocation failed its response contract"
755 ) from None
756 record["invoke_evidence"] = {
757 "framework": plan.runtime.framework,
758 "generated_text_non_empty": True,
759 "generated_text_length": len(generated_text),
760 "replayed": False,
761 }
762 self._set_phase(record, "invoked")
764 def _prepare_incarnation_for_resume(
765 self,
766 plan: EndpointPlan,
767 record: dict[str, Any],
768 ) -> None:
769 """Rotate a fully cleaned pre-invocation lifecycle before redeploying."""
770 lifecycle_id = record.get("lifecycle_id")
771 if (
772 record.get("cleanup_phase") != "absent"
773 or not isinstance(lifecycle_id, str)
774 or not lifecycle_id
775 or isinstance(record.get("invoke_evidence"), dict)
776 or isinstance(record.get("invoke_journal"), dict)
777 ):
778 return
779 # Re-prove strong DDB and Kubernetes absence before closing the old incarnation.
780 evidence = self.prove_absence(record)
781 if evidence.get("stable_absence_observations") != 2:
782 raise ManagedInferenceValidationError(
783 "cleaned inference incarnation did not retain stable absence"
784 )
785 closed = record.setdefault("closed_incarnations", [])
786 if not isinstance(closed, list):
787 raise ManagedInferenceValidationError("inference closed-incarnation state is invalid")
788 closed.append(
789 {
790 "number": record.get("incarnation", 1),
791 "lifecycle_id": lifecycle_id,
792 "invoked": False,
793 "cleanup_phase": "absent",
794 "absence_evidence": evidence,
795 "commands": record.get("commands", []),
796 "backend_probe_attempts": record.get("backend_probe_attempts", []),
797 "tunnel_heartbeats": record.get("tunnel_heartbeats", []),
798 "cleanup_attempts": record.get("cleanup_attempts", []),
799 "failures": record.get("failures", []),
800 }
801 )
802 record.update(
803 {
804 "incarnation": int(record.get("incarnation", 1)) + 1,
805 "lifecycle_id": None,
806 "owned": False,
807 "phase": "planned-resume-incarnation",
808 "cleanup_phase": "not-started",
809 "validation_complete": False,
810 "validation_steps_complete": False,
811 "absence_proven": True,
812 "commands": [],
813 "cleanup_attempts": [],
814 "failures": [],
815 }
816 )
817 for key in (
818 "backend_probe_evidence",
819 "backend_probe_attempts",
820 "tunnel_heartbeats",
821 "invoke_evidence",
822 "invoke_journal",
823 "structured_decision",
824 "hpa_stability_observations",
825 "last_hpa_replica_observation",
826 "last_readiness",
827 "last_failed_phase",
828 ):
829 record.pop(key, None)
830 self.state["phase"] = f"endpoint-{plan.ordinal}:incarnation-rotated"
831 self._persist()
833 def run_endpoint(self, plan: EndpointPlan, record: dict[str, Any]) -> bool:
834 """Run one endpoint's validation phases, or safely continue a durable resume."""
835 if record.get("validation_complete") is True:
836 self.prove_absence(record)
837 return False
838 invocation_finished = (
839 record.get("phase") == "invoked"
840 and isinstance(record.get("invoke_evidence"), dict)
841 and record["invoke_evidence"].get("generated_text_non_empty") is True
842 )
843 if record.get("validation_steps_complete") is True or invocation_finished:
844 # A crash after invoke or cleanup must never recreate or reinvoke;
845 # the caller's finally only needs to finish/prove cleanup.
846 record["validation_steps_complete"] = True
847 record["phase"] = "validation-complete-resume"
848 self._persist()
849 return False
850 journal = record.get("invoke_journal")
851 if isinstance(journal, dict) and not isinstance(record.get("invoke_evidence"), dict):
852 # Invocation intent is one-way: recover durable success without a request;
853 # any other persisted state fails closed for cleanup.
854 self.invoke(plan, record)
855 record["validation_steps_complete"] = True
856 self._persist()
857 return False
858 self._prepare_incarnation_for_resume(plan, record)
859 record["absence_proven"] = False
860 self._set_phase(record, "starting")
861 self.ensure_owned_endpoint(plan, record)
862 self.wait_for_ddb_running(plan, record)
863 self.wait_for_kubernetes_ready(plan, record)
864 if plan.autoscaling:
865 self.verify_hpa_stability(plan, record)
866 self.verify_backend_probes(plan, record)
867 self.invoke(plan, record)
868 # The vLLM baseline then asks /v1/systemone through the same path.
869 self.verify_structured_decision(plan, record)
870 # The store scenario then changes its configuration and must roll
871 # onto it; the rolled pods are what the restart audit below sees.
872 self.verify_store_reconfigure(plan, record)
873 # Last, so it covers the whole leg: a container that crashed or was
874 # killed by a probe anywhere between creation and invocation fails
875 # the leg even though every wait above was satisfied in between.
876 self.verify_no_container_restarts(plan, record)
877 record["validation_steps_complete"] = True
878 self._persist()
879 return True
881 def cleanup_endpoint(self, plan: EndpointPlan, record: dict[str, Any]) -> dict[str, Any]:
882 """Delete only a matching marker, then prove strong DDB and full K8s absence."""
883 attempts = record.setdefault("cleanup_attempts", [])
884 if not isinstance(attempts, list):
885 attempts = []
886 record["cleanup_attempts"] = attempts
887 attempt: dict[str, Any] = {"started_at_monotonic": time.monotonic()}
888 attempts.append(attempt)
889 record["cleanup_phase"] = "checking-ownership"
890 self._persist()
892 delete_error: Exception | None = None
893 item = self._strong_get(record)
894 if item is not None:
895 if not self._is_owned(item):
896 attempt["refused_collision"] = True
897 self._persist()
898 raise ManagedInferenceValidationError(
899 "managed inference cleanup refused a colliding endpoint"
900 )
901 self._verify_item_contract(plan, item, record)
902 record["owned"] = True
903 if item.get("desired_state") != "deleted":
904 record["cleanup_phase"] = "delete-requested"
905 self._persist()
906 lifecycle_id = record.get("lifecycle_id")
907 if not isinstance(lifecycle_id, str) or not lifecycle_id:
908 raise ManagedInferenceValidationError(
909 "inference cleanup has no checkpointed lifecycle identity"
910 )
911 try:
912 self._run_command(
913 record,
914 "delete",
915 build_delete_command(
916 self.settings,
917 plan,
918 self.owner_nonce,
919 lifecycle_id,
920 ),
921 )
922 except _CommandFailure as exc:
923 delete_error = exc
924 self._record_failure(record, "delete", exc)
925 replacement = self._strong_get(record)
926 if replacement is not None and (
927 not self._is_owned(replacement)
928 or replacement.get("lifecycle_id") != record.get("lifecycle_id")
929 ):
930 attempt["refused_replacement_race"] = True
931 self._persist()
932 raise ManagedInferenceValidationError(
933 "managed inference cleanup refused a replacement endpoint"
934 ) from None
936 try:
937 evidence = self.prove_absence(record)
938 repository = self.cleanup_model_repository(plan, record)
939 except (Exception, KeyboardInterrupt) as exc:
940 attempt["completed"] = False
941 attempt["error"] = f"{type(exc).__name__}: {exc}"
942 if delete_error is not None:
943 attempt["delete_error"] = f"{type(delete_error).__name__}: {delete_error}"
944 self._persist()
945 raise
946 if repository is not None:
947 evidence = {**evidence, "model_repository": repository.get("cleanup")}
948 attempt["completed"] = True
949 attempt["delete_recovered"] = delete_error is not None
950 record["cleanup_phase"] = "absent"
951 self._persist()
952 return evidence
954 def execute(self) -> dict[str, Any]:
955 """Run strictly sequential endpoints and aggregate only after all cleanup attempts."""
956 primary_error: Exception | KeyboardInterrupt | None = None
957 cleanup_errors: list[tuple[int, Exception | KeyboardInterrupt]] = []
958 validated = 0
959 self.state["phase"] = "running"
960 self._persist()
962 try:
963 for plan, record in zip(self.plans, self.records, strict=True):
964 endpoint_error: Exception | KeyboardInterrupt | None = None
965 ran = False
966 try:
967 ran = self.run_endpoint(plan, record)
968 except (Exception, KeyboardInterrupt) as exc:
969 endpoint_error = exc
970 self._record_failure(record, str(record.get("phase", "endpoint")), exc)
971 finally:
972 try:
973 self.cleanup_endpoint(plan, record)
974 except (Exception, KeyboardInterrupt) as exc:
975 cleanup_errors.append((plan.ordinal, exc))
976 self._record_failure(record, "endpoint-finally-cleanup", exc)
977 if endpoint_error is None:
978 endpoint_error = ManagedInferenceValidationError(
979 "managed inference endpoint cleanup failed"
980 )
981 if endpoint_error is not None:
982 primary_error = endpoint_error
983 break
984 record["validation_complete"] = True
985 record["phase"] = "complete"
986 record["absence_proven"] = True
987 self._persist()
988 validated += 1 if ran else 0
989 finally:
990 self.state["phase"] = "final-cleanup"
991 self._persist()
992 for plan, record in zip(self.plans, self.records, strict=True):
993 try:
994 self.cleanup_endpoint(plan, record)
995 except (Exception, KeyboardInterrupt) as exc:
996 cleanup_errors.append((plan.ordinal, exc))
997 self._record_failure(record, "aggregate-finally-cleanup", exc)
999 if primary_error is not None or cleanup_errors:
1000 self.state["phase"] = "failed"
1001 self.state["cleanup_failure_count"] = len(cleanup_errors)
1002 self._persist()
1003 if primary_error is not None and not isinstance(primary_error, Exception):
1004 raise primary_error
1005 raise ManagedInferenceValidationError(
1006 "managed inference validation failed after all cleanup attempts; "
1007 f"cleanup failures: {len(cleanup_errors)}; inspect the private checkpoint"
1008 ) from None
1010 self.state["phase"] = "complete"
1011 self._persist()
1012 completed = sum(1 for record in self.records if record.get("validation_complete") is True)
1013 frameworks: dict[str, dict[str, Any]] = {}
1014 model_registry = {"uploaded": 0, "removed": 0}
1015 mooncake_store = {"cold_tier_verified": 0, "reconfigure_rolled": 0}
1016 structured_decisions = {"required": 0, "verified": 0}
1017 for plan, record in zip(self.plans, self.records, strict=True):
1018 framework = frameworks.setdefault(
1019 plan.runtime.framework,
1020 {**dict.fromkeys(plan.runtime.roles, False), "invocations": 0, "model_info": 0},
1021 )
1022 framework[plan.role] = record.get("validation_complete") is True
1023 if isinstance(record.get("invoke_evidence"), dict):
1024 framework["invocations"] += 1
1025 probes = record.get("backend_probe_evidence")
1026 if isinstance(probes, dict) and isinstance(probes.get("model_info"), dict):
1027 framework["model_info"] += 1
1028 repository = record.get("model_repository")
1029 if isinstance(repository, dict):
1030 model_registry["uploaded"] += repository.get("phase") == "uploaded"
1031 model_registry["removed"] += repository.get("cleanup") == "absent"
1032 if plan.structured_decision:
1033 structured_decisions["required"] += 1
1034 decision = record.get("structured_decision")
1035 structured_decisions["verified"] += (
1036 isinstance(decision, dict) and decision.get("phase") == "verified"
1037 )
1038 if plan.mooncake_store:
1039 mooncake_store["cold_tier_verified"] += record.get("cold_tier_verified") is True
1040 reconfigure = record.get("store_reconfigure")
1041 mooncake_store["reconfigure_rolled"] += (
1042 isinstance(reconfigure, dict) and reconfigure.get("phase") == "verified"
1043 )
1044 shared_proxy = self.state.get("shared_proxy_autoscaling")
1045 return {
1046 "endpoint_count": len(self.plans),
1047 "validated_or_resumed": completed,
1048 "newly_validated": validated,
1049 "execution": "strictly-sequential",
1050 "frameworks": frameworks,
1051 "model_registry": model_registry,
1052 "mooncake_store": mooncake_store,
1053 "structured_decisions": structured_decisions,
1054 "shared_proxy_autoscaling_verified": (
1055 isinstance(shared_proxy, dict) and shared_proxy.get("phase") == "verified"
1056 ),
1057 "all_endpoints_absent": all(
1058 record.get("absence_proven") is True for record in self.records
1059 ),
1060 "invocations": {
1061 "required_non_empty_generated_text": True,
1062 "completed": sum(
1063 1 for record in self.records if isinstance(record.get("invoke_evidence"), dict)
1064 ),
1065 },
1066 "hpa": {
1067 "target_kind": "Deployment",
1068 "min_replicas": self.settings.hpa_min_replicas,
1069 "max_replicas": self.settings.hpa_max_replicas,
1070 "stable_monitor_intervals": self.settings.hpa_stability_intervals,
1071 },
1072 }