Validate training completion handoff
GeoIntel release gates / Compile, test, contracts and builds (push) Canceled after 0s
GeoIntel release gates / Python and npm vulnerability policy (push) Canceled after 0s
GeoIntel release gates / GIS image, SBOM and container scan (push) Canceled after 0s

This commit is contained in:
Jens
2026-07-29 15:48:51 +02:00
parent d6502140da
commit f18a23c060
3 changed files with 30 additions and 4 deletions
+11 -2
View File
@@ -37,6 +37,13 @@ def write_state(path: Path, payload: dict) -> None:
temporary.replace(path)
def load_completion_command(path: Path) -> list[str]:
command = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(command, list) or not command or not all(isinstance(x, str) and x for x in command):
raise ValueError("completion command must be a non-empty JSON list of non-empty strings")
return command
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--container", required=True)
@@ -64,9 +71,11 @@ def main() -> int:
write_state(state_path, state)
return 0
if args.completion_command_json:
command = json.loads(args.completion_command_json.read_text(encoding="utf-8"))
if not isinstance(command, list) or not command or not all(isinstance(x, str) and x for x in command):
try:
command = load_completion_command(args.completion_command_json)
except (OSError, json.JSONDecodeError, ValueError) as exc:
state["status"] = "invalid_completion_command"
state["completion_handoff_error"] = str(exc)
write_state(state_path, state)
return 4
result = subprocess.run(command, check=False)