AI Flow

"""Issue #141: the last route's numbers and every command that ended with findings or an error show
in `status`, in run.json and on the timeline's data endpoint (GET /state), and stay there across a
board change until the command runs again. A small synthetic board whose net J7-J8 cannot be routed
(J8 sits outside the outline), a fake kicad-cli that reports one unconnected item on the routed board;
no dashboard, no bridge. Usage: python3 tests/attention-workflow.py <adom-aiflow binary>"""
import json, os, pathlib, socket, subprocess, sys, tempfile, time, urllib.request

binary = str(pathlib.Path(sys.argv[1]).resolve())


def board_text(shift=0.0, j8=(40, 40)):
    pads = [("J1", "A", 2 + shift, 2), ("J2", "A", 6, 2), ("J3", "B", 2, 8), ("J4", "B", 18, 8), ("J7", "C", 10, 5), ("J8", "C", *j8)]
    t = '(kicad_pcb\n\t(version 20260206)\n\t(generator "test")\n\t(layers\n\t\t(0 "F.Cu" signal)\n\t\t(2 "B.Cu" signal)\n\t\t(25 "Edge.Cuts" user)\n\t\t(31 "F.CrtYd" user)\n\t)\n'
    for r, net, x, y in pads:
        t += f'\t(footprint "T:{r}"\n\t\t(layer "F.Cu")\n\t\t(at {x} {y})\n\t\t(property "Reference" "{r}")\n\t\t(fp_rect\n\t\t\t(start -0.9 -0.9)\n\t\t\t(end 0.9 0.9)\n\t\t\t(stroke\n\t\t\t\t(width 0.05)\n\t\t\t\t(type solid)\n\t\t\t)\n\t\t\t(fill no)\n\t\t\t(layer "F.CrtYd")\n\t\t)\n\t\t(pad "1" thru_hole circle\n\t\t\t(at 0 0)\n\t\t\t(size 1.2 1.2)\n\t\t\t(drill 0.6)\n\t\t\t(layers "*.Cu")\n\t\t\t(net "{net}")\n\t\t)\n\t)\n'
    t += '\t(gr_rect\n\t\t(start 0 0)\n\t\t(end 20 10)\n\t\t(layer "Edge.Cuts")\n\t)\n)\n'
    return t


