ProjectIntakeTower/verifier/read_only_audit.py

#!/usr/bin/env python3
import json, subprocess, hashlib
from pathlib import Path
from datetime import datetime, timezone

ROOT = Path(__file__).resolve().parents[1]
TRIAGE = ROOT / 'triage'
E = ROOT / 'inventory' / 'evidence'
JOBS = ROOT / 'jobs' / 'pending'
OUT = TRIAGE / 'READ_ONLY_AUDIT.md'
OUT_JSON = TRIAGE / 'READ_ONLY_AUDIT.json'

REQUIRED_TRIAGE = [
    'PROJECT_PRIORITIZATION.md', 'PROJECT_PRIORITIZATION.json', 'PROJECT_PRIORITIZATION_SUMMARY.json',
    'KEEP.md', 'MERGE.md', 'PAUSE.md', 'ARCHIVE.md', 'URGENT.md',
    'TOKEN_BURN_CANDIDATE.md', 'HUMAN_DECISION_REQUIRED.md', 'TOP_10_NEXT_ACTIONS.md'
]
REQUIRED_EVIDENCE = [
    'git_repos.json', 'readmes.json', 'todos_fixmes.json', 'old_plans.json',
    'uncommitted_changes.json', 'open_branches.json', 'tests.json',
    'broken_projects.json', 'duplicate_projects.json', 'theme_clusters.json'
]

FORBIDDEN_JOB_WORDS = ['commit', 'push', 'merge', 'delete', 'rm -rf', 'create github issues', 'gh issue create']
REQUIRED_JOB_WORDS = ['read-only', 'projectintaketower']


