1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Run one treatment arm (E–J) on a subsample of properties and write the standard artefacts:34results/runs/<run_id>/5 predictions.parquet one row per (tx_id, aggregation rule) — contract in docs/architecture.md §16 transcripts/<tx_id>.json DebateTranscript (every turn, tokens, cost, prompt hash, errors)7 packets/<tx_id>.json the full EvidencePacket shown (filtered per role) to the agents8 manifest.json arm, committee, models, effort, prompt hashes, config hashes, timings, cost, counts910Resumable: properties with an existing transcript are skipped and predictions are rebuilt from all11transcripts in the directory. Mock provider unless ``live=True`` AND keys are present. The ledger's12spend cap (AV_MAX_SPEND_USD) stops the run gracefully: what exists is written, nothing is fabricated.1314Feature functions (comparables, market, geo) are imported lazily from ``agentic_valuation.features``;15when that package is not available yet, thin local fallbacks (``_simple_comps``, ``_simple_market``)16keep the smoke tests running on the synthetic parquet.17"""18from __future__ import annotations1920import asyncio21import hashlib22import json23import math24import platform25import subprocess26from dataclasses import asdict, dataclass27from datetime import date, datetime, timedelta, timezone28from pathlib import Path29from typing import Any3031import polars as pl3233from .. import __version__34from ..agents.packets import build_packet35from ..agents.roles import Role, placebo_roles, role_prompt_hashes, roles_for36from ..agents.runner import ProviderPool37from ..config import CONFIGS, PROCESSED, RESULTS, ROOT, SEED, SYNTHETIC, load_config, provider_mode38from ..debate.aggregation import aggregate_intervals, load_weights39from ..debate.committees import COMMITTEES, assign, committee_models40from ..debate.protocols import cost_matched_single, independent, structured_debate41from ..providers import CostLedger42from ..providers.ledger import SpendCapExceeded43from ..schemas import Comparable, DebateTranscript, EvidencePacket, GeoContext, MarketContext4445PREDICTION_COLUMNS: tuple[str, ...] = ("tx_id", "split", "arm", "committee", "aggregation", "point", "q05", "q25", "q50", "q75", "q95", "cost_usd", "latency_s", "tokens", "run_id")46LLM_ARMS = ("E_single_llm", "F_independent", "G_debate", "H_debate_skeptic", "I_placebo", "J_cost_matched", "G_fullview")474849@dataclass50class RunSpec:51 arm: str = "F_independent"52 rounds: int | None = None53 committee: str = "all_anthropic"54 split: str = "validation"55 n: int = 2456 live: bool = False57 tier: str | None = None58 run_id: str | None = None59 synthetic: bool = False60 seed: int = SEED61 effort: str | None = None62 property_concurrency: int = 463 out_dir: Path | None = None64 weights_path: Path | None = None65 max_tokens: int = 409666 data_dir: Path | None = None # override of data/<processed|synthetic> (e.g. the prospective bundle); None = default layout6768 def resolved_run_id(self) -> str:69 if self.run_id:70 return self.run_id71 parts = [self.arm]72 if self.rounds is not None and self.arm in ("G_debate", "H_debate_skeptic", "I_placebo", "G_fullview"):73 parts.append(f"r{self.rounds}")74 parts += [self.committee, self.split, "synthetic" if self.synthetic else "real", "live" if self.live else "mock"]75 rid = "_".join(parts)76 # mock or synthetic runs are never empirical results: a leading underscore makes scripts/20_evaluate.py skip them77 return rid if (self.live and not self.synthetic) else f"_{rid}"787980# ---------------------------------------------------------------------------------------------------81# data82# ---------------------------------------------------------------------------------------------------83def _assign_split_by_date(df: pl.DataFrame, data_cfg: dict[str, Any]) -> pl.DataFrame:84 """Synthetic data has no splits parquet: derive the chronological split from configs/data.yaml."""85 expr = pl.lit("unused")86 for name, w in data_cfg["splits"].items():87 if not isinstance(w, dict) or "start" not in w:88 continue89 s, e = date.fromisoformat(w["start"]), date.fromisoformat(w["end"])90 expr = pl.when((pl.col("date") >= s) & (pl.col("date") <= e)).then(pl.lit(name)).otherwise(expr)91 return df.with_columns(expr.alias("split"))929394def data_base(synthetic: bool, data_dir: Path | None = None) -> Path:95 return Path(data_dir) if data_dir is not None else (SYNTHETIC if synthetic else PROCESSED)969798def load_analytic(synthetic: bool, data_dir: Path | None = None) -> pl.DataFrame:99 """data/<base>/analytic.parquet joined with splits.parquet (base = synthetic | processed | ``data_dir``). When the100 synthetic analytic layer has not been built yet, it is derived on the fly from synthetic_transactions.parquet."""101 base = data_base(synthetic, data_dir)102 path = base / "analytic.parquet"103 if path.exists():104 df = pl.read_parquet(path)105 elif synthetic:106 raw = SYNTHETIC / "synthetic_transactions.parquet"107 if not raw.exists():108 from ..data.synthetic import write_default109110 write_default()111 from ..data.analytic_schema import add_derived_columns112113 df = add_derived_columns(pl.read_parquet(raw))114 else:115 raise FileNotFoundError(f"{path} not found: run `av build` (scripts/01_build_analytic.py) first, or use --synthetic")116 if "split" not in df.columns:117 sp = base / "splits.parquet"118 if sp.exists():119 df = df.join(pl.read_parquet(sp).select("tx_id", "split"), on="tx_id", how="left")120 elif synthetic:121 df = _assign_split_by_date(df, load_config("data"))122 else:123 raise FileNotFoundError(f"{sp} not found: run `av splits` first")124 return df125126127def select_subsample(df: pl.DataFrame, split: str, n: int, seed: int, synthetic: bool, data_dir: Path | None = None) -> pl.DataFrame:128 """Properties to value: the first n (sorted by tx_id) of data/<base>/llm_subsamples.parquet (tx_id, split,129 stratum — drawn once by scripts/02_splits.py); when that file is absent, a seeded random draw within the split."""130 sub_path = data_base(synthetic, data_dir) / "llm_subsamples.parquet"131 if sub_path.exists():132 sub = pl.read_parquet(sub_path).filter(pl.col("split") == split)133 ids = sorted(sub["tx_id"].to_list())[:n]134 out = df.filter(pl.col("tx_id").is_in(ids)).sort("tx_id")135 if out.height:136 return out137 pool = df.filter(pl.col("split") == split)138 if "exclude_reason" in pool.columns:139 pool = pool.filter(pl.col("exclude_reason").is_null())140 if pool.height == 0:141 raise ValueError(f"no property in split {split!r}")142 k = min(n, pool.height)143 return pool.sort("tx_id").sample(n=k, seed=seed, shuffle=True).sort("tx_id")144145146# ---------------------------------------------------------------------------------------------------147# features: lazy import with local fallbacks148# ---------------------------------------------------------------------------------------------------149def _haversine_expr(lat: float, lng: float) -> pl.Expr:150 r = 6371.0151 lat1, lng1 = math.radians(lat), math.radians(lng)152 dlat = pl.col("lat").radians() - lat1153 dlng = pl.col("lng").radians() - lng1154 a = (dlat / 2).sin() ** 2 + math.cos(lat1) * pl.col("lat").radians().cos() * (dlng / 2).sin() ** 2155 return 2 * r * a.sqrt().arcsin()156157158def _simple_comps(analytic: pl.DataFrame, subject: dict[str, Any], cutoff: date, cfg: dict[str, Any]) -> list[Comparable]:159 """Fallback retrieval: same class, date < cutoff, within the lookback and 10 km, nearest first, top max_comps."""160 c = cfg.get("comparables", {})161 radius = max(c.get("radius_km", [10.0])) if isinstance(c.get("radius_km"), list) else float(c.get("radius_km", 10.0))162 lookback = int(c.get("lookback_days", 540))163 max_comps = int(c.get("max_comps", 12))164 val_date = cutoff + timedelta(days=1)165 df = (166 analytic.filter(167 (pl.col("date") < cutoff)168 & (pl.col("date") >= cutoff - timedelta(days=lookback))169 & (pl.col("property_class") == subject["property_class"])170 & (pl.col("tx_id") != subject["tx_id"])171 )172 )173 if "parcel_id" in df.columns and subject.get("parcel_id") is not None:174 df = df.filter(pl.col("parcel_id") != subject["parcel_id"])175 if "exclude_reason" in df.columns:176 df = df.filter(pl.col("exclude_reason").is_null())177 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)178 out: list[Comparable] = []179 for i, r in enumerate(df.iter_rows(named=True)):180 area = r.get("floor_area_m2")181 roll = r.get("roll_total_value")182 sub = r.get("subtype") or "unknown"183 out.append(184 Comparable(185 evidence_id=f"C{i + 1:02d}",186 sale_date=r["date"],187 price=float(r["price"]),188 distance_km=round(float(r["distance_km"]), 2),189 property_class=r["property_class"],190 subtype=sub if sub in ("detached", "semi", "row", "integrated", "unknown") else "unknown",191 municipality=str(r.get("mun_name") or "unknown"),192 floor_area_m2=_num(area),193 land_area_m2=_num(r.get("land_area_m2")),194 year_built=(int(r["year_built"]) if r.get("year_built") is not None else None),195 storeys=_num(r.get("storeys")),196 n_units=(int(r["n_units"]) if r.get("n_units") is not None else None),197 roll_total_value=_num(roll),198 price_per_m2=(round(float(r["price"]) / float(area), 1) if area else None),199 assessment_ratio=(round(float(r["price"]) / float(roll), 3) if roll else None),200 months_before_valuation=round((val_date - r["date"]).days / 30.44, 1),201 )202 )203 return out204205206def _num(v: Any) -> float | None:207 if v is None:208 return None209 try:210 x = float(v)211 except (TypeError, ValueError):212 return None213 return None if math.isnan(x) else x214215216def _median(s: pl.Series) -> float | None:217 s = s.drop_nulls()218 return float(s.median()) if s.len() else None219220221def _simple_market(analytic: pl.DataFrame, subject: dict[str, Any], cutoff: date, cfg: dict[str, Any]) -> MarketContext:222 """Fallback market context from sales dated <= cutoff only."""223 past = analytic.filter(pl.col("date") <= cutoff)224 if "exclude_reason" in past.columns:225 past = past.filter(pl.col("exclude_reason").is_null())226 y1 = cutoff - timedelta(days=365)227 y2 = cutoff - timedelta(days=730)228 mun = past.filter(pl.col("mun_name") == subject.get("mun_name"))229 mun12 = mun.filter(pl.col("date") > y1)230 reg = past.filter(pl.col("region_name") == subject.get("region_name"))231 reg12 = reg.filter(pl.col("date") > y1)232 reg_prev = reg.filter((pl.col("date") > y2) & (pl.col("date") <= y1))233 local = past.with_columns(_haversine_expr(float(subject["lat"]), float(subject["lng"])).alias("d")).filter((pl.col("d") <= 3.0) & (pl.col("date") > y1))234 ppm2 = mun12.filter(pl.col("floor_area_m2") > 0).with_columns((pl.col("price") / pl.col("floor_area_m2")).alias("ppm2"))235 ratio = local.filter(pl.col("roll_total_value") > 0).with_columns((pl.col("price") / pl.col("roll_total_value")).alias("ar"))236 m_now, m_prev = _median(reg12["price"]), _median(reg_prev["price"])237 yoy = (m_now / m_prev - 1.0) if (m_now and m_prev) else None238 # monthly municipal index: median price per m2 by month relative to the first month with >= 5 sales239 idx: list[dict[str, Any]] = []240 if mun.height:241 mm = (242 mun.filter(pl.col("floor_area_m2") > 0)243 .with_columns((pl.col("price") / pl.col("floor_area_m2")).alias("ppm2"), pl.col("date").dt.strftime("%Y-%m").alias("month"))244 .group_by("month")245 .agg(pl.col("ppm2").median().alias("med"), pl.len().alias("n"))246 .filter(pl.col("n") >= 5)247 .sort("month")248 )249 if mm.height:250 base = float(mm["med"][0])251 idx = [{"month": r["month"], "index": round(float(r["med"]) / base * 100, 1)} for r in mm.tail(36).iter_rows(named=True)]252 return MarketContext(253 as_of=cutoff,254 municipality_median_price_12m=_median(mun12["price"]),255 municipality_median_ppm2_12m=(round(_median(ppm2["ppm2"]), 1) if ppm2.height else None),256 municipality_n_sales_12m=int(mun12.height),257 municipality_price_index=idx,258 region_median_price_12m=m_now,259 region_yoy_change=(round(yoy, 4) if yoy is not None else None),260 local_liquidity_n_sales_3km_12m=int(local.height),261 local_median_assessment_ratio_12m=(round(_median(ratio["ar"]), 3) if ratio.height else None),262 mortgage_rate_5y_fixed=None,263 )264265266def _prior_sale(analytic: pl.DataFrame, row: dict[str, Any], cutoff: date) -> tuple[float | None, date | None]:267 if "parcel_id" not in analytic.columns or row.get("parcel_id") is None:268 return None, None269 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)270 if prev.height == 0:271 return None, None272 return float(prev["price"][0]), prev["date"][0]273274275@dataclass276class FeatureBackend:277 comps: Any278 market: Any279 geo: Any280 source: str281 geo_cache: pl.DataFrame | None = None282283284def feature_backend(synthetic: bool, data_dir: Path | None = None) -> FeatureBackend:285 """Lazy import of agentic_valuation.features; local fallbacks when it is not importable."""286 geo_cache = None287 gp = data_base(synthetic, data_dir) / "geo_features.parquet"288 if gp.exists():289 try:290 geo_cache = pl.read_parquet(gp)291 except Exception:292 geo_cache = None293 try:294 from ..features.comparables import retrieve_comparables # type: ignore295 from ..features.market import market_context # type: ignore296297 try:298 from ..features.geo import geo_context # type: ignore299 except Exception:300 geo_context = None301 return FeatureBackend(retrieve_comparables, market_context, geo_context, "agentic_valuation.features", geo_cache)302 except Exception as e:303 return FeatureBackend(_simple_comps, _simple_market, None, f"run_arm fallback (_simple_comps/_simple_market, no geo; features import failed: {type(e).__name__})", None)304305306def build_packet_for(analytic: pl.DataFrame, row: dict[str, Any], data_cfg: dict[str, Any], fb: FeatureBackend) -> EvidencePacket:307 cutoff = row["date"] - timedelta(days=1)308 row = dict(row)309 if row.get("prior_sale_price") is None:310 row["prior_sale_price"], row["prior_sale_date"] = _prior_sale(analytic, row, cutoff)311 comps = fb.comps(analytic, row, cutoff, data_cfg)312 market = fb.market(analytic, row, cutoff, data_cfg)313 geo: GeoContext | None = None314 if fb.geo is not None and fb.geo_cache is not None:315 try:316 geo = fb.geo(row, None, cached=fb.geo_cache)317 except Exception:318 geo = None319 return build_packet(row, comps, market, geo, data_cfg)320321322# ---------------------------------------------------------------------------------------------------323# arm configuration324# ---------------------------------------------------------------------------------------------------325def arm_config(spec: RunSpec) -> dict[str, Any]:326 arms = load_config("treatments")["arms"]327 if spec.arm not in arms:328 raise KeyError(f"arm {spec.arm!r} not in configs/treatments.yaml (expected one of {list(arms)})")329 a = dict(arms[spec.arm])330 if a.get("family") != "llm":331 raise ValueError(f"arm {spec.arm!r} is a baseline arm; use `av train` for A–D")332 rounds_cfg = a.get("rounds", 0)333 if spec.rounds is not None:334 rounds = int(spec.rounds)335 elif isinstance(rounds_cfg, list):336 rounds = 2337 a["rounds_note"] = f"rounds grid {rounds_cfg} in treatments.yaml; defaulted to 2 (pass --rounds)"338 else:339 rounds = int(rounds_cfg)340 a["rounds_used"] = rounds341 roles_spec = a.get("roles", "default12")342 if spec.arm == "I_placebo":343 roles = placebo_roles(12)344 else:345 roles = roles_for(roles_spec)346 a["role_ids"] = [r.id for r in roles]347 a["_roles"] = roles348 a["protocol_name"] = a.get("protocol", "independent")349 # arm label used in predictions/transcripts: the rounds grid of G is spelled out (G_debate_r1/r2/r4), as the350 # evaluation layer and the preregistration expect351 a["arm_label"] = f"{spec.arm}_r{rounds}" if isinstance(rounds_cfg, list) else spec.arm352 if spec.arm == "J_cost_matched":353 a["protocol_name"] = "cost_matched_single"354 a["k"] = int(a.get("repeats", 12))355 return a356357358async def _run_property(packet: EvidencePacket, spec: RunSpec, acfg: dict[str, Any], pool: ProviderPool, weights: dict[str, Any] | None, tier: str, effort: str) -> DebateTranscript:359 roles: list[Role] = acfg["_roles"]360 role_ids = [r.id for r in roles]361 if acfg.get("blinded_skeptic") and "skeptic" not in role_ids:362 role_ids.append("skeptic")363 assignment = assign(role_ids, spec.committee, spec.seed, packet.subject.tx_id, tier)364 common = dict(arm=acfg["arm_label"], committee=spec.committee, effort=effort, max_tokens=spec.max_tokens, weights=weights)365 name = acfg["protocol_name"]366 if name == "independent":367 return await independent(packet, roles, assignment, pool, **common)368 if name == "structured_debate":369 return await structured_debate(370 packet, roles, assignment, pool, rounds=int(acfg["rounds_used"]),371 blinded_skeptic=bool(acfg.get("blinded_skeptic", False)), evidence_verification=bool(acfg.get("evidence_verification", False)),372 placebo=(spec.arm == "I_placebo"), **common,373 )374 if name == "cost_matched_single":375 return await cost_matched_single(packet, int(acfg["k"]), assignment, pool, **common)376 raise KeyError(f"unknown protocol {name!r}")377378379# ---------------------------------------------------------------------------------------------------380# artefacts381# ---------------------------------------------------------------------------------------------------382def prediction_rows(tr: DebateTranscript, split: str, run_id: str) -> list[dict[str, Any]]:383 """One row per aggregation rule (plus 'mixture', whose quantiles are the mixture quantiles)."""384 if not tr.aggregated:385 return []386 iv = aggregate_intervals(tr.final_quantiles)387 q_default = iv.get("median_of_quantiles") or [float("nan")] * 5388 rows = []389 for rule, point in tr.aggregated.items():390 rows.append(_row(tr, split, run_id, rule, point, q_default))391 if "mixture" in iv:392 mq = iv["mixture"]393 rows.append(_row(tr, split, run_id, "mixture", mq[2], mq))394 return rows395396397def _row(tr: DebateTranscript, split: str, run_id: str, rule: str, point: float, q: list[float]) -> dict[str, Any]:398 return {399 "tx_id": tr.tx_id, "split": split, "arm": tr.arm, "committee": tr.committee, "aggregation": rule, "point": float(point),400 "q05": float(q[0]), "q25": float(q[1]), "q50": float(q[2]), "q75": float(q[3]), "q95": float(q[4]),401 "cost_usd": float(tr.total_cost_usd), "latency_s": float(tr.total_latency_s), "tokens": int(tr.total_tokens), "run_id": run_id,402 }403404405def predictions_from_dir(run_dir: Path, split_of: dict[str, str], run_id: str) -> pl.DataFrame:406 rows: list[dict[str, Any]] = []407 for f in sorted((run_dir / "transcripts").glob("*.json")):408 tr = DebateTranscript.model_validate_json(f.read_text())409 rows += prediction_rows(tr, split_of.get(tr.tx_id, "unknown"), run_id)410 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}411 return pl.DataFrame(rows, schema=schema).select(list(PREDICTION_COLUMNS))412413414def _sha(path: Path) -> str:415 return hashlib.sha256(path.read_bytes()).hexdigest()[:16]416417418def _git_commit() -> str | None:419 try:420 return subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"], capture_output=True, text=True, timeout=5).stdout.strip() or None421 except Exception:422 return None423424425# ---------------------------------------------------------------------------------------------------426# main entry point427# ---------------------------------------------------------------------------------------------------428async def run_arm_async(spec: RunSpec) -> Path:429 run_id = spec.resolved_run_id()430 run_dir = (spec.out_dir or (RESULTS / "runs")) / run_id431 (run_dir / "transcripts").mkdir(parents=True, exist_ok=True)432 (run_dir / "packets").mkdir(parents=True, exist_ok=True)433 t_start = datetime.now(timezone.utc)434 acfg = arm_config(spec)435 models_cfg = load_config("models")436 tier = spec.tier or load_config("treatments").get("providers", {}).get("tier") or models_cfg.get("default_tier", "mid")437 effort = spec.effort or models_cfg["request_defaults"].get("anthropic", {}).get("effort", "medium")438 if spec.committee not in COMMITTEES:439 raise KeyError(f"committee {spec.committee!r} not in {COMMITTEES}")440 live = bool(spec.live) and provider_mode() == "live"441 if spec.live and not live:442 print("[run-arm] --live requested but no API key / AV_PROVIDER_MODE=mock: running in MOCK mode")443 ledger = CostLedger(run_id=run_id)444 pool = ProviderPool(ledger=ledger, force_mock=not live)445 weights = load_weights(spec.weights_path) if (spec.weights_path or spec.split in ("test", "holdout", "prospective")) else None446 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)"447448 data_cfg = load_config("data")449 analytic = load_analytic(spec.synthetic, spec.data_dir)450 sample = select_subsample(analytic, spec.split, spec.n, spec.seed, spec.synthetic, spec.data_dir)451 fb = feature_backend(spec.synthetic, spec.data_dir)452 split_of = dict(zip(sample["tx_id"].to_list(), sample["split"].to_list()))453454 done = {f.stem for f in (run_dir / "transcripts").glob("*.json")}455 todo = [r for r in sample.iter_rows(named=True) if r["tx_id"] not in done]456 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}")457458 sem = asyncio.Semaphore(max(1, spec.property_concurrency))459 stop = asyncio.Event()460 errors: list[dict[str, str]] = []461 n_ok = 0462 n_failed = 0463 cap_hit = False464465 async def worker(row: dict[str, Any]) -> None:466 nonlocal n_ok, n_failed, cap_hit467 async with sem:468 if stop.is_set():469 errors.append({"tx_id": row["tx_id"], "error": "skipped: spend cap reached"})470 n_failed += 1471 return472 tx = row["tx_id"]473 try:474 packet = await asyncio.to_thread(build_packet_for, analytic, row, data_cfg, fb)475 (run_dir / "packets" / f"{tx}.json").write_text(packet.model_dump_json(indent=1))476 tr = await _run_property(packet, spec, acfg, pool, weights, tier, effort)477 except SpendCapExceeded as e:478 cap_hit = True479 stop.set()480 errors.append({"tx_id": tx, "error": f"SpendCapExceeded: {e}"[:300]})481 n_failed += 1482 return483 except Exception as e: # never silently fill484 errors.append({"tx_id": tx, "error": f"{type(e).__name__}: {e}"[:300]})485 n_failed += 1486 return487 (run_dir / "transcripts" / f"{tx}.json").write_text(tr.to_json())488 if tr.final_estimates:489 n_ok += 1490 else:491 n_failed += 1492 errors.append({"tx_id": tx, "error": "no valid estimate (all turns failed)"})493494 await asyncio.gather(*(worker(r) for r in todo))495496 preds = predictions_from_dir(run_dir, split_of, run_id)497 preds.write_parquet(run_dir / "predictions.parquet")498 transcripts = [DebateTranscript.model_validate_json(f.read_text()) for f in sorted((run_dir / "transcripts").glob("*.json"))]499 t_end = datetime.now(timezone.utc)500 manifest = {501 "run_id": run_id,502 "arm": spec.arm,503 "arm_label": acfg["arm_label"],504 "protocol": acfg["protocol_name"],505 "rounds": acfg["rounds_used"],506 "rounds_note": acfg.get("rounds_note"),507 "k_repeats": acfg.get("k"),508 "committee": spec.committee,509 "tier": tier,510 "models": committee_models(spec.committee, tier),511 "effort": effort,512 "max_tokens": spec.max_tokens,513 "mode": "live" if live else "mock",514 "split": spec.split,515 "n_requested": spec.n,516 "n_sampled": sample.height,517 "synthetic": spec.synthetic,518 "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")),519 "data_dir": (str(spec.data_dir) if spec.data_dir is not None else None),520 "features_source": fb.source,521 "geo_cache": (str(data_base(spec.synthetic, spec.data_dir) / "geo_features.parquet") if fb.geo_cache is not None else None),522 "roles": acfg["role_ids"],523 "prompt_hashes": role_prompt_hashes(acfg["_roles"]) | ({"skeptic": role_prompt_hashes(roles_for(["skeptic"]))["skeptic"]} if acfg.get("blinded_skeptic") else {}),524 "config_hashes": {p.name: _sha(p) for p in sorted(CONFIGS.glob("*.yaml"))},525 "schema_hash": _sha(ROOT / "src" / "agentic_valuation" / "schemas.py"),526 "package_version": __version__,527 "git_commit": _git_commit(),528 "python": platform.python_version(),529 "host": platform.node(),530 "seed": spec.seed,531 "weights_source": weights_source,532 "spend_cap_usd": ledger.cap,533 "spend_cap_hit": cap_hit,534 "start_time": t_start.isoformat(),535 "end_time": t_end.isoformat(),536 "wall_clock_s": round((t_end - t_start).total_seconds(), 1),537 "total_cost_usd_estimated": round(sum(t.total_cost_usd for t in transcripts), 4),538 "total_tokens": int(sum(t.total_tokens for t in transcripts)),539 "n_transcripts": len(transcripts),540 "n_ok_this_session": n_ok,541 "n_failed_this_session": n_failed,542 "n_resumed_skipped": len(done),543 "n_turn_errors": int(sum(1 for t in transcripts for u in t.turns if u.error)),544 "stopped_early": [t.tx_id for t in transcripts if any(n.startswith("stopped_early_at") for n in t.notes)],545 "errors": errors,546 "prediction_columns": list(PREDICTION_COLUMNS),547 "spec": {k: (str(v) if isinstance(v, Path) else v) for k, v in asdict(spec).items()},548 }549 (run_dir / "manifest.json").write_text(json.dumps(manifest, indent=1, ensure_ascii=False, default=str))550 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 ''}")551 return run_dir552553554def run_arm(spec: RunSpec) -> Path:555 return asyncio.run(run_arm_async(spec))556