# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Run one treatment arm (E–J) on a subsample of properties and write the standard artefacts:

results/runs/<run_id>/
    predictions.parquet      one row per (tx_id, aggregation rule) — contract in docs/architecture.md §1
    transcripts/<tx_id>.json DebateTranscript (every turn, tokens, cost, prompt hash, errors)
    packets/<tx_id>.json     the full EvidencePacket shown (filtered per role) to the agents
    manifest.json            arm, committee, models, effort, prompt hashes, config hashes, timings, cost, counts

Resumable: properties with an existing transcript are skipped and predictions are rebuilt from all
transcripts in the directory. Mock provider unless ``live=True`` AND keys are present. The ledger's
spend cap (AV_MAX_SPEND_USD) stops the run gracefully: what exists is written, nothing is fabricated.

Feature functions (comparables, market, geo) are imported lazily from ``agentic_valuation.features``;
when that package is not available yet, thin local fallbacks (``_simple_comps``, ``_simple_market``)
keep the smoke tests running on the synthetic parquet.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import math
import platform
import subprocess
from dataclasses import asdict, dataclass
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import polars as pl

from .. import __version__
from ..agents.packets import build_packet
from ..agents.roles import Role, placebo_roles, role_prompt_hashes, roles_for
from ..agents.runner import ProviderPool
from ..config import CONFIGS, PROCESSED, RESULTS, ROOT, SEED, SYNTHETIC, load_config, provider_mode
from ..debate.aggregation import aggregate_intervals, load_weights
from ..debate.committees import COMMITTEES, assign, committee_models
from ..debate.protocols import cost_matched_single, independent, structured_debate
from ..providers import CostLedger
from ..providers.ledger import SpendCapExceeded
from ..schemas import Comparable, DebateTranscript, EvidencePacket, GeoContext, MarketContext

PREDICTION_COLUMNS: tuple[str, ...] = ("tx_id", "split", "arm", "committee", "aggregation", "point", "q05", "q25", "q50", "q75", "q95", "cost_usd", "latency_s", "tokens", "run_id")
LLM_ARMS = ("E_single_llm", "F_independent", "G_debate", "H_debate_skeptic", "I_placebo", "J_cost_matched", "G_fullview")


@dataclass
class RunSpec:
    arm: str = "F_independent"
    rounds: int | None = None
    committee: str = "all_anthropic"
    split: str = "validation"
    n: int = 24
    live: bool = False
    tier: str | None = None
    run_id: str | None = None
    synthetic: bool = False
    seed: int = SEED
    effort: str | None = None
    property_concurrency: int = 4
    out_dir: Path | None = None
    weights_path: Path | None = None
    max_tokens: int = 4096
    data_dir: Path | None = None  # override of data/<processed|synthetic> (e.g. the prospective bundle); None = default layout

    def resolved_run_id(self) -> str:
        if self.run_id:
            return self.run_id
        parts = [self.arm]
        if self.rounds is not None and self.arm in ("G_debate", "H_debate_skeptic", "I_placebo", "G_fullview"):
            parts.append(f"r{self.rounds}")
        parts += [self.committee, self.split, "synthetic" if self.synthetic else "real", "live" if self.live else "mock"]
        rid = "_".join(parts)
        # mock or synthetic runs are never empirical results: a leading underscore makes scripts/20_evaluate.py skip them
        return rid if (self.live and not self.synthetic) else f"_{rid}"


# ---------------------------------------------------------------------------------------------------
# data
# ---------------------------------------------------------------------------------------------------
def _assign_split_by_date(df: pl.DataFrame, data_cfg: dict[str, Any]) -> pl.DataFrame:
    """Synthetic data has no splits parquet: derive the chronological split from configs/data.yaml."""
    expr = pl.lit("unused")
    for name, w in data_cfg["splits"].items():
        if not isinstance(w, dict) or "start" not in w:
            continue
        s, e = date.fromisoformat(w["start"]), date.fromisoformat(w["end"])
        expr = pl.when((pl.col("date") >= s) & (pl.col("date") <= e)).then(pl.lit(name)).otherwise(expr)
    return df.with_columns(expr.alias("split"))


