AI Flow

"""Issue #60 through the CLI: a declared mechanical review becomes a step before finish, finish refuses
it missing, a missing tool is refused with how to get it, a failing review records nothing, a passing
review (a fake adom-aiflow-mechanical on PATH speaking the prototype's contract) satisfies finish, a
different board voids it, and a spec that declares nothing is unchanged. No CAD, desktop or dashboard."""
import json, os, pathlib, subprocess, sys, tempfile
binary = str(pathlib.Path(sys.argv[1]).resolve())
HEAD = '(kicad_pcb (version 20240108) (net 0 "") (layers (0 "F.Cu" signal) (31 "B.Cu" signal)) (gr_circle (center 10 10) (end 19 10) (layer "Edge.Cuts")) (gr_circle (center 10 10) (end 14 10) (layer "F.Cu") (width 0.5))'
def hole(x):
    return f' (footprint "MountingHole:MountingHole_3.2mm_M3" (layer "F.Cu") (at {x} 10) (property "Reference" "H1") (pad "" np_thru_hole circle (at 0 0) (size 3.2 3.2) (drill 3.2) (layers "*.Cu" "*.Mask")))'
FAKE = '''#!/usr/bin/env python3
import json, sys
if len(sys.argv) != 3 or sys.argv[1] != "lint": sys.exit(2)
v = json.load(open(sys.argv[2]))
bad = [f for f in v.get("fasteners", []) if not f.get("tool")]
print(json.dumps({"schemaVersion": 1, "passed": not bad and bool(v.get("fasteners")), "findings": [{"severity": "error", "id": f.get("id", "?"), "check": "tool", "hint": "Specify the actual tool."} for f in bad], "animationRequired": False}))
sys.exit(1 if bad or not v.get("fasteners") else 0)
'''
with tempfile.TemporaryDirectory(prefix='aiflow-review-') as td:
    p = pathlib.Path(td)
    tools = p / 'tools'; tools.mkdir(); fake = tools / 'adom-aiflow-mechanical'; fake.write_text(FAKE); fake.chmod(0o755)
    base = {k: v for k, v in os.environ.items() if k not in ('AIFLOW_MECHANICAL_SRC', 'AIFLOW_WIRING_SRC')}
    bare = dict(base, AIFLOW_DASHBOARD='off', PATH='/usr/bin:/bin')
    with_tool = dict(bare, PATH=f'{tools}:/usr/bin:/bin')
    assert not any((pathlib.Path(d) / 'adom-aiflow-mechanical').exists() for d in ('/usr/bin', '/bin'))
    def call(run, *args, env=bare):
        return subprocess.run([binary, '--run', str(run), '--ai-thread', 'review-test', *args], env=env, capture_output=True, text=True, timeout=60)
    def attach(run, board, spec):
        state = run / 'run.json'; d = json.loads(state.read_text()); d['boards'] = [str(board)]; d['spec'] = str(spec); state.write_text(json.dumps(d))
    def data(run): return json.loads((run / 'run.json').read_text())
    board = p / 'plate.kicad_pcb'; board.write_text(HEAD + hole(15) + ')\n')
    spec = p / 'spec.json'; spec.write_text(json.dumps({'mechanicalReview': {'schemaVersion': 1, 'input': 'mech.json', 'refs': ['H1']}}))
    mech = p / 'mech.json'

    run = p / 'declared'
    assert call(run, 'intake', '--prompt-time', '2026-10-10T00:00:00Z', '--engine', 'test', '--kind', 'pattern').returncode == 0
    attach(run, board, spec)
    r = call(run, 'step', 'rules'); assert r.returncode == 0, (r.stdout, r.stderr)
    seq = data(run)['sequence']; assert seq.index('mechanical-review') == seq.index('finish') - 1 and 'wiring-review' not in seq, seq
    r = call(run, 'finish'); out = r.stdout + r.stderr
    assert r.returncode == 0 and 'mechanical review not recorded' in out, out

    # tool not installed: a clear refusal that says it is unreleased and names the repo
    mech.write_text(json.dumps({'schemaVersion': 1, 'fasteners': [{'id': 'H1-screw', 'ref': 'H1', 'tool': '2.5 mm hex'}]}))
    r = call(run, 'review', 'mechanical'); out = r.stdout + r.stderr
    assert r.returncode != 0 and 'not installed' in out and 'adom/adom-aiflow-mechanical' in out and 'unreleased' in out and 'AIFLOW_MECHANICAL_SRC' in out, out
    # a review file without the contract version is refused before the tool runs
    mech.write_text(json.dumps({'fasteners': [{'id': 'H1-screw', 'tool': 'x'}]}))
    r = call(run, 'review', 'mechanical', env=with_tool); assert r.returncode != 0 and 'schemaVersion' in r.stdout + r.stderr, r.stdout + r.stderr
    # a review naming a part the board does not have is stale
    mech.write_text(json.dumps({'schemaVersion': 1, 'fasteners': [{'id': 's', 'ref': 'H7', 'tool': 'x'}]}))
    r = call(run, 'review', 'mechanical', env=with_tool); assert r.returncode != 0 and 'H7' in r.stdout + r.stderr, r.stdout + r.stderr
    # a failing review records nothing
    mech.write_text(json.dumps({'schemaVersion': 1, 'fasteners': [{'id': 'H1-screw', 'ref': 'H1'}]}))
    r = call(run, 'review', 'mechanical', env=with_tool); out = r.stdout + r.stderr
    assert r.returncode == 0 and 'FINDINGS' in out and 'did not pass' in out and 'Specify the actual tool' in out, out
    assert 'mechanical-review' not in data(run)['outcomes']
    # a passing review, during its step, bound to the review file, spec and board
    mech.write_text(json.dumps({'schemaVersion': 1, 'fasteners': [{'id': 'H1-screw', 'ref': 'H1', 'tool': '2.5 mm hex'}]}))
    d = data(run); d['steps']['mechanical-review'] = {'visits': [{'at': '2026-10-10T00:00:01Z'}]}; d['currentStep'] = 'mechanical-review'; (run / 'run.json').write_text(json.dumps(d))
    r = call(run, 'review', 'mechanical', env=with_tool); assert r.returncode == 0, (r.stdout, r.stderr)
    o = data(run)['outcomes']['mechanical-review']
    assert o['pass'] and o['schemaVersion'] == 1 and o['boardSha256'] and o['inputSha256'] and o['specSha256'] and o['refs'] == ['H1'], o
    assert [f['ref'] for f in o['serviceFeatures']] == ['H1'] and o['tool']['mode'] == 'path', o
    r = call(run, 'finish'); out = r.stdout + r.stderr
    assert r.returncode == 0 and 'mechanical review' not in out and 'native DRC' in out, out
    # the review file edited afterwards: stale
    mech.write_text(json.dumps({'schemaVersion': 1, 'fasteners': [{'id': 'H1-screw', 'ref': 'H1', 'tool': '3 mm hex'}]}))
    r = call(run, 'finish'); assert 'review file changed' in r.stdout + r.stderr, r.stdout + r.stderr
    mech.write_text(json.dumps({'schemaVersion': 1, 'fasteners': [{'id': 'H1-screw', 'ref': 'H1', 'tool': '2.5 mm hex'}]}))
    r = call(run, 'finish'); assert 'mechanical review' not in r.stdout + r.stderr, r.stdout + r.stderr
    # a different board (the hole moved): the review no longer counts
    moved = p / 'moved.kicad_pcb'; moved.write_text(HEAD + hole(16) + ')\n')
    d = data(run); d['boards'].append(str(moved)); (run / 'run.json').write_text(json.dumps(d))
    r = call(run, 'finish'); out = r.stdout + r.stderr
    assert r.returncode == 0 and 'mechanical review is of another board' in out, out

    # not declared: no review step, finish says nothing about reviews
    plain = p / 'plain'; empty = p / 'empty.json'; empty.write_text('{}')
    assert call(plain, 'intake', '--prompt-time', '2026-10-10T00:00:00Z', '--engine', 'test', '--kind', 'pattern').returncode == 0
    attach(plain, board, empty)
    assert call(plain, 'step', 'rules').returncode == 0
    assert data(plain)['sequence'] == ['intake', 'rules', 'models', 'outline', 'silkscreen', '3d', 'finish', 'export'], data(plain)['sequence']
    r = call(plain, 'finish'); open_items = data(plain)['summary']['open']; assert not any('review' in x.lower() for x in open_items), open_items
    # an undeclared review can still be run (advisory), it says finish does not need it
    r = call(plain, 'review', 'mechanical', '--input', str(mech), env=with_tool); assert r.returncode == 0 and 'Not declared' in r.stdout + r.stderr, r.stdout + r.stderr
    assert data(plain)['sequence'] == ['intake', 'rules', 'models', 'outline', 'silkscreen', '3d', 'finish', 'export']
print('PASS: declared review is a step before finish, missing/failing/stale reviews and a missing tool are refused, a passing bound review satisfies finish, nothing changes when undeclared')