Recover active training loop checkpoints

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:

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")

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

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/<iteration>/weights/last.pt` exists, the orchestrator uses YOLO's
+  exact `resume=<checkpoint>` 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.

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.
 

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,7 +260,11 @@ 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(
+        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,
@@ -282,9 +290,13 @@ def main() -> int:
             hsv_h=args.hsv_h,
             hsv_s=args.hsv_s,
                 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"],

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())