source/scripts/select_strata_headquotient_v1_1_groups.py

#!/usr/bin/env python3
"""Solve the exact linearized joint group-selection problem."""

from __future__ import annotations

import argparse
from datetime import datetime, timezone
import hashlib
import itertools
import json
from pathlib import Path
import sys

import numpy as np
from scipy.optimize import Bounds, LinearConstraint, milp
from scipy.sparse import coo_matrix


ROOT = Path(__file__).resolve().parents[1]
CONFIG = ROOT / "configs/experiments/strata_headquotient_v1_1.json"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("frontier", choices=("Q10", "Q20", "Q25", "Q30"))
    parser.add_argument("--config", type=Path, default=CONFIG)
    parser.add_argument("--time-limit", type=float, default=900)
    parser.add_argument("--relative-gap", type=float, default=0.30)
    parser.add_argument(
        "--allow-direct-replication",
        action="store_true",
        help="Allow a fresh replication to solve Q25 without rerunning Q10/Q20.",
    )
    return parser.parse_args()


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(16 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def group(row: dict) -> tuple[int, int]:
    return int(row["layer"]), int(row["head"])


def fit_interaction_scale(
    nodes: dict[tuple[int, int], dict],
    pairs: dict[tuple[tuple[int, int], tuple[int, int]], dict],
    triples: list[dict],
    metric: str,
) -> dict[str, float]:
    pair_sum = []
    residual = []
    for triple in triples:
        members = [group(row) for row in triple["groups"]]
        node_cost = sum(float(nodes[item]["lm_delta"][metric]) for item in members)
        interaction = sum(
            float(pairs[tuple(sorted(edge))]["lm_interaction"][metric])
            for edge in itertools.combinations(members, 2)
        )
        pair_sum.append(interaction)
        residual.append(float(triple["lm_delta"][metric]) - node_cost)
    x = np.asarray(pair_sum)
    y = np.asarray(residual)
    denominator = max(float(x @ x), 1e-12)
    scale = float((x @ y) / denominator)
    prediction = scale * x
    correlation = float(np.corrcoef(prediction, y)[0, 1]) if np.std(prediction) else 0.0
    return {
        "scale": scale,
        "correlation": correlation,
        "rmse": float(np.sqrt(np.mean((prediction - y) ** 2))),
    }


def main() -> None:
    args = parse_args()
    config = json.loads(args.config.read_text(encoding="utf-8"))
    output = Path(config["output_root"])
    interactions = json.loads((output / "interactions/result.json").read_text(encoding="utf-8"))
    if not interactions["passed"]:
        raise RuntimeError("joint selection is closed because the interaction graph is incomplete")
    frontier_index = config["frontier"].index(args.frontier)
    direct_replication = False
    if frontier_index > 0:
        prior_name = config["frontier"][frontier_index - 1]
        prior = output / f"frontier/{prior_name}/result.json"
        direct_replication = bool(
            args.allow_direct_replication
            and args.frontier == "Q25"
            and config["scoped_training"].get("fresh_initialization", False)
            and "REP" in config["program"]
        )
        if (
            not direct_replication
            and (
                not prior.is_file()
                or not json.loads(prior.read_text(encoding="utf-8"))["quality_pass"]
            )
        ):
            raise RuntimeError(f"{args.frontier} is closed until {prior_name} passes")

    node_payload = json.loads(Path(interactions["artifacts"]["nodes"]["path"]).read_text(encoding="utf-8"))
    node_rows = node_payload["nodes"]
    candidates = [group(row) for row in node_rows]
    nodes = {group(row): row for row in node_rows}
    pair_rows = [
        json.loads(line)
        for line in Path(interactions["artifacts"]["pairs"]["path"]).read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]
    pairs = {
        tuple(sorted((group(row["groups"][0]), group(row["groups"][1])))): row
        for row in pair_rows
    }
    expected_pairs = len(candidates) * (len(candidates) - 1) // 2
    if len(pairs) != expected_pairs:
        raise RuntimeError(f"interaction graph has {len(pairs)} unique pairs, expected {expected_pairs}")
    triples = json.loads(Path(interactions["artifacts"]["triples"]["path"]).read_text(encoding="utf-8"))
    aggregate_scale = fit_interaction_scale(nodes, pairs, triples, "aggregate")
    late_scale = fit_interaction_scale(nodes, pairs, triples, "4096-8192")

    budget = int(config["budgets"][args.frontier])
    if budget > len(candidates):
        raise RuntimeError(f"{args.frontier} budget exceeds the frozen candidate set")
    n = len(candidates)
    pair_keys = list(itertools.combinations(range(n), 2))
    pair_offset = n
    layer_offset = n + len(pair_keys)
    layers = sorted({layer for layer, _head in candidates})
    layer_index = {layer: layer_offset + index for index, layer in enumerate(layers)}
    variables = layer_offset + len(layers)
    objective = np.zeros(variables)
    semantic_weight = 20.0
    for index, item in enumerate(candidates):
        row = nodes[item]
        objective[index] = (
            float(row["lm_delta"]["aggregate"])
            + 0.5 * float(row["lm_delta"]["4096-8192"])
            + semantic_weight * max(0.0, float(row["semantic_cost"]))
        )
    for pair_position, (first, second) in enumerate(pair_keys):
        row = pairs[tuple(sorted((candidates[first], candidates[second])))]
        objective[pair_offset + pair_position] = (
            aggregate_scale["scale"] * float(row["lm_interaction"]["aggregate"])
            + 0.5 * late_scale["scale"] * float(row["lm_interaction"]["4096-8192"])
            + semantic_weight * max(0.0, float(row["semantic_interaction"]))
        )
    for layer in layers:
        objective[layer_index[layer]] = 5e-3

    row_indices: list[int] = []
    column_indices: list[int] = []
    values: list[float] = []
    lower: list[float] = []
    upper: list[float] = []

    def constraint(coefficients: dict[int, float], lb: float, ub: float) -> None:
        row = len(lower)
        for column, value in coefficients.items():
            row_indices.append(row)
            column_indices.append(column)
            values.append(value)
        lower.append(lb)
        upper.append(ub)

    constraint({index: 1.0 for index in range(n)}, budget, budget)
    for pair_position, (first, second) in enumerate(pair_keys):
        y = pair_offset + pair_position
        constraint({y: 1, first: -1}, -np.inf, 0)
        constraint({y: 1, second: -1}, -np.inf, 0)
        constraint({first: 1, second: 1, y: -1}, -np.inf, 1)
    for index, (layer, _head) in enumerate(candidates):
        constraint({index: 1, layer_index[layer]: -1}, -np.inf, 0)
    for layer in layers:
        members = [
            index for index, item in enumerate(candidates) if item[0] == layer
        ]
        constraint(
            {**{index: 1 for index in members}, layer_index[layer]: -2},
            0,
            np.inf,
        )
    maximum_layers = {"Q10": 5, "Q20": 8, "Q25": 10, "Q30": 15}[args.frontier]
    constraint({layer_index[layer]: 1 for layer in layers}, -np.inf, maximum_layers)

    graph_layer = 23
    graph_indices = [index for index, item in enumerate(candidates) if item[0] == graph_layer]
    constraint({index: 1 for index in graph_indices}, min(10, budget), np.inf)
    for program in range(5):
        eligible = [
            index for index in graph_indices
            if program in [
                int(value) for value in nodes[candidates[index]]["frozen_h1"].get("programs", [])
            ]
        ]
        # The v1 H1 rows predate scoped programs; recover assignments from the
        # frozen scoped node field when no explicit program list exists.
        if not eligible:
            eligible = [
                index for index in graph_indices
                if nodes[candidates[index]]["program"] == (
                    "READ_EVENT_ARG0", "READ_EVENT_ARG1", "READ_EVENT_ARG2",
                    "READ_EVENT_TIME", "READ_EVENT_LOCATION",
                )[program]
                or program == 0
            ]
        if len(eligible) < 2:
            raise RuntimeError(f"program {program} has fewer than two candidate readers")
        constraint({index: 1 for index in eligible}, 2, np.inf)

    parent_similarity = json.loads((
        Path(config["parent_manifest"]).parent
        / "h1/qualification/functional_similarity_graph.json"
    ).read_text(encoding="utf-8"))
    total_by_cluster: dict[int, int] = {}
    cluster_by_group = {}
    for row in parent_similarity:
        cluster = int(row["cluster"])
        total_by_cluster[cluster] = total_by_cluster.get(cluster, 0) + 1
        cluster_by_group[group(row)] = cluster
    for cluster, total in total_by_cluster.items():
        member_indices = [
            index for index, item in enumerate(candidates)
            if cluster_by_group[item] == cluster
        ]
        if member_indices:
            constraint({index: 1 for index in member_indices}, -np.inf, total - 1)

    matrix = coo_matrix(
        (values, (row_indices, column_indices)), shape=(len(lower), variables),
    ).tocsr()
    result = milp(
        c=objective,
        integrality=np.ones(variables),
        bounds=Bounds(np.zeros(variables), np.ones(variables)),
        constraints=LinearConstraint(matrix, np.asarray(lower), np.asarray(upper)),
        options={
            "time_limit": float(args.time_limit),
            "mip_rel_gap": float(args.relative_gap),
            "presolve": True,
            "disp": True,
        },
    )
    if result.x is None or result.status not in (0, 1):
        raise RuntimeError(
            f"branch-and-bound produced no feasible selection: "
            f"status={result.status}, message={result.message}"
        )
    selected = [candidates[index] for index in range(n) if result.x[index] > 0.5]
    if len(selected) != budget:
        raise AssertionError("solver returned the wrong replacement cardinality")
    selected_layers = sorted({layer for layer, _head in selected})
    groups = []
    for item in selected:
        row = nodes[item]
        groups.append({
            "layer": item[0],
            "head": item[1],
            "mode": "LOCAL_GRAPH" if item[0] == graph_layer else "LOCAL",
            "program": row["program"] if item[0] == graph_layer else "NO_GRAPH_READ",
            "node_objective": float(objective[candidates.index(item)]),
            "cluster": cluster_by_group[item],
        })
    plan = {
        "program": config["program"],
        "stage": f"{args.frontier}_exact_joint_selection",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "frontier": args.frontier,
        "budget": budget,
        "candidate_groups": n,
        "groups": groups,
        "selected_layers": selected_layers,
        "selected_layer_count": len(selected_layers),
        "graph_groups": sum(row["mode"] == "LOCAL_GRAPH" for row in groups),
        "local_groups": sum(row["mode"] == "LOCAL" for row in groups),
        "replacement_fraction": budget / 384,
        "global_kv_reduction": 384 / (384 - budget),
        "solver": {
            "method": "scipy.optimize.milp branch-and-bound over linearized signed QUBO",
            "status": int(result.status),
            "message": result.message,
            "objective": float(result.fun),
            "mip_gap": float(result.mip_gap),
            "mip_node_count": int(result.mip_node_count),
            "requested_relative_gap": float(args.relative_gap),
            "globally_optimal": bool(float(result.mip_gap) <= 1e-9),
            "feasible": True,
        },
        "triple_calibration": {
            "aggregate": aggregate_scale,
            "late": late_scale,
        },
        "constraints": {
            "exact_budget": budget,
            "minimum_layer23_graph_groups": min(10, budget),
            "minimum_readers_per_program": 2,
            "retained_representative_per_functional_cluster": True,
            "compact_layer_penalty": 5e-3,
            "maximum_localized_layers": maximum_layers,
            "minimum_groups_per_localized_layer": 2,
            "direct_replication_without_lower_frontiers": bool(direct_replication),
        },
        "interaction_artifacts": interactions["artifacts"],
    }
    result_root = output / f"selection/{args.frontier}"
    result_root.mkdir(parents=True, exist_ok=True)
    destination = result_root / "plan.json"
    destination.write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    manifest = {
        "plan": {"path": str(destination), "sha256": sha256(destination)},
        "optimal": plan["solver"]["globally_optimal"],
        "certified_relative_gap": plan["solver"]["mip_gap"],
        "feasible": True,
        "next_stage": f"EXPORT_{args.frontier}",
    }
    (result_root / "result.json").write_text(
        json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8",
    )
    print(json.dumps({
        "output": str(destination),
        "objective": result.fun,
        "selected_layers": selected_layers,
        "graph_groups": plan["graph_groups"],
        "global_kv_reduction": plan["global_kv_reduction"],
    }, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()