AI Flow

"""Issue #193 end to end, with a stand-in adom-wiki on PATH (no wiki call):
part-request records attempts and turns away the same inputs, --refresh reads the factory's answers,
parts uses a page built under the orderable number, supply-risk reports what the assessment leaves out,
--choose records the human's pick, and parts leads with each decision.
Run: python3 tools/test_supply_risk.py [path/to/adom-aiflow]"""
import json, os, stat, subprocess, sys, tempfile, unittest
from pathlib import Path

BIN = sys.argv[1] if len(sys.argv) > 1 else str(Path(__file__).resolve().parent.parent / 'bin' / 'adom-aiflow')

FAKE = r'''#!/usr/bin/env python3
import json, sys
a = sys.argv[1:]
if a[:1] == ['search']:
    print(json.dumps({"data": {"results": []}})); sys.exit(0)
if a[:2] == ['repo', 'show']:
    if a[2] == 'adom/r9x-1':
        print(json.dumps({"slug": "r9x-1", "component": {"mpn": "R9X-1"}})); sys.exit(0)
    sys.exit(1)
if a[:2] == ['part', 'request']:
    mpn = a[2]
    ans = {"U7X": {"state": "cannot_build", "reason": "The datasheet has no land-pattern drawing."},
           "U7X-T": {"state": "cannot_build", "reason": "No distributor sells this exact part number."},
           "R9X": {"state": "cannot_build", "reason": "Stopped at the 3D model step."},
           "R9X-1": {"state": "queued", "status_url": "/part-request/r9"}}[mpn]
    print(json.dumps({"status": "ok", "data": dict(ans, mpn=mpn)})); sys.exit(0)
if a[:2] == ['part', 'status']:
    print(json.dumps({"status": "ok", "data": {"requests": [
        {"mpn": "R9X-1", "state": "built", "page_url": "/adom/r9x-1", "status_url": "/part-request/r9", "requested_at": "2026-11-10 10:00:00"}]}}))
    sys.exit(0)
sys.exit(2)
'''


class SupplyRisk(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix='aiflow-supply-'))
        fake = self.tmp / 'bin' / 'adom-wiki'
        fake.parent.mkdir()
        fake.write_text(FAKE)
        fake.chmod(fake.stat().st_mode | stat.S_IEXEC)
        run = self.tmp / 'run'
        run.mkdir()
        (run / 'run.json').write_text(json.dumps({"version": 1, "id": "supplytest", "outcomes": {}, "turns": []}))
        (self.tmp / 'bom.csv').write_text('Ref,MPN\nU7,U7X\nR9,R9X\n')
        self.env = dict(os.environ, PATH=f"{fake.parent}:{os.environ['PATH']}", AIFLOW_RUN=str(run), AIFLOW_DASHBOARD='off')

    def aiflow(self, *args):
        p = subprocess.run([BIN, '--ai-thread', 'test', *args], cwd=self.tmp, env=self.env, capture_output=True, text=True, stdin=subprocess.DEVNULL)
        return p.stdout + p.stderr

    def test_retry_assess_choose(self):
        out = self.aiflow('parts', '--bom', 'bom.csv')
        self.assertIn('U7 (U7X): no page request yet', out)
        self.assertIn('part-request U7X', out)

        out = self.aiflow('part-request', 'U7X')
        self.assertIn('attempt 1 for U7X: cannot_build', out)
        self.assertIn('retry once', out)
        out = self.aiflow('part-request', 'U7X')
        self.assertIn('ERROR: not sent: attempt 1 for U7X sent these same inputs', out)
        out = self.aiflow('part-request', 'U7X', '--as', 'U7X-T')
        self.assertIn('attempt 2 for U7X (sent as U7X-T), changed: part number: cannot_build', out)
        self.assertIn('supply-risk', out)

        # R9X: refused, the retry under the orderable number is queued, then built
        self.aiflow('part-request', 'R9X')
        out = self.aiflow('part-request', 'R9X', '--as', 'R9X-1')
        self.assertIn('queued', out)
        out = self.aiflow('part-request', '--refresh')
        self.assertIn('1 request(s) changed state', out)

        design = self.tmp / 'design'
        design.mkdir()
        fig = lambda v: {"value": v, "source": "https://www.mouser.com/u7x", "date": "2026-11-10"}
        (design / 'supply-risk.json').write_text(json.dumps({"parts": [{
            "mpn": "U7X", "risk": "high", "lifecycle": fig("NRND"), "leadTime": fig("26 weeks"), "package": fig("maker-only QFN"),
            "datasheetPublished": fig("yes"), "landPatternPublished": {"value": "no", "source": "https://maker/u7x"},
            "distributors": [{"name": "Mouser", "stock": 300, "source": "https://www.mouser.com/u7x", "date": "2026-11-10"}], "secondSources": [],
            "alternatives": [{"mpn": "ALT1", "lifecycle": "active", "stock": 12000, "source": "https://www.digikey.com/alt1", "date": "2026-11-10", "wikiPage": None, "boardChanges": "QFN-20 to SOT-23-6 footprint"}]}]}))
        out = self.aiflow('supply-risk')
        self.assertIn('FINDINGS', out)
        self.assertIn('U7X: land pattern published: no date', out)
        self.assertIn('U7 (U7X): refused twice (The datasheet has no land-pattern drawing; No distributor sells this exact part number); supply risk high (NRND, 1 distributor); alternatives ALT1', out)
        self.assertIn('ask the human', out)

        out = self.aiflow('supply-risk', '--choose', 'U7X', '--pick', 'ALT1', '--why', 'Colby: take the multi-sourced part')
        self.assertIn("the human's choice for U7X: ALT1", out)
        self.assertIn('put ALT1 in the BOM', out)

        out = self.aiflow('parts', '--bom', 'bom.csv')
        self.assertIn('1 of 2 parts have no Adom wiki component page', out)
        self.assertIn('chosen: ALT1', out)
        self.assertIn('adom/r9x-1 was built by the part factory on attempt 2 (requested as R9X-1)', out)
        rec = json.loads((self.tmp / 'run' / 'run.json').read_text())
        self.assertEqual(len(rec['partRequests']['U7X']), 2)
        self.assertEqual(rec['outcomes']['parts']['decisions'][0]['supplyRisk'], 'high')
        self.assertEqual(rec['supplyChoices']['U7X']['pick'], 'ALT1')


if __name__ == '__main__':
    unittest.main(argv=sys.argv[:1])