AI Flow

"""`sourcing check` through the CLI: any distributor passes on any profile as consigned parts, with one
warning and a suggestion per row. The lookup order per distinct MPN (issue #111): the wiki component
page first, then the pnp-inventory vendor stock cache, and a live adom-parts-search only for what those
did not answer. A fake adom-wiki (two component pages) and a fake adom-parts-search on PATH return
fixed JSON and log each call, so the test never touches the wiki or spends the shared vendor quota. It
proves: a page with the number and a dated stock count answers with no live search; a page without the
preferred source's number falls back to the live search; a number with no stock is "stock not
recorded"; both caches in the run folder (a rerun reads and searches nothing); AIFLOW_NO_PART_SEARCH=1
(no live search, pages still read) and AIFLOW_NO_PART_LOOKUP=1 (nothing). pnp-inventory is not on PATH,
so that step is skipped quietly. No CAD, desktop or dashboard."""
import json, os, pathlib, subprocess, sys, tempfile, datetime
binary = str(pathlib.Path(sys.argv[1]).resolve())

SEARCH = r'''#!/usr/bin/env python3
import json, os, sys
open(os.environ["FAKE_LOG"], "a").write(" ".join(sys.argv[1:]) + "\n")
mpn = sys.argv[2]
mouser = {"TPS54202DDCR": [{"mpn": "TPS54202DDCR", "vendor_pn": "595-TPS54202DDCR", "stock": 1200}],
          "AP2112K-3.3TRG1": [{"mpn": "AP2112K-3.3TRG1", "vendor_pn": "621-AP2112K-3.3TRG1", "stock": 5000}]}.get(mpn, [])
jlc = {"TPS54202DDCR": [{"mpn": "TPS54202DDCR", "vendor_pn": "C191884", "stock": 299988, "flags": ["extended"]}]}.get(mpn, [])
print(json.dumps({"query": mpn, "mouser": {"ok": True, "hits": mouser}, "digikey": {"ok": True, "hits": []}, "jlcpcb": {"ok": True, "hits": jlc}}))
'''

# page.json the way real component pages store distributor numbers: a string, {part, url, stock, checked}, or false
WIKI = r'''#!/usr/bin/env python3
import json, os, sys
open(os.environ["FAKE_WIKI_LOG"], "a").write(" ".join(sys.argv[1:]) + "\n")
pages = {
  "adom/tps54202ddcr": {"type": "component", "component": {"mpn": "TPS54202DDCR", "distributor_links": {
      "mouser": {"part": "595-TPS54202DDCR", "url": "https://www.mouser.com/x", "stock": 160, "checked": "2026-10-20"},
      "digikey": "296-TPS54202DDCRTR-ND", "lcsc": False}}},
  "adom/ap2112k-3-3trg1": {"type": "component", "component": {"mpn": "AP2112K-3.3TRG1", "distributor_links": {
      "mouser": "621-AP2112K-3.3TRG1",
      "lcsc": {"part": "C51118", "url": "https://www.lcsc.com/x", "stock": 12535, "checked": "2026-10-19", "tier": "basic"}}}},
}
if sys.argv[1:3] == ["repo", "show"] and sys.argv[4:] == ["page.json"] and sys.argv[3] in pages:
    print(json.dumps(pages[sys.argv[3]])); sys.exit(0)
print(json.dumps({"http_status": 404, "response": {"error": "Page not found", "code": "PAGE_NOT_FOUND"}}))
print("error [PAGE_NOT_FOUND]: no such page", file=sys.stderr); sys.exit(1)
'''

today = datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')

