AI Flow

"""Hands-on AI Flow end to end (docs/hands-on.md): the plan blocks design work until the human
confirms it on the timeline (a real POST to the timeline server), the first confirmed plan is saved
as their default profile, a check-in step holds the run until they approve, a human step until they
say done, finish-guard reads an open check-in as the human's turn, and the next run starts from the
default profile. No CAD, bridge, notifications or dashboard tab."""
import json, os, pathlib, socket, subprocess, sys, tempfile, time, urllib.request

binary = str(pathlib.Path(sys.argv[1]).resolve())
BOARD = """(kicad_pcb (version 20241229) (generator "test") (general (thickness 1.6))
 (layers (0 "F.Cu" signal) (2 "B.Cu" signal) (25 "Edge.Cuts" user) (37 "F.SilkS" user))
 (net 0 "") (net 1 "A") (net 2 "B")
 (gr_rect (start 0 0) (end 30 20) (layer "Edge.Cuts"))
 (footprint "R:R0603" (layer "F.Cu") (at 10 10) (property "Reference" "R1" (at 0 -1.5) (layer "F.SilkS"))
  (pad "1" smd rect (at -0.8 0) (size 0.8 0.9) (layers "F.Cu") (net 1 "A"))
  (pad "2" smd rect (at 0.8 0) (size 0.8 0.9) (layers "F.Cu") (net 2 "B"))))
"""
WHY = "test fixture: this step's proof is not what this workflow test is about"

