Add guarded promoted YOLO activation

scripts/README.md

@@ -648,6 +648,24 @@ files are present without `--model-file`. It writes only
 `GEOINTEL_INSTALL_AI=true`, `YOLO_ENABLED=true`, `YOLO_MODELS_DIR=/app/models`
 and the mounted `YOLO_MODEL_PATH`.
 
+When a split-background promotion report recommends a specific candidate, use
+the guarded activation helper instead of choosing a model path manually. The
+helper validates the exact `candidate_key`, promotion status and local model
+asset before writing anything, and it mutates `.env` only with `--apply`:
+
+```bash
+python scripts/activate_promoted_yolo_candidate.py \
+  --promotion-report /mnt/user/appdata/geointel/artifacts/detection-model-promotion/split-aware/aoi1024bg512r3e50-high-threshold-split-20260710T222934Z/detection_model_promotion_report.json \
+  --candidate-key 'geointel-building-yolov8s-aoi1024bg512r3e50-pt|512|64|0.35' \
+  --models-dir /mnt/user/appdata/geointel/models \
+  --env-file /mnt/user/appdata/geointel/.env \
+  --json
+```
+
+Add `--apply` only after reviewing the emitted updates. The helper never
+downloads weights, loads the model or runs inference; restart or rebuild the
+container after applying because `YOLO_MODEL_PATH` is read from the environment.
+
 Tower-local model evaluation status:
 
 - `geointel-building-yolov8s-hardneg160r4e50.pt` is available as an evaluated

scripts/activate_promoted_yolo_candidate.py