def data_base(synthetic: bool, data_dir: Path | None = None) -> Path:
    return Path(data_dir) if data_dir is not None else (SYNTHETIC if synthetic else PROCESSED)


def load_analytic(synthetic: bool, data_dir: Path | None = None) -> pl.DataFrame:
    """data/<base>/analytic.parquet joined with splits.parquet (base = synthetic | processed | ``data_dir``). When the
    synthetic analytic layer has not been built yet, it is derived on the fly from synthetic_transactions.parquet."""
    base = data_base(synthetic, data_dir)
    path = base / "analytic.parquet"
    if path.exists():
        df = pl.read_parquet(path)
    elif synthetic:
        raw = SYNTHETIC / "synthetic_transactions.parquet"
        if not raw.exists():
            from ..data.synthetic import write_default

            write_default()
        from ..data.analytic_schema import add_derived_columns

        df = add_derived_columns(pl.read_parquet(raw))
    else:
        raise FileNotFoundError(f"{path} not found: run `av build` (scripts/01_build_analytic.py) first, or use --synthetic")
    if "split" not in df.columns:
        sp = base / "splits.parquet"
        if sp.exists():
            df = df.join(pl.read_parquet(sp).select("tx_id", "split"), on="tx_id", how="left")
        elif synthetic:
            df = _assign_split_by_date(df, load_config("data"))
        else:
            raise FileNotFoundError(f"{sp} not found: run `av splits` first")
    return df


def select_subsample(df: pl.DataFrame, split: str, n: int, seed: int, synthetic: bool, data_dir: Path | None = None) -> pl.DataFrame:
    """Properties to value: the first n (sorted by tx_id) of data/<base>/llm_subsamples.parquet (tx_id, split,
    stratum — drawn once by scripts/02_splits.py); when that file is absent, a seeded random draw within the split."""
    sub_path = data_base(synthetic, data_dir) / "llm_subsamples.parquet"
    if sub_path.exists():
        sub = pl.read_parquet(sub_path).filter(pl.col("split") == split)
        ids = sorted(sub["tx_id"].to_list())[:n]
        out = df.filter(pl.col("tx_id").is_in(ids)).sort("tx_id")
        if out.height:
            return out
    pool = df.filter(pl.col("split") == split)
    if "exclude_reason" in pool.columns:
        pool = pool.filter(pl.col("exclude_reason").is_null())
    if pool.height == 0:
        raise ValueError(f"no property in split {split!r}")
    k = min(n, pool.height)
    return pool.sort("tx_id").sample(n=k, seed=seed, shuffle=True).sort("tx_id")


# ---------------------------------------------------------------------------------------------------
# features: lazy import with local fallbacks
# ---------------------------------------------------------------------------------------------------
def _haversine_expr(lat: float, lng: float) -> pl.Expr:
    r = 6371.0
    lat1, lng1 = math.radians(lat), math.radians(lng)
    dlat = pl.col("lat").radians() - lat1
    dlng = pl.col("lng").radians() - lng1
    a = (dlat / 2).sin() ** 2 + math.cos(lat1) * pl.col("lat").radians().cos() * (dlng / 2).sin() ** 2
    return 2 * r * a.sqrt().arcsin()


