"""Prepare four new research batches, review them, and serve live progress.""" import argparse import hashlib import json import subprocess import sys import threading from collections import Counter from datetime import UTC, datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from dataset import plot_run, write_jsonl from quality_batch import audit_records, build_cases ROOT = Path("data/luna-expansion-2000-001") def read_rows(path): return [json.loads(line) for line in path.read_text().splitlines()] def save_state(phase, error=None): path = ROOT / "state.json" tmp = path.with_suffix(".tmp") tmp.write_text( json.dumps( { "phase": phase, "error": error, "updated_at": datetime.now(UTC).isoformat(), } ) ) tmp.replace(path) def progress(): state = ( json.loads((ROOT / "state.json").read_text()) if (ROOT / "state.json").exists() else {"phase": "Preparing inputs"} ) batches = [] for i in range(1, 5): results = [] for path in (ROOT / f"batch-{i}" / "case-audits").glob("*.json"): try: results.append(json.loads(path.read_text())) except json.JSONDecodeError: continue # A worker may still be writing this case. accepted = sum(r["status"] == "accepted" for r in results) batches.append( { "batch": i, "reviewed": len(results), "accepted": accepted, "quarantined": len(results) - accepted, } ) return { **state, "total": 2000, "reviewed": sum(b["reviewed"] for b in batches), "accepted": sum(b["accepted"] for b in batches), "quarantined": sum(b["quarantined"] for b in batches), "batches": batches, } def prepare(): sample = ROOT / "sources" if not sample.exists(): subprocess.run( [ sys.executable, "prepare_sources.py", "--per-class", "500", "--seed", "20261015", "--output", str(sample), ], check=True, ) pilot = read_rows(Path("data/quality-500-004/candidates-500.jsonl")) used = { ( r["audit"]["source_record"]["source"], r["audit"]["source_record"]["original_id"], ) for r in pilot } sources = read_rows(sample / "source_records.jsonl") by_source = { name: [ r for r in sources if r["source"] == name and (name, r["original_id"]) not in used ][:800] for name in ("saml_d", "baf") } if any(len(rows) != 800 for rows in by_source.values()): raise ValueError("Not enough unused source rows") all_rows = [] for i in range(1, 5): directory = ROOT / f"batch-{i}" directory.mkdir(exist_ok=True) source = directory / "source-records.jsonl" write_jsonl( source, [r for rows in by_source.values() for r in rows[(i - 1) * 200 : i * 200]], ) rows = build_cases(source, seed=20261015 + i, expansion_batch=i) path = directory / "inputs.jsonl" if path.exists() and read_rows(path) != rows: raise ValueError("Existing expansion plan differs; do not overwrite") write_jsonl(path, rows) all_rows.extend(rows) check = audit_records(pilot + all_rows) if check["failures"]: raise ValueError(str(check["failures"])) assert len({r["input"]["case_id"] for r in pilot + all_rows}) == 2500 assert Counter(r["audit"]["mixture_class"] for r in all_rows) == { "suspicious": 1000, "ordinary": 500, "benign_unusual": 500, } source_ids = [ ( r["audit"]["source_record"]["source"], r["audit"]["source_record"]["original_id"], ) for r in all_rows if r["metadata"]["source"]["origin"] in by_source ] assert len(set(source_ids)) == 1600 and not set(source_ids) & used manifest = { "new_cases": 2000, "mixture": dict(Counter(r["audit"]["mixture_class"] for r in all_rows)), "sources": dict(Counter(r["metadata"]["source"]["origin"] for r in all_rows)), "validation": check, "source_sample_manifest": json.loads((sample / "manifest.json").read_text()), "input_hashes": { str(i): hashlib.sha256( (ROOT / f"batch-{i}" / "inputs.jsonl").read_bytes() ).hexdigest() for i in range(1, 5) }, "notes": "Synthetic policy candidates; no human review or observed private-share outcomes. New source rows; existing source-group/month holdouts retained. Template families remain shared across splits.", } (ROOT / "plan.json").write_text(json.dumps(manifest, indent=2)) def run(): try: save_state("Preparing 2,000 new inputs") prepare() for i in range(1, 5): directory = ROOT / f"batch-{i}" limits = (50, 500) if i == 1 else (500,) for limit in limits: save_state(f"Reviewing batch {i}/4 ({limit}-case checkpoint)") with (directory / f"run-{limit}.log").open("a") as log: subprocess.run( [ sys.executable, "review_batch.py", "--backend", "droid-luna", "--explicit-envelope", "--input", str(directory / "inputs.jsonl"), "--output", str(directory), "--limit", str(limit), "--workers", "12", ], stdout=log, stderr=subprocess.STDOUT, check=True, ) report = json.loads((directory / f"report-{limit}.json").read_text()) with Path("devlogs.md").open("a") as log: log.write( f"\nExpansion batch {i}, checkpoint {limit}: {report['accepted']} accepted, {report['rejected']} quarantined; {report['initial_schema_valid']} initial schema-valid; {report['reported_factory_credits']} cumulative batch credits.\n" ) if report["accepted"] / limit < 0.85: raise ValueError( f"Batch {i} acceptance below 85%; inspect checkpoint before scaling further" ) save_state("Verifying and exporting") accepted, quarantined, audits, reports = [], [], [], [] for i in range(1, 5): directory = ROOT / f"batch-{i}" accepted.extend(read_rows(directory / "candidates-500.jsonl")) quarantined.extend(read_rows(directory / "quarantined-500.jsonl")) audits.extend(read_rows(directory / "audits-500.jsonl")) reports.append(json.loads((directory / "report-500.json").read_text())) assert len(accepted) + len(quarantined) == 2000 assert all(r["input_evidence_unchanged"] for r in reports) check = audit_records(accepted) assert not check["failures"], check report = { "attempted": 2000, "accepted": len(accepted), "rejected": len(quarantined), "mixture": dict(Counter(r["audit"]["mixture_class"] for r in accepted)), "fraud_types": dict( Counter(r["metadata"]["label"]["fraud_type"] for r in accepted) ), "risk_levels": dict( Counter(str(r["target"]["risk_level"]) for r in accepted) ), "splits": dict(Counter(r["metadata"]["split"] for r in accepted)), "reported_factory_credits": sum( r["reported_factory_credits"] for r in reports ), "initial_schema_valid": sum(r["initial_schema_valid"] for r in reports), "first_pass_accepted": sum(r["first_pass_accepted"] for r in reports), "batches": reports, "validation": check, } write_jsonl(ROOT / "candidates.jsonl", accepted) write_jsonl(ROOT / "quarantined.jsonl", quarantined) write_jsonl(ROOT / "audits.jsonl", audits) (ROOT / "report.json").write_text(json.dumps(report, indent=2)) artifacts = Path(".amp/in/artifacts") artifacts.mkdir(parents=True, exist_ok=True) for name in ( "candidates.jsonl", "quarantined.jsonl", "audits.jsonl", "report.json", "plan.json", ): (artifacts / ("luna-expansion-" + name)).write_bytes( (ROOT / name).read_bytes() ) plot_run(report, artifacts / "luna-expansion-distribution.png") with Path("devlogs.md").open("a") as log: log.write( f"\nExpansion finished: {len(accepted)}/2000 admitted; {len(quarantined)} quarantined. Report and plots exported. No training or publication.\n" ) save_state("Complete") except Exception as exc: save_state("Stopped: inspect run logs before resuming", str(exc)) with Path("devlogs.md").open("a") as log: log.write( f"\nExpansion stopped: {exc}. Cached responses preserved; no blind retry.\n" ) raise PAGE = """Luna dataset expansion

Reviewing 2,000 new research cases with Luna

Connecting to live progress...

0 / 2,000 reviewed

Accepted: 0

Quarantined: 0

Batch	Reviewed	Accepted	Quarantined

Waiting for status

Refreshes every 5 seconds from saved case audits. Accepted means synthetic-policy consistent, not confirmed real-world fraud. The original 480 candidates are not included in this bar.

""" class Handler(BaseHTTPRequestHandler): def do_GET(self): if self.path.split("?")[0] not in {"/", "/progress.json"}: self.send_error(404) return is_json = self.path.split("?")[0] == "/progress.json" body = json.dumps(progress()).encode() if is_json else PAGE.encode() self.send_response(200) self.send_header( "Content-Type", "application/json" if is_json else "text/html; charset=utf-8", ) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def log_message(self, *args): pass if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--run", action="store_true") parser.add_argument("--prepare-only", action="store_true") parser.add_argument("--port", type=int, default=8765) args = parser.parse_args() ROOT.mkdir(parents=True, exist_ok=True) if args.prepare_only: prepare() else: if args.run: threading.Thread(target=run, daemon=True).start() ThreadingHTTPServer(("0.0.0.0", args.port), Handler).serve_forever()