scripts/eval_frontier_adv.py

"""Mac-side: evaluate adversarial images (produced on the Sparks) against the
held-out FRONTIER models gpt-5.5 / gemini-3.5-flash, which were never in the
attack ensemble. Applies JPEG Q85 first (scraper realism), then asks each model
to label clean vs adv and scores semantic distance from the truth.

  python scripts/eval_frontier_adv.py --run runs/v1 --manifest examples/testset.csv \
      --images examples/testset --limit 6
"""

from __future__ import annotations

import argparse
import io
import json
import time
from pathlib import Path

from PIL import Image

import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from veil_pgd.config import get_settings  # noqa: E402
from veil_pgd.fitness.embed import Embedder  # noqa: E402
from veil_pgd.fitness.semantic import embedding_distance  # noqa: E402
from veil_pgd.targets.base import LabelPrompt  # noqa: E402
from veil_pgd.targets.registry import Registry  # noqa: E402

FLIP_TAU = 0.5


def jpeg(img: Image.Image, q=85) -> Image.Image:
    buf = io.BytesIO()
    img.convert("RGB").save(buf, format="JPEG", quality=q)
    buf.seek(0)
    return Image.open(buf).convert("RGB")


def log(m):
    print(f"[{time.strftime('%H:%M:%S')}] {m}", flush=True)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--run", required=True)
    ap.add_argument("--manifest", required=True)
    ap.add_argument("--images", required=True)
    ap.add_argument("--limit", type=int, default=6)
    args = ap.parse_args()

    s = get_settings()
    reg = Registry(s)
    emb = Embedder(reg.embeddings(), s.klaus3_vision_service_url)
    prompt = LabelPrompt()
    blackbox = reg.all_blackbox()
    advdir = Path(args.run) / "adv"

    rows = []
    for line in Path(args.manifest).read_text().splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            p, t = line.split(",", 1)
            rows.append((Path(p).name, t.strip()))
    step = max(1, len(rows) // args.limit)
    rows = rows[::step][:args.limit]

    out = []
    tallies = {m.name: [0, 0] for m in blackbox}
    for fname, truth in rows:
        stem = Path(fname).stem
        adv_fp = advdir / f"{stem}.png"
        if not adv_fp.exists():
            log(f"no adv for {fname}, skip")
            continue
        clean = jpeg(Image.open(Path(args.images) / fname).convert("RGB"))
        adv = jpeg(Image.open(adv_fp).convert("RGB"))
        rec = {"image": fname, "truth": truth, "models": {}}
        for m in blackbox:
            cp = m.label(clean, prompt).parsed_label
            ap_ = m.label(adv, prompt).parsed_label
            cd = embedding_distance(emb, cp, truth)
            ad = embedding_distance(emb, ap_, truth)
            flip = ad >= FLIP_TAU and cd < FLIP_TAU
            rec["models"][m.name] = {"clean": cp, "adv": ap_,
                                     "clean_dist": round(cd, 3), "adv_dist": round(ad, 3),
                                     "flip": flip}
            tallies[m.name][0] += flip; tallies[m.name][1] += 1
            log(f"{fname} {m.name.split('/')[-1]}: {cp!r}->{ap_!r} "
                f"d {cd:.2f}->{ad:.2f} {'FLIP' if flip else ''}")
        out.append(rec)

    (Path(args.run) / "frontier_eval.json").write_text(json.dumps(out, indent=2))
    print("\n===== FRONTIER (held-out) TRANSFER, post-JPEG =====")
    for name, (f, n) in tallies.items():
        print(f"  {name}: flip {f}/{n}")
    print("===================================================")
    reg.close()


if __name__ == "__main__":
    main()