"""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()