with tempfile.TemporaryDirectory(prefix='aiflow-attention-') as td:
    p = pathlib.Path(td)
    (p / 'b.kicad_pcb').write_text(board_text())
    (p / 'b2.kicad_pcb').write_text(board_text(0.5))
    (p / 'b3.kicad_pcb').write_text(board_text(0.5, (14, 5)))
    (p / 'spec.json').write_text(json.dumps({"board": "attention fixture", "clearance": 0.2}))
    fake = p / 'kicad-cli'
    # one unconnected item on the routed board only, so the gate ends with findings
    fake.write_text('#!/usr/bin/env python3\nimport sys, json\na = sys.argv[1:]\n'
                    'if "--help" in a: print("--refill-zones"); sys.exit(0)\n'
                    'out = a[a.index("--output") + 1]\nrouted = any("routed" in x for x in a)\n'
                    'json.dump({"violations": [], "unconnected_items": ([{"type": "unconnected_items", "description": "Missing connection between items", "items": [{"description": "PTH pad 1 [C] of J8"}]}] if routed else [])}, open(out, "w"))\n')
    fake.chmod(0o755)
    run = p / 'run'
    env = dict(os.environ, AIFLOW_DASHBOARD='off', ADOM_AIFLOW_KICAD_CLI=str(fake))
    env.pop('KICAD_REMOTE_TARGET', None)

    def call(*args):
        return subprocess.run([binary, '--run', str(run), '--ai-thread', 'attention-test', *args], env=env, capture_output=True, text=True, timeout=180)

    def data():
        return json.loads((run / 'run.json').read_text())

    def timeline():
        # with the dashboard off nothing refreshes the file by itself: `timeline write` does
        assert call('timeline', 'write').returncode == 0
        return json.loads((run / 'timeline' / 'timeline.json').read_text())

    r = call('start', '--board', str(p / 'b.kicad_pcb'), '--spec', str(p / 'spec.json'), '--engine', 'test')
    assert r.returncode == 0, r.stdout + r.stderr
    # route ends with findings (J8 is off the board): exit 0, never a refusal
    r = call('route', '--passes', '2')
    assert r.returncode == 0, r.stdout + r.stderr
    assert r.stdout.startswith('FINDINGS'), r.stdout
    d = data()
    lr = d['lastRoute']
    assert lr['unrouted'] >= 1 and lr['connections'] >= lr['unrouted'] and lr['routed'] == lr['connections'] - lr['unrouted'], lr
    assert 'C' in lr['netsUnrouted'] and lr['vias'] is not None and lr['boardRevision'] == 1, lr
    chk = d['checks']['route']
    assert chk['kind'] == 'findings' and chk['step'] == 'routing' and chk['count'] >= 1 and 'routing closed' in chk['lines'][0], chk
    led = [json.loads(l) for l in (run / 'run.jsonl').read_text().splitlines()]
    assert any(e.get('event') == 'command-outcome' and e['key'] == 'route' and e['outcome'] == 'findings' for e in led)
    # an error (could not run) is recorded as an error, distinct from findings
    r = call('gate', '--plan', str(p / 'missing-plan.json'))
    assert r.returncode != 0 and data()['checks']['gate']['kind'] == 'error', r.stdout
    # the gate on the route's plan ends with findings and replaces the error
    r = call('gate')
    assert r.returncode == 0 and r.stdout.startswith('FINDINGS'), r.stdout + r.stderr
    d = data()
    assert d['checks']['gate']['kind'] == 'findings' and d['lastGate']['unconnected'] == 1, (d['checks'], d.get('lastGate'))
    # status names the numbers and the open checks; a status call itself clears nothing
    r = call('status')
    if os.environ.get('SHOW'): print(r.stdout)
    assert r.returncode == 0, r.stdout + r.stderr
    assert 'last route' in r.stdout and f"{lr['routed']} of {lr['connections']} connection(s) routed, {lr['unrouted']} unrouted" in r.stdout, r.stdout
    assert 'measured on the current board' in r.stdout and 'needs a look (2 command(s)' in r.stdout and '[findings] `route`' in r.stdout, r.stdout
    assert 'needs a look (' in r.stdout.split('Hint:', 1)[1], 'the step table names the routing step'
    # the timeline: the routing step carries the badge, the route card the numbers
    tl = timeline()
    if os.environ.get('SHOW'): print(json.dumps({k: tl[k] for k in ('route', 'checks')}, indent=1), [(s['name'], s['attention'] and s['attention']['kind']) for s in tl['steps']])
    routing = [s for s in tl['steps'] if s['name'] == 'routing']
    assert routing, [s['name'] for s in tl['steps']]
    if routing:
        a = routing[0]['attention']
        assert a and a['kind'] == 'findings' and a['count'] >= 2 and {i['key'] for i in a['items']} == {'route', 'gate'}, a
    assert tl['route']['route']['unrouted'] == lr['unrouted'] and tl['route']['route']['onCurrentBoard'] is True, tl['route']
    assert {c['key'] for c in tl['checks']} == {'route', 'gate'}, tl['checks']
    # the board changes: outcomes are voided, the route card stays, labelled with its board
    r = call('adopt-board', '--board', str(p / 'b2.kicad_pcb'), '--evidence', 'test: moved J1 by 0.5 mm')
    assert r.returncode == 0, r.stdout + r.stderr
    d = data()
    assert d['lastRoute'] == lr, 'a new board keeps the last route record'
    r = call('status')
    assert 'previous board' in r.stdout and f"{lr['unrouted']} unrouted" in r.stdout and '[findings] `route`' in r.stdout, r.stdout
    tl = timeline()
    assert tl['route']['route']['onCurrentBoard'] is False and 'previous board' in tl['route']['route']['measuredOn'], tl['route']
    # the data endpoint serves the same: GET /state from `timeline serve`
    with socket.socket() as sock:
        sock.bind(('127.0.0.1', 0)); port = sock.getsockname()[1]
    proc = subprocess.Popen([binary, '--run', str(run), 'timeline', 'serve', '--port', str(port)], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        for _ in range(100):
            try: state = json.load(urllib.request.urlopen(f'http://127.0.0.1:{port}/state', timeout=3)); break
            except OSError: time.sleep(.05)
        else: raise AssertionError('server did not start')
        st = state['timeline']
        assert st['route']['route']['unrouted'] == lr['unrouted'] and st['route']['route']['onCurrentBoard'] is False, st['route']
        assert any(c['key'] == 'route' and c['kind'] == 'findings' for c in st['checks']), st['checks']
    finally:
        proc.terminate(); proc.wait(timeout=5)
    # the next route replaces the record and, ending with findings again, keeps the check; a clean
    # run of a command clears its own entry only
    r = call('route', '--passes', '2')
    d = data()
    assert r.returncode == 0 and d['lastRoute']['boardRevision'] == 2 and d['lastRoute']['board'] != lr['board'], d['lastRoute']
    assert d['checks']['route']['kind'] == 'findings', d['checks']
    tl = timeline()
    assert tl['route']['route']['onCurrentBoard'] is True, tl['route']
    # J8 moved onto the board: the route closes, its check clears; the gate's stays until the gate runs
    assert call('adopt-board', '--board', str(p / 'b3.kicad_pcb'), '--evidence', 'test: J8 onto the board').returncode == 0
    r = call('route', '--passes', '4')
    assert r.returncode == 0 and r.stdout.startswith('OK'), r.stdout + r.stderr
    d = data()
    assert d['lastRoute']['unrouted'] == 0 and 'route' not in d['checks'] and 'gate' in d['checks'], (d['lastRoute'], d['checks'])
    r = call('status')
    assert 'last route' in r.stdout and ', 0 unrouted;' in r.stdout and '`route`' not in r.stdout and '`gate`' in r.stdout, r.stdout
    print('PASS: route findings, gate error then findings, status lines, step badge, route card, board change labelled "previous board", GET /state carries them, a clean route clears its check only')