AI Flow

"""Issues #82, #83, #84 through the CLI: silkscreen-audit shows the checker's own error, project-page
refuses a stale parts record by name, the fields step is recorded as not applicable on a spec with
nothing to solve, and the pattern lane (intake --kind pattern) with its refusal for boards that need
routing. 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 "") (net 1 "GND") (net 2 "SIG") (layers (0 "F.Cu" signal) (31 "B.Cu" signal)) (gr_circle (center 10 10) (end 19 10) (layer "Edge.Cuts"))'
def pin(ref, x, net):
    n = {'GND': '(net 1 "GND")', 'SIG': '(net 2 "SIG")'}.get(net, '')
    return f' (footprint "MP" (layer "F.Cu") (at {x} 10) (property "Reference" "{ref}") (pad "1" thru_hole circle (at 0 0) (size 2 2) (drill 1) (layers "*.Cu") {n}))'

with tempfile.TemporaryDirectory(prefix='aiflow-board-kind-') as td:
    p = pathlib.Path(td); env = dict(os.environ, AIFLOW_DASHBOARD='off')
    def call(run, *args):
        return subprocess.run([binary, '--run', str(run), '--ai-thread', 'board-kind-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())
    spec = p / 'spec.json'; spec.write_text('{}')

    # --- #84 reply: the pattern lane -------------------------------------------------------------
    assert call(p / 'x', 'intake', '--prompt-time', '2026-10-10T00:00:00Z', '--engine', 'test', '--kind', 'pattern', '--eda', 'fusion').returncode != 0, 'pattern is KiCad only'
    run = p / 'pattern'
    r = call(run, 'intake', '--prompt-time', '2026-10-10T00:00:00Z', '--engine', 'test', '--kind', 'pattern'); assert r.returncode == 0, r.stderr
    routed = p / 'routed.kicad_pcb'; routed.write_text(HEAD + pin('MP1', 5, 'GND') + pin('MP2', 3, 'GND') + pin('J1', 15, 'SIG') + pin('J2', 17, 'SIG') + ')\n')
    attach(run, routed, spec)
    r = call(run, 'step', 'rules'); assert r.returncode == 0 and '2 nets' in r.stdout + r.stderr, (r.stdout, r.stderr)
    # #86: a carrier plate, 4 machine pins on a GND pour and 8 one-pad unconnected-(SDn-Pad1) seats
    nets = ''.join(f' (net {i + 1} "unconnected-(SD{i}-Pad1)")' for i in range(1, 9))
    seat = lambda ref, x, n, name: f' (footprint "MC" (layer "F.Cu") (at {x} 12) (property "Reference" "{ref}") (pad "1" smd rect (at 0 0) (size 1 1) (layers "F.Cu") (net {n} "{name}")))'
    seats = ''.join(seat(f'SD{i}', 2 + 2 * i, i + 1, f'unconnected-(SD{i}-Pad1)') for i in range(1, 9))
    pins = ''.join(pin(f'MP{i}', x, 'GND') for i, x in enumerate([3, 7, 13, 17], 1))
    zone = ' (zone (net 1) (net_name "GND") (layer "B.Cu") (uuid "z1") (polygon (pts (xy 1 1) (xy 19 1) (xy 19 19) (xy 1 19))))'
    carrier = p / 'carrier.kicad_pcb'; carrier.write_text(HEAD.replace('(net 2 "SIG")', nets) + pins + seats + zone + ')\n')
    attach(run, carrier, spec)
    r = call(run, 'step', 'rules'); assert r.returncode == 0, (r.stdout, r.stderr)
    carrier.write_text(HEAD.replace('(net 2 "SIG")', nets) + pins + seats + seat('SD9', 18, 4, 'unconnected-(SD3-Pad1)') + zone + ')\n')
    r = call(run, 'step', 'rules'); out = r.stdout + r.stderr
    assert r.returncode == 0 and '2 nets' in out and 'unconnected-(SD3-Pad1)' in out, out
    disc = p / 'disc.kicad_pcb'; disc.write_text(HEAD + ' (gr_circle (center 10 10) (end 14 10) (layer "F.Cu") (width 0.5)))\n')
    attach(run, disc, spec)
    r = call(run, 'step', 'rules'); assert r.returncode == 0, (r.stdout, r.stderr)
    assert data(run)['sequence'] == ['intake', 'rules', 'models', 'outline', 'silkscreen', '3d', 'finish', 'export'], data(run)['sequence']
    r = call(run, 'finish'); out = r.stdout + r.stderr; assert r.returncode == 0 and 'open item' in out, out
    assert 'native DRC' in out and 'land route' not in out and 'analyze-current' not in out and 'measure (' not in out, out
    # a run stuck at `rules` on the electrical sequence switches once requirements say boardKind pattern
    stuck = p / 'stuck'; assert call(stuck, 'intake', '--prompt-time', '2026-10-10T00:00:00Z', '--engine', 'test').returncode == 0
    attach(stuck, disc, spec)
    assert call(stuck, 'step', 'rules').returncode == 0 and 'routing' in data(stuck)['sequence']
    req = p / 'requirements.json'; req.write_text(json.dumps({'boardKind': 'pattern'}))
    d = data(stuck); d['requirements'] = str(req); (stuck / 'run.json').write_text(json.dumps(d))
    r = call(stuck, 'step', 'rules'); assert r.returncode == 0, r.stderr
    assert data(stuck)['sequence'][2] == 'models', data(stuck)['sequence']

    # --- #84 part 1: fields not applicable ---------------------------------------------------------
    plate = p / 'plate.kicad_pcb'
    plate.write_text(HEAD + pin('MP1', 5, 'GND') + pin('MP2', 15, 'GND') + ' (zone (net 1) (net_name "GND") (layer "F.Cu") (uuid "z1") (polygon (pts (xy 1 1) (xy 19 1) (xy 19 19) (xy 1 19)))))\n')
    run = p / 'fields'; assert call(run, 'intake', '--prompt-time', '2026-10-10T00:00:00Z', '--engine', 'test').returncode == 0
    attach(run, plate, spec)
    r = call(run, 'tour', 'fields'); assert r.returncode == 0 and 'not applicable' in r.stdout, (r.stdout, r.stderr)
    o = data(run)['outcomes']['tour-fields']; assert o['notApplicable'] and o['reason'] == 'no loads, hot parts or signal checks in the spec', o
    loaded = p / 'loaded.json'; loaded.write_text(json.dumps({'loads': {'GND': {'amps': 1}}}))
    attach(run, plate, loaded)
    # (a loaded spec goes on to the real Fields walkthrough, which needs the desktop: unit-tested in boardkind.rs)
    r = call(run, 'finish'); assert 'fields: recorded as not applicable, but now spec.loads names loads' in r.stdout + r.stderr, r.stdout + r.stderr

    # --- #82: silkscreen-audit shows the checker's error --------------------------------------------
    repro = p / 'audit.json'
    repro.write_text(json.dumps({"obstacles": [{"id": "R1:body", "kind": "body", "layer": "F.SilkS", "boxes": [[0, 0, 1, 1]]}],
                                 "inventory": [{"id": "R1:body", "kind": "body", "layers": ["F.SilkS"]}],
                                 "marks": [{"id": "R1", "layer": "F.SilkS", "text": "R1", "box": [2, 2, 3, 3]}]}))
    r = call(run, 'silkscreen-audit', '--input', str(repro), '--out', str(p / 'audit-out.json')); assert r.returncode == 0, (r.stdout, r.stderr)
    bad = json.loads(repro.read_text()); del bad['obstacles'][0]['boxes']; repro.write_text(json.dumps(bad))
    r = call(run, 'silkscreen-audit', '--input', str(repro), '--out', str(p / 'audit-out.json'))
    assert r.returncode != 0 and "obstacle 'R1:body' has neither 'box' nor 'boxes'" in r.stdout + r.stderr, (r.stdout, r.stderr)

    # --- #83 part 1: a stale parts record is named -------------------------------------------------
    bom = p / 'bom.csv'; bom.write_text('Ref,MPN\nMP1,PIN-1\n')
    d = data(run); d['outcomes']['parts'] = {'pass': True, 'bom': str(bom), 'parts': [{'mpn': 'PIN-1', 'refs': ['MP1'], 'wiki': 'adom/pin-1'}, {'mpn': 'RC0402FR-0733KL', 'refs': ['R9'], 'wiki': 'adom/rc0402'}]}
    (run / 'run.json').write_text(json.dumps(d))
    r = call(run, 'project-page', '--page', 'someone/board', '--no-push'); out = r.stdout + r.stderr
    assert r.returncode != 0 and 'part RC0402FR-0733KL is in the parts record but not in the BOM' in out and 'run `parts` again' in out.replace('Run `parts` again', 'run `parts` again'), out
print('PASS: pattern lane chosen and refused for a routed board, one-pad unconnected nets are no nets, finish skips the electrical gates and asks for live DRC; fields n/a recorded and refused at finish once the spec names a load; silkscreen-audit names the bad item; stale parts record named')