def _simple_comps(analytic: pl.DataFrame, subject: dict[str, Any], cutoff: date, cfg: dict[str, Any]) -> list[Comparable]:
    """Fallback retrieval: same class, date < cutoff, within the lookback and 10 km, nearest first, top max_comps."""
    c = cfg.get("comparables", {})
    radius = max(c.get("radius_km", [10.0])) if isinstance(c.get("radius_km"), list) else float(c.get("radius_km", 10.0))
    lookback = int(c.get("lookback_days", 540))
    max_comps = int(c.get("max_comps", 12))
    val_date = cutoff + timedelta(days=1)
    df = (
        analytic.filter(
            (pl.col("date") < cutoff)
            & (pl.col("date") >= cutoff - timedelta(days=lookback))
            & (pl.col("property_class") == subject["property_class"])
            & (pl.col("tx_id") != subject["tx_id"])
        )
    )
    if "parcel_id" in df.columns and subject.get("parcel_id") is not None:
        df = df.filter(pl.col("parcel_id") != subject["parcel_id"])
    if "exclude_reason" in df.columns:
        df = df.filter(pl.col("exclude_reason").is_null())
    df = df.with_columns(_haversine_expr(float(subject["lat"]), float(subject["lng"])).alias("distance_km")).filter(pl.col("distance_km") <= radius).sort("distance_km").head(max_comps)
    out: list[Comparable] = []
    for i, r in enumerate(df.iter_rows(named=True)):
        area = r.get("floor_area_m2")
        roll = r.get("roll_total_value")
        sub = r.get("subtype") or "unknown"
        out.append(
            Comparable(
                evidence_id=f"C{i + 1:02d}",
                sale_date=r["date"],
                price=float(r["price"]),
                distance_km=round(float(r["distance_km"]), 2),
                property_class=r["property_class"],
                subtype=sub if sub in ("detached", "semi", "row", "integrated", "unknown") else "unknown",
                municipality=str(r.get("mun_name") or "unknown"),
                floor_area_m2=_num(area),
                land_area_m2=_num(r.get("land_area_m2")),
                year_built=(int(r["year_built"]) if r.get("year_built") is not None else None),
                storeys=_num(r.get("storeys")),
                n_units=(int(r["n_units"]) if r.get("n_units") is not None else None),
                roll_total_value=_num(roll),
                price_per_m2=(round(float(r["price"]) / float(area), 1) if area else None),
                assessment_ratio=(round(float(r["price"]) / float(roll), 3) if roll else None),
                months_before_valuation=round((val_date - r["date"]).days / 30.44, 1),
            )
        )
    return out


def _num(v: Any) -> float | None:
    if v is None:
        return None
    try:
        x = float(v)
    except (TypeError, ValueError):
        return None
    return None if math.isnan(x) else x


def _median(s: pl.Series) -> float | None:
    s = s.drop_nulls()
    return float(s.median()) if s.len() else None


def _simple_market(analytic: pl.DataFrame, subject: dict[str, Any], cutoff: date, cfg: dict[str, Any]) -> MarketContext:
    """Fallback market context from sales dated <= cutoff only."""
    past = analytic.filter(pl.col("date") <= cutoff)
    if "exclude_reason" in past.columns:
        past = past.filter(pl.col("exclude_reason").is_null())
    y1 = cutoff - timedelta(days=365)
    y2 = cutoff - timedelta(days=730)
    mun = past.filter(pl.col("mun_name") == subject.get("mun_name"))
    mun12 = mun.filter(pl.col("date") > y1)
    reg = past.filter(pl.col("region_name") == subject.get("region_name"))
    reg12 = reg.filter(pl.col("date") > y1)
    reg_prev = reg.filter((pl.col("date") > y2) & (pl.col("date") <= y1))
    local = past.with_columns(_haversine_expr(float(subject["lat"]), float(subject["lng"])).alias("d")).filter((pl.col("d") <= 3.0) & (pl.col("date") > y1))
    ppm2 = mun12.filter(pl.col("floor_area_m2") > 0).with_columns((pl.col("price") / pl.col("floor_area_m2")).alias("ppm2"))
    ratio = local.filter(pl.col("roll_total_value") > 0).with_columns((pl.col("price") / pl.col("roll_total_value")).alias("ar"))
    m_now, m_prev = _median(reg12["price"]), _median(reg_prev["price"])
    yoy = (m_now / m_prev - 1.0) if (m_now and m_prev) else None
    # monthly municipal index: median price per m2 by month relative to the first month with >= 5 sales
    idx: list[dict[str, Any]] = []
    if mun.height:
        mm = (
            mun.filter(pl.col("floor_area_m2") > 0)
            .with_columns((pl.col("price") / pl.col("floor_area_m2")).alias("ppm2"), pl.col("date").dt.strftime("%Y-%m").alias("month"))
            .group_by("month")
            .agg(pl.col("ppm2").median().alias("med"), pl.len().alias("n"))
            .filter(pl.col("n") >= 5)
            .sort("month")
        )
        if mm.height:
            base = float(mm["med"][0])
            idx = [{"month": r["month"], "index": round(float(r["med"]) / base * 100, 1)} for r in mm.tail(36).iter_rows(named=True)]
    return MarketContext(
        as_of=cutoff,
        municipality_median_price_12m=_median(mun12["price"]),
        municipality_median_ppm2_12m=(round(_median(ppm2["ppm2"]), 1) if ppm2.height else None),
        municipality_n_sales_12m=int(mun12.height),
        municipality_price_index=idx,
        region_median_price_12m=m_now,
        region_yoy_change=(round(yoy, 4) if yoy is not None else None),
        local_liquidity_n_sales_3km_12m=int(local.height),
        local_median_assessment_ratio_12m=(round(_median(ratio["ar"]), 3) if ratio.height else None),
        mortgage_rate_5y_fixed=None,
    )


