#!/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()