"""One qualified native GRPO training arm in a managed Determined trial. Selection and held-out assessment remain in the original experiment driver. Core calls are isolated from training RNG; native seals stay authoritative. """ from __future__ import annotations import argparse from contextlib import contextmanager import copy import fcntl import hashlib import importlib.metadata import json import math import os from pathlib import Path import random import signal from types import SimpleNamespace from typing import Any import uuid import time from .managed_trial import ManagedTrial, _digest, _read_json, _write_json, CANONICAL_ROOT from . import budget_lineage @contextmanager def preserve_rng(): """Management retries and storage code may not advance policy RNG.""" python_state = random.getstate() import numpy as np import torch numpy_state = copy.deepcopy(np.random.get_state()) cpu_state = torch.get_rng_state().clone() cuda_states = [state.clone() for state in torch.cuda.get_rng_state_all()] if torch.cuda.is_available() else None try: yield finally: random.setstate(python_state) np.random.set_state(numpy_state) torch.set_rng_state(cpu_state) if cuda_states is not None: torch.cuda.set_rng_state_all(cuda_states) @contextmanager def _wall_guard(seconds: float): """Bound this allocation by both the original deadline and remaining GPU budget.""" if not math.isfinite(seconds) or seconds <= 0: raise TimeoutError("No frozen managed GPU budget remains") previous_handler = signal.getsignal(signal.SIGALRM) previous_timer = signal.getitimer(signal.ITIMER_REAL) if previous_timer[0]: raise RuntimeError("Managed training cannot replace another active wall timer") def expired(signum, frame): raise TimeoutError("Managed trial reached its frozen wall-clock bound") signal.signal(signal.SIGALRM, expired) signal.setitimer(signal.ITIMER_REAL, seconds) try: yield finally: signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, previous_handler) def _require_gpu(info) -> None: import torch if (info is None or info.task_type != "TRIAL" or len(info.slot_ids) != 1 or len(info.container_addrs) != 1 or not torch.cuda.is_available() or torch.cuda.device_count() != 1 or not torch.cuda.is_bf16_supported()): raise ValueError("Managed training requires one allocated BF16 CUDA GPU") def _load_backend(name: str, plan_path: Path, service_python: Path): if name == "nemo": from . import gym_experiment as driver from . import gym_training_long as engine plan = driver.verify_plan(plan_path) if plan.get("backend") is not None: raise ValueError("NeMo entrypoint received another backend's plan") elif name == "openenv": from . import openenv_experiment plan = openenv_experiment.verify_plan(plan_path) if str(service_python) != plan["sdk"]["python"]: raise ValueError("OpenEnv service Python must match the qualified native SDK") from .openenv_client import sdk_identity if sdk_identity(service_python) != plan["sdk"]["runtime"]: raise ValueError("OpenEnv SDK changed after qualification") driver = openenv_experiment._engine(service_python) engine = driver._openenv_training else: raise ValueError("Backend must be nemo or openenv") return SimpleNamespace(name=name, plan=plan, driver=driver, engine=engine, service_python=service_python) def _identity(bundle, plan_path: Path, config: dict) -> dict: return {"backend": bundle.name, "core_api": _core_identity(), "experiment_plan_sha256": _digest(plan_path), "training_config_sha256": _digest(Path(bundle.plan["training_config"])), "source_sha256": bundle.plan["source_sha256"], "managed_sources": {"managed_training.py": _digest(Path(__file__)), "managed_trial.py": _digest(Path(__file__).with_name("managed_trial.py"))}, "service_python": str(bundle.service_python), "service_python_sha256": _digest(bundle.service_python.resolve()), "canonical_output": config["output_dir"]} def _core_identity() -> dict: import inspect from determined import core version = importlib.metadata.version("determined") if version != "0.35.0": raise ValueError("Managed trials require pinned determined==0.35.0") objects = [core.init, core.Context, core.CheckpointContext, core.TrainContext, core.SearcherContext, core.PreemptContext] return {"version": version, "sources": {value.__name__: _digest(Path(inspect.getfile(value))) for value in objects}} def _qualification_bundle(bundle): """Change only the output path of the verified native continuous-four config.""" root = Path(bundle.plan["root"]) original = _read_json(root / "resume-qualification/continuous.json") if original["training"]["max_steps"] != 4 or original["seed"] != 43: raise ValueError("Managed qualification requires the native four-step protocol") destination = root / "managed-qualification" config_path = destination / "training.json" config = copy.deepcopy(original) config["output_dir"] = str(destination / "training") if config_path.exists(): if _read_json(config_path) != config: raise ValueError("Existing derived managed qualification configuration changed") else: _write_json(config_path, config) derived = SimpleNamespace(**vars(bundle)) derived.plan = copy.deepcopy(bundle.plan) derived.plan.update(training_config=str(config_path), training_config_sha256=_digest(config_path), checkpoint_interval=2) return derived, config def _restore_receipt_path(root, trial_id, allocation_id): name = "restore-" + hashlib.sha256(allocation_id.encode()).hexdigest() + ".json" return root / "managed-trials" / str(trial_id) / name def _record_qualification_restore(managed, info, root): """Keep the selected scheduler envelope even after server checkpoint GC.""" if not info.latest_checkpoint: return with preserve_rng(): with managed.context.checkpoint.restore_path(info.latest_checkpoint) as snapshot: envelope = _read_json(Path(snapshot) / "snapshot.json") if (envelope["scheduler_uuid"] != info.latest_checkpoint or envelope["binding"] != managed.binding or envelope["steps_completed"] != managed.last_step): raise ValueError("Selected scheduler restore envelope changed") receipt = {"format": "hessian-managed-gpu-restore-v1", "experiment_id": info.trial.experiment_id, "trial_id": info.trial.trial_id, "allocation_id": info.allocation_id, "scheduler_uuid": info.latest_checkpoint, "optimizer_step": managed.last_step, "snapshot": envelope} path = _restore_receipt_path(root, info.trial.trial_id, info.allocation_id) if path.exists() and _read_json(path) != receipt: raise ValueError("Existing allocation restore receipt changed") _write_json(path, receipt) def _qualification_proof(bundle, original_plan_path, native_proof, *, managed=None, info=None, restored_step=None): """Validate or create proof of a real same-trial scheduler restoration.""" root = Path(bundle.plan["root"]) proof_path = root / "managed-qualification/proof.json" derived, config = _qualification_bundle(bundle) output = Path(config["output_dir"]) checkpoint = output / "checkpoint-4" identity = _identity(derived, original_plan_path, config) expected = {"format": "hessian-managed-gpu-resume-v1", "status": "passed", "backend": bundle.name, "experiment_plan_sha256": _digest(original_plan_path), "native_resume_proof_sha256": _digest(root / "resume-qualification/proof.json"), "source_sha256": bundle.plan["source_sha256"], "managed_sources": identity["managed_sources"], "core_api": identity["core_api"], "derived_config_sha256": _digest(Path(derived.plan["training_config"])), "model": config["model"], "runtime_binding": native_proof["runtime_binding"], "gpu": native_proof["gpu"]} if proof_path.exists(): proof = _read_json(proof_path) if any(proof.get(key) != value for key, value in expected.items()): raise ValueError("Managed GPU qualification source/config/runtime proof changed") else: if managed is None or info is None or not info.latest_checkpoint or restored_step not in (2, 4): raise ValueError("Managed GPU qualification requires actual scheduler restore at step two") selected = _read_json(_restore_receipt_path(root, info.trial.trial_id, info.allocation_id)) # A crash after terminal publication can finish qualification using the # publishing allocation's durable pre-training restore receipt. allocation = (info.allocation_id if restored_step == 2 else selected["snapshot"]["publishing_allocation_id"]) restore_path = _restore_receipt_path(root, info.trial.trial_id, allocation) restored = _read_json(restore_path) envelope = restored["snapshot"] previous_allocation = envelope["publishing_allocation_id"] previous_path = root / "managed-trials" / str(info.trial.trial_id) / ( hashlib.sha256(previous_allocation.encode()).hexdigest() + ".json") proof = expected | {"experiment_id": info.trial.experiment_id, "trial_id": info.trial.trial_id, "previous_allocation_id": previous_allocation, "allocation_id": allocation, "restored_from_uuid": restored["scheduler_uuid"], "restore_receipt": str(restore_path), "restore_receipt_sha256": _digest(restore_path), "preempt_receipt": str(previous_path), "preempt_receipt_sha256": _digest(previous_path)} trial_root = root / "managed-trials" / str(proof["trial_id"]) restore_path, receipt_path = Path(proof["restore_receipt"]), Path(proof["preempt_receipt"]) if (restore_path != _restore_receipt_path(root, proof["trial_id"], proof["allocation_id"]) or receipt_path != trial_root / (hashlib.sha256(proof["previous_allocation_id"].encode()).hexdigest() + ".json") or _digest(restore_path) != proof["restore_receipt_sha256"] or _digest(receipt_path) != proof["preempt_receipt_sha256"] or proof["previous_allocation_id"] == proof["allocation_id"] or not proof["restored_from_uuid"]): raise ValueError("Managed GPU restoration/preemption evidence changed") restored, previous = _read_json(restore_path), _read_json(receipt_path) envelope = restored["snapshot"] binding = _read_json(output / "managed-binding.json") if (restored.get("format") != "hessian-managed-gpu-restore-v1" or restored.get("trial_id") != proof["trial_id"] or restored.get("experiment_id") != proof["experiment_id"] or restored.get("allocation_id") != proof["allocation_id"] or restored.get("optimizer_step") != 2 or restored.get("scheduler_uuid") != proof["restored_from_uuid"] or envelope.get("scheduler_uuid") != proof["restored_from_uuid"] or envelope.get("steps_completed") != 2 or envelope.get("binding") != binding or envelope.get("publishing_allocation_id") != proof["previous_allocation_id"] or previous.get("status") != "preempted_at_checkpoint" or previous.get("trial_id") != proof["trial_id"] or previous.get("experiment_id") != proof["experiment_id"] or previous.get("allocation_id") != proof["previous_allocation_id"] or previous.get("optimizer_step") != 2 or previous.get("scheduler_uuid") != proof["restored_from_uuid"]): raise ValueError("Managed GPU qualification lacks a durable same-trial step-two preemption and restore") native = bundle.engine.validate_checkpoint(output, checkpoint, expected_step=4) if (binding["trial_id"] != proof["trial_id"] or binding["experiment_id"] != proof["experiment_id"] or binding["identity"] != identity): raise ValueError("Managed GPU checkpoint belongs to a different qualified trial") if _read_json(output / "runtime-binding.json") != native_proof["runtime_binding"]: raise ValueError("Managed GPU checkpoint runtime differs from native qualification") manifest = _read_json(output / "training-manifest.json") if manifest.get("status") != "completed" or manifest.get("gpu") != native_proof["gpu"]: raise ValueError("Managed GPU qualification did not finalize native training") with preserve_rng(): bundle.driver._assert_checkpoint_states_equal( [root / "resume-qualification/continuous/checkpoint-4", checkpoint], "Managed scheduler GPU resume") evidence = _digest(output / "rollout-evidence.jsonl") if evidence != _digest(root / "resume-qualification/continuous/rollout-evidence.jsonl"): raise RuntimeError("Managed scheduler resume changed sampled token/tool/reward evidence") fields = {"adapter_optimizer_scheduler_rng_equal": True, "exact_rollout_evidence_equal": True, "evidence_sha256": evidence, "managed_checkpoint": native} if proof_path.exists(): if any(proof.get(key) != value for key, value in fields.items()): raise ValueError("Managed GPU checkpoint/evidence proof changed") else: proof.update(fields) _write_json(proof_path, proof) return proof def _qualified_runtime(bundle, proof): """Reject another GPU or upstream runtime before loading policy/optimizer.""" import torch if torch.cuda.get_device_name() != proof["gpu"]: raise ValueError("Allocated GPU differs from the exact native qualification") with preserve_rng(): installed = bundle.engine.require_versions() runtime = bundle.engine.legacy._runtime() if bundle.engine._runtime_identity(runtime, installed) != proof["runtime_binding"]: raise ValueError("Current upstream runtime differs from native qualification") def _remaining(journal, managed: ManagedTrial) -> float: journal.save() managed.check_deadline() remaining = min(budget_lineage.remaining(journal.plan, journal.state, time.time()), managed.binding["deadline"] - managed.clock()) if not math.isfinite(remaining) or remaining <= 0: raise TimeoutError("The original cumulative managed GPU budget is exhausted") return remaining class TrainingReporter: """Keep all management side effects outside the policy RNG stream.""" def __init__(self, managed: ManagedTrial, journal, operation=None): self.managed, self.journal, self.operation = managed, journal, operation @property def binding(self): return self.managed.binding def report(self, step, metrics): with preserve_rng(): _remaining(self.journal, self.managed) self.managed.report(step, metrics, self.operation) def preempt_requested(self, step): with preserve_rng(): _remaining(self.journal, self.managed) return self.managed.preempt_requested(step) def publish(self, output, checkpoint): with preserve_rng(): _remaining(self.journal, self.managed) result = self.managed.publish(output, checkpoint) self.journal.save() return result def _canonical_config(bundle, plan_path: Path) -> tuple[Path, dict]: root = Path(bundle.plan["root"]).resolve() if not root.is_relative_to(CANONICAL_ROOT) or plan_path != root / "experiment-plan.json": raise ValueError("Managed plan must use the canonical JAIF project") config_path = Path(bundle.plan["training_config"]).resolve() config = _read_json(config_path) if _digest(config_path) != bundle.plan["training_config_sha256"]: raise ValueError("Frozen managed training configuration changed") if Path(config["output_dir"]) != root / "training": raise ValueError("Managed training output differs from the frozen experiment") for path in (config["model"]["path"], config["dataset"]["path"]): if not Path(path).resolve().is_relative_to(CANONICAL_ROOT): raise ValueError("Managed model and data must stay in the JAIF project") return root, config def _restore_validator(bundle, managed, config, proof): engine = bundle.engine def validate(staged_root, canonical_output, checkpoint_name): frozen = _read_json(staged_root / "run-plan.json") binding = _read_json(staged_root / "managed-binding.json") expected = {"binding_sha256": engine._json_hash(managed.binding)} if (frozen["output_dir"] != str(canonical_output) or canonical_output != Path(config["output_dir"]) or frozen["config_path"] != bundle.plan["training_config"] or frozen["config_sha256"] != bundle.plan["training_config_sha256"] or _digest(Path(frozen["config_path"])) != frozen["config_sha256"] or engine._sources() != frozen["long_source_sha256"] or frozen.get("managed_training") != expected or binding != managed.binding or frozen["model"] != config["model"] or frozen["dataset"] != config["dataset"] or frozen["training"] != config["training"] or _read_json(staged_root / "runtime-binding.json") != proof["runtime_binding"]): raise ValueError("Managed scheduler snapshot native provenance changed") engine._verify_checkpoint(staged_root / checkpoint_name, staged_root, frozen) return validate def _fresh_output(output: Path, managed: ManagedTrial) -> None: """An unregistered local checkpoint never replaces scheduler checkpoint selection.""" if not output.exists(): return if output.is_symlink() or not output.is_dir(): raise ValueError("Managed output is not a regular run directory") entries = list(output.iterdir()) if not entries: output.rmdir() return path = output / "managed-binding.json" if not path.is_file() or path.is_symlink() or _read_json(path) != managed.binding: raise ValueError("Local partial output belongs to another or unbound trial") history = output.parent / (output.name + ".managed-recovery") if history.is_symlink(): raise ValueError("Managed recovery directory may not be a symlink") history.mkdir(exist_ok=True) os.replace(output, history / ("no-scheduler-checkpoint-" + uuid.uuid4().hex)) def _terminal_manifest(bundle, output: Path, checkpoint: Path, config: dict, proof: dict) -> dict: """Finalize a verified full horizon without claiming another optimizer execution.""" sealed = bundle.engine.validate_checkpoint(output, checkpoint, expected_step=config["training"]["max_steps"]) path = output / "training-manifest.json" report = _read_json(path) if (report.get("status") != "completed" or report.get("gpu") != proof["gpu"] or report.get("model") != config["model"] or report.get("config_sha256") != bundle.plan["training_config_sha256"] or _read_json(output / "runtime-binding.json") != proof["runtime_binding"] or report.get("cleanup_errors")): raise ValueError("Restored terminal checkpoint lacks its qualified training manifest") learning = _read_json(checkpoint / "learning-evidence.json") frozen = _read_json(output / "run-plan.json") groups = bundle.engine.evidence_groups(bundle.engine._read_evidence(output / "rollout-evidence.jsonl"), frozen, sealed["optimizer_step"]) trainer_state = _read_json(checkpoint / "trainer_state.json") if (trainer_state.get("global_step") != sealed["optimizer_step"] or trainer_state.get("max_steps") != config["training"]["max_steps"] or not isinstance(trainer_state.get("log_history"), list) or type(frozen.get("rows")) is not int or frozen["rows"] < 1): raise ValueError("Terminal checkpoint lacks complete native trainer/coverage state") retained = [] entries = [entry for entry in output.glob("checkpoint-*") if entry.name.removeprefix("checkpoint-").isdigit()] for entry in sorted(entries, key=lambda item: int(item.name.removeprefix("checkpoint-"))): if entry.name.removeprefix("checkpoint-").isdigit(): native = bundle.engine.validate_checkpoint(output, entry) retained.append({"path": str(entry), "optimizer_step": native["optimizer_step"], "seal_sha256": _digest(entry / bundle.engine.SEAL), "learning": _read_json(entry / "learning-evidence.json")}) update = learning["from_initial_adapter"] | { key: learning["cumulative"][key] for key in ("mixed_reward_groups", "nonzero_gradient_steps")} report.update(status="completed", optimizer_step=sealed["optimizer_step"], checkpoint=str(checkpoint), checkpoint_sha256=sealed["checkpoint_sha256"], learning_evidence=learning, update_evidence=update, reward_groups=groups, sealed_checkpoints=retained, task_coverage={"training_pool_rows": frozen["rows"], "optimizer_groups": len(groups), "total_rollouts": len(groups) * config["training"]["num_generations"], "consumed_unique_task_ids": sorted({task for group in groups for task in group["task_ids"]})}, evidence_sha256=_digest(output / "rollout-evidence.jsonl"), native_tool_diagnostics_sha256=_digest(output / "native-tool-diagnostics.jsonl"), managed_completion_origin="verified_terminal_scheduler_checkpoint") report.setdefault("train_log_history", trainer_state["log_history"]) bundle.engine.require_learning_update(report) # Existing HF metrics, timing and logs are preserved if present. No duration # or Trainer.train() metrics are synthesized from the checkpoint step count. _write_json(path, report) return report def _operation(context, horizon): with preserve_rng(): operations = iter(context.searcher.operations()) operation = next(operations, None) if operation is not None and operation.length != horizon: raise ValueError("Single searcher horizon differs from frozen optimizer steps") return operations, operation def run(context: Any, info: Any, plan_path: str | Path, *, backend: str, service_python: str | Path, max_wall_seconds: int = 28800, qualification_only: bool = False) -> dict: _require_gpu(info) plan_path = Path(plan_path).resolve() service_python = Path(service_python).absolute() if (not service_python.is_file() or not os.access(service_python, os.X_OK) or not service_python.resolve().is_relative_to(CANONICAL_ROOT)): raise ValueError("Use the qualified service Python from the canonical JAIF project") bundle = _load_backend(backend, plan_path, service_python) root, config = _canonical_config(bundle, plan_path) if type(max_wall_seconds) is not int or not 0 < max_wall_seconds <= bundle.plan["max_gpu_wall_seconds"]: raise ValueError("Managed deadline may not exceed the frozen GPU wall budget") horizon, output = config["training"]["max_steps"], Path(config["output_dir"]) with (root / ".experiment.lock").open("a") as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise RuntimeError("Another controller owns this frozen experiment") from None journal = bundle.driver._Journal(plan_path, resume=True) receipt = None try: if not journal.complete("resume-qualification"): raise ValueError("Managed training requires the exact completed GPU resume qualification") proof = bundle.driver._verify_resume_proof(journal) _qualified_runtime(bundle, proof) if qualification_only: bundle, config = _qualification_bundle(bundle) horizon, output = config["training"]["max_steps"], Path(config["output_dir"]) else: if not journal.complete("managed-resume-qualification"): raise ValueError("Production training requires completed managed GPU resume qualification") _qualification_proof(bundle, plan_path, proof) managed = ManagedTrial(context, info, identity=_identity(bundle, plan_path, config), binding_path=root / "managed-trials" / str(info.trial.trial_id) / "binding.json", max_wall_seconds=max_wall_seconds, native_validator=bundle.engine.validate_checkpoint) reporter = TrainingReporter(managed, journal) with preserve_rng(): checkpoint = managed.restore(output, _restore_validator(bundle, managed, config, proof)) if checkpoint is None: _fresh_output(output, managed) remaining = _remaining(journal, managed) restored_step = managed.last_step if checkpoint is not None else 0 if qualification_only and checkpoint is not None: _record_qualification_restore(managed, info, root) terminal = checkpoint is not None and managed.last_step == horizon operations, operation = (None, None) if terminal else _operation(context, horizon) if not terminal and operation is None: raise ValueError("Searcher ended before the frozen training horizon") reporter.operation = operation with _wall_guard(remaining): if terminal: report = _terminal_manifest(bundle, output, checkpoint, config, proof) else: with bundle.driver._service(root, bundle.service_python) as (gym_url, _): report = bundle.engine.run(bundle.plan["training_config"], gym_url, checkpoint_interval=bundle.plan["checkpoint_interval"], resume_from_checkpoint=checkpoint, managed_reporter=reporter) # Service context exits before the terminal snapshot or any acknowledgement. manifest = _read_json(output / "training-manifest.json") if (manifest.get("status") != report.get("status") or manifest.get("optimizer_step") != report.get("optimizer_step") or manifest.get("cleanup_errors")): raise ValueError("Managed trainer did not finish its native cleanup/manifest") checkpoint = Path(report["checkpoint"]) status = report["status"] step = report["optimizer_step"] if (status not in {"completed", "preempted_at_checkpoint"} or (status == "completed" and step != horizon) or (status == "preempted_at_checkpoint" and managed.preempt_step != step)): raise ValueError("Managed trainer ended at an unexpected optimizer boundary") if not terminal: reporter.publish(output, checkpoint) # Sole stopping snapshot follows all cleanup. # A restored terminal UUID is already registered. Reconstructed local # manifest metadata is not a new scheduler snapshot. if qualification_only and status == "completed": _qualification_proof(bundle, plan_path, proof, managed=managed, info=info, restored_step=restored_step) journal.done("managed-resume-qualification", [root / "managed-qualification/proof.json"]) stage = ("managed-preempted" if status == "preempted_at_checkpoint" else "managed-qualification-completed" if qualification_only else "managed-training-completed") journal.state.update(status="running", stage=stage, managed_training={"experiment_id": info.trial.experiment_id, "trial_id": info.trial.trial_id, "allocation_id": info.allocation_id, "optimizer_step": step, "scheduler_uuid": managed.last_uuid, "native_status": status, "qualification_only": qualification_only, "snapshot_reused": terminal, "heldout_assessment": "pending_original_driver"}) receipt = {"status": status, **journal.state["managed_training"], "restored_from_uuid": info.latest_checkpoint, "deadline": managed.binding["deadline"], "selection_and_heldout_assessment": "pending_original_driver"} except BaseException as error: journal.state.update(status="failed", error={"type": type(error).__name__, "message": str(error)}) raise finally: with preserve_rng(): journal.close() if receipt is not None: receipt["gpu_wall_seconds"] = journal.state["gpu_wall_seconds"] name = hashlib.sha256(info.allocation_id.encode()).hexdigest() _write_json(root / "managed-trials" / str(info.trial.trial_id) / (name + ".json"), receipt) if receipt["status"] == "preempted_at_checkpoint": with preserve_rng(): managed.acknowledge_preemption() return receipt if operations is None: operations, operation = _operation(context, horizon) if operation is not None: with preserve_rng(): context.train.report_validation_metrics(steps_completed=horizon, metrics={"completed_optimizer_steps": horizon}) operation.report_progress(horizon) operation.report_completed(float(horizon)) if next(operations, None) is not None: raise ValueError("Managed entrypoint requires a single searcher operation") return receipt def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--plan", type=Path, required=True) parser.add_argument("--backend", choices=("nemo", "openenv"), required=True) parser.add_argument("--service-python", type=Path, required=True) parser.add_argument("--max-wall-seconds", type=int, default=28800) parser.add_argument("--qualification-only", action="store_true") args = parser.parse_args() if importlib.metadata.version("determined") != "0.35.0": parser.error("Use pinned determined==0.35.0") import determined as det from determined import core info = det.get_cluster_info() with core.init(tensorboard_mode=core.TensorboardMode.MANUAL) as context: result = run(context, info, args.plan, backend=args.backend, service_python=args.service_python, max_wall_seconds=args.max_wall_seconds, qualification_only=args.qualification_only) print(json.dumps(result, sort_keys=True)) if __name__ == "__main__": main()