def _prior_sale(analytic: pl.DataFrame, row: dict[str, Any], cutoff: date) -> tuple[float | None, date | None]:
    if "parcel_id" not in analytic.columns or row.get("parcel_id") is None:
        return None, None
    prev = analytic.filter((pl.col("parcel_id") == row["parcel_id"]) & (pl.col("date") < cutoff) & (pl.col("tx_id") != row["tx_id"])).sort("date", descending=True).head(1)
    if prev.height == 0:
        return None, None
    return float(prev["price"][0]), prev["date"][0]


@dataclass
class FeatureBackend:
    comps: Any
    market: Any
    geo: Any
    source: str
    geo_cache: pl.DataFrame | None = None


def feature_backend(synthetic: bool, data_dir: Path | None = None) -> FeatureBackend:
    """Lazy import of agentic_valuation.features; local fallbacks when it is not importable."""
    geo_cache = None
    gp = data_base(synthetic, data_dir) / "geo_features.parquet"
    if gp.exists():
        try:
            geo_cache = pl.read_parquet(gp)
        except Exception:
            geo_cache = None
    try:
        from ..features.comparables import retrieve_comparables  # type: ignore
        from ..features.market import market_context  # type: ignore

        try:
            from ..features.geo import geo_context  # type: ignore
        except Exception:
            geo_context = None
        return FeatureBackend(retrieve_comparables, market_context, geo_context, "agentic_valuation.features", geo_cache)
    except Exception as e:
        return FeatureBackend(_simple_comps, _simple_market, None, f"run_arm fallback (_simple_comps/_simple_market, no geo; features import failed: {type(e).__name__})", None)


def build_packet_for(analytic: pl.DataFrame, row: dict[str, Any], data_cfg: dict[str, Any], fb: FeatureBackend) -> EvidencePacket:
    cutoff = row["date"] - timedelta(days=1)
    row = dict(row)
    if row.get("prior_sale_price") is None:
        row["prior_sale_price"], row["prior_sale_date"] = _prior_sale(analytic, row, cutoff)
    comps = fb.comps(analytic, row, cutoff, data_cfg)
    market = fb.market(analytic, row, cutoff, data_cfg)
    geo: GeoContext | None = None
    if fb.geo is not None and fb.geo_cache is not None:
        try:
            geo = fb.geo(row, None, cached=fb.geo_cache)
        except Exception:
            geo = None
    return build_packet(row, comps, market, geo, data_cfg)


# ---------------------------------------------------------------------------------------------------
# arm configuration
# ---------------------------------------------------------------------------------------------------
def arm_config(spec: RunSpec) -> dict[str, Any]:
    arms = load_config("treatments")["arms"]
    if spec.arm not in arms:
        raise KeyError(f"arm {spec.arm!r} not in configs/treatments.yaml (expected one of {list(arms)})")
    a = dict(arms[spec.arm])
    if a.get("family") != "llm":
        raise ValueError(f"arm {spec.arm!r} is a baseline arm; use `av train` for A–D")
    rounds_cfg = a.get("rounds", 0)
    if spec.rounds is not None:
        rounds = int(spec.rounds)
    elif isinstance(rounds_cfg, list):
        rounds = 2
        a["rounds_note"] = f"rounds grid {rounds_cfg} in treatments.yaml; defaulted to 2 (pass --rounds)"
    else:
        rounds = int(rounds_cfg)
    a["rounds_used"] = rounds
    roles_spec = a.get("roles", "default12")
    if spec.arm == "I_placebo":
        roles = placebo_roles(12)
    else:
        roles = roles_for(roles_spec)
    a["role_ids"] = [r.id for r in roles]
    a["_roles"] = roles
    a["protocol_name"] = a.get("protocol", "independent")
    # arm label used in predictions/transcripts: the rounds grid of G is spelled out (G_debate_r1/r2/r4), as the
    # evaluation layer and the preregistration expect
    a["arm_label"] = f"{spec.arm}_r{rounds}" if isinstance(rounds_cfg, list) else spec.arm
    if spec.arm == "J_cost_matched":
        a["protocol_name"] = "cost_matched_single"
        a["k"] = int(a.get("repeats", 12))
    return a


