"""Run independent historical reviews while preserving all observable evidence. Completed cells are never overwritten. Each run has its own source copy, explicit model/effort, caller prompt, final artifact, event stream, and full local session capture. A failed cell is retained for diagnosis, not silently retried. The caller monitors progress; elapsed time never terminates a review. """ import hashlib import json import os from pathlib import Path import shlex import shutil import subprocess import sys import argparse import time import uuid def save(path,value): """Atomically replace status metadata so interruption cannot leave partial JSON.""" tmp=path.with_suffix(path.suffix+'.tmp') tmp.write_text(json.dumps(value,indent=2)+'\n');tmp.replace(path) def tree_hash(path): """Detect changes to names, object kinds, executable bits, bytes and link targets. Type tags distinguish a file from a symlink containing the same text. Length prefixes keep arbitrary binary contents from being confused with neighboring entries; directory entries also expose newly created empty directories. """ h = hashlib.sha256() for p in sorted(path.rglob('*')): mode = p.lstat().st_mode & 0o111 if p.is_symlink(): kind, data = b'link', os.readlink(p).encode() elif p.is_file(): kind, data = b'file', p.read_bytes() elif p.is_dir(): kind, data = b'directory', b'' else: raise ValueError('Unsupported input object: ' + str(p)) for field in (str(p.relative_to(path)).encode(), kind, str(mode).encode(), data): h.update(len(field).to_bytes(8, 'big')) h.update(field) return h.hexdigest() def parse_lines(path): """Keep raw JSONL separately; tolerate a partial trailing event during diagnosis.""" items=[] for line in path.read_text().splitlines(): try:items.append(json.loads(line)) except json.JSONDecodeError:pass return items def archive_session(state,directory,events): """Retain exact locally recorded instructions rather than reconstructing prompts. Claude snapshots contain the system prompt and tool schema; Codex session metadata contains base instructions and injected messages. Provider-side hidden instructions are outside either local record and are not claimed to be captured. """ rid=state['run_id'];model=state['model'] sid=state['thread_id'] if model=='claude-opus-5-5': project=str(Path(state['workdir'])).replace('/','-').replace('.','-') candidates=list((Path.home()/'.claude/projects'/project).glob(sid+'.jsonl')) if not candidates:candidates=list((Path.home()/'.claude/projects').rglob(sid+'.jsonl')) else: candidates=list((Path.home()/'.codex/sessions').rglob('*'+sid+'*.jsonl')) if len(candidates)!=1:raise RuntimeError(f'Expected one session for {rid}: {len(candidates)}') session=directory/(rid+'.session.jsonl');shutil.copy2(candidates[0],session) entries=parse_lines(session) if model=='claude-opus-5-5': instructions=[e for e in entries if e.get('type') in ['attachment','user','system']] snapshots=[e['attachment'] for e in entries if e.get('attachment',{}).get('type')=='prompt_snapshot'] assert snapshots and any(s.get('systemPrompt') for s in snapshots) assert any(s.get('tools') for s in snapshots) actual=[e for e in entries if e.get('type')=='assistant'] assert actual and all(e.get('message',{}).get('model')==model for e in actual) assert all(e.get('effort')=='high' for e in actual) assert any(e.get('message',{}).get('content')==PROMPT.decode() for e in entries if e.get('type')=='user') state['recorded_model']=model;state['recorded_effort']='high';state['system_prompt_snapshot_saved']=True state['tools_used']=sorted({c.get('name') for e in actual for c in e.get('message',{}).get('content',[]) if c.get('type')=='tool_use'}) state['usage']=next(e for e in reversed(events) if e.get('type')=='result') else: instructions=[e for e in entries if e.get('type')=='session_meta' or (e.get('type')=='response_item' and e.get('payload',{}).get('role') in ['system','developer','user'])] contexts=[e['payload'] for e in entries if e.get('type')=='turn_context'] assert contexts and all(c.get('model')==model and c.get('effort')=='high' for c in contexts) messages=[c.get('text','') for e in instructions if e.get('payload',{}).get('role')=='user' for c in e['payload'].get('content',[])] assert PROMPT.decode() in messages assert any(e.get('payload',{}).get('base_instructions') for e in entries if e.get('type')=='session_meta') state['recorded_model']=model;state['recorded_effort']='high' state['tools_used']=sorted({e['payload'].get('name') for e in entries if e.get('type')=='response_item' and e['payload'].get('type') in ['custom_tool_call','function_call']}) usages=[e['payload']['info']['total_token_usage'] for e in entries if e.get('type')=='event_msg' and e.get('payload',{}).get('type')=='token_count' and e['payload'].get('info')] state['usage']=usages[-1] if usages else None assert 'exec' in state['tools_used'] save(directory/(rid+'.instructions.json'),instructions) state['instructions_verified']=True def run_cell(cell): """Execute and audit one cell; preserve failed attempts without substituting models.""" case,model,repeat=cell rid=f"{case['id']}--{model}--r{repeat}" directory=OUT/'runs'/rid;directory.mkdir(parents=True,exist_ok=True) status=directory/(rid+'.status.json') if status.exists(): old=json.loads(status.read_text()) if old['status'] in ['completed','inconclusive']:return old raise RuntimeError('Existing unfinished/failed attempt: '+rid) work=ROOT/'work'/rid work.parent.mkdir(exist_ok=True) subprocess.run(['cp','-a','--reflink=auto',str(ROOT/'inputs'/case['id']),str(work)],check=True) prompt=directory/(rid+'.prompt.md');prompt.write_bytes(PROMPT) review=directory/(rid+'.review.md');eventpath=directory/(rid+'.events.jsonl');stderr=directory/(rid+'.stderr.log') env=os.environ.copy() if model=='claude-opus-5-5': sid=str(uuid.uuid4()) argv=['claude','-p','--safe-mode','--strict-mcp-config','--model','opus','--effort','high','--system-prompt-snapshot','on','--session-id',sid,'--output-format','stream-json','--verbose','--dangerously-skip-permissions',PROMPT.decode()] env['CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS']='0' else: sid=None argv=['codex','-c','model_reasoning_effort=high','-c','project_doc_max_bytes=0','-c','model_catalog_json='+json.dumps(str(CATALOG)),'exec','--yolo','--ignore-rules','--skip-git-repo-check','--json','-m',model,'-C',str(work),'-o',str(review),PROMPT.decode()] started=time.time() state={'run_id':rid,'case':case['id'],'model':model,'effort':'high','repeat':repeat,'status':'running','started_unix':started,'hard_deadline_seconds':None,'expected_minutes':8,'workdir':str(work),'command':argv,'thread_id':sid,'prompt_sha256':hashlib.sha256(PROMPT).hexdigest(),'source_hash_before':tree_hash(work)} save(status,state) print('START '+rid,flush=True) try: with eventpath.open('w') as stdout,stderr.open('w') as err: proc=subprocess.Popen(['bash','-c','exec '+shlex.join(argv)+' < /dev/null'],cwd=work,env=env,stdin=subprocess.DEVNULL,stdout=stdout,stderr=err,start_new_session=True) state['pid']=proc.pid;save(status,state) # This scheduler runs under a monitored shell session. The user # explicitly requires transcript-based judgment for slow reviews, # so a wait has no deadline and never kills an active reviewer. rc=proc.wait() state['exit_code']=rc events=parse_lines(eventpath) if model=='claude-opus-5-5': results=[e for e in events if e.get('type')=='result'] assert results and not results[-1].get('is_error'),str(results[-1:])[:300] text=results[-1].get('result','') review.write_bytes(text.encode()) else: state['thread_id']=next(e['thread_id'] for e in events if e.get('type')=='thread.started') assert any(e.get('type')=='turn.completed' for e in events) assert 'model metadata' not in stderr.read_text().lower() text=review.read_text() assert rc==0 and len(text)>100 assert 'Background tasks still running after' not in text state['source_hash_after']=tree_hash(work) assert state['source_hash_before']==state['source_hash_after'],'Input mutation' state['source_unchanged']=True archive_session(state,directory,events) state['report_sections_present']=all(word in text.lower() for word in ['scope','findings','uncertain','coverage']) state['review_sha256']=hashlib.sha256(review.read_bytes()).hexdigest() state['review_bytes']=review.stat().st_size state['status']='completed' except Exception as exc: state['status']='failed';state['error']=type(exc).__name__+': '+str(exc) events=parse_lines(eventpath) if not state.get('thread_id'): state['thread_id']=next((e.get('thread_id') for e in events if e.get('type')=='thread.started'),None) if state.get('thread_id'): try:archive_session(state,directory,events) except Exception as capture_error:state['partial_capture_note']=str(capture_error) state['elapsed_seconds']=round(time.time()-started,2);save(status,state) # Keep the isolated work tree for inspection; cleanup is a separate choice. print(f"END {rid} {state['status']} {state['elapsed_seconds']}s "+state.get('error',''),flush=True) return state def main(): """Run one explicitly selected cell after the caller prepares pinned inputs. No batch launches implicitly. A fresh process supplies independent context; completed outputs are preserved, and failed attempts require a fresh output root or explicit archival by the operator before retrying the missing cell. """ global ROOT, OUT, PROMPT, CATALOG parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--root', type=Path, required=True, help='Experiment directory containing inputs/') parser.add_argument('--case', required=True, choices=[c['id'] for c in json.loads(Path(__file__).with_name('cases.json').read_text())]) parser.add_argument('--model', required=True, choices=json.loads(Path(__file__).with_name('experiment.json').read_text())['models']) parser.add_argument('--repeat', type=int, required=True, choices=range(1,5)) parser.add_argument('--catalog', type=Path, help='Pinned model catalog; required for Codex') args = parser.parse_args() ROOT = args.root.resolve() OUT = ROOT / 'outputs' OUT.mkdir(parents=True, exist_ok=True) PROMPT = Path(__file__).with_name('reviewer-prompt.txt').read_bytes() CATALOG = args.catalog.resolve() if args.catalog else None if args.model.startswith('gpt-') and (CATALOG is None or not CATALOG.is_file()): parser.error('--catalog must name a readable frozen catalog for Codex') case = json.loads((ROOT / 'inputs' / args.case / 'case.json').read_text()) if case['id'] != args.case: raise ValueError('Input case identity does not match the selected cell') for name in ['base', 'subject', 'change.diff']: if not (ROOT / 'inputs' / args.case / name).exists(): raise ValueError('Missing review input: ' + name) result = run_cell((case, args.model, args.repeat)) if result['status'] != 'completed': raise SystemExit(1) if __name__ == '__main__': main()