scripts/task2_synthetic_patch_audit.py

#!/usr/bin/env python3
"""Derived Task-2 audit on real released VCG-Bench mxGraph diagrams.

This does not recreate the missing Task-2 release. It constructs deterministic,
machine-checkable edits from real Task-1 XML and tests execution, preservation,
and instruction discrimination for correct/no-op/wrong-target patches.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import random
import subprocess
import time
import xml.etree.ElementTree as ET
from collections import Counter
from pathlib import Path

import pandas as pd

SEED = 20260722
COLORS = {"easy": "#ff6666", "medium": "#6699ff", "hard": "#66cc99"}


def render(drawio: Path, xml_path: Path, png_path: Path) -> tuple[bool, str]:
    """Use the exact Draw.io CLI protocol used by the existing VCG audit."""
    command = [
        "xvfb-run", "-a", str(drawio), "-x", "-f", "png", "-s", "1.0",
        "-o", str(png_path), "--no-sandbox", str(xml_path),
    ]
    try:
        completed = subprocess.run(command, capture_output=True, text=True, timeout=90)
        success = completed.returncode == 0 and png_path.exists() and png_path.stat().st_size > 0
        return success, (completed.stdout + "\n" + completed.stderr).strip()[-2000:]
    except subprocess.TimeoutExpired:
        return False, "Draw.io timed out after 90 seconds"


def set_style(style: str, key: str, value: str) -> str:
    parts = [part for part in style.split(";") if part]
    mapping = {}
    order = []
    for part in parts:
        if "=" not in part:
            continue
        name, old = part.split("=", 1)
        if name not in mapping:
            order.append(name)
        mapping[name] = old
    if key not in mapping:
        order.append(key)
    mapping[key] = value
    return ";".join(f"{name}={mapping[name]}" for name in order) + ";"


def eligible_rows(frame: pd.DataFrame, per_domain: int) -> pd.DataFrame:
    rng = random.Random(SEED)
    selected = []
    for domain, group in sorted(frame.groupby("domain_l1")):
        candidates = []
        for index, row in group.iterrows():
            try:
                root = ET.fromstring(str(row.restored_xml))
            except ET.ParseError:
                continue
            vertices = [cell for cell in root.iter("mxCell") if cell.get("vertex") == "1"]
            if len(vertices) >= 2 and all(vertex.find("mxGeometry") is not None for vertex in vertices[:2]):
                candidates.append(index)
        rng.shuffle(candidates)
        if len(candidates) < per_domain:
            raise RuntimeError((domain, len(candidates)))
        selected.extend(candidates[:per_domain])
    return frame.loc[selected].reset_index(drop=True)


def apply_operations(xml: str, difficulty: str, target_position: int, marker: str) -> tuple[str, dict]:
    root = ET.fromstring(xml)
    vertices = [cell for cell in root.iter("mxCell") if cell.get("vertex") == "1"]
    target = vertices[target_position]
    geometry = target.find("mxGeometry")
    assert geometry is not None
    original = {
        "id": target.get("id"),
        "value": target.get("value", ""),
        "style": target.get("style", ""),
        "x": float(geometry.get("x", "0")),
        "y": float(geometry.get("y", "0")),
        "width": float(geometry.get("width", "0")),
        "height": float(geometry.get("height", "0")),
    }
    expected = {"target_id": target.get("id"), "color": COLORS[difficulty]}
    target.set("style", set_style(target.get("style", ""), "fillColor", COLORS[difficulty]))
    operations = ["fill_color"]
    if difficulty in {"medium", "hard"}:
        expected["value"] = f"{original['value']} [{marker}]"
        expected["width"] = round(original["width"] * 1.2, 6)
        target.set("value", expected["value"])
        geometry.set("width", f"{expected['width']:g}")
        operations.extend(["change_text", "resize_width"])
    if difficulty == "hard":
        expected["x"] = round(original["x"] + 30.0, 6)
        expected["y"] = round(original["y"] + 20.0, 6)
        expected["height"] = round(original["height"] * 1.1, 6)
        geometry.set("x", f"{expected['x']:g}")
        geometry.set("y", f"{expected['y']:g}")
        geometry.set("height", f"{expected['height']:g}")
        operations.extend(["move_x", "move_y", "resize_height"])
    expected["operations"] = operations
    expected["original_target"] = original
    return ET.tostring(root, encoding="unicode"), expected


def style_value(style: str, key: str) -> str | None:
    for part in style.split(";"):
        if part.startswith(key + "="):
            return part.split("=", 1)[1]
    return None


def exact_xdrfr(xml: str, expected: dict) -> tuple[float, list[dict]]:
    root = ET.fromstring(xml)
    target = next(cell for cell in root.iter("mxCell") if cell.get("id") == expected["target_id"])
    geometry = target.find("mxGeometry")
    assert geometry is not None
    checks = [("fill_color", style_value(target.get("style", ""), "fillColor") == expected["color"])]
    if "value" in expected:
        checks.extend(
            [
                ("change_text", target.get("value") == expected["value"]),
                ("resize_width", abs(float(geometry.get("width", "nan")) - expected["width"]) < 1e-6),
            ]
        )
    if "x" in expected:
        checks.extend(
            [
                ("move_x", abs(float(geometry.get("x", "nan")) - expected["x"]) < 1e-6),
                ("move_y", abs(float(geometry.get("y", "nan")) - expected["y"]) < 1e-6),
                ("resize_height", abs(float(geometry.get("height", "nan")) - expected["height"]) < 1e-6),
            ]
        )
    details = [{"question": name, "is_satisfied": bool(ok)} for name, ok in checks]
    return sum(ok for _, ok in checks) / len(checks), details


def cell_signatures(xml: str) -> dict[str, str]:
    root = ET.fromstring(xml)
    return {
        cell.get("id"): ET.tostring(cell, encoding="unicode")
        for cell in root.iter("mxCell")
        if cell.get("id") is not None
    }


def preservation_score(original: str, modified: str, target_id: str) -> float:
    left, right = cell_signatures(original), cell_signatures(modified)
    ids = sorted((set(left) & set(right)) - {target_id})
    return sum(left[item] == right[item] for item in ids) / len(ids) if ids else 1.0


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--parquet", type=Path, default=Path("dataset/train.parquet"))
    parser.add_argument("--drawio", type=Path, default=Path("tools/squashfs-root/drawio"))
    parser.add_argument("--output", type=Path, default=Path("results/task2_derived_audit"))
    parser.add_argument("--per-domain", type=int, default=3)
    args = parser.parse_args()
    root_dir = Path(__file__).resolve().parent
    parquet = (root_dir / args.parquet).resolve()
    drawio = (root_dir / args.drawio).resolve()
    output = (root_dir / args.output).resolve()
    xml_dir, png_dir = output / "xml", output / "rendered"
    xml_dir.mkdir(parents=True, exist_ok=True)
    png_dir.mkdir(parents=True, exist_ok=True)

    selected = eligible_rows(pd.read_parquet(parquet), args.per_domain)
    rows = []
    for sample_index, row in selected.iterrows():
        original = str(row.restored_xml)
        for difficulty in ("easy", "medium", "hard"):
            case_id = f"{row.image_id}__{difficulty}"
            correct, expected = apply_operations(original, difficulty, 0, case_id)
            wrong, _ = apply_operations(original, difficulty, 1, case_id)
            variants = {"correct_patch": correct, "no_op": original, "wrong_target": wrong}
            for variant, xml in variants.items():
                score, checks = exact_xdrfr(xml, expected)
                rows.append(
                    {
                        "case_id": case_id,
                        "image_id": str(row.image_id),
                        "domain_l1": str(row.domain_l1),
                        "domain_l2": str(row.domain_l2),
                        "difficulty": difficulty,
                        "operation_count": len(expected["operations"]),
                        "variant": variant,
                        "exact_xdrfr": score,
                        "satisfied": sum(item["is_satisfied"] for item in checks),
                        "questions": len(checks),
                        "untouched_cell_preservation": preservation_score(original, xml, expected["target_id"]),
                        "xml_parse_success": True,
                    }
                )
            xml_path = xml_dir / f"{case_id}.drawio"
            png_path = png_dir / f"{case_id}.png"
            xml_path.write_text(correct, encoding="utf-8")
            started = time.perf_counter()
            success, log = render(drawio, xml_path, png_path)
            rows[-3]["render_success"] = success
            rows[-3]["render_seconds"] = time.perf_counter() - started
            rows[-3]["render_sha256"] = sha256(png_path) if success else None
            rows[-3]["render_log_tail"] = log[-400:]
            print(f"[{sample_index + 1}/{len(selected)}] {case_id} render={success}", flush=True)

    frame = pd.DataFrame(rows)
    frame.to_csv(output / "per_task.csv", index=False)
    correct = frame[frame.variant == "correct_patch"]
    summary = {
        "scope": "derived synthetic Task-2 edits on real released Task-1 mxGraph XML; not the missing official Task-2 benchmark",
        "seed": SEED,
        "source_rows": len(selected),
        "domains": dict(Counter(selected.domain_l1)),
        "derived_edit_tasks": len(correct),
        "difficulty_counts": dict(Counter(correct.difficulty)),
        "operation_counts": sorted(correct.operation_count.unique().tolist()),
        "correct_patch": {
            "drawio_render_successes": int(correct.render_success.sum()),
            "execution_success_rate": float(correct.render_success.mean()),
            "mean_exact_xdrfr": float(correct.exact_xdrfr.mean()),
            "mean_untouched_cell_preservation": float(correct.untouched_cell_preservation.mean()),
        },
        "controls": {
            variant: {
                "mean_exact_xdrfr": float(group.exact_xdrfr.mean()),
                "perfect_instruction_following": int((group.exact_xdrfr == 1).sum()),
            }
            for variant, group in frame[frame.variant != "correct_patch"].groupby("variant")
        },
        "discrimination_gap_correct_minus_no_op": float(
            correct.exact_xdrfr.mean() - frame[frame.variant == "no_op"].exact_xdrfr.mean()
        ),
        "limitations": [
            "Task-2 source parquet and author model outputs were not released.",
            "Edits are deterministic oracle patches, not LLM-generated patches.",
            "XDRFR questions are evaluated by exact XML property checks rather than the paper's Gemini judge.",
            "Untouched-cell XML preservation is a deterministic style-preservation proxy, not VLM SCS.",
        ],
    }
    (output / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
    assert summary["correct_patch"]["execution_success_rate"] == 1.0
    assert summary["correct_patch"]["mean_exact_xdrfr"] == 1.0
    assert summary["correct_patch"]["mean_untouched_cell_preservation"] == 1.0
    assert summary["controls"]["no_op"]["mean_exact_xdrfr"] == 0.0
    assert summary["controls"]["wrong_target"]["mean_exact_xdrfr"] == 0.0
    print(json.dumps(summary, indent=2), flush=True)


if __name__ == "__main__":
    main()