| |
| import fnmatch |
| import json |
| import os |
| import re |
| import subprocess |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| CONFIG = ROOT / 'config' |
| INV = ROOT / 'inventory' |
| TRIAGE = ROOT / 'triage' |
| JOBS = ROOT / 'jobs' / 'pending' |
| GH_DRAFTS = ROOT / 'github' / 'issue_drafts' |
| EVIDENCE = INV / 'evidence' |
| CARDS = INV / 'project_cards' |
|
|
| TODO_RE = re.compile(r'\b(TODO|FIXME|HACK|XXX|BUG|WIP|REVIEW)\b|^- \[ \]', re.I) |
| AGENT_NAMES = {'AGENTS.md', 'CLAUDE.md', '.cursorrules', '.hermes', '.claude', '.codex', '.openclaude'} |
| PLAN_RE = re.compile(r'(plan|roadmap|backlog|idea|brainstorm)', re.I) |
| TEXT_EXT = {'.md','.txt','.yaml','.yml','.json','.jsonl','.toml','.ini','.py','.js','.ts','.tsx','.jsx','.rs','.go','.sh'} |
| IGNORE_DIRS = {'.git/objects','node_modules','.venv','venv','__pycache__','.cache','dist','build','.next','target','vendor'} |
| SENSITIVE = ['.env','.env.*','*.pem','id_rsa','id_ed25519','credentials*','secrets*','*token*','*apikey*'] |
|
|
| def sh(args, cwd=None, timeout=8): |
| 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()[:2000], 'stderr': p.stderr.strip()[:1000]} |
| except Exception as e: |
| return {'exit_code': 124, 'stdout': '', 'stderr': str(e)[:1000]} |
|
|
| def parse_roots(): |
| roots = [] |
| current = None |
| for line in (CONFIG / 'scan_roots.yaml').read_text(encoding='utf-8').splitlines(): |
| s = line.strip() |
| if s.startswith('- path:'): |
| current = {'path': s.split(':',1)[1].strip(), 'enabled': False, 'purpose': ''} |
| roots.append(current) |
| elif current and s.startswith('enabled:'): |
| current['enabled'] = s.split(':',1)[1].strip().lower() == 'true' |
| elif current and s.startswith('purpose:'): |
| current['purpose'] = s.split(':',1)[1].strip() |
| return roots |
|
|
| def is_sensitive(path: Path): |
| name = path.name |
| return any(fnmatch.fnmatch(name, pat) for pat in SENSITIVE) |
|
|
| def skip_dir(path: Path): |
| parts = set(path.parts) |
| if any(part in parts for part in ['node_modules','.venv','venv','__pycache__','.cache','dist','build','.next','target','vendor']): |
| return True |
| return False |
|
|
| def safe_rel(path): |
| try: |
| return str(path) |
| except Exception: |
| return repr(path) |
|
|
| def iter_files(root: Path, max_files=12000): |
| count = 0 |
| for dirpath, dirnames, filenames in os.walk(root): |
| d = Path(dirpath) |
| dirnames[:] = [x for x in dirnames if not skip_dir(d / x) and x != 'objects'] |
| if skip_dir(d): |
| continue |
| for name in filenames: |
| count += 1 |
| if count > max_files: |
| return |
| yield d / name |
|
|
| def find_git_repos(enabled_roots): |
| repos = set() |
| for r in enabled_roots: |
| root = Path(r['path']) |
| if not root.exists(): |
| continue |
| if (root / '.git').exists(): |
| repos.add(root) |
| for dirpath, dirnames, filenames in os.walk(root): |
| d = Path(dirpath) |
| if skip_dir(d): |
| dirnames[:] = [] |
| continue |
| if '.git' in dirnames: |
| repos.add(d) |
| dirnames[:] = [x for x in dirnames if x != '.git'] |
| dirnames[:] = [x for x in dirnames if x not in {'node_modules','.venv','venv','__pycache__','.cache','dist','build','.next','target','vendor'}] |
| return sorted(repos, key=lambda p: str(p).lower()) |
|
|
| def repo_info(repo: Path): |
| status = sh(['git','status','--porcelain'], cwd=repo) |
| remote = sh(['git','remote','get-url','origin'], cwd=repo) |
| branch = sh(['git','rev-parse','--abbrev-ref','HEAD'], cwd=repo) |
| last = sh(['git','log','-1','--format=%cI|%s'], cwd=repo) |
| files = list(repo.iterdir()) if repo.exists() else [] |
| names = {p.name for p in files} |
| typ = [] |
| if 'package.json' in names: typ.append('node') |
| if 'pyproject.toml' in names or 'requirements.txt' in names: typ.append('python') |
| if 'Cargo.toml' in names: typ.append('rust') |
| if 'go.mod' in names: typ.append('go') |
| if '.hermes' in names or 'AGENTS.md' in names or 'CLAUDE.md' in names: typ.append('agentized') |
| dirty = bool(status['stdout']) if status['exit_code'] == 0 else None |
| last_date = last['stdout'].split('|',1)[0] if last['exit_code'] == 0 and last['stdout'] else '' |
| return { |
| 'path': str(repo), 'exists': repo.exists(), 'status': 'unknown', |
| 'git': {'branch': branch['stdout'], 'remote': remote['stdout'], 'dirty': dirty, 'last_commit': last['stdout']}, |
| 'project_type': typ or ['unknown'], 'signals': [] |
| } |
|
|
| def verification(repo: Path, typ): |
| |
| |
| results = [] |
| gs = sh(['git','status','--short'], cwd=repo, timeout=6) |
| results.append({'name':'git_status_readable', 'status':'passed' if gs['exit_code']==0 else 'failed', 'exit_code':gs['exit_code'], 'summary': (gs['stderr'] or 'git status readable')[:500]}) |
| manifest_checks = { |
| 'python': ['pyproject.toml', 'requirements.txt', 'setup.py'], |
| 'node': ['package.json'], |
| 'rust': ['Cargo.toml'], |
| 'go': ['go.mod'], |
| } |
| for kind, files in manifest_checks.items(): |
| if kind in typ: |
| present = [f for f in files if (repo / f).exists()] |
| results.append({'name':f'{kind}_manifest_present', 'status':'passed' if present else 'skipped', 'exit_code':0, 'summary': ', '.join(present) or 'not found'}) |
| return results |
|
|
| def classify(info, todos, verifies): |
| dirty = info['git'].get('dirty') |
| failed = any(v['status']=='failed' for v in verifies) |
| todo_count = len(todos) |
| last = info['git'].get('last_commit','')[:10] |
| year = int(last[:4]) if re.match(r'\d{4}', last) else 0 |
| if failed: |
| return 'needs_human' |
| if todo_count >= 5 and year >= 2025: |
| return 'active' |
| if dirty or todo_count: |
| return 'quick_win' |
| if year and year < 2024: |
| return 'archive_candidate' |
| return 'unknown' |
|
|
| def main(): |
| for d in [INV, TRIAGE, JOBS, GH_DRAFTS, EVIDENCE, CARDS]: d.mkdir(parents=True, exist_ok=True) |
| roots = parse_roots() |
| enabled = [r for r in roots if r['enabled']] |
| now = datetime.now(timezone.utc).isoformat() |
| raw = [] |
| projects = [] |
| repo_todos = {} |
| agent_traces = [] |
| plan_hits = [] |
|
|
| repos = find_git_repos(enabled) |
| repo_by_path = {str(r): r for r in repos} |
|
|
| for r in enabled: |
| root = Path(r['path']) |
| raw.append({'type':'root','path':str(root),'exists':root.exists(),'purpose':r['purpose'],'ts':now}) |
| if not root.exists(): |
| continue |
| for path in iter_files(root): |
| if is_sensitive(path): |
| raw.append({'type':'sensitive_skipped','path':str(path)}) |
| continue |
| if path.name in AGENT_NAMES or any(part in AGENT_NAMES for part in path.parts): |
| agent_traces.append({'path':str(path), 'kind':'agent_trace'}) |
| if PLAN_RE.search(path.name): |
| plan_hits.append({'path':str(path), 'kind':'plan_or_idea'}) |
| if path.suffix.lower() not in TEXT_EXT: |
| continue |
| try: |
| text = path.read_text(encoding='utf-8', errors='ignore') |
| except Exception: |
| continue |
| for i, line in enumerate(text.splitlines()[:2000], 1): |
| if TODO_RE.search(line): |
| item = {'path':str(path), 'line':i, 'text':line.strip()[:240]} |
| repo_todos.setdefault(str(root), []).append(item) |
|
|
| ver_matrix = [] |
| for repo in repos: |
| info = repo_info(repo) |
| todos = repo_todos.get(str(repo), [])[:50] |
| verifies = verification(repo, info['project_type']) |
| info['status'] = classify(info, todos, verifies) |
| info['signals'] = {'todo_count': len(repo_todos.get(str(repo), [])), 'verification': verifies} |
| projects.append(info) |
| ver_matrix.append({'path':str(repo), 'checks':verifies}) |
| slug = re.sub(r'[^A-Za-z0-9_.-]+','_', repo.name)[:80] |
| card = CARDS / f'{slug}.md' |
| card.write_text(f"# {repo.name}\n\nPath: {repo}\nStatus: {info['status']}\nType: {', '.join(info['project_type'])}\nDirty: {info['git']['dirty']}\nLast: {info['git']['last_commit']}\nTODOs: {len(repo_todos.get(str(repo), []))}\nVerification: {verifies[0]['status'] if verifies else 'skipped'}\n", encoding='utf-8') |
| if info['status'] in {'active','quick_win','needs_human'}: |
| job = { |
| 'job_id': f"work-{datetime.now().date()}-{slug.lower()}-handoff", |
| 'task_type': 'verify' if info['status']=='needs_human' else 'repair', |
| 'objective': f"Turn inventory findings for {repo.name} into one verified next action.", |
| 'scope': {'allowed_paths':[str(repo), str(ROOT)], 'allowed_tools':['terminal','read_file','search_files','write_file'], 'allowed_actions':['read repo','run non-destructive tests','write artifact under ProjectIntakeTower'], 'forbidden_actions':['commit','push','delete files','create GitHub issues','print secrets']}, |
| 'inputs': {'project_card': str(card)}, |
| 'required_artifacts': [f'inventory/evidence/{slug}_handoff_result.json'], |
| 'verifier': {'type':'command','command':'python3 ProjectIntakeTower/verifier/verify_artifacts.py'}, |
| 'timeout_minutes': 30, 'max_iterations': 2, |
| 'if_blocked': 'write needs_human result and stop', |
| 'rejoin_policy': 'summarize artifact, verifier result, next action' |
| } |
| (JOBS / f'{slug}.json').write_text(json.dumps(job, indent=2), encoding='utf-8') |
| draft_dir = GH_DRAFTS / slug |
| draft_dir.mkdir(parents=True, exist_ok=True) |
| (draft_dir / 'intake-next-action.md').write_text(f"Title: Intake: next action for {repo.name}\nRepo: {info['git']['remote'] or 'NEEDS_REPO_MAPPING'}\nLabels: intake:discovered,agent:handoff-ready\n\n## Evidence\n\n- Path: `{repo}`\n- Status: {info['status']}\n- TODO count: {len(repo_todos.get(str(repo), []))}\n- Verification: {verifies[0]['status'] if verifies else 'skipped'}\n\n## Acceptance criteria\n\n- One concrete next action selected.\n- Verifier command documented.\n- No secrets exposed.\n", encoding='utf-8') |
|
|
| (INV / 'raw_scan.jsonl').write_text('\n'.join(json.dumps(x, ensure_ascii=False) for x in raw) + ('\n' if raw else ''), encoding='utf-8') |
| (INV / 'projects.jsonl').write_text('\n'.join(json.dumps(x, ensure_ascii=False) for x in projects) + ('\n' if projects else ''), encoding='utf-8') |
| (EVIDENCE / 'todo_inventory.json').write_text(json.dumps(repo_todos, indent=2, ensure_ascii=False), encoding='utf-8') |
| (EVIDENCE / 'agent_traces.json').write_text(json.dumps(agent_traces[:2000], indent=2, ensure_ascii=False), encoding='utf-8') |
| (EVIDENCE / 'plan_hits.json').write_text(json.dumps(plan_hits[:2000], indent=2, ensure_ascii=False), encoding='utf-8') |
| (EVIDENCE / 'verification_matrix.json').write_text(json.dumps(ver_matrix, indent=2, ensure_ascii=False), encoding='utf-8') |
|
|
| ranked = sorted(projects, key=lambda p: ({'active':0,'quick_win':1,'needs_human':2,'unknown':3,'archive_candidate':4}.get(p['status'],5), -p['signals']['todo_count'], p['path'])) |
| backlog = ['# Backlog\n'] |
| archive = ['# Archive Candidates\n\nNo deletion implied.\n'] |
| needs = ['# Needs Human\n'] |
| top = ['# Top 10 Next Actions\n', '| Rank | Action | Project/Path | Why valuable | Effort | Owner | Verifier | Risk/Human Gate |', '|---:|---|---|---|---|---|---|---|'] |
| for i,p in enumerate(ranked[:10],1): |
| action = 'Run bounded handoff job' if p['status'] != 'archive_candidate' else 'Review archive candidacy' |
| top.append(f"| {i} | {action} | `{p['path']}` | status={p['status']}, todos={p['signals']['todo_count']} | M | Agent | see evidence/verification_matrix.json | {'yes' if p['status']=='needs_human' else 'no'} |") |
| for p in ranked: |
| entry = f"\n## {p['path']}\nStatus: {p['status']}\nValue: 3\nEffort: 3\nRisk: 2\nConfidence: 3\nEvidence:\n- TODO count: {p['signals']['todo_count']}\n- Verification: {p['signals']['verification'][0]['status']}\nNext action:\n- Review project card in inventory/project_cards.\nVerifier:\n- evidence/verification_matrix.json\n" |
| backlog.append(entry) |
| if p['status']=='archive_candidate': archive.append(entry) |
| if p['status']=='needs_human': needs.append(entry) |
| (TRIAGE / 'BACKLOG.md').write_text('\n'.join(backlog), encoding='utf-8') |
| (TRIAGE / 'ARCHIVE_CANDIDATES.md').write_text('\n'.join(archive), encoding='utf-8') |
| (TRIAGE / 'NEEDS_HUMAN.md').write_text('\n'.join(needs), encoding='utf-8') |
| (TRIAGE / 'TOP_10_NEXT_ACTIONS.md').write_text('\n'.join(top)+'\n', encoding='utf-8') |
| (ROOT / 'OWNER_DIGEST.md').write_text(f"# Owner Digest\n\nStatus: scanned\nTimestamp: {now}\n\nScan roots enabled:\n" + ''.join(f"- {r['path']} ({r['purpose']})\n" for r in enabled) + f"\nProjects found: {len(projects)}\nHandoff jobs pending: {len(list(JOBS.glob('*.json')))}\nIssue drafts: {len(list(GH_DRAFTS.glob('**/*.md')))}\n\nRead next:\n- triage/TOP_10_NEXT_ACTIONS.md\n- triage/BACKLOG.md\n- triage/NEEDS_HUMAN.md\n", encoding='utf-8') |
| print(json.dumps({'status':'succeeded','projects':len(projects),'jobs_pending':len(list(JOBS.glob('*.json'))),'issue_drafts':len(list(GH_DRAFTS.glob('**/*.md')))}, indent=2)) |
|
|
| if __name__ == '__main__': |
| main() |
|
|