async def _run_property(packet: EvidencePacket, spec: RunSpec, acfg: dict[str, Any], pool: ProviderPool, weights: dict[str, Any] | None, tier: str, effort: str) -> DebateTranscript:
    roles: list[Role] = acfg["_roles"]
    role_ids = [r.id for r in roles]
    if acfg.get("blinded_skeptic") and "skeptic" not in role_ids:
        role_ids.append("skeptic")
    assignment = assign(role_ids, spec.committee, spec.seed, packet.subject.tx_id, tier)
    common = dict(arm=acfg["arm_label"], committee=spec.committee, effort=effort, max_tokens=spec.max_tokens, weights=weights)
    name = acfg["protocol_name"]
    if name == "independent":
        return await independent(packet, roles, assignment, pool, **common)
    if name == "structured_debate":
        return await structured_debate(
            packet, roles, assignment, pool, rounds=int(acfg["rounds_used"]),
            blinded_skeptic=bool(acfg.get("blinded_skeptic", False)), evidence_verification=bool(acfg.get("evidence_verification", False)),
            placebo=(spec.arm == "I_placebo"), **common,
        )
    if name == "cost_matched_single":
        return await cost_matched_single(packet, int(acfg["k"]), assignment, pool, **common)
    raise KeyError(f"unknown protocol {name!r}")


# ---------------------------------------------------------------------------------------------------
# artefacts
# ---------------------------------------------------------------------------------------------------
def prediction_rows(tr: DebateTranscript, split: str, run_id: str) -> list[dict[str, Any]]:
    """One row per aggregation rule (plus 'mixture', whose quantiles are the mixture quantiles)."""
    if not tr.aggregated:
        return []
    iv = aggregate_intervals(tr.final_quantiles)
    q_default = iv.get("median_of_quantiles") or [float("nan")] * 5
    rows = []
    for rule, point in tr.aggregated.items():
        rows.append(_row(tr, split, run_id, rule, point, q_default))
    if "mixture" in iv:
        mq = iv["mixture"]
        rows.append(_row(tr, split, run_id, "mixture", mq[2], mq))
    return rows


def _row(tr: DebateTranscript, split: str, run_id: str, rule: str, point: float, q: list[float]) -> dict[str, Any]:
    return {
        "tx_id": tr.tx_id, "split": split, "arm": tr.arm, "committee": tr.committee, "aggregation": rule, "point": float(point),
        "q05": float(q[0]), "q25": float(q[1]), "q50": float(q[2]), "q75": float(q[3]), "q95": float(q[4]),
        "cost_usd": float(tr.total_cost_usd), "latency_s": float(tr.total_latency_s), "tokens": int(tr.total_tokens), "run_id": run_id,
    }


def predictions_from_dir(run_dir: Path, split_of: dict[str, str], run_id: str) -> pl.DataFrame:
    rows: list[dict[str, Any]] = []
    for f in sorted((run_dir / "transcripts").glob("*.json")):
        tr = DebateTranscript.model_validate_json(f.read_text())
        rows += prediction_rows(tr, split_of.get(tr.tx_id, "unknown"), run_id)
    schema = {"tx_id": pl.Utf8, "split": pl.Utf8, "arm": pl.Utf8, "committee": pl.Utf8, "aggregation": pl.Utf8, "point": pl.Float64, "q05": pl.Float64, "q25": pl.Float64, "q50": pl.Float64, "q75": pl.Float64, "q95": pl.Float64, "cost_usd": pl.Float64, "latency_s": pl.Float64, "tokens": pl.Int64, "run_id": pl.Utf8}
    return pl.DataFrame(rows, schema=schema).select(list(PREDICTION_COLUMNS))


