Blame for ProjectIntakeTower/verifier/generate_archive_plan.py

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

ROOT=Path(__file__).resolve().parents[1]
T=ROOT/'triage'; OUT=ROOT/'archive_plan'; CAND=OUT/'candidates'; JOBS=ROOT/'jobs/pending'
for d in [OUT,CAND,JOBS]: d.mkdir(parents=True,exist_ok=True)
MAX_CANDIDATES=80

def load(p): return json.loads(Path(p).read_text(encoding='utf-8'))
def slug(s): return re.sub(r'[^A-Za-z0-9_.-]+','_',s.replace('/home/smlflg/',''))[:110]
def archive_action(row):
    path=row['path'].lower(); cats=row.get('categories',[])
    if '/repo_cache/' in path or '/eval/fixtures/' in path or 'fixture' in path:
        return 'ARCHIVE_AS_GENERATED_CACHE_OR_FIXTURE'
    if 'MERGE' in cats:
        return 'ARCHIVE_AFTER_CONSOLIDATION_TARGET_APPROVED'
    if row.get('score',0) < 20 and not row.get('readme_present'):
        return 'ARCHIVE_LOW_SIGNAL_AFTER_FINAL_DIFF'
    return 'PAUSE_THEN_ARCHIVE_IF_NO_OWNER_VALUE'

def risk(row):
    risks=[]
    if row.get('dirty_count',0): risks.append(f'uncommitted_changes={row["dirty_count"]}')
    if row.get('branch_count',0)>1: risks.append(f'branches={row["branch_count"]}')
    if row.get('readme_present'): risks.append('has_readme')
    if 'KEEP' in row.get('categories',[]): risks.append('also_KEEP_conflict')
    return risks or ['low visible risk']

def candidate_doc(i,row):
    act=archive_action(row); risks=risk(row)
    return f"""# ARCHIVE_PLAN Candidate — {row['name']}

Project: `{row['path']}`
Categories: {', '.join(row['categories'])}
Recommended action: {act}
Score: {row['score']}
Dirty: {row['dirty_count']}
Branches: {row['branch_count']}
TODO/FIXME: {row['todo_fixme_count']}
Plans: {row['plan_hit_count']}

## Why archive/pause

- Archive category or low-signal/generated-cache heuristics selected this project.
""" + ''.join(f'- Risk: {x}\n' for x in risks) + f"""
## Required pre-archive checks

1. Read-only final diff/listing: file count, latest commit, dirty state.
2. Preserve path, remote, branch, and last commit in archive manifest.
3. If dirty: export patch before any archive decision.
4. If branch_count > 1: list branches and ask Samuel before archive.
5. If duplicated: confirm consolidation target first.
6. Never delete. Archive means mark/plan/move only after explicit approval.

## Human gate

Status = needs_human
Reason = Archive/move/delete is destructive or semantically destructive.
Options:
1. Keep as PAUSE.
2. Archive after manifest + patch export.
3. Merge into approved target, then archive source.
Recommendation: default to Option 1 unless this is cache/fixture.
"""

def main():
    pri=load(T/'PROJECT_PRIORITIZATION.json')
    archive=pri['categories'].get('ARCHIVE',[])
    # high-confidence archive first: caches/fixtures, then low score, then no readme/dirty-low
    archive=sorted(archive,key=lambda r:(0 if ('/repo_cache/' in r['path'] or 'fixture' in r['path'].lower()) else 1, r.get('score',0), r['path']))[:MAX_CANDIDATES]
    index=['# ARCHIVE_PLAN','',f'Generated: {datetime.now(timezone.utc).isoformat()}','',f'Candidates included: {len(archive)}','', 'Mode: READ_ONLY_ARCHIVE_PLAN — no moves, deletes, commits, pushes, or source edits.', '', 'Archive requires explicit Samuel approval per candidate or batch.', '']
    manifest=[]; jobs=[]
    for i,row in enumerate(archive,1):
        s=slug(row['path']); doc=CAND/f'{i:02d}_{s}.md'; doc.write_text(candidate_doc(i,row),encoding='utf-8')
        rec={'rank':i,'name':row['name'],'path':row['path'],'categories':row['categories'],'recommended_action':archive_action(row),'risks':risk(row),'candidate_doc':str(doc),'human_gate':True}
        manifest.append(rec)
        job={
          'job_id':f'archiveplan-2026-05-04-{i:02d}-{s.lower()}',
          'project_name':row['name'],'project_path':row['path'],'mode':'review_only',
          'allowed_actions':['read files','inspect git status/branches/remotes','write ProjectIntakeTower archive manifest only','write ProjectIntakeTower patch-export plan only'],
          'forbidden_actions':['modify source project files','commit','push','merge','delete files','move files','archive files','install dependencies','create GitHub issues','print secrets'],
          'allowed_files':[f'{row["path"]} read-only','ProjectIntakeTower/archive_plan write scope'],
          'expected_artifacts':[str(doc)],
          'verifier_commands':['python3 ProjectIntakeTower/verifier/verify_archive_plan.py'],
          'stop_conditions':['archive/delete/move requested','dirty changes require human approval','multiple branches require human approval','consolidation target unclear'],
          'objective':f'Prepare read-only archive decision manifest for {row["name"]}; do not archive anything.',
          'recommended_action':archive_action(row)
        }
        jp=JOBS/f'archiveplan_{i:02d}_{s}.json'; jp.write_text(json.dumps(job,indent=2,ensure_ascii=False),encoding='utf-8'); jobs.append(str(jp))
        index.append(f"- `{row['path']}` — {archive_action(row)} — {', '.join(risk(row))} — doc `{doc}`")
    result={'status':'succeeded','mode':'READ_ONLY_ARCHIVE_PLAN','candidates':len(manifest),'jobs_created':len(jobs),'manifest':manifest,'jobs':jobs}
    (OUT/'ARCHIVE_PLAN.md').write_text('\n'.join(index)+'\n',encoding='utf-8')
    (OUT/'archive_manifest.json').write_text(json.dumps(manifest,indent=2,ensure_ascii=False),encoding='utf-8')
    (OUT/'result.json').write_text(json.dumps(result,indent=2,ensure_ascii=False),encoding='utf-8')
    print(json.dumps({'status':'succeeded','candidates':len(manifest),'jobs_created':len(jobs)},indent=2))
if __name__=='__main__': main()