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

68 lines
import datetime, hashlib, json, math, os, pathlib, plistlib, subprocess, sys, urllib.request, urllib.error

P = pathlib.Path(__file__).resolve().parent
ROOT = pathlib.Path('/Users/macstudio3/Projects/ticket-system')
checks, commands = [], []
def run(argv):
    r = subprocess.run(argv, capture_output=True, text=True, timeout=45)
    commands.append(dict(argv=argv, exit_code=r.returncode, stdout=r.stdout, stderr=r.stderr))
    return r
def check(name, ok, detail):
    checks.append(dict(check=name, verdict='PASS' if ok else 'FAIL', detail=detail))
def sha(path):
    h = hashlib.sha256()
    with open(path, 'rb') as f:
        for block in iter(lambda: f.read(8*1024*1024), b''): h.update(block)
    return h.hexdigest()

baseline = json.loads((P/'preconditions.json').read_text())
check('zero-cost environment and canonical guard', os.getenv('DTD_ZERO_COST') == '1' and (ROOT/'config/dtd-cost-mode').read_text() == 'ZERO_COST_REQUIRED\n', 'Read only; controls never modified')
for path, digest in baseline['files'].items():
    check('protected bytes '+path, sha(path) == digest, digest)
for path, want in [('/healthz', 200), ('/api/tickets', 401)]:
    try:
        with urllib.request.urlopen('http://127.0.0.1:9794'+path, timeout=10) as r: code, body = r.status, r.read().decode()
    except urllib.error.HTTPError as e: code, body = e.code, e.read().decode()
    check('ticket API '+path, code == want, dict(status=code, body=body[:160]))
r = run(['du', '-sk', '/Users/macstudio3/Projects/.git'])
check('current root storage measured', r.returncode == 0, r.stdout.strip())
kib = int(r.stdout.split()[0]); mb = math.floor(kib/1024+0.5)
r = run(['node', '--input-type=module', '-e', "import {classifyRepo,KNOWN_LEGIT} from '/Users/macstudio3/.claude/skills/dw-backup-canary/lib.mjs'; console.log(JSON.stringify({floor:KNOWN_LEGIT['(projects-root)'].floor_mb, result:classifyRepo({repo:'(projects-root)',mb:"+str(mb)+",remote:null,upstream:null,ahead:0})}));"])
classification = json.loads(r.stdout)
check('actual canary classification boundary', r.returncode == 0 and classification['floor'] == 3000 and classification['result']['verdict'] == ('FAIL' if mb > 3000 else 'WARN' if mb >= 2850 else 'PASS'), dict(measured_mb=mb, **classification))
r = run(['launchctl','list','com.steve.dw-backup-canary'])
check('backup scheduler state observed', r.returncode == 0, 'Loaded state only; last exit retained in commands and is not recovery proof')
source_refs = run(['git','--git-dir=/Users/macstudio3/Projects/.git','show-ref']).stdout
source_index = sha('/Users/macstudio3/Projects/.git/index')
check('root master remains preserved', 'c539b5a08a4f84a551d8825a0fc73fa588bcbce8 refs/heads/master' in source_refs, dict(refs=source_refs, index_sha256=source_index, snapshot_changed_since_baseline=source_refs != baseline['source_refs'], index_changed_since_baseline=source_index != baseline['source_index_sha256'], scope='Existing snapshot job can change source; retained original is historical'))
plist = pathlib.Path('/Users/macstudio3/Library/LaunchAgents/com.steve.codex-yoloforever.plist')
schedule = plistlib.loads(plist.read_bytes())
r = run(['launchctl','list','com.steve.codex-yoloforever'])
check('existing 600-second runner remains active', schedule.get('StartInterval') == 600 and r.returncode == 0 and not (ROOT/'data/codex-yoloforever/STOPPED').exists(), dict(interval=schedule.get('StartInterval'), plist_sha256=sha(plist), future_worker_exit='Observed by runner after return'))
if '--pricing' in sys.argv:
    sql = P/'11250-pricing.sql'
    r = run(['/opt/homebrew/bin/psql','-X','-w','-h','/tmp','-d','dw_unified','-At','--set','ON_ERROR_STOP=1','-f',str(sql)])
    rows = [json.loads(s) for s in r.stdout.splitlines() if s.startswith('{')]
    check('pricing read-only transaction and rollback', r.returncode == 0 and len(rows) == 2 and rows[0]['read_only'] == 'on' and r.stdout.rstrip().endswith('ROLLBACK'), rows)
    check('known mirror has no usable prices', len(rows) == 2 and rows[0]['rows'] == 1470 and rows[0]['positive_cost'] == rows[0]['positive_price'] == rows[1]['price_columns'] == 0, 'Local mirror only; authoritative vendor price lists and identity still missing')
reconstruction = P/'parent-reconstruction-proof.json'
if reconstruction.exists():
    accepted = json.loads(reconstruction.read_text())
    candidate = accepted['candidate']
    r = run(['git','--git-dir='+candidate,'-c','core.commitGraph=false','fsck','--connectivity-only','--no-dangling'])
    check('final corrected candidate native connectivity', r.returncode == 0 and accepted['verdict'] == 'PASS', 'Historical candidate only; full independent restoration proof retained')
    r = run(['du','-sk',candidate]); candidate_kib = int(r.stdout.split()[0])
    allocation = json.loads((P/'parent-allocation-drift-proof.json').read_text())
    check('final corrected candidate allocation below2850MB with verified content identity', r.returncode == 0 and math.floor(candidate_kib/1024+.5) < 2850 and allocation['verdict'] == 'PASS', dict(kib=candidate_kib, canary_mb=math.floor(candidate_kib/1024+.5), earlier_kib=1492032, allocation_drift_proof=str(P/'parent-allocation-drift-proof.json'), physically_freed=False))
    cli = json.loads((P/'parent-independent-cli-proof.json').read_text())
    check('tested verifier bytes remain unchanged', sha(P/'ops-object-verifier-v1.py') == cli['verifier_sha256'] and cli['passes'] == 6, 'Actual CLI independent tests retained; no unnecessary full60GB repeat')
else:
    checks.append(dict(check='full restored repository and five-path physical size target',verdict='SKIP',detail='Historical restoration/size proof still pending'))
for name, reason in [('full fsck semantics and external Gitlink recovery','Object/hash/connectivity and byte-complete historical Git restore do not prove external repositories or every full-fsck semantic'),('live source recovery and scheduler success','Exact approval absent; existing failure remains visible'),('current authoritative pricing and sellability','No external catalog action authorized'),('worker exit','Runner observes after this turn ends')]:
    checks.append(dict(check=name, verdict='SKIP', detail=reason))
out = dict(timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(),intent='Prove bounded local progress and read-only operational monitoring',risk_tier='R1 local evidence; R4 source recovery held',environment='macstudio3',build_identity=baseline['repository_head'],baseline=str(P/'preconditions.json'),checks=checks,commands=commands,counts={v:sum(c['verdict']==v for c in checks) for v in ['PASS','FAIL','SKIP']},ui='No runnable UI produced; real Git/canary classifier/API/auth-negative and optional READ ONLY SQL substitute for CTA and screenrecord',cleanup='All scratch retained; pricing transaction rolls back; no source/runtime edits')
name = next((a.split('=',1)[1] for a in sys.argv if a.startswith('--output=')), 'parent-e2e-proof.json')
(P/name).write_text(json.dumps(out,indent=2)+'\n')
print(json.dumps(dict(evidence=str(P/name),counts=out['counts'],current_root_mb=mb)))
sys.exit(bool(out['counts']['FAIL']))