with tempfile.TemporaryDirectory(prefix="aiflow-handson-") as td:
    p = pathlib.Path(td)
    env = dict(os.environ, AIFLOW_DASHBOARD="off", AIFLOW_NOTIFY="off", AIFLOW_PREFS_DIR=str(p / "prefs"),
               AIFLOW_NO_PART_SEARCH="1", AIFLOW_NO_PART_LOOKUP="1")
    (p / "board.kicad_pcb").write_text(BOARD)
    (p / "spec.json").write_text("{}")

    def call(*args, run="run"):
        return subprocess.run([binary, "--run", run, "--ai-thread", "handson-test", *args], cwd=p, env=env, capture_output=True, text=True, timeout=120)

    def data(run="run"):
        return json.loads((p / run / "run.json").read_text())

    def step_past(name):
        """`step name`, deviating any proof the fixture cannot give; returns the final answer."""
        for _ in range(12):
            r = call("step", name)
            if r.returncode == 0 or "waits for the human" in r.stdout:
                return r
            items = data().get("lastRefusalItems", [])
            assert items, r.stdout
            for i in items:
                assert call("deviate", i, "--why", WHY).returncode == 0
        raise AssertionError("could not reach " + name)

    r = call("handson", "profiles")
    assert r.returncode == 0 and "Full AI flow" in r.stdout and "1 flow profile" in r.stdout, r.stdout
    r = call("start", "--board", "board.kicad_pcb", "--spec", "spec.json", "--engine", "test")
    assert r.returncode == 0, r.stdout + r.stderr
    assert "handsOn" not in data(), "no default profile yet: the run is not hands-on"
    assert call("step", "intake").returncode == 0
    seq = data()["sequence"]
    a, b, c = seq[1], seq[2], seq[3]

    # the plan is open: the AI waits for the human to confirm it (AI Flow never refuses: a step started now is noted)
    r = call("handson", "plan")
    assert r.returncode == 0 and "waiting for the human" in r.stdout, r.stdout
    r = call("step", a)
    assert r.returncode == 0 and "hands-on plan" in r.stdout + r.stderr, r.stdout + r.stderr
    assert call("finish-guard", "--json").stdout.count('"waiting-for-human"') == 1

    # the human confirms it on the timeline: a real POST to the run's timeline server
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0)); port = s.getsockname()[1]
    server = subprocess.Popen([binary, "--run", "run", "timeline", "serve", "--port", str(port)], cwd=p, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        def post(body, header=True):
            req = urllib.request.Request(f"http://127.0.0.1:{port}/handson", data=json.dumps(body).encode(), method="POST",
                                         headers={"Content-Type": "application/json", **({"X-Aiflow-Handson": "1"} if header else {})})
            try:
                with urllib.request.urlopen(req, timeout=5) as resp: return json.loads(resp.read())
            except urllib.error.HTTPError as e: return json.loads(e.read())
        for _ in range(50):
            try:
                urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=1); break
            except Exception: time.sleep(0.1)
        assert post({"action": "approved", "step": a}, header=False)["ok"] is False, "the header is required"
        assert post({"action": "plan", "plan": {"steps": {a: {"mode": "custom"}}}})["ok"] is False, "custom needs instructions"
        plan = {"steps": {a: {"mode": "checkin"}, b: {"mode": "human"}, c: {"mode": "custom", "instructions": "keep R1 near the edge"}},
                "notify": "none", "screen": "hands-off", "saveAs": "Test flow", "makeDefault": True}
        assert post({"action": "plan", "plan": plan})["ok"] is True
        r = call("handson", "wait", "--timeout", "5")
        assert r.returncode == 0 and "HUMAN (timeline): confirmed" in r.stdout and "Test flow" in r.stdout, r.stdout
        h = data()["handsOn"]
        assert h["state"] == "active" and h["steps"][b]["mode"] == "human" and h["screen"] == "hands-off", h
        assert json.loads((p / "prefs" / "preferences.json").read_text())["defaultProfile"] == "test-flow"

        # a check-in step: done, shown, held until the human approves
        r = step_past(a)
        assert r.returncode == 0 and "HANDS-ON CHECK-IN" in r.stdout, r.stdout
        # going on without the OK is never refused: it is noted, logged and listed at finish
        r = call("step", b)
        assert r.returncode == 0 and f"handson ask {a}" in r.stdout + r.stderr, r.stdout + r.stderr
        assert '"handson-proceeded-without-ok"' in (p / "run" / "run.jsonl").read_text()
        assert call("handson", "ask", a, "--summary", "look at R1").returncode == 0
        assert '"waiting-for-human"' in call("finish-guard", "--json").stdout
        r = call("step", b); assert "waiting for the human's review" in r.stdout + r.stderr, r.stdout + r.stderr
        assert call("handson", "approve", a).returncode == 2, "a chat approval quotes the human"
        assert post({"action": "changes", "step": a})["ok"] is False, "changes say what to change"
        assert post({"action": "changes", "step": a, "note": "move R1 left"})["ok"] is True
        r = call("step", b)
        assert r.returncode == 0 and "move R1 left" in r.stdout + r.stderr and "HUMAN (timeline): changes" in r.stdout, r.stdout + r.stderr
        assert call("handson", "ask", a, "--summary", "R1 moved").returncode == 0
        assert post({"action": "approved", "step": a, "note": "good"})["ok"] is True

        # a human step: handed over, held until they say done
        r = step_past(b)
        assert r.returncode == 0 and "HANDS-ON HUMAN STEP" in r.stdout and "Do NOT do its work" in r.stdout, r.stdout
        r = call("step", c)
        assert r.returncode == 0 and "not said they are done" in r.stdout + r.stderr, r.stdout + r.stderr
        r = call("handson", "done", b, "--quote", "routed it myself")
        assert r.returncode == 0 and "proof" in r.stdout, r.stdout
        r = step_past(c)
        assert r.returncode == 0 and "keep R1 near the edge" in r.stdout, r.stdout

        # the timeline carries the plan, the modes and the profiles for the page
        call("timeline", "write")
        tl = json.loads((p / "run" / "timeline" / "timeline.json").read_text())
        assert tl["handsOn"]["state"] == "active" and any(x["slug"] == "test-flow" and x["default"] for x in tl["handsOn"]["profiles"]), tl["handsOn"]
        assert {s["name"]: s["mode"] for s in tl["steps"]}[b] == "human"
        assert "hands-on" in (p / "run" / "timeline" / "index.html").read_text().lower()
    finally:
        server.terminate(); server.wait(timeout=10)

    # the next board starts from the human's default flow; switching is one command
    r = call("start", "--board", "board.kicad_pcb", "--spec", "spec.json", "--engine", "test", run="run2")
    assert r.returncode == 0 and "Test flow" in r.stdout, r.stdout
    assert data("run2")["handsOn"]["steps"][b]["mode"] == "human"
    r = call("handson", "use", "full-ai", run="run2")
    assert r.returncode == 0 and "every step auto" in r.stdout, r.stdout
    assert call("handson", "profiles").stdout.count("* Test flow") == 1
print("ok: hands-on workflow")