feat(rectification): submit varga-resolution implementation for review

Review-only snapshot for BUG-1115 through BUG-1117; not merge-ready. New opening and append-turn PostgreSQL permission failures remain blocked. Persisted joint replay has zero completed questions; segment ordering remains off by default. The existing offline replay JSON is retained stale and unchanged after a denied overwrite, including its CRLF line endings. Browser/provider validation and final serial gates remain pending. No deployment, role permission changes, or staging/main push.

Co-Authored-By: Claude Code <noreply@anthropic.com>

scripts/api_heavy_compute_gate.py

@@ -23,6 +23,7 @@ HEAVY_COMPUTE_PATHS = frozenset(
         "/api/active_rectification_events_v4",
         "/api/rectification/v5/candidate-features",
         "/api/rectification/v5/score",
+        "/api/rectification/v5/varga-segments",
         "/api/rectification/v5/diagnostics",
         "/api/rectification/v5/vedastro-validate",
         "/api/rectification/v5/range_reading",

scripts/jyotish_api_server.py

@@ -3560,6 +3560,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, VedastroEvidenceMixin, SynastryM
             elif path == '/api/active_rectification_events_v4':
                 result = self._compute_active_rectification_events_v4(body)
                 self._json(result)
+            elif path == '/api/rectification/v5/varga-segments':
+                _load_local_module('rectification_varga_segments_api').handle_varga_segments(self, body)
             elif path == '/api/rectification/v5/candidate-features':
                 self._json(self._compute_rectification_v5_candidate_features(body))
             elif path == '/api/rectification/v5/score':

scripts/rectification_varga_segments_api.py

@@ -0,0 +1,110 @@
+"""Minute-exact ascendant segments. This endpoint does not score or interpret."""
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import os
+import tempfile
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+from scripts import domain_calculation_service, varga
+
+TARGETS = ("D1", "D9", "D10")
+VERSION = "rectification-varga-segments-v1"
+
+
+def _number(body, name, low, high):
+    value = body.get(name)
+    if isinstance(value, bool):
+        raise ValueError(f"invalid_{name}")
+    try:
+        result = float(value)
+    except (ValueError, TypeError) as exc:
+        raise ValueError(f"invalid_{name}") from exc
+    if not math.isfinite(result) or not low <= result <= high:
+        raise ValueError(f"invalid_{name}")
+    return result
+
+
+def scan_varga_segments(body: dict[str, Any], *, cache_dir: Path | None = None) -> dict[str, Any]:
+    if not isinstance(body, dict):
+        raise ValueError("invalid_segment_request")
+    try:
+        start = datetime.strptime(str(body["start_at"]), "%Y-%m-%dT%H:%M")
+        end = datetime.strptime(str(body["end_at"]), "%Y-%m-%dT%H:%M")
+    except (KeyError, ValueError) as exc:
+        raise ValueError("invalid_segment_window") from exc
+    width = int((end - start).total_seconds() / 60) + 1
+    if width < 1 or width > 1441:
+        raise ValueError("invalid_segment_window")
+    targets = body.get("target_charts", list(TARGETS))
+    if (not isinstance(targets, list) or not targets
+            or any(not isinstance(target, str) or target not in TARGETS for target in targets)
+            or len(set(targets)) != len(targets)):
+        raise ValueError("invalid_target_charts")
+    profile = {
+        "lat": _number(body, "latitude", -90, 90),
+        "lon": _number(body, "longitude", -180, 180),
+        "tz": _number(body, "timezone_offset", -14, 14),
+        "ayanamsa": "raman", "node_mode": "mean",
+    }
+    key = hashlib.sha256(json.dumps({"version": VERSION, "start": str(start), "end": str(end), "targets": targets, **profile}, sort_keys=True).encode()).hexdigest()
+    root = cache_dir if cache_dir is not None else Path(os.environ.get("JYOTISH_CACHE_DIR", str(Path(__file__).resolve().parents[1] / "scratch" / "local"))) / "varga_segments"
+    path = root / f"{key}.json"
+    if path.exists():
+        try:
+            cached = json.loads(path.read_text(encoding="utf-8"))
+            if isinstance(cached, dict) and cached.get("window_key") == key and cached.get("version") == VERSION:
+                return cached
+        except (ValueError, OSError):
+            pass
+    minutes = []
+    segments = {target: [] for target in targets}
+    for offset in range(width):
+        moment = start + timedelta(minutes=offset)
+        chart = domain_calculation_service.compute_chart({
+            **profile, "year": moment.year, "month": moment.month, "day": moment.day,
+            "hour": moment.hour, "minute": moment.minute,
+        })
+        ascendant = float(chart["ascendant"]["lon"])
+        signs = {"D1": int(ascendant // 30) % 12} if "D1" in targets else {}
+        divisions = [int(target[1:]) for target in targets if target != "D1"]
+        if divisions:
+            planets = {name: float(item["lon"]) for name, item in chart.get("planets", {}).items() if isinstance(item, dict) and isinstance(item.get("lon"), (int, float))}
+            computed = varga.calc_all_vargas(planets, ascendant, divisions=divisions)
+            for value in computed.values():
+                signs[f"D{value['_meta']['div']}"] = int(value["Ascendant"]["sign_idx"])
+        stamp = moment.strftime("%Y-%m-%dT%H:%M")
+        minutes.append({"offset": offset, "time": stamp[11:], "date": stamp[:10], "signs": signs})
+        for target in targets:
+            rows = segments[target]
+            sign = signs[target]
+            if rows and rows[-1]["sign"] == sign:
+                rows[-1].update(end_at=stamp, end=offset, length=rows[-1]["length"] + 1)
+            else:
+                rows.append({"index": len(rows), "start": offset, "end": offset, "start_at": stamp, "end_at": stamp, "sign": sign, "length": 1})
+    payload = {"version": VERSION, "window_key": key, "start_at": body["start_at"], "end_at": body["end_at"],
+               "window_minutes": width, "target_charts": targets, "minutes": minutes,
+               "segments": segments, "full_window_unique": {target: len({row['sign'] for row in rows}) == 1 for target, rows in segments.items()},
+               "ayanamsa": "raman", "node_mode": "mean", "step_minutes": 1}
+    root.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=root, delete=False) as handle:
+        json.dump(payload, handle, ensure_ascii=False, sort_keys=True)
+        temporary = Path(handle.name)
+    try:
+        temporary.replace(path)
+    finally:
+        temporary.unlink(missing_ok=True)
+    return payload
+
+
+def handle_varga_segments(handler, body):
+    from scripts.offline_compute_mixins import BadRequest
+    try:
+        payload = scan_varga_segments(body)
+    except ValueError as exc:
+        raise BadRequest(str(exc)) from exc
+    handler._json(payload)

scripts/research/generate_varga_api_golden.py

@@ -0,0 +1,60 @@
+"""Capture real loopback score/scan API responses for the optional metadata contract.
+
+Uses the existing public AA request, never a private profile or a fabricated
+engine payload. Run in the calibrated Linux runtime for final evidence.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import platform
+import threading
+import urllib.request
+from http.server import ThreadingHTTPServer
+from pathlib import Path
+
+from scripts.jyotish_api_server import DEFAULT_ALLOWED_HOSTS, DEFAULT_ALLOWED_ORIGINS, JyotishAPIHandler
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def main():
+    source = ROOT / "frontend/tests/fixtures/rectification-grounding-aa-30min.golden.json"
+    request = json.loads(source.read_text(encoding="utf-8"))["request"]
+    server = ThreadingHTTPServer(("127.0.0.1", 0), JyotishAPIHandler)
+    server.daemon_threads = True
+    origin = f"http://127.0.0.1:{server.server_port}"
+    server.allowed_origins = set(DEFAULT_ALLOWED_ORIGINS) | {origin}
+    server.allowed_hosts = set(DEFAULT_ALLOWED_HOSTS)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    def post(path, body):
+        req = urllib.request.Request(origin + path, data=json.dumps(body).encode(), method="POST",
+                                     headers={"Content-Type": "application/json", "Origin": origin})
+        with urllib.request.urlopen(req, timeout=180) as response:
+            return json.load(response)
+
+    try:
+        response = post("/api/rectification/v5/score", request)
+        intervals = response["decision_receipt"]["candidate_intervals"]
+        assert len(intervals) == 1
+        scan_request = {"start_at": intervals[0]["start_at"], "end_at": intervals[0]["end_at"],
+                        "latitude": request["lat"], "longitude": request["lon"],
+                        "timezone_offset": request["tz"], "target_charts": ["D1", "D9", "D10"]}
+        scan = post("/api/rectification/v5/varga-segments", scan_request)
+        payload = {"source": "real loopback API; existing public AA calibration request",
+                   "python_version": platform.python_version(),
+                   "source_request_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
+                   "request": request, "response": response, "scan_request": scan_request, "scan": scan}
+        target = ROOT / "frontend/tests/fixtures/varga-api-response.golden.json"
+        target.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
+        print("captured real score and minute scan API responses")
+    finally:
+        server.shutdown()
+        thread.join(timeout=10)
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()

scripts/research/generate_varga_segment_goldens.py

@@ -0,0 +1,56 @@
+"""Generate public v5 engine goldens; expected values come from accepted research."""
+from __future__ import annotations
+
+import json
+import hashlib
+import platform
+from pathlib import Path
+
+from scripts.research import varga_resolution_lib as vr
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def main():
+    cache = ROOT / "artifacts/varga-impl-validation-20260930/replay-cache"
+    accepted_path = ROOT / "docs/research/varga_resolution_research_2026_09_30.json"
+    accepted = json.loads(accepted_path.read_text(encoding="utf-8"))
+    records = []
+    for case in vr.load_cases()[:12]:
+        for radius in (10, 30):
+            replay = vr.CaseReplay(case, radius, cache_dir=cache)
+            state = vr.replay_scores(replay.public, replay.probes, replay.true_time)
+            for chart in ("D1", "D9", "D10"):
+                segments = replay.segments((chart,))
+                ordered = vr.select_probes_by_segment(replay.public, replay.probes, segments, replay.true_time)
+                summary = vr.segment_shares(vr.minute_weights(state, "raw"), segments, replay.true_time)
+                accepted_cell = next(row for row in accepted["m1"]["by_radius"][str(radius)][chart]["raw"]["per_case"]
+                                     if row["case_id"] == replay.case_id)
+                actual_cell = {key: summary[key] for key in ("top_share", "top_is_truth", "truth_retained", "alive_segments")}
+                actual_cell["segments_in_window"] = len(segments)
+                if any(value != accepted_cell[key] for key, value in actual_cell.items()):
+                    raise ValueError(f"accepted_raw_parity_mismatch:{replay.case_id}:{radius}:{chart}; use the calibrated Python runtime")
+                records.append({
+                    "case_id": replay.case_id, "radius": radius, "chart": chart,
+                    "rows": replay.public,
+                    "posterior_rows": state["public"], "eliminated": sorted(state["eliminated"]),
+                    "probes": replay.probes, "segments": segments,
+                    "offsets": {minute: vr.offset_of(minute, replay.true_time) for minute in replay.times},
+                    "truth_offset": 0, "truth_time": replay.true_time,
+                    "expected_summary": summary, "accepted_cell": actual_cell,
+                    "expected_order": [probe["semantic_key"] for probe in ordered["asked"]],
+                    "answers": {probe["semantic_key"]: vr.optimal_answer(probe, replay.true_time) for probe in replay.probes},
+                })
+    path = ROOT / "frontend/tests/fixtures/varga-segment-v5-golden.json"
+    path.write_text(vr.stable_json({
+        "source": "v5 public AA engine responses; raw cells strictly verified against accepted research",
+        "python_version": platform.python_version(),
+        "accepted_sha256": hashlib.sha256(accepted_path.read_bytes()).hexdigest(),
+        "order_source": "sequential accepted research function; m2 JSON does not store semantic-key lists",
+        "records": records,
+    }), encoding="utf-8")
+    print(f"generated {len(records)} chart goldens across 12 public cases")
+
+
+if __name__ == "__main__":
+    main()

scripts/research/varga_resolution_method_replay.py

@@ -0,0 +1,105 @@
+"""Method-selection + direct reducer diagnostic (not answer persistence).
+
+Public holdout-v5 cases only. Truth goes exclusively to the offline answer
+oracle; the frozen collection, holdout, delivery and stopping gates stay live.
+The bridge does not exercise option classification, owned contrast registration
+or persistV9ChoiceAction. Early stop or a missing probe is not six-question
+validation. This is not the accepted single-chart M2 experiment.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import platform
+from datetime import timedelta
+from pathlib import Path
+
+from scripts.research import varga_resolution_lib as vr
+from scripts.research.varga_resolution_production_bridge import ProductionSegmentBridge
+from scripts.rectification.api_service import score_candidates
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dataset", default=str(vr.HOLDOUT_V5))
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--cache-dir", required=True)
+    parser.add_argument("--limit", type=int)
+    parser.add_argument("--radii", default="10,30,60")
+    args = parser.parse_args()
+    cases = vr.load_cases(Path(args.dataset))
+    if args.limit:
+        cases = cases[:args.limit]
+    cache = Path(args.cache_dir)
+    cache.mkdir(parents=True, exist_ok=True)
+    bridge = ProductionSegmentBridge()
+    records = []
+    try:
+        for case in cases:
+            for radius in map(int, args.radii.split(",")):
+                replay = vr.CaseReplay(case, radius, cache_dir=cache)
+                key = hashlib.sha256(json.dumps(replay.request, sort_keys=True).encode()).hexdigest()
+                response_file = cache / f"score-api-{key}.json"
+                if response_file.exists():
+                    response = json.loads(response_file.read_text(encoding="utf-8"))
+                else:
+                    response = score_candidates(replay.request)
+                    response_file.write_text(json.dumps(response, ensure_ascii=False), encoding="utf-8")
+                intervals = response["decision_receipt"]["candidate_intervals"]
+                birth = case["birth"]
+                origin = vr.birth_datetime(case)
+                snapshot = {"birth_date": str(birth["date"]), "reported_birth_time": replay.true_time,
+                            "birth_time_source": "approximate", "timezone_offset": replay.request["tz"],
+                            "timezone_id": replay.request.get("timezone_id"),
+                            "latitude": replay.request["lat"], "longitude": replay.request["lon"]}
+                evidence = [{"id": str(event["id"]), "status": "confirmed", "domain": event["domain"],
+                             "datePrecision": event.get("precision", "day"), "occurredFrom": event.get("date_start", event.get("date")),
+                             "occurredTo": event.get("date_end", event.get("date")), "eventKind": event.get("event_kind", "dated_event"),
+                             "summary": event.get("description", "Public calibration event")}
+                            for event in replay.request["events"]]
+                minutes = [{"offset": offset + radius, "date": (origin + timedelta(minutes=offset)).strftime("%Y-%m-%d"),
+                            "time": (origin + timedelta(minutes=offset)).strftime("%H:%M"),
+                            "signs": {chart: signs[chart] for chart in ("D1", "D9", "D10")}}
+                           for offset, signs in sorted(replay.signs.items())]
+                for enabled in (False, True):
+                    result = bridge.call({"operation": "method_replay", "request": replay.request,
+                                          "response": response, "snapshot": snapshot,
+                                          "range": {"start_time": replay.request["start_time"], "end_time": replay.request["end_time"],
+                                                    "candidate_intervals": intervals},
+                                          "evidence": evidence, "minutes": minutes, "targets": ["D1", "D9", "D10"],
+                                          "enabled": enabled, "true_time": replay.true_time, "ask_count": vr.ASK_COUNT})
+                    state = result.pop("state")
+                    raw = bridge.call({"operation": "weights", "rows": [{"time": row["time"], "score": row.get("raw_posterior_score", 0),
+                                       "cluster_times": row.get("cluster_times")} for row in state["candidates"]],
+                                       "eliminated": [row["time"] for row in state["candidates"] if row.get("raw_eliminated")]})
+                    metrics = {}
+                    for chart in ("D9", "D10"):
+                        segments = replay.segments([chart])
+                        summary = bridge.call({"operation": "summary_weights", "weights": raw, "segments": segments,
+                                               "offsets": {stamp: vr.offset_of(stamp, replay.true_time) for stamp in raw}, "truth_offset": 0})
+                        metrics[chart] = summary
+                    records.append({"case_id": replay.case_id, "radius": radius, "strategy": "joint" if enabled else "frozen",
+                                    "api_response_sha256": hashlib.sha256(response_file.read_bytes()).hexdigest(),
+                                    **result, "metrics": metrics})
+                print(f"method replay completed radius={radius}", flush=True)
+    finally:
+        bridge.close()
+    aggregate = {}
+    for radius in map(int, args.radii.split(",")):
+        for strategy in ("frozen", "joint"):
+            rows = [row for row in records if row["radius"] == radius and row["strategy"] == strategy]
+            aggregate[f"{radius}:{strategy}"] = {"cases": len(rows), "asked": sum(len(row["asked"]) for row in rows),
+                "D9": {"truth_kept": sum(row["metrics"]["D9"]["truth_retained"] for row in rows),
+                       "top_hit": sum(row["metrics"]["D9"]["top_is_truth"] for row in rows)},
+                "D10": {"truth_kept": sum(row["metrics"]["D10"]["truth_retained"] for row in rows),
+                        "top_hit": sum(row["metrics"]["D10"]["top_is_truth"] for row in rows)}}
+    output = {"scope": "method-selection + direct reducer diagnostic (not answer persistence); real score API/parser/inference/catalog/method/decision, default joint targets; early stop/missing probe is not six-question validation",
+              "python_version": platform.python_version(), "dataset_sha256": hashlib.sha256(Path(args.dataset).read_bytes()).hexdigest(),
+              "aggregate": aggregate, "records": records}
+    Path(args.output).write_text(json.dumps(output, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/research/varga_resolution_persisted_replay.py

@@ -0,0 +1,80 @@
+"""Public holdout replay through live API and production PostgreSQL answer paths.
+
+Explicit fictional Case seeds are not successful product opening. No turn,
+evidence, focus or answer is seeded: a refused production RPC stops that row.
+Off is the default; on changes only segment ordering. Truth is oracle-only.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import platform
+import subprocess
+import threading
+from http.server import ThreadingHTTPServer
+from pathlib import Path
+
+from scripts.jyotish_api_server import DEFAULT_ALLOWED_HOSTS, DEFAULT_ALLOWED_ORIGINS, JyotishAPIHandler
+from scripts.research import varga_resolution_lib as vr
+from scripts.research.minute_resolution_sweep import scoring_request_for
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dataset", default=str(vr.HOLDOUT_V5))
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--limit", type=int, default=10)
+    parser.add_argument("--radii", default="10,30,60")
+    args = parser.parse_args()
+    target = Path(args.output)
+    if target.exists():
+        parser.error("--output already exists; retain prior evidence and choose a unique label")
+    cases = vr.load_cases(Path(args.dataset))
+    if args.limit:
+        cases = cases[:args.limit]
+    rows = []
+    for case in cases:
+        birth = case["birth"]
+        for radius in map(int, args.radii.split(",")):
+            request = scoring_request_for(case, radius)
+            rows.append({"case_key": case["case_id"], "radius": radius,
+                         "snapshot": {"birth_date": birth["date"], "reported_birth_time": str(birth["time"])[:5],
+                                      "birth_time_source": "approximate", "active_birth_time": None,
+                                      "uncertainty_before_minutes": radius, "uncertainty_after_minutes": radius,
+                                      "latitude": request["lat"], "longitude": request["lon"],
+                                      "timezone_offset": request["tz"], "timezone_id": birth.get("timezone_id")},
+                         "range": {"start_time": request["start_time"], "end_time": request["end_time"]},
+                         "events": request["events"], "oracle_truth_minute": str(birth["time"])[:5]})
+    server = ThreadingHTTPServer(("127.0.0.1", 0), JyotishAPIHandler)
+    server.daemon_threads = True
+    origin = f"http://127.0.0.1:{server.server_port}"
+    server.allowed_hosts = set(DEFAULT_ALLOWED_HOSTS)
+    server.allowed_origins = set(DEFAULT_ALLOWED_ORIGINS) | {origin}
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        env = {**os.environ, "JYOTISH_API_BASE": origin}
+        result = subprocess.run(["node", "--import", "tsx", "scripts/rectification-segment-persisted-replay.ts"],
+                                cwd=ROOT / "frontend", env=env, input=json.dumps({"rows": rows}),
+                                text=True, encoding="utf-8", stdout=subprocess.PIPE, check=False)
+        payload = json.loads(result.stdout)
+        payload.update({"python_version": platform.python_version(),
+                        "dataset_sha256": hashlib.sha256(Path(args.dataset).read_bytes()).hexdigest(),
+                        "runner_exit": result.returncode})
+        target.parent.mkdir(parents=True, exist_ok=True)
+        with target.open("x", encoding="utf-8") as stream:
+            stream.write(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n")
+        print(f"persisted replay rows={len(payload.get('records', []))}; runner_exit={result.returncode}", flush=True)
+        return result.returncode
+    finally:
+        server.shutdown()
+        thread.join(timeout=10)
+        server.server_close()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/research/varga_resolution_probe.py

@@ -32,6 +32,8 @@ Usage::
 from __future__ import annotations
 
 import argparse
+import hashlib
+import platform
 import statistics
 import sys
 import time
@@ -411,7 +413,11 @@ def main() -> int:
     parser.add_argument("--baseline-out", default=str(BASELINE_JSON))
     parser.add_argument("--json-out", default=str(REPORT_JSON))
     parser.add_argument("--quiet", action="store_true")
+    parser.add_argument("--posterior", choices=("research", "production"), default="research")
     args = parser.parse_args()
+    if args.posterior == "production":
+        from scripts.research.varga_resolution_production_bridge import ProductionSegmentBridge
+        ProductionSegmentBridge().install(vr)
     stages = {s.strip() for s in args.stages.split(",") if s.strip()}
     radii = tuple(int(r) for r in args.radii.split(",") if r.strip())
     cache_dir = Path(args.cache_dir) if args.cache_dir else None
@@ -440,6 +446,18 @@ def main() -> int:
         "lmt_era_before_year": vr.LMT_ERA_BEFORE_YEAR,
         "open_set_not_blind": True,
     }
+    if args.posterior == "production":
+        sources = ["frontend/scripts/rectification-segment-bridge.ts",
+                   "frontend/src/lib/rectification-agentic/core/build-state.ts",
+                   "frontend/src/lib/rectification-agentic/core/apply-probe-outcome.ts",
+                   "frontend/src/lib/rectification-agentic/core/segment-summary.ts",
+                   "frontend/src/lib/rectification-agentic/core/segment-probe-order.ts",
+                   "scripts/research/varga_resolution_production_bridge.py"]
+        meta["implementation_identity"] = {
+            "posterior": args.posterior, "python_version": platform.python_version(),
+            "scope": "m1 raw answers via buildInferenceState; mass/summary/gain via production TS; m2 loop remains research experiment, including >61",
+            "source_sha256": {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in sources},
+        }
     if "m0" in stages:
         baseline = {**meta, **stage_m0(replays, signs60, radii)}
         Path(args.baseline_out).write_text(vr.stable_json(baseline), encoding="utf-8")

scripts/research/varga_resolution_production_bridge.py

@@ -0,0 +1,81 @@
+"""Offline transport to the production TypeScript segment functions, not a replica."""
+from __future__ import annotations
+
+import atexit
+import json
+import os
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+class ProductionSegmentBridge:
+    def __init__(self):
+        node = os.environ.get("VARGA_REPLAY_NODE", "node")
+        self.process = subprocess.Popen(
+            [node, "--import", "tsx", "scripts/rectification-segment-bridge.ts"],
+            cwd=ROOT / "frontend", stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+            stderr=None, text=True, encoding="utf-8", bufsize=1,
+        )
+        atexit.register(self.close)
+
+    def close(self):
+        if self.process.poll() is None:
+            self.process.terminate()
+            self.process.wait(timeout=10)
+
+    def call(self, payload):
+        self.process.stdin.write(json.dumps(payload) + "\n")
+        self.process.stdin.flush()
+        line = self.process.stdout.readline()
+        if not line:
+            raise RuntimeError("production TS bridge exited")
+        value = json.loads(line)
+        if "error" in value:
+            raise RuntimeError(value["error"])
+        return value
+
+    def install(self, vr):
+        self.vr = vr
+        # The caller has already distributed raw/percent/uniform mass. Passing
+        # one row per minute retains that exact mass (lead is deliberately not
+        # applied a second time here); production reducer still owns all math.
+        def summary(weights, segments, true_time):
+            value = self.call({
+                "operation": "summary_weights", "weights": weights,
+                "segments": segments,
+                "offsets": {stamp: vr.offset_of(stamp, true_time) for stamp in weights},
+                "truth_offset": 0,
+            })
+            value.pop("tied", None)
+            return value
+
+        def gain(probe, weights, segments, true_time):
+            return self.call({
+                "operation": "gain_weights", "weights": weights,
+                "probes": [probe], "segments": segments,
+                "offsets": {stamp: vr.offset_of(stamp, true_time) for stamp in weights},
+            })["gains"][0]
+
+        original_weights = vr.minute_weights
+
+        def minute_weights(state, mode, *, valid_only=True):
+            if mode != "raw" or not valid_only:
+                return original_weights(state, mode, valid_only=valid_only)
+            return self.call({
+                "operation": "weights", "rows": state["public"],
+                "eliminated": sorted(state["eliminated"]),
+            })
+
+        def replay_scores(public, probes, true_time, *, answers=None, ask_count=vr.ASK_COUNT):
+            asked = list(probes)[:ask_count]
+            given = list(answers) if answers is not None else [vr.optimal_answer(p, true_time) for p in asked]
+            result = self.call({"operation": "replay_raw", "rows": public, "probes": asked, "answers": given})
+            return vr.finish_state(public, result["scores"], set(result["eliminated"]),
+                                   sum(answer is not None for answer in given[:len(asked)]))
+
+        vr.replay_scores = replay_scores
+        vr.minute_weights = minute_weights
+        vr.segment_shares = summary
+        vr.segment_information_gain = gain

scripts/run_quality_gate.py

@@ -64,6 +64,8 @@ CORE_PYTEST_TARGETS = [
     # stale window_scan assertion in this glob stayed red on origin/staging
     # until listed here.
     "tests/test_rectification_*.py",
+    "tests/test_varga_segments_api.py",
+    "tests/test_varga_resolution_research.py",
     # This file regexes frontend source. Home-split and other page.tsx moves must keep it green.
     "tests/test_supabase_user_data_contract.py",
     # Consultation birth-accuracy invariants are pytest-style functions; unittest