Coverage for scripts / live_release_validation / actions / inference.py: 100.00%

1"""First-class managed inference validation action.""" 

2 

3from __future__ import annotations 

4 

5import sys 

6from typing import Any 

7 

8from scripts.example_job_validation import kube 

9 

10from ..checks.inference import ( 

11 ManagedInferenceLifecycle, 

12 ManagedInferenceValidationError, 

13 initialize_run_state, 

14) 

15from ..models import RunContext 

16 

17 

18def action_inference(ctx: RunContext) -> dict[str, Any]: 

19 """Run four strictly sequential runtime scenarios and prove stable absence.""" 

20 settings = ctx.settings 

21 if not settings.inference_enabled: 

22 raise ManagedInferenceValidationError( 

23 "inference action requires the main RunSettings inference contract" 

24 ) 

25 plans, state = initialize_run_state(ctx, settings) 

26 if settings.selected_region not in ctx.deployment_regions: 

27 state["session_error"] = "selected region is not in the deployed regional topology" 

28 ctx.persist() 

29 raise ManagedInferenceValidationError( 

30 "inference selected region is not part of this deployment" 

31 ) 

32 

33 kubeconfig_path = settings.kubeconfig_path 

34 if kubeconfig_path.parent != settings.report_dir: 

35 raise ManagedInferenceValidationError("isolated kubeconfig escaped the private report dir") 

36 cluster_name = f"{ctx.config.project_name}-{settings.selected_region}" 

37 

38 try: 

39 with kube.cluster_session( 

40 settings.repo_root, 

41 cluster_name, 

42 settings.selected_region, 

43 kubeconfig_path=kubeconfig_path, 

44 gco_command=(sys.executable, "-m", "cli.main"), 

45 ) as kubectl: 

46 lifecycle = ManagedInferenceLifecycle( 

47 ctx=ctx, 

48 settings=settings, 

49 plans=plans, 

50 state=state, 

51 kubectl=kubectl, 

52 kubeconfig_path=kubeconfig_path, 

53 ) 

54 lifecycle.verify_shared_proxy_autoscaling(state) 

55 return lifecycle.execute() 

56 except ManagedInferenceValidationError: 

57 raise 

58 except (Exception, KeyboardInterrupt) as exc: 

59 state["session_error"] = f"{type(exc).__name__}: {exc}" 

60 ctx.persist() 

61 if not isinstance(exc, Exception): 

62 raise 

63 raise ManagedInferenceValidationError( 

64 "inference cluster session failed; inspect the private checkpoint" 

65 ) from None