`data/codex-yoloforever/cycle-20260908T1322Z.m4gNE5/finalize-cycle.py`

55 lines
"""Persist this bounded cycle once; run only after final review and empirical proof."""
import datetime, fcntl, hashlib, json, os
from pathlib import Path

P = Path(__file__).resolve().parent
R = P.parents[2]
read = lambda n: json.loads((P/n).read_text())
assessment = read('final-assessment.json')
proof = read('parent-final-proof.json')
assert assessment['cycle_id'] == P.name
assert assessment['execution_iterations'] <= 6
assert [d['position'] for d in assessment['ordered_dispositions']] == [1, 2]
assert [d['ticket'].split('-')[1] for d in assessment['ordered_dispositions']] == ['10928', '11250']
assert proof['counts']['FAIL'] == 0
assert os.environ.get('DTD_ZERO_COST') == '1'
assert (R/'config/dtd-cost-mode').read_text() == 'ZERO_COST_REQUIRED\n'
assert assessment['final_dtd']['verdict'] in ['SHIP','FIX-THEN-SHIP','HOLD-FOR-STEVE','KILL']
assert assessment['final_dtd']['post_decision'] in ['KEEP','OVERTURN','inconclusive']
record = dict(assessment)
record.update(timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(),tests_evidence={'closing_proof':str(P/'parent-final-proof.json'),'counts':proof['counts'],'cycle_directory':str(P)},scheduled_next='Existing launchd600s; no schedule mutation; runner observes exit after return')
manifest = {}
for p in sorted(P.rglob('*')):
    if p.is_file() and p.name not in ['artifact-manifest.json','ledger-append-proof.json','commit-proof.json']:
        manifest[str(p.relative_to(P))] = {'bytes':p.stat().st_size,'sha256':hashlib.sha256(p.read_bytes()).hexdigest()}
(P/'artifact-manifest.json').write_text(json.dumps(manifest,indent=2)+'\n')
record['artifact_manifest'] = {'path':str(P/'artifact-manifest.json'),'sha256':hashlib.sha256((P/'artifact-manifest.json').read_bytes()).hexdigest()}
report = R/'verification/cycle-20260908T1322Z-m4gNE5.json'
report.write_text(json.dumps(record,indent=2)+'\n')
md = R/'verification/cycle-20260908T1322Z-m4gNE5.md'
lines = ['Bounded Codex cycle — '+P.name,'',assessment['headline'],'', 'Ordered dispositions:']
for d in assessment['ordered_dispositions']:
    lines.append(f"- {d['position']}. {d['ticket']} (rank{d['rank']}): {d['disposition']}. {d['reason']}")
lines += ['',f"Execution iterations: {assessment['execution_iterations']}/6. Implementation: {assessment['implementation_progress']}. Operational outcomes completed: {assessment['operational_outcomes_completed']}.",'',f"Cody: {assessment['cody']['verdict']}. Top action: {assessment['cody']['top_fix']}.",f"Final DTD: {assessment['final_dtd']['verdict']}; {assessment['final_dtd']['vote']}; post-decision {assessment['final_dtd']['post_decision']}.",'',f"Closing empirical checks: {proof['counts']}. No runnable UI was produced; actual Git/CLI/canary/API/auth-negative/READ ONLY SQL checks replace CTA and screenrecord.",'',f"Next seed: {assessment['next_seed']}",'','All originals and scratch are retained. Direct paid API/external spend $0. Existing600-second launchd schedule remains active; this cycle returns normally.','',f"Evidence: {P}"]
md.write_text('\n'.join(lines)+'\n')
ledger = R/'data/codex-yoloforever/ledger.jsonl'
payload = (json.dumps(record,separators=(',',':'))+'\n').encode()
with ledger.open('a+b',buffering=0) as f:
    fcntl.flock(f,fcntl.LOCK_EX)
    f.seek(0); before = f.read()
    assert not before or before.endswith(b'\n'), 'Refuse appending after torn existing ledger tail'
    old = [json.loads(line) for line in before.splitlines() if line.strip()]
    assert not any(x.get('cycle_id') == P.name for x in old), 'Cycle already appended'
    offset = 0
    while offset < len(payload):
        wrote = f.write(payload[offset:]); assert wrote > 0; offset += wrote
    os.fsync(f.fileno()); f.seek(0); after = f.read()
    assert after == before+payload
    parsed = [json.loads(line) for line in after.splitlines() if line.strip()]
    assert len(parsed) == len(old)+1 and parsed[-1] == record
    assert sum(x.get('cycle_id') == P.name for x in parsed) == 1
    append_proof = {'verdict':'PASS','cycle_id':P.name,'prior_bytes':len(before),'appended_bytes':len(payload),'prior_sha256':hashlib.sha256(before).hexdigest(),'line_sha256':hashlib.sha256(payload).hexdigest(),'prefix_preserved':True,'fsynced':True,'exactly_once':True,'record_count':len(parsed)}
    fcntl.flock(f,fcntl.LOCK_UN)
(P/'ledger-append-proof.json').write_text(json.dumps(append_proof,indent=2)+'\n')
print(json.dumps({'report':str(md),'ledger_append':append_proof}))