pcb-afm-simulation-system/scripts/test_p3_closed_loop.py

"""P3-M5 integration: real HTTP closed loop with a mock local executor.

Run:  python scripts/test_p3_closed_loop.py   (exit 0 = PASS)

Spins up the real FastAPI web backend on an isolated temp SQLite DB, drives
the AdaptiveOrchestrator, and lets a real TaskExecutor (mock solver) poll and
execute adaptive batches over HTTP. Verifies the full chain:

    orchestrator.start_loop -> task created (HTTP) -> executor claims ->
    mock simulation -> report_results (point_id) -> orchestrator.advance_loop
    feeds back -> next batch -> ... -> budget exhausted.

Requires: requests, uvicorn, fastapi. No real Motor-CAD.
"""
import json
import os
import subprocess
import sys
import tempfile
import time
import urllib.request

_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_BACKEND = os.path.join(_ROOT, "web", "backend")
_SCRIPTS = os.path.join(_ROOT, "scripts")
_TMP = tempfile.mkdtemp(prefix="p3loop_")
_DB = os.path.join(_TMP, "web.db")
_STATE = os.path.join(_TMP, "loops")
_PORT = 8137
_BASE = "http://127.0.0.1:%d" % _PORT

os.environ["AFM_DB_PATH"] = _DB
os.environ["KIMI_API_KEY"] = ""  # keep test hermetic: no AI calls
sys.path.insert(0, _BACKEND)
sys.path.insert(0, _SCRIPTS)
sys.path.insert(0, _ROOT)  # src.* (l0 re-export) resolves from repo root

from app.database import init_db  # noqa: E402
init_db()

from app.services.strategy_orchestrator import AdaptiveOrchestrator  # noqa: E402
from app.services.task_manager import get_task_manager  # noqa: E402

WEB_ENV = dict(os.environ, AFM_DB_PATH=_DB, AFM_PORT=str(_PORT))


def wait_health(timeout=30):
    url = _BASE + "/api/monitor/health"
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(url, timeout=3) as r:
                if r.status == 200:
                    return True
        except Exception:
            time.sleep(0.5)
    return False


# ---- start web ----
print("[0] starting web backend on :%d (temp DB)" % _PORT, flush=True)
web = subprocess.Popen(
    [sys.executable, "-m", "uvicorn", "app.main:app",
     "--host", "127.0.0.1", "--port", str(_PORT), "--log-level", "warning"],
    cwd=_BACKEND, env=WEB_ENV,
    stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
)
try:
    assert wait_health(), "web backend did not become healthy"
    print("[1] web backend healthy", flush=True)

    tm = get_task_manager()
    orch = AdaptiveOrchestrator(state_dir=_STATE)

    # ---- orchestrator start ----
    res = orch.start_loop(
        loop_id="loop-http",
        parameters=[
            {"name": "airgap_mm", "min_value": 0.8, "max_value": 2.0, "step": 0.1, "unit": "mm"},
            {"name": "current_a", "min_value": 5.0, "max_value": 20.0, "step": 0.5},
        ],
        total_budget=8, batch_size=4, initial_samples=4,
        objective_metric="tavg_nm", objective_direction="maximize",
    )
    assert res["phase"] == "running" and res["current_task_id"], res
    print("[2] orchestrator started: task=%s" % res["current_task_id"], flush=True)

    # ---- local executor (mock) over real HTTP ----
    import task_executor as te_mod
    te_mod.requests = None if not True else te_mod.requests  # keep real requests
    from task_executor import TaskExecutor

    ex = TaskExecutor(web_base_url=_BASE, enable_mock=True,
                      executor_id="motorcad-mock-m5")
    thread = ex.start_polling(interval=2)
    print("[3] mock executor polling started", flush=True)

    # ---- drive the loop ----
    steps = 0
    max_steps = 12
    batches = set()
    while steps < max_steps:
        steps += 1
        view = orch.get_loop_status("loop-http")
        phase = view["phase"]
        if phase in ("converged", "budget_exhausted", "failed"):
            print("[4] terminal phase=%s after %d advance steps" % (phase, steps), flush=True)
            break
        tid = view.get("current_task_id")
        if not tid:
            view = orch.advance_loop("loop-http")
            continue
        # wait for the executor to finish this batch over HTTP
        deadline = time.time() + 90
        while time.time() < deadline:
            t = tm.get_task(tid)
            if t and t["status"] in ("completed", "failed", "cancelled"):
                break
            time.sleep(1)
        else:
            raise RuntimeError("batch task %s did not finish in time" % tid)
        view = orch.advance_loop("loop-http")
        batches.add(view.get("current_batch"))
        print("[5] advance -> batch=%s task=%s phase=%s n_results=%s"
              % (view.get("current_batch"), view.get("current_task_id"),
                 view["phase"], view["n_results"]), flush=True)

    ex.stop()
    thread.join(timeout=5)

    final = orch.get_loop_status("loop-http")
    print("[6] FINAL phase=%s batches=%s n_results=%s"
          % (final["phase"], sorted(batches), final["n_results"]), flush=True)
    assert final["phase"] in ("converged", "budget_exhausted"), final
    assert final["n_results"] >= 4, final
    ss = final.get("search_state") or {}
    assert ss.get("completed_points", 0) >= 4, ss
    # every reported point must carry point_id and be fed back
    assert ss.get("used_budget", 0) > 0, ss

    print("\nALL P3-M5 HTTP CLOSED LOOP TESTS PASSED", flush=True)
finally:
    # cleanup
    try:
        ex.stop()
    except Exception:
        pass
    web.terminate()
    try:
        web.wait(timeout=10)
    except Exception:
        web.kill()
    print("[cleanup] web stopped", flush=True)