AI Flow

#!/usr/bin/env python3
"""Read-only wiki PR preflight. A passing report is not native acceptance."""
import argparse
import difflib
import hashlib
import json
from pathlib import Path
import re
import subprocess
import tempfile


def inspect(path, before, after):
    problems = []
    old, new = before.splitlines(True), after.splitlines(True)
    if path.endswith('.rs'):
        # Conservative lexical inventory; production compilation is still mandatory.
        pattern = r'(?m)^\s*(?:pub(?:\([^)]*\))?\s+)?(?:async\s+)?(fn|mod)\s+(\w+)'
        removed = set(re.findall(pattern, before)) - set(re.findall(pattern, after))
        if removed:
            problems.append('removed Rust declarations: ' + repr(sorted(removed)))
    if path.endswith('bridge.json') and before:
        removed = set(json.loads(before).get('verbs', [])) - set(json.loads(after).get('verbs', []))
        if removed:
            problems.append('removed manifest verbs: ' + repr(sorted(removed)))
    deleted = sum(i2-i1 for op,i1,i2,j1,j2 in difflib.SequenceMatcher(None,old,new,autojunk=False).get_opcodes() if op in ('delete','replace'))
    if len(old) >= 30 and deleted / len(old) > 0.5:
        problems.append('more than half of existing lines replaced/deleted; split or review separately')
    diff = ''.join(difflib.unified_diff(old,new,fromfile='a/'+path,tofile='b/'+path,n=3))
    return problems, diff


def wiki(*args):
    p = subprocess.run(['adom-wiki', *args, '--json'], capture_output=True, text=True)
    if p.returncode:
        raise RuntimeError(p.stdout + p.stderr)
    return json.loads(p.stdout)['data']


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--repo', required=True)
    p.add_argument('--base-sha', required=True)
    p.add_argument('--base-dir', type=Path, required=True)
    p.add_argument('--work-dir', type=Path, required=True)
    p.add_argument('--out', type=Path, required=True)
    p.add_argument('--files', nargs='+', required=True)
    a = p.parse_args()
    if not re.fullmatch(r'[0-9a-f]{40}', a.base_sha):
        p.error('--base-sha must be a full immutable commit hash')
    head = lambda: wiki('repo','log',a.repo,'--limit','1')['log'][0]['hash']
    if head() != a.base_sha:
        raise SystemExit('REFUSED: live HEAD moved; rebase and rerun combined checks')
    problems, diffs, hashes = [], [], {}
    with tempfile.TemporaryDirectory() as tmp:
        for i, path in enumerate(a.files):
            rel = Path(path)
            if rel.is_absolute() or '..' in rel.parts:
                raise SystemExit('REFUSED: paths must stay inside the working copy')
            base, work = a.base_dir / rel, a.work_dir / rel
            if base.is_symlink() or work.is_symlink():
                raise SystemExit('REFUSED: symlink input')
            saved = Path(tmp) / str(i)
            # Independently verify each claimed base against the immutable wiki blob.
            result = subprocess.run(['adom-wiki','repo','show',a.repo,path,'--ref',a.base_sha,'--out',str(saved),'--json'],capture_output=True,text=True)
            if result.returncode:
                if base.exists() or 'FILE_NOT_FOUND' not in result.stdout + result.stderr:
                    raise RuntimeError(result.stdout + result.stderr)
                before = ''
            else:
                if not base.exists() or saved.read_bytes() != base.read_bytes():
                    raise SystemExit('REFUSED: base snapshot mismatch: '+path)
                before = base.read_text()
            after = work.read_text()  # Missing proposed files fail; deletions need separate review.
            errors, diff = inspect(path,before,after)
            problems.extend(path+': '+e for e in errors)
            diffs.append(diff)
            hashes[path] = hashlib.sha256(work.read_bytes()).hexdigest()
    if head() != a.base_sha:
        problems.append('live HEAD moved during preflight')
    a.out.mkdir(parents=True,exist_ok=True)
    (a.out/'review.diff').write_text(''.join(diffs))
    (a.out/'report.json').write_text(json.dumps({'repo':a.repo,'base':a.base_sha,'filesSha256':hashes,'problems':problems,'status':'refused' if problems else 'ready-for-human-diff-review','limits':'Lexical checks only. Run production compilation, tests, semantic diff review and native acceptance separately.'},indent=2)+'\n')
    print(json.dumps({'problems':problems,'report':str(a.out/'report.json')}))
    raise SystemExit(1 if problems else 0)


if __name__ == '__main__':
    main()