diff --git a/backend/tests/test_belgium_training_loop.py b/backend/tests/test_belgium_training_loop.py index ca8eb260..78c2b52d 100644 --- a/backend/tests/test_belgium_training_loop.py +++ b/backend/tests/test_belgium_training_loop.py @@ -85,6 +85,13 @@ def test_failed_iteration_builds_train_only_sampling_for_next_checkpoint(tmp_pat assert command[command.index("--output-dir") + 1].endswith("failure-driven-training") +def test_partial_iteration_resume_uses_exact_checkpoint_and_cuda(tmp_path: Path) -> None: + checkpoint = tmp_path / "runs" / "iteration-002" / "weights" / "last.pt" + assert MODULE.resumable_training_command("yolo", checkpoint) == [ + "yolo", "train", f"resume={checkpoint}", "device=0" + ] + + def test_dry_run_can_gate_existing_checkpoint_without_training(tmp_path: Path) -> None: audit = tmp_path / "audit.json" audit.write_text(json.dumps({ @@ -107,7 +114,9 @@ def test_dry_run_can_gate_existing_checkpoint_without_training(tmp_path: Path) - ], capture_output=True, text=True, check=False, ) assert result.returncode == 0 - assert json.loads(result.stdout) == {"training_command": None, "evaluate_existing": True} + assert json.loads(result.stdout) == { + "training_command": None, "evaluate_existing": True, "resume_partial": False + } def test_existing_checkpoint_iteration_directory_can_be_created_without_yolo(tmp_path: Path) -> None: diff --git a/backend/tests/test_training_loop_supervisor.py b/backend/tests/test_training_loop_supervisor.py new file mode 100644 index 00000000..23f195f5 --- /dev/null +++ b/backend/tests/test_training_loop_supervisor.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path + + +SCRIPT = Path(__file__).parents[2] / "scripts" / "supervise_container_training_loop.py" +SPEC = importlib.util.spec_from_file_location("loop_supervisor", SCRIPT) +assert SPEC and SPEC.loader +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + + +def test_loop_status_is_fail_closed_and_recognises_completion(tmp_path: Path) -> None: + state = tmp_path / "state.json" + assert MODULE.read_loop_status(state) is None + state.write_text("not-json", encoding="utf-8") + assert MODULE.read_loop_status(state) == "invalid" + state.write_text(json.dumps({"status": "training_complete"}), encoding="utf-8") + assert MODULE.read_loop_status(state) == "training_complete" + + +def test_process_detection_requires_exact_marker(monkeypatch) -> None: + class Result: + returncode = 0 + stdout = "12 python api.py\n13 python run_belgium_building_training_loop.py --output-dir /runs/v37\n" + + monkeypatch.setattr(MODULE.subprocess, "run", lambda *args, **kwargs: Result()) + assert MODULE.process_active("geointel", "run_belgium_building_training_loop.py") + assert not MODULE.process_active("geointel", "other_loop.py") diff --git a/docs/BELGIUM_BUILDING_TRAINING_LOOP.md b/docs/BELGIUM_BUILDING_TRAINING_LOOP.md index 24d5316c..d849647e 100644 --- a/docs/BELGIUM_BUILDING_TRAINING_LOOP.md +++ b/docs/BELGIUM_BUILDING_TRAINING_LOOP.md @@ -106,6 +106,13 @@ versioned JSON argv list. The handoff starts the orchestrator detached exactly once; shell strings are not accepted. Subsequent iterations retain the frozen 180-degree aerial rotation, vertical/horizontal flip, scale and translation parameters rather than silently reverting to generic augmentation defaults. +Before fitting a state-pending iteration, the orchestrator checks its canonical +run directory for `weights/last.pt`. If present, it resumes that exact CUDA +checkpoint instead of restarting from the prior candidate. A host-side parent +supervisor monitors the exact loop-process marker and persisted loop state; on +container recreation it restores the current orchestrator script and launches +the versioned JSON argv command. Invalid state and bounded launch exhaustion +fail closed. The orchestrator refuses to start unless every automated frozen-dataset gate passes and the corpus contains zero blank/low-variance positive tiles. The audit status may remain `needs_human_review` while training and objective diff --git a/docs/CODEX_EXECUTION_LOG.md b/docs/CODEX_EXECUTION_LOG.md index 16411365..6d327adf 100644 --- a/docs/CODEX_EXECUTION_LOG.md +++ b/docs/CODEX_EXECUTION_LOG.md @@ -11827,6 +11827,15 @@ Deployment evidence: cleanly from the same candidate and checksummed sampling with `warmup_epochs=1`, `warmup_bias_lr=0.01`, HSV `0.01/0.2/0.15`, patience 18 and the existing aerial rotation/flip contract. +- Added checkpoint-aware recovery for the state-pending current iteration. + When `runs//weights/last.pt` exists, the orchestrator uses YOLO's + exact `resume=` CUDA path and records that provenance after the + iteration is assessed. +- Added and activated a host-side parent-loop supervisor. It monitors the exact + orchestrator marker and fail-closed JSON loop state, restores the current + script into a recreated container and relaunches only the versioned argv + command. Its live state is `monitoring`, with zero relaunches and one active + iteration-2 GPU process. - Added those aerial augmentation parameters to the orchestrator CLI and training command, preventing later failure-driven checkpoints from silently reverting to generic orientation assumptions. diff --git a/docs/TODO.md b/docs/TODO.md index 9f24a682..00fb73d3 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -979,6 +979,7 @@ This file now starts with the current implementation status. Older preparation/b - [x] Complete v37 by early stopping at epoch 23, reject its best epoch-5 checkpoint calibration-first, and start closed-loop iteration 2 from checksummed failure-driven sampling. - [x] Make orchestrator patience explicit and retain the frozen value `18` in every subsequent CUDA iteration. - [x] Preserve v37 warmup and aerial HSV augmentation in every loop iteration; archive and invalidate the pre-epoch-2 run that exposed generic Ultralytics warmup drift. +- [x] Resume a partially written current loop iteration from its exact `last.pt` after container recreation and activate a host-side parent-loop supervisor. - [x] Evaluate the completed v36 YOLO11x checkpoint calibration-first on the rotated v30 holdouts; reject it before opening test/background because the regional calibration gate failed. - [ ] Finish and assess the leak-free v37 YOLO11x failure-driven CUDA iteration; open test/background evidence only if every calibration gate passes. diff --git a/scripts/run_belgium_building_training_loop.py b/scripts/run_belgium_building_training_loop.py index de8df250..3c23b919 100644 --- a/scripts/run_belgium_building_training_loop.py +++ b/scripts/run_belgium_building_training_loop.py @@ -166,6 +166,10 @@ def failure_sampling_command( ] +def resumable_training_command(yolo: str, checkpoint: Path) -> list[str]: + return [yolo, "train", f"resume={checkpoint}", "device=0"] + + def run(command: list[str], log_path: Path | None = None, *, allowed: set[int] = {0}) -> int: if log_path: log_path.parent.mkdir(parents=True, exist_ok=True) @@ -256,8 +260,12 @@ def main() -> int: iteration_dir.mkdir(parents=True, exist_ok=True) train_run = args.output_dir / "runs" / name evaluate_existing = args.evaluate_initial_model and offset == 0 and not state["iterations"] - command = None if evaluate_existing else training_command( - args.yolo, + partial_checkpoint = train_run / "weights" / "last.pt" + resume_partial = not evaluate_existing and partial_checkpoint.is_file() + command = None if evaluate_existing else ( + resumable_training_command(args.yolo, partial_checkpoint) + if resume_partial else training_command( + args.yolo, model=model, data=train_yaml, project=args.output_dir / "runs", @@ -281,10 +289,14 @@ def main() -> int: warmup_bias_lr=args.warmup_bias_lr, hsv_h=args.hsv_h, hsv_s=args.hsv_s, - hsv_v=args.hsv_v, - ) + hsv_v=args.hsv_v, + )) if args.dry_run: - print(json.dumps({"training_command": command, "evaluate_existing": evaluate_existing}, indent=2)) + print(json.dumps({ + "training_command": command, + "evaluate_existing": evaluate_existing, + "resume_partial": resume_partial, + }, indent=2)) return 0 if evaluate_existing: best = model @@ -396,6 +408,7 @@ def main() -> int: "candidate": str(candidate), "candidate_sha256": sha256(candidate), "training_skipped_for_existing_checkpoint": evaluate_existing, + "training_resumed_from_partial_checkpoint": resume_partial, "assessment": str(assessment), "status": decision["status"], "failures": decision["failures"], diff --git a/scripts/supervise_container_training_loop.py b/scripts/supervise_container_training_loop.py new file mode 100644 index 00000000..3e90551e --- /dev/null +++ b/scripts/supervise_container_training_loop.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Keep a checkpoint-aware training-loop parent alive across container recreation.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import time +from datetime import UTC, datetime +from pathlib import Path + +TERMINAL_STATUSES = {"training_complete"} + + +def read_loop_status(path: Path) -> str | None: + if not path.is_file(): + return None + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return "invalid" + return value.get("status") if isinstance(value, dict) else "invalid" + + +def process_active(container: str, marker: str) -> bool: + result = subprocess.run( + ["docker", "top", container, "-eo", "pid,args"], + capture_output=True, text=True, check=False, + ) + return result.returncode == 0 and any(marker in line for line in result.stdout.splitlines()) + + +def write_state(path: Path, state: dict) -> None: + state["updated_at"] = datetime.now(UTC).isoformat() + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(state, indent=2), encoding="utf-8") + temporary.replace(path) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--container", required=True) + parser.add_argument("--process-marker", required=True) + parser.add_argument("--loop-state", type=Path, required=True) + parser.add_argument("--command-json", type=Path, required=True) + parser.add_argument("--host-script", type=Path, required=True) + parser.add_argument("--container-script", required=True) + parser.add_argument("--state", type=Path, required=True) + parser.add_argument("--poll-seconds", type=int, default=30) + parser.add_argument("--max-launches", type=int, default=20) + parser.add_argument("--once", action="store_true") + args = parser.parse_args() + state = {"schema_version": 1, "status": "monitoring", "launch_count": 0} + if args.state.is_file(): + state.update(json.loads(args.state.read_text(encoding="utf-8"))) + while True: + loop_status = read_loop_status(args.loop_state) + state["loop_status"] = loop_status + if loop_status in TERMINAL_STATUSES: + state["status"] = "loop_finished" + write_state(args.state, state) + return 0 + if loop_status == "invalid": + state["status"] = "invalid_loop_state" + write_state(args.state, state) + return 2 + if not process_active(args.container, args.process_marker): + if int(state["launch_count"]) >= args.max_launches: + state["status"] = "launch_budget_exhausted" + write_state(args.state, state) + return 3 + command = json.loads(args.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): + state["status"] = "invalid_command" + write_state(args.state, state) + return 4 + copied = subprocess.run( + ["docker", "cp", str(args.host_script), f"{args.container}:{args.container_script}"], + check=False, + ) + launched = subprocess.run(command, check=False) if copied.returncode == 0 else copied + if launched.returncode != 0: + state["status"] = "launch_failed" + write_state(args.state, state) + return 5 + state["launch_count"] = int(state["launch_count"]) + 1 + state["status"] = "loop_relaunched" + else: + state["status"] = "monitoring" + write_state(args.state, state) + if args.once: + return 0 + time.sleep(args.poll_seconds) + + +if __name__ == "__main__": + raise SystemExit(main())