AI Flow

"""Deliberate deviations end to end: finish and step refuse with their items, `deviate` accepts one
on purpose with a reason, bound to the saved board and spec; finish completes and lists each one;
a changed board expires them. No CAD, bridge or dashboard."""
import json, os, pathlib, subprocess, sys, tempfile

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")))
 (footprint "R:R0603" (layer "F.Cu") (at 20 10) (property "Reference" "R2" (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 board is right even though the check refuses it"

with tempfile.TemporaryDirectory(prefix="aiflow-deviation-") as td:
    p = pathlib.Path(td)
    env = dict(os.environ, AIFLOW_DASHBOARD="off", 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):
        return subprocess.run([binary, "--run", "run", "--ai-thread", "deviation-test", *args], cwd=p, env=env, capture_output=True, text=True, timeout=120)

    def items():
        return json.loads((p / "run" / "run.json").read_text())["lastRefusalItems"]

    r = call("start", "--board", "board.kicad_pcb", "--spec", "spec.json", "--engine", "test")
    assert r.returncode == 0, r.stdout + r.stderr

    # never refused (John, 2026-11-04): deviate records even before anything was reported, with a note
    r = call("deviate", "gate", "--why", WHY)
    assert r.returncode == 0 and "finish" in r.stderr, r.stdout + r.stderr
    assert "removed" in call("deviate", "gate", "--clear").stdout

    # a step past undone steps proceeds, names them, and logs the skip
    r = call("step", "routing")
    assert r.returncode == 0 and "earlier steps not done yet" in r.stderr, r.stdout + r.stderr
    skipped = items()
    assert skipped and all(i.startswith("step ") for i in skipped), skipped
    log = (p / "run" / "run.jsonl").read_text()
    assert '"steps-skipped"' in log

    # finish completes as a report card: the open items, never a refusal
    r = call("finish")
    assert r.returncode == 0 and "open item(s)" in r.stdout and "Open items" in r.stderr, r.stdout + r.stderr
    open_items = items()
    assert open_items, "finish recorded no items"
    summary = json.loads((p / "run" / "run.json").read_text())["summary"]
    assert summary["open"], summary
    # a short reason and an unknown item are recorded too, with notes
    r = call("deviate", "not-an-item", "--why", "skip")
    assert r.returncode == 0 and "not an item" in r.stderr and "reason" in r.stderr, r.stderr
    assert "removed" in call("deviate", "not-an-item", "--clear").stdout
    # deviations move items from open to accepted-on-purpose
    for i in items():
        assert call("deviate", i, "--why", WHY).returncode == 0, i
    r = call("finish")
    assert r.returncode == 0 and "recorded deviation(s)" in r.stdout and "except" in r.stdout, r.stdout + r.stderr
    summary = json.loads((p / "run" / "run.json").read_text())["summary"]
    assert summary["deviations"] and all(d["why"] == WHY for d in summary["deviations"]) and not summary["open"], summary
    assert "deviation(s) on this board and spec" in call("deviate", "--list").stdout

    # the board changes: every deviation expires and the items are open again
    data = json.loads((p / "run" / "run.json").read_text())
    current = data["boards"][-1] if data.get("boards") else data["board"]
    cp = pathlib.Path(current if isinstance(current, str) else current["path"])
    cp = cp if cp.is_absolute() else p / cp
    cp.write_text(cp.read_text().replace("(at 20 10)", "(at 21 10)"))
    r = call("deviate", "--list")
    assert "0 deviation(s)" in r.stdout and "expired" in r.stdout, r.stdout
    r = call("finish")
    assert r.returncode == 0 and json.loads((p / "run" / "run.json").read_text())["summary"]["open"], "an expired deviation still hid an item"

    # clear removes one
    assert "removed" in call("deviate", open_items[0], "--clear").stdout
    print("PASS: never refused: step proceeds and logs skips, finish is a report card with open items, deviate records with notes and moves items to accepted, a changed board expires them, clear removes one")