with tempfile.TemporaryDirectory(prefix='aiflow-sourcing-') as td:
    p = pathlib.Path(td)
    fakebin = p / 'bin'; fakebin.mkdir()
    for name, body in (('adom-parts-search', SEARCH), ('adom-wiki', WIKI)):
        f = fakebin / name; f.write_text(body); f.chmod(0o755)
    log, wlog = p / 'searches.log', p / 'wiki.log'
    # PATH: the fakes, then only what python3 needs (no real adom-wiki or pnp-inventory)
    py = pathlib.Path(sys.executable).parent
    env = dict(os.environ, AIFLOW_DASHBOARD='off', PATH=f"{fakebin}{os.pathsep}{py}{os.pathsep}/usr/bin{os.pathsep}/bin", FAKE_LOG=str(log), FAKE_WIKI_LOG=str(wlog))
    env.pop('AIFLOW_NO_PART_SEARCH', None); env.pop('AIFLOW_NO_PART_LOOKUP', None)
    def call(run, *args, extra=None):
        return subprocess.run([binary, '--run', str(run), '--ai-thread', 'sourcing-test', *args], env=dict(env, **(extra or {})), capture_output=True, text=True, timeout=120)
    def outcome(run): return json.loads((run / 'run.json').read_text())['outcomes']['sourcing']
    def searches(): return log.read_text().splitlines() if log.exists() else []
    def reads(): return wlog.read_text().splitlines() if wlog.exists() else []

    bom = p / 'bom.csv'
    bom.write_text('ref,value,mpn,manufacturer,source,vendor_pn,package,stock_checked_2026-10-27\n'
                   'U1,buck,TPS54202DDCR,TI,DigiKey,296-TPS54202DDCRTR-ND,SOT-23-6,900\n'
                   'U2,buck,TPS54202DDCR,TI,DigiKey,296-TPS54202DDCRTR-ND,SOT-23-6,900\n'
                   'U3,op amp,ABC123,Acme,Arrow,ABC123,SOIC-8,800\n'
                   'U4,LDO,AP2112K-3.3TRG1,Diodes,DigiKey,AP2112K-3.3TRG1DITR-ND,SOT-23-5,700\n'
                   'R1,10k,CR0402-FX-1002GLF,Bourns,Adom stock,,0402,5000\n')
    run = p / 'run'
    r = call(run, 'intake', '--prompt-time', '2026-10-27T00:00:00Z', '--engine', 'test'); assert r.returncode == 0, r.stdout + r.stderr

    # AIFLOW_NO_PART_LOOKUP=1: no page read, no search, one note, still a pass
    r = call(run, 'sourcing', 'check', '--bom', str(bom), '--qty', '1', extra={'AIFLOW_NO_PART_LOOKUP': '1'})
    assert r.returncode == 0, r.stdout + r.stderr
    assert 'part suggestions skipped (AIFLOW_NO_PART_LOOKUP=1)' in r.stdout and not searches() and not reads(), r.stdout
    o = outcome(run); assert o['consignedRows']['count'] == 4 and o['suggestions'] == [], o

    # AIFLOW_NO_PART_SEARCH=1: the wiki pages still answer, no live search
    r = call(run, 'sourcing', 'check', '--bom', str(bom), '--qty', '1', extra={'AIFLOW_NO_PART_SEARCH': '1'})
    assert r.returncode == 0, r.stdout + r.stderr
    assert not searches(), searches()
    # one read per distinct MPN: AP2112K-3.3TRG1 tries ap2112k-3.3trg1 first, then the wiki slug form
    assert reads() == ['repo show adom/tps54202ddcr page.json', 'repo show adom/abc123 page.json',
                       'repo show adom/ap2112k-3.3trg1 page.json', 'repo show adom/ap2112k-3-3trg1 page.json'], reads()
    assert 'suggestion: U1 (TPS54202DDCR, DigiKey): Mouser has it, 595-TPS54202DDCR, 160 in stock (wiki page adom/tps54202ddcr, checked 2026-10-20); sourcing it there saves a consignment shipment' in r.stdout, r.stdout
    assert 'suggestion: U2 (TPS54202DDCR, DigiKey): Mouser has it, 595-TPS54202DDCR, 160 in stock (wiki page adom/tps54202ddcr' in r.stdout, r.stdout
    assert 'suggestion: U4 (AP2112K-3.3TRG1, DigiKey): Mouser has it, 621-AP2112K-3.3TRG1 (wiki page adom/ap2112k-3-3trg1), stock not recorded: `adom-parts-search show "AP2112K-3.3TRG1"` checks it live; sourcing it there saves a consignment shipment' in r.stdout, r.stdout
    assert 'live part searches skipped (AIFLOW_NO_PART_SEARCH=1) for 2 MPN(s) the wiki pages and the pnp-inventory cache did not fully answer: ABC123, AP2112K-3.3TRG1' in r.stdout, r.stdout
    assert 'suggestion: U3' not in r.stdout, r.stdout  # nothing known, nothing claimed
    o = outcome(run)
    assert o['suggestionLookups']['wikiReads'] == 3 and o['suggestionLookups']['wikiHits'] == 2 and o['suggestionLookups']['liveSearches'] == 0, o['suggestionLookups']
    assert (run / 'sourcing-lookup-cache.json').is_file()

    # fab with live search: the page answers TPS54202DDCR, so only the rest is searched
    r = call(run, 'sourcing', 'check', '--bom', str(bom), '--qty', '1')
    assert r.returncode == 0, r.stdout + r.stderr
    assert len(reads()) == 4, reads()  # the page answers came from the run's cache
    assert sum('warning:' in l and 'consigned parts' in l for l in r.stdout.splitlines()) == 1, r.stdout
    assert searches() == ['search ABC123', 'search AP2112K-3.3TRG1'], searches()
    assert 'suggestion: U1 (TPS54202DDCR, DigiKey): Mouser has it, 595-TPS54202DDCR, 160 in stock (wiki page adom/tps54202ddcr, checked 2026-10-20)' in r.stdout, r.stdout
    assert f'suggestion: U4 (AP2112K-3.3TRG1, DigiKey): Mouser has it, 621-AP2112K-3.3TRG1, 5000 in stock (live search, checked {today}); sourcing it there saves a consignment shipment' in r.stdout, r.stdout
    assert 'suggestion: U3 (ABC123, Arrow): no exact match at Mouser (live search); `adom-parts-search show "op amp SOIC-8"` to look for an equivalent, or send this part as consigned' in r.stdout, r.stdout
    lines = r.stdout.splitlines()
    w = next(i for i, l in enumerate(lines) if 'warning:' in l and 'consigned parts' in l)
    assert 'suggestion:' in lines[w + 1], lines[w:w + 3]
    o = outcome(run)
    assert o['pass'] and o['consignedRows']['count'] == 4 and len(o['suggestions']) == 4, o
    assert o['suggestionLookups']['wikiFromCache'] == 3 and o['suggestionLookups']['liveSearches'] == 2, o['suggestionLookups']
    assert {s['refs']: s['from'] for s in o['suggestions']} == {'U1': 'wiki', 'U2': 'wiki', 'U3': 'search', 'U4': 'search'}, o['suggestions']
    assert (run / 'sourcing-search-cache.json').is_file()

    # a rerun reads both caches: no page read, no search
    r = call(run, 'sourcing', 'check', '--bom', str(bom), '--qty', '1')
    assert r.returncode == 0 and len(searches()) == 2 and len(reads()) == 4, (searches(), reads())

    # jlcpcb: the TPS54202DDCR page has no LCSC number (false), so it falls back to the live search;
    # the AP2112K page has its LCSC number and dated stock, so no search for it
    r = call(run, 'sourcing', 'check', '--bom', str(bom), '--qty', '1', '--profile', 'jlcpcb')
    assert r.returncode == 0, r.stdout + r.stderr
    assert searches()[2:] == ['search TPS54202DDCR', 'search CR0402-FX-1002GLF'], searches()
    assert reads()[4:] == ['repo show adom/cr0402-fx-1002glf page.json'], reads()
    assert 'are not JLCPCB/LCSC parts' in r.stdout, r.stdout
    assert f'U1 (TPS54202DDCR, DigiKey): LCSC has it, C191884 (extended), 299988 in stock (live search, checked {today}); add it to the lcsc column to save a consignment shipment' in r.stdout, r.stdout
    assert 'U4 (AP2112K-3.3TRG1, DigiKey): LCSC has it, C51118 (basic), 12535 in stock (wiki page adom/ap2112k-3-3trg1, checked 2026-10-19); add it to the lcsc column to save a consignment shipment' in r.stdout, r.stdout
    assert outcome(run)['consignedRows']['count'] == 5  # the Adom stock row is consigned on jlcpcb

    # a real problem still fails: no MPN
    bad = p / 'bad.csv'; bad.write_text('ref,mpn,source,stock_checked_2026-10-27\nU1,,DigiKey,10\n')
    r = call(run, 'sourcing', 'check', '--bom', str(bad), '--qty', '1')
    assert r.returncode == 0 and 'FINDINGS' in r.stdout and 'U1: no MPN' in r.stdout, r.stdout
print('sourcing-suggestions: ok')