避免四舍五入和缺区间把还可能的时间从目前范围里藏掉。

Co-Authored-By: Grok (agent loop)

scripts/rectification/living_display_intervals.py

@@ -1,79 +0,0 @@
-"""Display-only merge of living candidate intervals.
-
-Decision width stays on ``interval_union_width`` / ``unionWindowIntervals``.
-This module is not part of the sealed scoring identity: scoring must not import it.
-"""
-from __future__ import annotations
-
-from datetime import datetime
-from typing import Any
-
-LIVING_DISPLAY_MERGE_GAP_MINUTES = 4
-LIVING_DISPLAY_MAX_SEGMENTS = 3
-
-
-def _stamp(value: str) -> datetime:
-    return datetime.fromisoformat(value)
-
-
-def merge_living_display_intervals(
-    intervals: list[dict[str, Any]],
-    gap_minutes: int = LIVING_DISPLAY_MERGE_GAP_MINUTES,
-) -> list[dict[str, str]]:
-    """Merge intervals whose gap is <= gap_minutes. Gap is the open minutes between them."""
-    rows = []
-    for row in intervals:
-        start, end = str(row.get("start_at") or ""), str(row.get("end_at") or "")
-        if len(start) < 16 or len(end) < 16 or start > end:
-            continue
-        try:
-            _stamp(start)
-            _stamp(end)
-        except ValueError:
-            continue
-        rows.append({"start_at": start[:16], "end_at": end[:16]})
-    rows.sort(key=lambda item: (item["start_at"], item["end_at"]))
-    merged: list[dict[str, str]] = []
-    for row in rows:
-        if not merged:
-            merged.append(row)
-            continue
-        previous = merged[-1]
-        gap = int((_stamp(row["start_at"]) - _stamp(previous["end_at"])).total_seconds() // 60) - 1
-        if gap <= gap_minutes:
-            if row["end_at"] > previous["end_at"]:
-                previous["end_at"] = row["end_at"]
-        else:
-            merged.append(row)
-    return merged
-
-
-def inclusive_interval_minutes(intervals: list[dict[str, str]]) -> int:
-    total = 0
-    for row in intervals:
-        width = int((_stamp(row["end_at"]) - _stamp(row["start_at"])).total_seconds() // 60) + 1
-        if width > 0:
-            total += width
-    return total
-
-
-def living_display_label(
-    intervals: list[dict[str, str]],
-    birth_date: str | None = None,
-    max_segments: int = LIVING_DISPLAY_MAX_SEGMENTS,
-) -> str:
-    def label(value: str) -> str:
-        date, clock = value[:10], value[11:16]
-        if not birth_date:
-            return f"{date} {clock}"
-        if date == birth_date:
-            return clock
-        return f"{'前一天' if date < birth_date else '后一天'} {clock}"
-
-    def text(row: dict[str, str]) -> str:
-        return f"{label(row['start_at'])}–{label(row['end_at'])}"
-
-    if len(intervals) <= max_segments:
-        return "、".join(text(row) for row in intervals)
-    shown = "、".join(text(row) for row in intervals[:max_segments])
-    return f"{shown}，等 {len(intervals)} 段，共约 {inclusive_interval_minutes(intervals)} 分钟"

scripts/research/varga_resolution_persisted_replay.py

@@ -29,6 +29,10 @@ def main() -> int:
     parser.add_argument("--output", required=True)
     parser.add_argument("--limit", type=int, default=10)
     parser.add_argument("--radii", default="10,30,60")
+    parser.add_argument("--study", action="store_true")
+    parser.add_argument("--modes", default="all_correct,flip_1,flip_3,flip_5")
+    parser.add_argument("--orderings", default="off")
+    parser.add_argument("--code-sha", default="")
     args = parser.parse_args()
     target = Path(args.output)
     if target.exists():
@@ -58,13 +62,32 @@ def main() -> int:
     thread.start()
     try:
         env = {**os.environ, "JYOTISH_API_BASE": origin}
+        body: dict = {"rows": rows}
+        if args.study:
+            body["study"] = {
+                "modes": [item for item in args.modes.split(",") if item],
+                "orderings": [item for item in args.orderings.split(",") if item],
+                "code_sha": args.code_sha,
+            }
         result = subprocess.run(["node", "--import", "tsx", "tests/research/rectification-segment-persisted-replay.ts"],
-                                cwd=ROOT / "frontend", env=env, input=json.dumps({"rows": rows}),
+                                cwd=ROOT / "frontend", env=env, input=json.dumps(body),
                                 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})
+        if args.study:
+            for row in payload.get("records") or []:
+                if isinstance(row, dict):
+                    for key in ("asked", "gates", "ledger_before", "ledger_after", "metrics", "targets", "domain"):
+                        row.pop(key, None)
+            payload["rerun"] = {
+                "command": "python scripts/research/varga_resolution_persisted_replay.py --study "
+                           f"--modes {args.modes} --orderings {args.orderings} --radii {args.radii} "
+                           f"--limit {args.limit} --code-sha {args.code_sha or '<HEAD>'} --output <new-file>",
+                "max_rounds": "DEFAULT_MAX_DISCRIMINATION_ROUNDS",
+                "truth_clock": "omitted; real-case calibration stores containment booleans only",
+            }
         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")