scripts/label_anchor.py

"""Phase 5 plan 05-05: interactive label confirmation → reality_anchor.parquet.

Reads JSONL from ``export_dogfood.py``; for each row, surfaces the v1.0.0
classifier's predicted ``top_class`` and asks the owner to confirm or correct.
Writes a Parquet file matching the ``data/eval.parquet`` column schema PLUS a
``consent_level`` column (D-ANCHOR-03).

Reality Anchor parquet column schema (mirrors eval.parquet + consent_level):
    ts                  : double
    predicted_class     : string
    true_class          : string
    consent_level       : string
    schema_version      : string
    telemetry_json      : string (the full opted-in telemetry window)
    verdict_json        : string (the v1.0.0 verdict at diagnosis time)

The columns intentionally do NOT mirror eval.parquet's per-feature columns
(``rssi_dbm``, ``bssid``, etc) verbatim — eval.parquet stores one row per
telemetry frame, whereas the Reality Anchor stores one row per *diagnosis*
(a whole window of frames). The eval-pipeline schema verified at task-1 step 1
informs the auxiliary lints; the Reality-Anchor file is the diagnosis-grained
projection ``eval_reality_anchor.py`` needs.

Exit codes:
- 0 — parquet written
- 2 — input JSONL missing OR empty (no rows to label)
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq

# Canonical class slug list — mirrors model.features.CLASSES (Phase 2 D-CAL-08).
# Kept in lockstep with eval_reality_anchor.CLASSES.
CLASSES = [
    "auth_8021x_eap_fail",
    "ap_roam_rekey_fail",
    "radius_timeout",
    "captive_portal_expiry",
    "mac_randomization_reject",
    "dhcp_lease_churn",
    "dns_resolver_fail",
    "driver_power_save_wake",
    "rf_sticky_client",
    "isp_upstream_fail",
]


def _prompt_label(predicted: str, headline: str) -> str:
    print(f"\n  predicted: {predicted}")
    print(f"  narrator: {headline[:120]}")
    ans = input("  accept predicted? [Y/n/? to list classes]: ").strip().lower()
    if ans in ("?", "list", "show"):
        for i, c in enumerate(CLASSES):
            print(f"    [{i}] {c}")
        idx = input("  enter class index: ").strip()
        return CLASSES[int(idx)]
    if ans in ("", "y", "yes"):
        return predicted
    print("  classes:")
    for i, c in enumerate(CLASSES):
        print(f"    [{i}] {c}")
    idx = input("  enter class index: ").strip()
    return CLASSES[int(idx)]


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(
        description="Label opted-in real diagnoses → data/reality_anchor.parquet"
    )
    ap.add_argument("--in", dest="inp", required=True, type=Path,
                    help="Input JSONL from export_dogfood.py")
    ap.add_argument("--out", required=True, type=Path,
                    help="Output Parquet path (e.g. data/reality_anchor.parquet)")
    ap.add_argument("--non-interactive", action="store_true",
                    help="Accept every predicted top_class as the label (CI smoke).")
    args = ap.parse_args(argv)

    if not args.inp.exists():
        print(f"input JSONL not found: {args.inp}", file=sys.stderr)
        return 2

    rows: list[dict] = []
    with args.inp.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            raw = json.loads(line)
            v = json.loads(raw["verdict_json"])
            predicted = v.get("top_class", "")
            headline = v.get("headline", "")
            if args.non_interactive:
                label = predicted
            else:
                label = _prompt_label(predicted, headline)
            rows.append(
                {
                    "ts": float(raw["ts"]),
                    "predicted_class": predicted,
                    "true_class": label,
                    "consent_level": raw["consent_level"],
                    "schema_version": raw["schema_version"],
                    "telemetry_json": raw["telemetry_json"],
                    "verdict_json": raw["verdict_json"],
                }
            )

    if not rows:
        print("no rows to label (input JSONL was empty)", file=sys.stderr)
        return 2

    table = pa.Table.from_pylist(rows)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    pq.write_table(table, args.out)
    print(f"wrote {len(rows)} rows to {args.out}")
    return 0


if __name__ == "__main__":
    sys.exit(main())