def _sha(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]


def _git_commit() -> str | None:
    try:
        return subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"], capture_output=True, text=True, timeout=5).stdout.strip() or None
    except Exception:
        return None


# ---------------------------------------------------------------------------------------------------
# main entry point
# ---------------------------------------------------------------------------------------------------
async def run_arm_async(spec: RunSpec) -> Path:
    run_id = spec.resolved_run_id()
    run_dir = (spec.out_dir or (RESULTS / "runs")) / run_id
    (run_dir / "transcripts").mkdir(parents=True, exist_ok=True)
    (run_dir / "packets").mkdir(parents=True, exist_ok=True)
    t_start = datetime.now(timezone.utc)
    acfg = arm_config(spec)
    models_cfg = load_config("models")
    tier = spec.tier or load_config("treatments").get("providers", {}).get("tier") or models_cfg.get("default_tier", "mid")
    effort = spec.effort or models_cfg["request_defaults"].get("anthropic", {}).get("effort", "medium")
    if spec.committee not in COMMITTEES:
        raise KeyError(f"committee {spec.committee!r} not in {COMMITTEES}")
    live = bool(spec.live) and provider_mode() == "live"
    if spec.live and not live:
        print("[run-arm] --live requested but no API key / AV_PROVIDER_MODE=mock: running in MOCK mode")
    ledger = CostLedger(run_id=run_id)
    pool = ProviderPool(ledger=ledger, force_mock=not live)
    weights = load_weights(spec.weights_path) if (spec.weights_path or spec.split in ("test", "holdout", "prospective")) else None
    weights_source = str(spec.weights_path or (RESULTS / "aggregation_weights.json")) if weights else "none (equal weights for inverse_variance/stacking; MAD scale for robust_bayes)"

    data_cfg = load_config("data")
    analytic = load_analytic(spec.synthetic, spec.data_dir)
    sample = select_subsample(analytic, spec.split, spec.n, spec.seed, spec.synthetic, spec.data_dir)
    fb = feature_backend(spec.synthetic, spec.data_dir)
    split_of = dict(zip(sample["tx_id"].to_list(), sample["split"].to_list()))

    done = {f.stem for f in (run_dir / "transcripts").glob("*.json")}
    todo = [r for r in sample.iter_rows(named=True) if r["tx_id"] not in done]
    print(f"[run-arm] {run_id}: arm={spec.arm} protocol={acfg['protocol_name']} rounds={acfg['rounds_used']} committee={spec.committee} tier={tier} mode={'live' if live else 'mock'} n={sample.height} (resume: {len(done)} done, {len(todo)} to run) features={fb.source}")

    sem = asyncio.Semaphore(max(1, spec.property_concurrency))
    stop = asyncio.Event()
    errors: list[dict[str, str]] = []
    n_ok = 0
    n_failed = 0
    cap_hit = False

    async def worker(row: dict[str, Any]) -> None:
        nonlocal n_ok, n_failed, cap_hit
        async with sem:
            if stop.is_set():
                errors.append({"tx_id": row["tx_id"], "error": "skipped: spend cap reached"})
                n_failed += 1
                return
            tx = row["tx_id"]
            try:
                packet = await asyncio.to_thread(build_packet_for, analytic, row, data_cfg, fb)
                (run_dir / "packets" / f"{tx}.json").write_text(packet.model_dump_json(indent=1))
                tr = await _run_property(packet, spec, acfg, pool, weights, tier, effort)
            except SpendCapExceeded as e:
                cap_hit = True
                stop.set()
                errors.append({"tx_id": tx, "error": f"SpendCapExceeded: {e}"[:300]})
                n_failed += 1
                return
            except Exception as e:  # never silently fill
                errors.append({"tx_id": tx, "error": f"{type(e).__name__}: {e}"[:300]})
                n_failed += 1
                return
            (run_dir / "transcripts" / f"{tx}.json").write_text(tr.to_json())
            if tr.final_estimates:
                n_ok += 1
            else:
                n_failed += 1
                errors.append({"tx_id": tx, "error": "no valid estimate (all turns failed)"})

    await asyncio.gather(*(worker(r) for r in todo))

    preds = predictions_from_dir(run_dir, split_of, run_id)
    preds.write_parquet(run_dir / "predictions.parquet")
    transcripts = [DebateTranscript.model_validate_json(f.read_text()) for f in sorted((run_dir / "transcripts").glob("*.json"))]
    t_end = datetime.now(timezone.utc)
    manifest = {
        "run_id": run_id,
        "arm": spec.arm,
        "arm_label": acfg["arm_label"],
        "protocol": acfg["protocol_name"],
        "rounds": acfg["rounds_used"],
        "rounds_note": acfg.get("rounds_note"),
        "k_repeats": acfg.get("k"),
        "committee": spec.committee,
        "tier": tier,
        "models": committee_models(spec.committee, tier),
        "effort": effort,
        "max_tokens": spec.max_tokens,
        "mode": "live" if live else "mock",
        "split": spec.split,
        "n_requested": spec.n,
        "n_sampled": sample.height,
        "synthetic": spec.synthetic,
        "data_source": (str(data_base(spec.synthetic, spec.data_dir) / "analytic.parquet") if spec.data_dir is not None else ("data/synthetic/synthetic_transactions.parquet (SYNTHETIC — not real transactions)" if spec.synthetic else "data/processed/analytic.parquet")),
        "data_dir": (str(spec.data_dir) if spec.data_dir is not None else None),
        "features_source": fb.source,
        "geo_cache": (str(data_base(spec.synthetic, spec.data_dir) / "geo_features.parquet") if fb.geo_cache is not None else None),
        "roles": acfg["role_ids"],
        "prompt_hashes": role_prompt_hashes(acfg["_roles"]) | ({"skeptic": role_prompt_hashes(roles_for(["skeptic"]))["skeptic"]} if acfg.get("blinded_skeptic") else {}),
        "config_hashes": {p.name: _sha(p) for p in sorted(CONFIGS.glob("*.yaml"))},
        "schema_hash": _sha(ROOT / "src" / "agentic_valuation" / "schemas.py"),
        "package_version": __version__,
        "git_commit": _git_commit(),
        "python": platform.python_version(),
        "host": platform.node(),
        "seed": spec.seed,
        "weights_source": weights_source,
        "spend_cap_usd": ledger.cap,
        "spend_cap_hit": cap_hit,
        "start_time": t_start.isoformat(),
        "end_time": t_end.isoformat(),
        "wall_clock_s": round((t_end - t_start).total_seconds(), 1),
        "total_cost_usd_estimated": round(sum(t.total_cost_usd for t in transcripts), 4),
        "total_tokens": int(sum(t.total_tokens for t in transcripts)),
        "n_transcripts": len(transcripts),
        "n_ok_this_session": n_ok,
        "n_failed_this_session": n_failed,
        "n_resumed_skipped": len(done),
        "n_turn_errors": int(sum(1 for t in transcripts for u in t.turns if u.error)),
        "stopped_early": [t.tx_id for t in transcripts if any(n.startswith("stopped_early_at") for n in t.notes)],
        "errors": errors,
        "prediction_columns": list(PREDICTION_COLUMNS),
        "spec": {k: (str(v) if isinstance(v, Path) else v) for k, v in asdict(spec).items()},
    }
    (run_dir / "manifest.json").write_text(json.dumps(manifest, indent=1, ensure_ascii=False, default=str))
    print(f"[run-arm] wrote {run_dir}: {len(transcripts)} transcripts, {preds.height} prediction rows, est. cost {manifest['total_cost_usd_estimated']} USD ({'mock' if not live else 'live'}), failures {n_failed}{' — SPEND CAP HIT' if cap_hit else ''}")
    return run_dir


def run_arm(spec: RunSpec) -> Path:
    return asyncio.run(run_arm_async(spec))