def sh(args, cwd=None, timeout=10):
    try:
        p = subprocess.run(args, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
        return {'exit_code': p.returncode, 'stdout': p.stdout.strip(), 'stderr': p.stderr.strip()}
    except Exception as e:
        return {'exit_code': 124, 'stdout': '', 'stderr': str(e)}


def sha(path):
    h=hashlib.sha256()
    with open(path,'rb') as f:
        for chunk in iter(lambda:f.read(65536), b''):
            h.update(chunk)
    return h.hexdigest()


def load_json(path):
    try:
        return json.loads(path.read_text(encoding='utf-8'))
    except Exception as e:
        return {'__error__': str(e)}


def main():
    checks=[]; failures=[]; warnings=[]

    for name in REQUIRED_TRIAGE:
        p=TRIAGE/name
        ok=p.exists() and p.stat().st_size>0
        checks.append({'check':'triage_artifact_exists','target':name,'ok':ok,'size':p.stat().st_size if p.exists() else 0})
        if not ok: failures.append(f'missing/empty triage artifact: {name}')

    for name in REQUIRED_EVIDENCE:
        p=E/name
        ok=p.exists() and p.stat().st_size>0
        checks.append({'check':'evidence_artifact_exists','target':name,'ok':ok,'size':p.stat().st_size if p.exists() else 0})
        if not ok: failures.append(f'missing/empty evidence artifact: {name}')
        elif p.suffix == '.json':
            data=load_json(p)
            if isinstance(data, dict) and '__error__' in data:
                failures.append(f'invalid json: {name}: {data["__error__"]}')

    summary=load_json(TRIAGE/'PROJECT_PRIORITIZATION_SUMMARY.json')
    prioritization=load_json(TRIAGE/'PROJECT_PRIORITIZATION.json')
    if isinstance(summary, dict) and isinstance(prioritization, dict) and 'categories' in prioritization:
        for cat,count in summary.items():
            if cat == 'total_projects': continue
            actual=len(prioritization['categories'].get(cat, []))
            ok=actual==count
            checks.append({'check':'category_count_matches','target':cat,'ok':ok,'expected':count,'actual':actual})
            if not ok: failures.append(f'category count mismatch {cat}: expected {count}, actual {actual}')
    else:
        failures.append('cannot validate prioritization summary/categories')

    # Audit pending jobs for review-only shape.
    job_files=sorted(JOBS.glob('*.json'))
    bad_jobs=[]
    for jp in job_files:
        data=load_json(jp)
        text=jp.read_text(encoding='utf-8', errors='ignore').lower()
        # Dangerous words are allowed only when they appear in forbidden_actions.
        fwords=' '.join(data.get('forbidden_actions',[]) + data.get('scope',{}).get('forbidden_actions',[])).lower() if isinstance(data,dict) else ''
        awords=' '.join(data.get('allowed_files',[]) + data.get('scope',{}).get('allowed_paths',[])).lower() if isinstance(data,dict) else ''
        allowed_actions=' '.join(data.get('allowed_actions',[]) + data.get('scope',{}).get('allowed_actions',[])).lower() if isinstance(data,dict) else ''
        mode=data.get('mode') if isinstance(data,dict) else None
        problems=[]
        if mode and mode != 'review_only': problems.append(f'mode={mode}')
        for word in ['commit','push','secret']:
            if word not in fwords: problems.append(f'forbidden missing {word}')
        if 'projectintaketower' not in awords and 'projectintaketower' not in text: problems.append('missing ProjectIntakeTower artifact scope')
        dangerous_allowed=[w for w in ['commit','push','merge','delete','rm -rf','create github issues','gh issue create'] if w in allowed_actions]
        if dangerous_allowed: problems.append('dangerous allowed_action: '+','.join(dangerous_allowed))
        if problems: bad_jobs.append({'path':str(jp),'problems':problems})
    if bad_jobs:
        warnings.append(f'{len(bad_jobs)} pending jobs need review-only contract cleanup')
    checks.append({'check':'pending_jobs_audited','target':'jobs/pending','ok':not bad_jobs,'jobs':len(job_files),'bad_jobs':len(bad_jobs)})

    # Git state of ProjectIntakeTower repo itself after generated artifacts.
    git_status=sh(['git','status','--porcelain=v1'], cwd=ROOT.parents[0])
    checks.append({'check':'current_repo_git_status_read','target':str(ROOT.parents[0]),'ok':git_status['exit_code']==0,'dirty_lines':len(git_status['stdout'].splitlines()) if git_status['stdout'] else 0})
    if git_status['exit_code'] != 0:
        warnings.append('could not read git status for current workspace')

    # Hash key outputs for stable handoff.
    hashes={}
    for name in REQUIRED_TRIAGE + REQUIRED_EVIDENCE:
        p=(TRIAGE/name) if (TRIAGE/name).exists() else (E/name)
        if p.exists() and p.is_file(): hashes[str(p.relative_to(ROOT))]=sha(p)

    status='passed' if not failures else 'failed'
    result={
        'status': status,
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'objective': 'READ_ONLY_AUDIT of ProjectIntakeTower prioritization/intake artifacts',
        'checks': checks,
        'failures': failures,
        'warnings': warnings,
        'bad_jobs': bad_jobs[:100],
        'hashes': hashes,
        'next_action': 'Review READ_ONLY_AUDIT.md; fix warnings only if Samuel wants stricter job-contract hygiene.' if status=='passed' else 'Fix failures before trusting prioritization.'
    }
    OUT_JSON.write_text(json.dumps(result,indent=2,ensure_ascii=False),encoding='utf-8')

    lines=[
        '# READ_ONLY_AUDIT', '',
        f'Status: {status}',
        f'Timestamp: {result["timestamp"]}', '',
        'Scope: ProjectIntakeTower artifacts only. No source-project edits, no commits, no GitHub issue creation.', '',
        '## Summary',
        f'- Checks: {len(checks)}',
        f'- Failures: {len(failures)}',
        f'- Warnings: {len(warnings)}',
        f'- Pending jobs audited: {len(job_files)}',
        f'- Bad job contracts: {len(bad_jobs)}', '',
        '## Failures'
    ]
    lines += [f'- {x}' for x in failures] or ['- none']
    lines += ['', '## Warnings']
    lines += [f'- {x}' for x in warnings] or ['- none']
    lines += ['', '## Category counts']
    if isinstance(summary, dict):
        for k,v in summary.items(): lines.append(f'- {k}: {v}')
    lines += ['', '## Next action', result['next_action'], '', 'JSON detail:', f'`{OUT_JSON}`']
    OUT.write_text('\n'.join(lines)+'\n',encoding='utf-8')
    print(json.dumps({'status':status,'failures':len(failures),'warnings':len(warnings),'bad_jobs':len(bad_jobs),'out':str(OUT)},indent=2))
    raise SystemExit(0 if status=='passed' else 1)

if __name__=='__main__': main()