Coverage for scripts / live_release_validation / checks / inference.py: 100.00%
541 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Owned, sequential managed-inference endpoint lifecycle and evidence checks."""
3from __future__ import annotations
5import json
6import os
7import secrets
8import subprocess
9import sys
10import time
11from dataclasses import dataclass
12from pathlib import Path
13from typing import Any, Literal, cast
15from scripts.example_job_validation.kube import KubectlRunner
17from ..models import (
18 INFERENCE_OWNER_LABEL,
19 InferenceRuntimeSpec,
20 RunContext,
21 RunSettings,
22)
23from .inference_common import (
24 InferenceCommandFailure as _CommandFailure,
25)
26from .inference_common import (
27 ManagedInferenceValidationError,
28)
29from .inference_inventory import KUBERNETES_INVENTORY_KINDS, InferenceInventoryMixin
30from .inference_runtime import InferenceRuntimeMixin
32__all__ = [
33 "KUBERNETES_INVENTORY_KINDS",
34 "EndpointPlan",
35 "ManagedInferenceLifecycle",
36 "ManagedInferenceValidationError",
37 "OWNER_LABEL",
38 "build_delete_command",
39 "build_deploy_command",
40 "build_endpoint_plans",
41 "build_health_command",
42 "build_invoke_command",
43 "build_models_command",
44 "extract_generated_text",
45 "initialize_run_state",
46]
48OWNER_LABEL = INFERENCE_OWNER_LABEL
50EndpointRole = Literal["baseline", "hpa"]
51_STATE_KEY = "inference_validation"
52_MAX_PRIVATE_OUTPUT = 64 * 1024
55@dataclass(frozen=True)
56class EndpointPlan:
57 """One immutable framework/role scenario in the sequential matrix."""
59 ordinal: int
60 role: EndpointRole
61 name: str
62 replicas: int
63 autoscaling: bool
64 runtime: InferenceRuntimeSpec
66 def private_dict(self, owner_nonce: str) -> dict[str, Any]:
67 return {
68 "ordinal": self.ordinal,
69 "framework": self.runtime.framework,
70 "role": self.role,
71 "name": self.name,
72 "replicas": self.replicas,
73 "autoscaling": self.autoscaling,
74 "owner_nonce": owner_nonce,
75 }
78def build_endpoint_plans(
79 settings: RunSettings,
80 owner_nonce: str,
81) -> tuple[EndpointPlan, ...]:
82 """Build vLLM and TGI baseline/HPA plans with peak concurrency of one."""
83 token = owner_nonce[:16]
84 plans: list[EndpointPlan] = []
85 for runtime in settings.inference_runtimes:
86 for role, replicas, autoscaling in (
87 ("baseline", settings.baseline_replicas, False),
88 ("hpa", settings.autoscale_initial_replicas, True),
89 ):
90 plans.append(
91 EndpointPlan(
92 ordinal=len(plans) + 1,
93 role=cast(EndpointRole, role),
94 name=f"gco-mi-{token}-{runtime.framework}-{role}",
95 replicas=replicas,
96 autoscaling=autoscaling,
97 runtime=runtime,
98 )
99 )
100 return tuple(plans)
103def initialize_run_state(
104 ctx: RunContext,
105 settings: RunSettings,
106) -> tuple[tuple[EndpointPlan, ...], dict[str, Any]]:
107 """Persist a random owner nonce and immutable plan before any create check."""
108 raw_state = ctx.checkpoint.state.get(_STATE_KEY)
109 if raw_state is None:
110 owner_nonce = secrets.token_hex(32)
111 plans = build_endpoint_plans(settings, owner_nonce)
112 expected_plan = [plan.private_dict(owner_nonce) for plan in plans]
113 state: dict[str, Any] = {
114 "contract_version": 3,
115 "owner_nonce": owner_nonce,
116 "phase": "planned",
117 "plan": expected_plan,
118 "endpoints": [
119 {
120 **plan.private_dict(owner_nonce),
121 "phase": "planned",
122 "incarnation": 1,
123 "closed_incarnations": [],
124 "cleanup_phase": "not-started",
125 "validation_complete": False,
126 "absence_proven": False,
127 "owned": False,
128 "commands": [],
129 "cleanup_attempts": [],
130 "failures": [],
131 }
132 for plan in plans
133 ],
134 }
135 ctx.checkpoint.state[_STATE_KEY] = state
136 ctx.persist()
137 return plans, state
139 if not isinstance(raw_state, dict):
140 raise ManagedInferenceValidationError("inference checkpoint state is invalid")
141 state = cast(dict[str, Any], raw_state)
142 raw_owner_nonce = state.get("owner_nonce")
143 if (
144 state.get("contract_version") != 3
145 or not isinstance(raw_owner_nonce, str)
146 or len(raw_owner_nonce) != 64
147 or any(character not in "0123456789abcdef" for character in raw_owner_nonce)
148 ):
149 raise ManagedInferenceValidationError("inference checkpoint owner nonce is invalid")
150 owner_nonce = raw_owner_nonce
151 plans = build_endpoint_plans(settings, owner_nonce)
152 expected_plan = [plan.private_dict(owner_nonce) for plan in plans]
153 if state.get("plan") != expected_plan:
154 raise ManagedInferenceValidationError("inference checkpoint plan changed")
155 records = state.get("endpoints")
156 if not isinstance(records, list) or len(records) != len(plans):
157 raise ManagedInferenceValidationError("inference endpoint checkpoint is invalid")
158 for plan, raw_record in zip(plans, records, strict=True):
159 if not isinstance(raw_record, dict):
160 raise ManagedInferenceValidationError("inference endpoint checkpoint is invalid")
161 expected = plan.private_dict(owner_nonce)
162 if any(raw_record.get(key) != value for key, value in expected.items()):
163 raise ManagedInferenceValidationError("inference checkpoint endpoint identity changed")
164 if (
165 not isinstance(raw_record.get("incarnation"), int)
166 or raw_record["incarnation"] < 1
167 or not isinstance(raw_record.get("closed_incarnations"), list)
168 ):
169 raise ManagedInferenceValidationError("inference checkpoint incarnation state changed")
170 return plans, state
173def _checkout_cli_prefix() -> list[str]:
174 """Bind every side effect to the interpreter and checkout preflight attested."""
175 return [sys.executable, "-m", "cli.main"]
178def build_deploy_command(
179 settings: RunSettings,
180 plan: EndpointPlan,
181 owner_nonce: str,
182) -> list[str]:
183 """Build the real argv-only ``gco inference deploy`` command."""
184 runtime = plan.runtime
185 command = [
186 *_checkout_cli_prefix(),
187 "--output",
188 "json",
189 "inference",
190 "deploy",
191 plan.name,
192 "--image",
193 runtime.image,
194 "--framework",
195 runtime.framework,
196 "--region",
197 settings.selected_region,
198 "--replicas",
199 str(plan.replicas),
200 "--gpu-count",
201 str(settings.gpu_count),
202 "--port",
203 str(runtime.port),
204 "--health-path",
205 settings.health_path,
206 "--namespace",
207 settings.namespace,
208 "--label",
209 f"{OWNER_LABEL}={owner_nonce}",
210 "--no-rewrite-image",
211 ]
212 for key, value in sorted(settings.framework_env(runtime).items()):
213 command.extend(("--env", f"{key}={value}"))
214 for value in settings.deploy_extra_args(runtime):
215 if value.startswith("-"):
216 command.append(f"--extra-args={value}")
217 else:
218 command.extend(("--extra-args", value))
219 if plan.autoscaling:
220 command.extend(
221 (
222 "--autoscale-metric",
223 f"cpu:{settings.hpa_cpu_target}",
224 "--min-replicas",
225 str(settings.hpa_min_replicas),
226 "--max-replicas",
227 str(settings.hpa_max_replicas),
228 )
229 )
230 return command
233def build_invoke_command(
234 settings: RunSettings,
235 plan: EndpointPlan,
236) -> list[str]:
237 """Build the deterministic, buffered real CLI invocation."""
238 return [
239 *_checkout_cli_prefix(),
240 "--output",
241 "json",
242 "inference",
243 "invoke",
244 plan.name,
245 "--data",
246 json.dumps(
247 settings.request_body(plan.runtime),
248 sort_keys=True,
249 separators=(",", ":"),
250 ),
251 "--path",
252 plan.runtime.request_path,
253 "--region",
254 settings.selected_region,
255 ]
258def build_health_command(settings: RunSettings, plan: EndpointPlan) -> list[str]:
259 """Build an authenticated health probe through the public CLI path."""
260 return [
261 *_checkout_cli_prefix(),
262 "--output",
263 "json",
264 "inference",
265 "health",
266 plan.name,
267 "--region",
268 settings.selected_region,
269 ]
272def build_models_command(settings: RunSettings, plan: EndpointPlan) -> list[str]:
273 """Build the framework-aware authenticated model-identity probe."""
274 return [
275 *_checkout_cli_prefix(),
276 "--output",
277 "json",
278 "inference",
279 "models",
280 plan.name,
281 "--framework",
282 plan.runtime.framework,
283 "--region",
284 settings.selected_region,
285 ]
288def build_delete_command(
289 settings: RunSettings,
290 plan: EndpointPlan,
291 owner_nonce: str,
292 lifecycle_id: str,
293) -> list[str]:
294 """Build deletion atomically bound to owner nonce and endpoint incarnation."""
295 del settings # Kept for a parallel command-builder interface.
296 return [
297 *_checkout_cli_prefix(),
298 "inference",
299 "delete",
300 plan.name,
301 "--expected-owner-label",
302 f"{OWNER_LABEL}={owner_nonce}",
303 "--expected-lifecycle-id",
304 lifecycle_id,
305 "--yes",
306 ]
309def _last_json_document(output: str) -> Any:
310 decoder = json.JSONDecoder()
311 candidate: Any = None
312 found = False
313 for index, character in enumerate(output):
314 if character not in "[{":
315 continue
316 try:
317 value, end = decoder.raw_decode(output, index)
318 except json.JSONDecodeError:
319 continue
320 if output[end:].strip():
321 continue
322 candidate = value
323 found = True
324 if not found:
325 raise ManagedInferenceValidationError(
326 "managed inference invoke returned no terminal JSON document"
327 )
328 return candidate
331def extract_generated_text(output: str, framework: str) -> str:
332 """Require the exact response schema for the request adapter in use."""
333 payload = _last_json_document(output)
334 text: Any = None
335 if framework == "vllm":
336 if isinstance(payload, dict):
337 choices = payload.get("choices")
338 if isinstance(choices, list) and choices and isinstance(choices[0], dict):
339 text = choices[0].get("text")
340 elif framework == "tgi":
341 if isinstance(payload, dict):
342 text = payload.get("generated_text")
343 else:
344 raise ManagedInferenceValidationError("unknown managed inference response framework")
345 if not isinstance(text, str) or not text.strip():
346 raise ManagedInferenceValidationError(
347 f"managed {framework} inference response did not match its non-empty text schema"
348 )
349 return text.strip()
352class ManagedInferenceLifecycle(InferenceInventoryMixin, InferenceRuntimeMixin):
353 """Run and clean the four-scenario runtime matrix with durable ownership."""
355 def __init__(
356 self,
357 *,
358 ctx: RunContext,
359 settings: RunSettings,
360 plans: tuple[EndpointPlan, ...],
361 state: dict[str, Any],
362 kubectl: KubectlRunner,
363 kubeconfig_path: Path,
364 ) -> None:
365 self.ctx = ctx
366 self.settings = settings
367 self.plans = plans
368 self.state = state
369 self.kubectl = kubectl
370 self.kubeconfig_path = kubeconfig_path
371 owner_nonce = state.get("owner_nonce")
372 if not isinstance(owner_nonce, str) or not owner_nonce:
373 raise ManagedInferenceValidationError("inference owner nonce is missing")
374 self.owner_nonce = owner_nonce
375 records = state.get("endpoints")
376 if not isinstance(records, list) or any(not isinstance(item, dict) for item in records):
377 raise ManagedInferenceValidationError("managed inference endpoint state is invalid")
378 self.records = cast(list[dict[str, Any]], records)
379 self._table: Any | None = None
381 def _persist(self) -> None:
382 self.ctx.persist()
384 def _set_phase(
385 self,
386 record: dict[str, Any],
387 phase: str,
388 **values: Any,
389 ) -> None:
390 record["phase"] = phase
391 record.update(values)
392 self.state["phase"] = f"endpoint-{record['ordinal']}:{phase}"
393 self._persist()
395 def _record_failure(
396 self,
397 record: dict[str, Any],
398 stage: str,
399 exc: BaseException,
400 ) -> None:
401 failures = record.setdefault("failures", [])
402 if not isinstance(failures, list):
403 failures = []
404 record["failures"] = failures
405 failures.append({"stage": stage, "error": f"{type(exc).__name__}: {exc}"})
406 record["last_failed_phase"] = stage
407 self._persist()
409 @property
410 def table(self) -> Any:
411 if self._table is None:
412 try:
413 resource = self.ctx.session.resource(
414 "dynamodb", region_name=self.ctx.config.global_region
415 )
416 self._table = resource.Table(f"{self.ctx.config.project_name}-inference-endpoints")
417 except Exception as exc:
418 self.state["ddb_setup_error"] = f"{type(exc).__name__}: {exc}"
419 self._persist()
420 raise ManagedInferenceValidationError(
421 "managed inference state store could not be opened"
422 ) from None
423 return self._table
425 def _strong_get(self, record: dict[str, Any]) -> dict[str, Any] | None:
426 try:
427 response = self.table.get_item(
428 Key={"endpoint_name": record["name"]},
429 ConsistentRead=True,
430 )
431 except Exception as exc:
432 self._record_failure(record, "ddb-strong-read", exc)
433 raise ManagedInferenceValidationError(
434 "managed inference strong state read failed; inspect the private checkpoint"
435 ) from None
436 item = response.get("Item") if isinstance(response, dict) else None
437 if item is None:
438 return None
439 if not isinstance(item, dict):
440 raise ManagedInferenceValidationError("managed inference state record is malformed")
441 return cast(dict[str, Any], item)
443 def _is_owned(self, item: dict[str, Any]) -> bool:
444 labels = item.get("labels")
445 return isinstance(labels, dict) and labels.get(OWNER_LABEL) == self.owner_nonce
447 def _verify_item_contract(
448 self,
449 plan: EndpointPlan,
450 item: dict[str, Any],
451 record: dict[str, Any],
452 ) -> None:
453 lifecycle_id = item.get("lifecycle_id")
454 if not isinstance(lifecycle_id, str) or not lifecycle_id:
455 raise ManagedInferenceValidationError(
456 "inference stored endpoint has no immutable lifecycle identity"
457 )
458 closed = record.get("closed_incarnations")
459 closed_ids = (
460 {
461 entry.get("lifecycle_id")
462 for entry in closed
463 if isinstance(entry, dict) and isinstance(entry.get("lifecycle_id"), str)
464 }
465 if isinstance(closed, list)
466 else set()
467 )
468 if lifecycle_id in closed_ids:
469 raise ManagedInferenceValidationError(
470 "inference endpoint reverted to a closed lifecycle incarnation"
471 )
472 observed_lifecycle = record.get("lifecycle_id")
473 if observed_lifecycle is None:
474 record["lifecycle_id"] = lifecycle_id
475 self._persist()
476 elif observed_lifecycle != lifecycle_id:
477 raise ManagedInferenceValidationError(
478 "inference endpoint incarnation changed; refusing replacement ownership"
479 )
480 spec = item.get("spec")
481 if not isinstance(spec, dict):
482 raise ManagedInferenceValidationError("managed inference stored spec is malformed")
483 runtime = plan.runtime
484 expected_base: dict[str, Any] = {
485 "image": runtime.image,
486 "framework": runtime.framework,
487 "port": runtime.port,
488 "replicas": plan.replicas,
489 "gpu_count": self.settings.gpu_count,
490 "health_check_path": self.settings.health_path,
491 "env": self.settings.framework_env(runtime),
492 }
493 extra_args = self.settings.deploy_extra_args(runtime)
494 if extra_args:
495 expected_base["args"] = list(extra_args)
496 for key, value in expected_base.items():
497 if spec.get(key) != value:
498 raise ManagedInferenceValidationError(
499 "managed inference stored endpoint contract does not match this run"
500 )
501 if item.get("target_regions") != [self.settings.selected_region]:
502 raise ManagedInferenceValidationError(
503 "managed inference stored target region does not match this run"
504 )
505 if item.get("namespace") != self.settings.namespace:
506 raise ManagedInferenceValidationError(
507 "managed inference stored namespace does not match this run"
508 )
509 autoscaling = spec.get("autoscaling")
510 if plan.autoscaling:
511 expected_autoscaling = {
512 "enabled": True,
513 "min_replicas": self.settings.hpa_min_replicas,
514 "max_replicas": self.settings.hpa_max_replicas,
515 "metrics": [{"type": "cpu", "target": self.settings.hpa_cpu_target}],
516 }
517 if autoscaling != expected_autoscaling:
518 raise ManagedInferenceValidationError(
519 "managed inference stored HPA contract does not match this run"
520 )
521 elif autoscaling is not None:
522 raise ManagedInferenceValidationError(
523 "managed inference baseline unexpectedly has autoscaling configured"
524 )
526 @staticmethod
527 def _truncated(value: str) -> str:
528 return value[-_MAX_PRIVATE_OUTPUT:]
530 def _set_invoke_journal_outcome(
531 self,
532 record: dict[str, Any],
533 status: str,
534 **values: Any,
535 ) -> None:
536 """Update the durable non-replay journal in the same checkpoint write."""
537 journal = record.get("invoke_journal")
538 if not isinstance(journal, dict):
539 raise ManagedInferenceValidationError("managed inference invoke journal is invalid")
540 journal["status"] = status
541 journal.update(values)
543 def _run_command(
544 self,
545 record: dict[str, Any],
546 stage: str,
547 command: list[str],
548 *,
549 deadline: float | None = None,
550 ) -> str:
551 environment = dict(os.environ)
552 environment["KUBECONFIG"] = str(self.kubeconfig_path)
553 command_timeout = float(self.settings.command_timeout_seconds)
554 if deadline is not None:
555 remaining = deadline - time.monotonic()
556 if remaining <= 0:
557 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
558 commands.append({"stage": stage, "argv": command, "deadline_exhausted": True})
559 self._persist()
560 raise _CommandFailure(f"{stage} phase deadline exhausted")
561 command_timeout = min(command_timeout, remaining)
562 try:
563 result = subprocess.run(
564 command,
565 cwd=self.settings.repo_root,
566 capture_output=True,
567 text=True,
568 timeout=command_timeout,
569 env=environment,
570 shell=False,
571 )
572 except subprocess.TimeoutExpired as exc:
573 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
574 commands.append(
575 {
576 "stage": stage,
577 "argv": command,
578 "timed_out": True,
579 "timeout_seconds": command_timeout,
580 "stdout": self._truncated(str(exc.stdout or "")),
581 "stderr": self._truncated(str(exc.stderr or "")),
582 }
583 )
584 if stage == "invoke":
585 self._set_invoke_journal_outcome(
586 record,
587 "ambiguous",
588 reason="timeout",
589 stdout=self._truncated(str(exc.stdout or "")),
590 )
591 self._persist()
592 raise _CommandFailure(f"{stage} timed out") from None
593 except (OSError, UnicodeError) as exc:
594 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
595 commands.append(
596 {
597 "stage": stage,
598 "argv": command,
599 "launch_error": f"{type(exc).__name__}: {exc}",
600 }
601 )
602 if stage == "invoke":
603 self._set_invoke_journal_outcome(
604 record,
605 "failed",
606 reason="launch-error",
607 )
608 self._persist()
609 raise _CommandFailure(f"{stage} could not start") from None
611 commands = cast(list[dict[str, Any]], record.setdefault("commands", []))
612 commands.append(
613 {
614 "stage": stage,
615 "argv": command,
616 "returncode": result.returncode,
617 "stdout": self._truncated(result.stdout),
618 "stderr": self._truncated(result.stderr),
619 }
620 )
621 if stage == "invoke":
622 self._set_invoke_journal_outcome(
623 record,
624 "succeeded" if result.returncode == 0 else "failed",
625 returncode=result.returncode,
626 stdout=self._truncated(result.stdout),
627 stderr=self._truncated(result.stderr),
628 )
629 self._persist()
630 if result.returncode != 0:
631 raise _CommandFailure(f"{stage} exited nonzero")
632 return result.stdout
634 def ensure_owned_endpoint(
635 self,
636 plan: EndpointPlan,
637 record: dict[str, Any],
638 ) -> None:
639 """Adopt only this run's marker, or prove collision-free absence then create."""
640 item = self._strong_get(record)
641 if item is not None:
642 if not self._is_owned(item):
643 raise ManagedInferenceValidationError(
644 "managed inference endpoint collision detected; refusing ownership"
645 )
646 self._verify_item_contract(plan, item, record)
647 record["owned"] = True
648 self._set_phase(record, "ownership-confirmed")
649 return
651 inventory = self.kubernetes_inventory(
652 record,
653 deadline=time.monotonic() + self.settings.readiness_timeout_seconds,
654 )
655 if any(inventory.values()):
656 raise ManagedInferenceValidationError(
657 "managed inference Kubernetes name collision detected; refusing creation"
658 )
659 self._set_phase(record, "collision-checked")
660 self._set_phase(record, "deploy-started")
661 try:
662 self._run_command(
663 record,
664 "deploy",
665 build_deploy_command(self.settings, plan, self.owner_nonce),
666 )
667 except _CommandFailure as exc:
668 self._record_failure(record, "deploy", exc)
669 item = self._strong_get(record)
670 if item is None or not self._is_owned(item):
671 raise ManagedInferenceValidationError(
672 "managed inference deploy failed; inspect the private checkpoint"
673 ) from None
674 self._wait_for_owned_record(plan, record)
676 def _wait_for_healthy_backend(
677 self,
678 plan: EndpointPlan,
679 record: dict[str, Any],
680 ) -> dict[str, Any]:
681 attempts = record.setdefault("backend_probe_attempts", [])
682 if not isinstance(attempts, list):
683 raise ManagedInferenceValidationError("managed backend probe history is invalid")
684 deadline = time.monotonic() + self.settings.readiness_timeout_seconds
685 heartbeat_at = time.monotonic()
686 while True:
687 if time.monotonic() >= deadline:
688 raise ManagedInferenceValidationError(
689 "managed health did not converge before timeout"
690 )
691 heartbeat_at = self.keep_cluster_tunnel_alive(record, heartbeat_at, deadline=deadline)
692 item = self._strong_get(record)
693 if item is None or not self._is_owned(item):
694 raise ManagedInferenceValidationError("managed ownership changed during health")
695 self._verify_item_contract(plan, item, record)
696 statuses = item.get("region_status")
697 regional = (
698 statuses.get(self.settings.selected_region) if isinstance(statuses, dict) else None
699 )
700 if (
701 item.get("desired_state") != "running"
702 or not isinstance(regional, dict)
703 or regional.get("state") != "running"
704 ):
705 raise ManagedInferenceValidationError(
706 "managed endpoint stopped running during health"
707 )
708 attempt: dict[str, Any] = {
709 "attempt": len(attempts) + 1,
710 "started_at_monotonic": time.monotonic(),
711 "classification": "started",
712 }
713 attempts.append(attempt)
715 def finish(classification: str, **values: Any) -> None:
716 attempt.update( # noqa: B023 - helper is called synchronously in this iteration
717 ended_at_monotonic=time.monotonic(), classification=classification, **values
718 )
719 self._persist()
721 self._persist()
722 try:
723 output = self._run_command(
724 record,
725 "health",
726 build_health_command(self.settings, plan),
727 deadline=deadline,
728 )
729 except _CommandFailure as exc:
730 finish("command-failed", error=str(exc))
731 raise
732 try:
733 health = _last_json_document(output)
734 except ManagedInferenceValidationError as exc:
735 finish("malformed-output", error=str(exc))
736 raise
737 status = health.get("status") if isinstance(health, dict) else None
738 http_status = health.get("http_status") if isinstance(health, dict) else None
739 if (
740 not isinstance(status, str)
741 or isinstance(http_status, bool)
742 or not isinstance(http_status, int)
743 ):
744 finish("malformed-contract")
745 raise ManagedInferenceValidationError(
746 "managed inference health probe returned a malformed contract"
747 )
748 attempt.update(
749 {
750 "status": status,
751 "http_status": http_status,
752 "path": health.get("path"),
753 "latency_ms": health.get("latency_ms"),
754 "body_summary": self._truncated(
755 json.dumps(health.get("body"), sort_keys=True, default=str)
756 ),
757 }
758 )
759 if status == "healthy" and 200 <= http_status < 300:
760 finish("healthy")
761 return cast(dict[str, Any], health)
762 retryable = status == "unhealthy" and (
763 http_status in {404, 429} or 500 <= http_status < 600
764 )
765 if not retryable:
766 finish("terminal-contract")
767 raise ManagedInferenceValidationError(
768 "managed inference health probe returned a terminal status or contract"
769 )
770 remaining = deadline - time.monotonic()
771 if remaining <= 0:
772 finish("deadline-exhausted")
773 raise ManagedInferenceValidationError(
774 "managed inference health endpoint did not converge before timeout"
775 )
776 finish("retryable-unhealthy")
777 time.sleep(min(float(self.settings.poll_interval_seconds), remaining))
779 def verify_backend_probes(
780 self,
781 plan: EndpointPlan,
782 record: dict[str, Any],
783 ) -> None:
784 """Require converged authenticated health and exact model discovery."""
785 try:
786 health = self._wait_for_healthy_backend(plan, record)
787 evidence: dict[str, Any] = {
788 "health": {
789 "healthy": True,
790 "http_status": health["http_status"],
791 "path": self.settings.health_path,
792 }
793 }
795 runtime = plan.runtime
796 model_output = self._run_command(
797 record,
798 "model-info",
799 build_models_command(self.settings, plan),
800 )
801 model_info = _last_json_document(model_output)
802 if runtime.framework == "vllm":
803 data = model_info.get("data") if isinstance(model_info, dict) else None
804 model_ids = (
805 [
806 item.get("id")
807 for item in data
808 if isinstance(item, dict) and isinstance(item.get("id"), str)
809 ]
810 if isinstance(data, list)
811 else []
812 )
813 if runtime.model_id not in model_ids:
814 raise ManagedInferenceValidationError(
815 "managed vLLM model inventory omitted the configured model"
816 )
817 evidence["model_info"] = {
818 "path": runtime.model_info_path,
819 "configured_model_present": True,
820 "model_revision_pinned": True,
821 "model_count": len(model_ids),
822 }
823 else:
824 if (
825 not isinstance(model_info, dict)
826 or model_info.get("model_id") != runtime.model_id
827 or model_info.get("model_sha") != runtime.model_revision
828 ):
829 raise ManagedInferenceValidationError(
830 "managed TGI /info did not report the exact model id and revision"
831 )
832 evidence["model_info"] = {
833 "path": runtime.model_info_path,
834 "configured_model_present": True,
835 "configured_revision_present": True,
836 }
837 except (ManagedInferenceValidationError, _CommandFailure) as exc:
838 self._record_failure(record, "backend-probes", exc)
839 raise ManagedInferenceValidationError(
840 "managed inference health/model probe failed its contract"
841 ) from None
843 record["backend_probe_evidence"] = evidence
844 self._set_phase(record, "backend-probes-verified")
846 def invoke(self, plan: EndpointPlan, record: dict[str, Any]) -> None:
847 """Invoke once, or recover a durable successful outcome without replay."""
848 command = build_invoke_command(self.settings, plan)
849 journal = record.get("invoke_journal")
850 if journal is None:
851 journal = {
852 "status": "started",
853 "framework": plan.runtime.framework,
854 "request_path": plan.runtime.request_path,
855 "argv": command,
856 }
857 record["invoke_journal"] = journal
858 self._persist()
859 output: str | None = None
860 elif not isinstance(journal, dict):
861 raise ManagedInferenceValidationError("managed inference invoke journal is invalid")
862 else:
863 if (
864 journal.get("framework") != plan.runtime.framework
865 or journal.get("request_path") != plan.runtime.request_path
866 or journal.get("argv") != command
867 ):
868 raise ManagedInferenceValidationError(
869 "managed inference invoke journal identity changed"
870 )
871 status = journal.get("status")
872 if status == "succeeded" and isinstance(journal.get("stdout"), str):
873 output = journal["stdout"]
874 elif status in {"started", "ambiguous", "failed"}:
875 raise ManagedInferenceValidationError(
876 "managed inference invocation has a non-replayable persisted outcome"
877 )
878 else:
879 raise ManagedInferenceValidationError(
880 "managed inference invoke journal status is invalid"
881 )
882 try:
883 if output is None:
884 output = self._run_command(record, "invoke", command)
885 generated_text = extract_generated_text(output, plan.runtime.framework)
886 except (ManagedInferenceValidationError, _CommandFailure) as exc:
887 self._record_failure(record, "invoke", exc)
888 raise ManagedInferenceValidationError(
889 "managed inference invocation failed its response contract"
890 ) from None
891 record["invoke_evidence"] = {
892 "framework": plan.runtime.framework,
893 "generated_text_non_empty": True,
894 "generated_text_length": len(generated_text),
895 "replayed": False,
896 }
897 self._set_phase(record, "invoked")
899 def _prepare_incarnation_for_resume(
900 self,
901 plan: EndpointPlan,
902 record: dict[str, Any],
903 ) -> None:
904 """Rotate a fully cleaned pre-invocation lifecycle before redeploying."""
905 lifecycle_id = record.get("lifecycle_id")
906 if (
907 record.get("cleanup_phase") != "absent"
908 or not isinstance(lifecycle_id, str)
909 or not lifecycle_id
910 or isinstance(record.get("invoke_evidence"), dict)
911 or isinstance(record.get("invoke_journal"), dict)
912 ):
913 return
914 # Re-prove strong DDB and Kubernetes absence before closing the old incarnation.
915 evidence = self.prove_absence(record)
916 if evidence.get("stable_absence_observations") != 2:
917 raise ManagedInferenceValidationError(
918 "cleaned inference incarnation did not retain stable absence"
919 )
920 closed = record.setdefault("closed_incarnations", [])
921 if not isinstance(closed, list):
922 raise ManagedInferenceValidationError("inference closed-incarnation state is invalid")
923 closed.append(
924 {
925 "number": record.get("incarnation", 1),
926 "lifecycle_id": lifecycle_id,
927 "invoked": False,
928 "cleanup_phase": "absent",
929 "absence_evidence": evidence,
930 "commands": record.get("commands", []),
931 "backend_probe_attempts": record.get("backend_probe_attempts", []),
932 "tunnel_heartbeats": record.get("tunnel_heartbeats", []),
933 "cleanup_attempts": record.get("cleanup_attempts", []),
934 "failures": record.get("failures", []),
935 }
936 )
937 record.update(
938 {
939 "incarnation": int(record.get("incarnation", 1)) + 1,
940 "lifecycle_id": None,
941 "owned": False,
942 "phase": "planned-resume-incarnation",
943 "cleanup_phase": "not-started",
944 "validation_complete": False,
945 "validation_steps_complete": False,
946 "absence_proven": True,
947 "commands": [],
948 "cleanup_attempts": [],
949 "failures": [],
950 }
951 )
952 for key in (
953 "backend_probe_evidence",
954 "backend_probe_attempts",
955 "tunnel_heartbeats",
956 "invoke_evidence",
957 "invoke_journal",
958 "hpa_stability_observations",
959 "last_hpa_replica_observation",
960 "last_readiness",
961 "last_failed_phase",
962 ):
963 record.pop(key, None)
964 self.state["phase"] = f"endpoint-{plan.ordinal}:incarnation-rotated"
965 self._persist()
967 def run_endpoint(self, plan: EndpointPlan, record: dict[str, Any]) -> bool:
968 """Run one endpoint's validation phases, or safely continue a durable resume."""
969 if record.get("validation_complete") is True:
970 self.prove_absence(record)
971 return False
972 invocation_finished = (
973 record.get("phase") == "invoked"
974 and isinstance(record.get("invoke_evidence"), dict)
975 and record["invoke_evidence"].get("generated_text_non_empty") is True
976 )
977 if record.get("validation_steps_complete") is True or invocation_finished:
978 # A crash after invoke or cleanup must never recreate or reinvoke;
979 # the caller's finally only needs to finish/prove cleanup.
980 record["validation_steps_complete"] = True
981 record["phase"] = "validation-complete-resume"
982 self._persist()
983 return False
984 journal = record.get("invoke_journal")
985 if isinstance(journal, dict) and not isinstance(record.get("invoke_evidence"), dict):
986 # Invocation intent is one-way: recover durable success without a request;
987 # any other persisted state fails closed for cleanup.
988 self.invoke(plan, record)
989 record["validation_steps_complete"] = True
990 self._persist()
991 return False
992 self._prepare_incarnation_for_resume(plan, record)
993 record["absence_proven"] = False
994 self._set_phase(record, "starting")
995 self.ensure_owned_endpoint(plan, record)
996 self.wait_for_ddb_running(plan, record)
997 self.wait_for_kubernetes_ready(plan, record)
998 if plan.autoscaling:
999 self.verify_hpa_stability(plan, record)
1000 self.verify_backend_probes(plan, record)
1001 self.invoke(plan, record)
1002 record["validation_steps_complete"] = True
1003 self._persist()
1004 return True
1006 def cleanup_endpoint(self, plan: EndpointPlan, record: dict[str, Any]) -> dict[str, Any]:
1007 """Delete only a matching marker, then prove strong DDB and full K8s absence."""
1008 attempts = record.setdefault("cleanup_attempts", [])
1009 if not isinstance(attempts, list):
1010 attempts = []
1011 record["cleanup_attempts"] = attempts
1012 attempt: dict[str, Any] = {"started_at_monotonic": time.monotonic()}
1013 attempts.append(attempt)
1014 record["cleanup_phase"] = "checking-ownership"
1015 self._persist()
1017 delete_error: Exception | None = None
1018 item = self._strong_get(record)
1019 if item is not None:
1020 if not self._is_owned(item):
1021 attempt["refused_collision"] = True
1022 self._persist()
1023 raise ManagedInferenceValidationError(
1024 "managed inference cleanup refused a colliding endpoint"
1025 )
1026 self._verify_item_contract(plan, item, record)
1027 record["owned"] = True
1028 if item.get("desired_state") != "deleted":
1029 record["cleanup_phase"] = "delete-requested"
1030 self._persist()
1031 lifecycle_id = record.get("lifecycle_id")
1032 if not isinstance(lifecycle_id, str) or not lifecycle_id:
1033 raise ManagedInferenceValidationError(
1034 "inference cleanup has no checkpointed lifecycle identity"
1035 )
1036 try:
1037 self._run_command(
1038 record,
1039 "delete",
1040 build_delete_command(
1041 self.settings,
1042 plan,
1043 self.owner_nonce,
1044 lifecycle_id,
1045 ),
1046 )
1047 except _CommandFailure as exc:
1048 delete_error = exc
1049 self._record_failure(record, "delete", exc)
1050 replacement = self._strong_get(record)
1051 if replacement is not None and (
1052 not self._is_owned(replacement)
1053 or replacement.get("lifecycle_id") != record.get("lifecycle_id")
1054 ):
1055 attempt["refused_replacement_race"] = True
1056 self._persist()
1057 raise ManagedInferenceValidationError(
1058 "managed inference cleanup refused a replacement endpoint"
1059 ) from None
1061 try:
1062 evidence = self.prove_absence(record)
1063 except (Exception, KeyboardInterrupt) as exc:
1064 attempt["completed"] = False
1065 attempt["error"] = f"{type(exc).__name__}: {exc}"
1066 if delete_error is not None:
1067 attempt["delete_error"] = f"{type(delete_error).__name__}: {delete_error}"
1068 self._persist()
1069 raise
1070 attempt["completed"] = True
1071 attempt["delete_recovered"] = delete_error is not None
1072 record["cleanup_phase"] = "absent"
1073 self._persist()
1074 return evidence
1076 def execute(self) -> dict[str, Any]:
1077 """Run strictly sequential endpoints and aggregate only after all cleanup attempts."""
1078 primary_error: Exception | KeyboardInterrupt | None = None
1079 cleanup_errors: list[tuple[int, Exception | KeyboardInterrupt]] = []
1080 validated = 0
1081 self.state["phase"] = "running"
1082 self._persist()
1084 try:
1085 for plan, record in zip(self.plans, self.records, strict=True):
1086 endpoint_error: Exception | KeyboardInterrupt | None = None
1087 ran = False
1088 try:
1089 ran = self.run_endpoint(plan, record)
1090 except (Exception, KeyboardInterrupt) as exc:
1091 endpoint_error = exc
1092 self._record_failure(record, str(record.get("phase", "endpoint")), exc)
1093 finally:
1094 try:
1095 self.cleanup_endpoint(plan, record)
1096 except (Exception, KeyboardInterrupt) as exc:
1097 cleanup_errors.append((plan.ordinal, exc))
1098 self._record_failure(record, "endpoint-finally-cleanup", exc)
1099 if endpoint_error is None:
1100 endpoint_error = ManagedInferenceValidationError(
1101 "managed inference endpoint cleanup failed"
1102 )
1103 if endpoint_error is not None:
1104 primary_error = endpoint_error
1105 break
1106 record["validation_complete"] = True
1107 record["phase"] = "complete"
1108 record["absence_proven"] = True
1109 self._persist()
1110 validated += 1 if ran else 0
1111 finally:
1112 self.state["phase"] = "final-cleanup"
1113 self._persist()
1114 for plan, record in zip(self.plans, self.records, strict=True):
1115 try:
1116 self.cleanup_endpoint(plan, record)
1117 except (Exception, KeyboardInterrupt) as exc:
1118 cleanup_errors.append((plan.ordinal, exc))
1119 self._record_failure(record, "aggregate-finally-cleanup", exc)
1121 if primary_error is not None or cleanup_errors:
1122 self.state["phase"] = "failed"
1123 self.state["cleanup_failure_count"] = len(cleanup_errors)
1124 self._persist()
1125 if primary_error is not None and not isinstance(primary_error, Exception):
1126 raise primary_error
1127 raise ManagedInferenceValidationError(
1128 "managed inference validation failed after all cleanup attempts; "
1129 f"cleanup failures: {len(cleanup_errors)}; inspect the private checkpoint"
1130 ) from None
1132 self.state["phase"] = "complete"
1133 self._persist()
1134 completed = sum(1 for record in self.records if record.get("validation_complete") is True)
1135 frameworks: dict[str, dict[str, Any]] = {}
1136 for plan, record in zip(self.plans, self.records, strict=True):
1137 framework = frameworks.setdefault(
1138 plan.runtime.framework,
1139 {"baseline": False, "hpa": False, "invocations": 0, "model_info": 0},
1140 )
1141 framework[plan.role] = record.get("validation_complete") is True
1142 if isinstance(record.get("invoke_evidence"), dict):
1143 framework["invocations"] += 1
1144 probes = record.get("backend_probe_evidence")
1145 if isinstance(probes, dict) and isinstance(probes.get("model_info"), dict):
1146 framework["model_info"] += 1
1147 shared_proxy = self.state.get("shared_proxy_autoscaling")
1148 return {
1149 "endpoint_count": len(self.plans),
1150 "validated_or_resumed": completed,
1151 "newly_validated": validated,
1152 "execution": "strictly-sequential",
1153 "frameworks": frameworks,
1154 "shared_proxy_autoscaling_verified": (
1155 isinstance(shared_proxy, dict) and shared_proxy.get("phase") == "verified"
1156 ),
1157 "all_endpoints_absent": all(
1158 record.get("absence_proven") is True for record in self.records
1159 ),
1160 "invocations": {
1161 "required_non_empty_generated_text": True,
1162 "completed": sum(
1163 1 for record in self.records if isinstance(record.get("invoke_evidence"), dict)
1164 ),
1165 },
1166 "hpa": {
1167 "target_kind": "Deployment",
1168 "min_replicas": self.settings.hpa_min_replicas,
1169 "max_replicas": self.settings.hpa_max_replicas,
1170 "stable_monitor_intervals": self.settings.hpa_stability_intervals,
1171 },
1172 }