@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+
+SUPPORTED_MODEL_SUFFIXES = {".pt", ".onnx", ".engine"}
+ENV_KEYS = ("GEOINTEL_INSTALL_AI", "YOLO_ENABLED", "YOLO_MODELS_DIR", "YOLO_MODEL_PATH")
+
+
+def _asset_id(path: Path) -> str:
+    raw = f"{path.stem}-{path.suffix.lower().lstrip('.')}"
+    normalized = re.sub(r"[^a-z0-9]+", "-", raw.lower()).strip("-")
+    return normalized or "model-asset"
+
+
+def _candidate_paths(models_dir: Path) -> list[Path]:
+    if not models_dir.exists() or not models_dir.is_dir():
+        return []
+    return sorted(
+        path.resolve()
+        for path in models_dir.rglob("*")
+        if path.is_file() and path.suffix.lower() in SUPPORTED_MODEL_SUFFIXES
+    )
+
+
+def _container_path(host_model_path: Path, models_dir: Path, container_model_dir: str) -> str:
+    relative = host_model_path.resolve().relative_to(models_dir.resolve())
+    base = container_model_dir.rstrip("/")
+    return f"{base}/{relative.as_posix()}" if relative.as_posix() else base
+
+
+def _read_env_lines(env_file: Path) -> list[str]:
+    if not env_file.exists():
+        return []
+    return env_file.read_text(encoding="utf-8").splitlines()
+
+
+def _update_env_file(env_file: Path, updates: dict[str, str]) -> None:
+    existing_lines = _read_env_lines(env_file)
+    seen: set[str] = set()
+    next_lines: list[str] = []
+
+    for line in existing_lines:
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or "=" not in line:
+            next_lines.append(line)
+            continue
+
+        key = line.split("=", 1)[0].strip()
+        if key in updates:
+            next_lines.append(f"{key}={updates[key]}")
+            seen.add(key)
+        else:
+            next_lines.append(line)
+
+    for key, value in updates.items():
+        if key not in seen:
+            next_lines.append(f"{key}={value}")
+
+    env_file.parent.mkdir(parents=True, exist_ok=True)
+    env_file.write_text("\n".join(next_lines).rstrip() + "\n", encoding="utf-8")
+
+
+def _load_report(path: Path) -> dict[str, Any]:
+    if not path.exists() or not path.is_file():
+        raise ValueError("Promotion report file does not exist")
+    payload = json.loads(path.read_text(encoding="utf-8-sig"))
+    if not isinstance(payload, dict):
+        raise ValueError("Promotion report must be a JSON object")
+    return payload
+
+
+def _candidate_from_report(report: dict[str, Any], candidate_key: str) -> dict[str, Any] | None:
+    recommended = report.get("recommended_candidate")
+    if isinstance(recommended, dict):
+        if recommended.get("candidate_key") == candidate_key:
+            return recommended
+        return None
+    if isinstance(recommended, str) and recommended == candidate_key:
+        for item in report.get("candidate_decisions") or []:
+            if isinstance(item, dict) and item.get("candidate_key") == candidate_key:
+                return item
+    for item in report.get("candidate_decisions") or []:
+        if isinstance(item, dict) and item.get("candidate_key") == candidate_key:
+            return item
+    return None
+
+
+def _gate_failures(report: dict[str, Any], candidate: dict[str, Any]) -> list[str]:
+    gates = report.get("gates") if isinstance(report.get("gates"), dict) else {}
+    failures: list[str] = []
+    if candidate.get("promotion_status") != "promote_candidate":
+        failures.append("candidate_not_promoted")
+    if candidate.get("rejection_reasons"):
+        failures.append("candidate_has_rejection_reasons")
+
+    min_positive_samples = int(gates.get("min_positive_samples") or 0)
+    min_background_samples = int(gates.get("min_background_samples") or 0)
+    min_mean_f1 = float(gates.get("min_mean_f1") or 0)
+    max_background_detections = int(gates.get("max_background_detections_per_sample") or 0)
+
+    if int(candidate.get("positive_sample_count") or 0) < min_positive_samples:
+        failures.append("insufficient_positive_samples")
+    if int(candidate.get("background_sample_count") or 0) < min_background_samples:
+        failures.append("insufficient_background_samples")
+    if float(candidate.get("mean_f1") or 0) < min_mean_f1:
+        failures.append("positive_mean_f1_below_gate")
+    if int(candidate.get("max_background_detections") or 0) > max_background_detections:
+        failures.append("background_false_positive_pressure")
+    return failures
+
+
+def _resolve_model_asset(models_dir: Path, model_asset_id: str) -> Path | None:
+    for path in _candidate_paths(models_dir):
+        if _asset_id(path) == model_asset_id:
+            return path
+    return None
+
+
+def _base_payload(args: argparse.Namespace) -> dict[str, Any]:
+    return {
+        "promotion_report": str(Path(args.promotion_report).resolve()),
+        "candidate_key": args.candidate_key,
+        "models_dir": str(Path(args.models_dir).resolve()),
+        "env_file": str(Path(args.env_file).resolve()),
+        "candidate": None,
+        "selected_host_model_path": None,
+        "selected_container_model_path": None,
+        "env_updates": {},
+        "apply": args.apply,
+        "applied": False,
+        "docker_restart_required": False,
+        "will_download_models": False,
+        "will_run_inference": False,
+    }
+
+
+def activate(args: argparse.Namespace) -> tuple[int, dict[str, Any]]:
+    payload = _base_payload(args)
+    try:
+        report = _load_report(Path(args.promotion_report))
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        payload.update({"status": "invalid_promotion_report", "message": str(exc)})
+        return 2, payload
+
+    candidate = _candidate_from_report(report, args.candidate_key)
+    if not candidate:
+        payload.update(
+            {
+                "status": "candidate_not_recommended",
+                "message": "Candidate key does not match the report recommended candidate.",
+            }
+        )
+        return 3, payload
+
+    payload["candidate"] = candidate
+    failures = _gate_failures(report, candidate)
+    if failures:
+        payload.update(
+            {
+                "status": "candidate_not_promoted",
+                "message": "Candidate did not pass the promotion gates.",
+                "rejection_reasons": failures,
+            }
+        )
+        return 3, payload
+
+    model_asset_id = str(candidate.get("model_asset_id") or "").strip()
+    if not model_asset_id:
+        payload.update({"status": "model_asset_missing", "message": "Candidate does not record model_asset_id."})
+        return 2, payload
+
+    models_dir = Path(args.models_dir).resolve()
+    selected_model = _resolve_model_asset(models_dir, model_asset_id)
+    if selected_model is None:
+        payload.update(
+            {
+                "status": "model_asset_not_found",
+                "message": "Promoted candidate model asset was not found in the local models directory.",
+                "model_asset_id": model_asset_id,
+            }
+        )
+        return 2, payload
+
+    selected_container_path = _container_path(selected_model, models_dir, args.container_model_dir)
+    updates = {
+        "GEOINTEL_INSTALL_AI": "true",
+        "YOLO_ENABLED": "true",
+        "YOLO_MODELS_DIR": args.container_model_dir.rstrip("/"),
+        "YOLO_MODEL_PATH": selected_container_path,
+    }
+    payload.update(
+        {
+            "status": "ready_to_apply",
+            "message": "Promoted YOLO candidate validated. Re-run with --apply to update the environment file.",
+            "selected_host_model_path": str(selected_model),
+            "selected_container_model_path": selected_container_path,
+            "env_updates": updates,
+            "docker_restart_required": True,
+        }
+    )
+
+    if args.apply:
+        _update_env_file(Path(args.env_file).resolve(), updates)
+        payload.update(
+            {
+                "status": "applied",
+                "message": "Environment file updated. Rebuild or restart the GeoIntel container to use the promoted model path.",
+                "applied": True,
+            }
+        )
+
+    return 0, payload
+
+
+def _emit(payload: dict[str, Any], *, as_json: bool) -> None:
+    if as_json:
+        print(json.dumps(payload, indent=2, sort_keys=True))
+        return
+
+    print(f"status: {payload['status']}")
+    print(f"message: {payload['message']}")
+    if payload.get("selected_host_model_path"):
+        print(f"host model: {payload['selected_host_model_path']}")
+        print(f"container model: {payload['selected_container_model_path']}")
+    if payload.get("env_updates"):
+        print("env updates:")
+        for key in ENV_KEYS:
+            if key in payload["env_updates"]:
+                print(f"  {key}={payload['env_updates'][key]}")
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Activate an existing local YOLO model only after a promotion report recommends the exact candidate key."
+    )
+    parser.add_argument("--promotion-report", required=True, help="Path to detection_model_promotion_report.json.")
+    parser.add_argument("--candidate-key", required=True, help="Exact promoted candidate key from the report.")
+    parser.add_argument(
+        "--models-dir",
+        default=os.environ.get("GEOINTEL_MODELS_PATH", "models"),
+        help="Host directory containing mounted local model files.",
+    )
+    parser.add_argument("--container-model-dir", default="/app/models", help="Container path where --models-dir is mounted.")
+    parser.add_argument("--env-file", default=".env", help="Environment file to update when --apply is supplied.")
+    parser.add_argument("--apply", action="store_true", help="Write YOLO env updates after all promotion gates pass.")
+    parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv or sys.argv[1:])
+    exit_code, payload = activate(args)
+    _emit(payload, as_json=args.json)
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_readiness_check.sh

@@ -47,6 +47,7 @@ ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_tile_dataset.py
 ${PYTHON_BIN} -m py_compile scripts/audit_operator_yolo_dataset_quality.py
 ${PYTHON_BIN} -m py_compile scripts/build_detection_model_promotion_report.py
 ${PYTHON_BIN} -m py_compile scripts/build_background_corpus_split_report.py
+${PYTHON_BIN} -m py_compile scripts/activate_promoted_yolo_candidate.py
 ${PYTHON_BIN} -m py_compile scripts/cleanup_demo_artifacts.py
 ${PYTHON_BIN} -m py_compile backend/scripts/cleanup_demo_artifacts.py
 ${PYTHON_BIN} -m compileall backend/app