| |
| 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') |
|
|
| |
| 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() |
| |
| 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_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') |
|
|
| |
| 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() |
|
|
