AI Flow

"""Actual artifact CLI regression: same basename, manifest collision and immutable history."""
import hashlib,json,os,pathlib,subprocess,sys,tempfile
binary=str(pathlib.Path(sys.argv[1]).resolve())
with tempfile.TemporaryDirectory(prefix='aiflow-artifact-cli-') as td:
    p=pathlib.Path(td);run=p/'run';env=dict(os.environ,AIFLOW_DASHBOARD='off')
    def call(*args):
        return subprocess.run([binary,'--run',str(run),'--ai-thread','artifact-regression',*args],env=env,capture_output=True,text=True,timeout=30)
    assert call('intake','--prompt-time','2026-10-10T00:00:00Z','--engine','test').returncode==0
    def artifacts():
        return [r for r in map(json.loads,(run/'run.jsonl').read_text().splitlines()) if r.get('event')=='artifact']
    for name,body in [('a','first'),('b','second')]:
        (p/name).mkdir();src=p/name/'review.json';src.write_text(body)
        result=call('artifact','--file',str(src),'--step','intake','--kind','engineering-review','--caption',name)
        assert result.returncode==0,result.stderr
    first,second=artifacts()[-2:]
    assert first['file']!=second['file']
    assert pathlib.Path(first['file']).read_text()=='first'
    assert pathlib.Path(second['file']).read_text()=='second'
    assert first['sha256']==hashlib.sha256(b'first').hexdigest()
    (p/'a/review.json').write_text('new first')
    assert pathlib.Path(first['file']).read_text()=='first'
    src=p/'run.json';src.write_text('not a run manifest')
    assert call('artifact','--file',str(src),'--step','intake','--caption','manifest-name').returncode==0
    assert json.loads((run/'run.json').read_text())['engine']=='test'
    assert call('artifact','--file',str(p/'a/review.json'),'--step','intake','--caption','updated','--replaces',first['file']).returncode==0
    assert artifacts()[-1]['replaces']==first['file']
    corrupted=pathlib.Path(second['file']);corrupted.write_text('externally changed')
    before=len(artifacts());result=call('artifact','--file',str(p/'b/review.json'),'--step','intake','--caption','must refuse')
    assert result.returncode!=0 and 'existing file preserved' in result.stdout+result.stderr
    assert len(artifacts())==before and corrupted.read_text()=='externally changed'
    print('PASS: CLI preserves distinct reviews and run manifest, binds hashes, supports replacement, refuses corrupted copies.')