root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(rectification): 有题就接着问，题问完才出卡；引导窗口不再硬贴领域

Browse Source 
出卡时机只看题源有没有空：撤回「门槛达标就短路采集线」的写法，同时
按 D2 保住「题源全空就按现行规则出卡」——门槛只在还有题可问时挡住
出卡，precision_gate_met 改成只上报（新挂在决策与公开投影上），不再
单独决定时机。引导窗口题在无领域轨道上改问开放题，一个时间窗只问一
次；录入卡提交的是「YYYY 年 M 月，<领域>方面有一件事」，不再是题干
的三选一列表。记忆化 golden 只补一个新键并冻结墙钟。离线回放改成注
入真值方向的边界事件，另跑一组反方向对照。Skill 10.0.28。

BUG-747～752

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JUei7K13cYxLHE3Axe4A45
This commit is contained in:
Jesse_Chen
2026-09-16 12:16:55 +00:00
co-authored by Claude Fable 5.1
parent a396bbe076
commit dc732825d8
61 changed files with 6251 additions and 2053 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/guided_collect_holdout_replay.py
+179 -30
View File
@@ -1,6 +1,12 @@
#!/usr/bin/env python3
"""Offline T2 replay: after six probes, inject guided_collect_windows events.
F5 (2026-09-16): the injected event now lands on the **true** candidate's own
boundary date for that window's track and index, so the replay measures what a
user who really lived through that window would report. A second, opposite
pass injects the boundary of the candidate furthest from the true minute, as a
control. Neither pass is a merge gate; both numbers go in the progress note.
Does not change production defaults. Writes a compact JSON summary for the
progress note. Not a merge gate.
"""
@@ -29,6 +35,15 @@ from scripts.active_rectification_event_engine import ( # noqa: E402
compute_candidate_static_contexts,
)
from scripts.rectification.event_probes import ( # noqa: E402
GUIDED_ANY_DOMAIN,
GUIDED_COLLECT_LIMIT,
_birth_year,
_context_time,
_guided_track_starts,
_guided_year_windows,
_remaining_contexts,
_scoreable,
_static_contexts,
discriminating_event_probes,
guided_collect_windows,
)
@@ -68,13 +83,23 @@ def load_cases() -> list[dict[str, Any]]:
return list(payload.get("cases") or [])
def synthetic_event(window: dict[str, Any], index: int) -> dict[str, Any]:
year = int(window["year"])
month = int(window["month_lo"])
last = monthrange(year, month)[1]
domain = str(window.get("domain") or "career")
#: Windows are asked open now (BUG-749); the replay still needs one concrete
#: domain to write into the ledger, so an open window uses this one.
DEFAULT_OPEN_DOMAIN = "career"
def synthetic_event(
window: dict[str, Any],
index: int,
*,
when: date,
) -> dict[str, Any]:
"""One month-precision event on `when`, labelled with the window's domain."""
last = monthrange(when.year, when.month)[1]
raw_domain = str(window.get("domain") or "")
domain = DEFAULT_OPEN_DOMAIN if raw_domain in {"", GUIDED_ANY_DOMAIN} else raw_domain
kind = KIND_BY_DOMAIN.get(domain, "career_change")
stamp = f"{year:04d}-{month:02d}"
stamp = f"{when.year:04d}-{when.month:02d}"
return {
"id": str(uuid5(NAMESPACE_URL, f"guided-collect:{index}:{domain}:{stamp}")),
"domain": domain,
@@ -86,6 +111,74 @@ def synthetic_event(window: dict[str, Any], index: int) -> dict[str, Any]:
}
def _minutes(value: str) -> int:
hour, minute = value.split(":")[:2]
return int(hour) * 60 + int(minute)
def window_boundary_dates(
request: dict[str, Any],
built: dict[str, Any],
*,
candidate_times: Sequence[str],
today: date,
) -> dict[tuple[int, int, int], dict[str, date]]:
"""Map each guided window to the per-candidate boundary date behind it.
Mirrors `guided_collect_windows`'s own loop so the (track, index) pairing
is the engine's, not a guess: for every window key (year, month_lo,
month_hi) it returns {candidate_time: that candidate's boundary date}.
"""
birth_date = str(request.get("birth_date") or "").strip()
birth_year = _birth_year(birth_date)
if birth_year is None:
return {}
remaining = _remaining_contexts(built, candidate_times)
if len(remaining) < 2:
remaining = _static_contexts(built)
work = [item for item in remaining if _scoreable(item)]
if len(work) < 2:
return {}
lo, hi = birth_year + 5, min(today.year, birth_year + 80)
vim_cache: dict[tuple[Any, ...], list[date]] = {}
narayana_cache: dict[tuple[Any, ...], list[date] | None] = {}
starts_by_time: dict[str, dict[str, list[date]]] = {}
for context in work:
stamp = _context_time(context)
if not stamp:
continue
starts_by_time[stamp] = _guided_track_starts(
context,
birth_date=birth_date,
lo=lo,
hi=hi,
include_pratyantar=True,
varga_narayana=True,
vim_cache=vim_cache,
narayana_cache=narayana_cache,
)
out: dict[tuple[int, int, int], dict[str, date]] = {}
for track in ("vim", "nara", "nara:d9", "nara:d10"):
lengths = [len(starts_by_time.get(str(_context_time(item)) or "", {}).get(track) or []) for item in work]
if not lengths or min(lengths) == 0:
continue
for index in range(min(lengths)):
per_time: dict[str, date] = {}
for context in work:
stamp = _context_time(context)
dates = (starts_by_time.get(stamp or "") or {}).get(track) or []
if stamp and index < len(dates):
per_time[stamp] = dates[index]
gathered = list(per_time.values())
if len(gathered) < 2 or min(gathered) == max(gathered):
continue
for year, month_lo, month_hi in _guided_year_windows(gathered):
if year < lo or year > hi:
continue
out.setdefault((year, month_lo, month_hi), per_time)
return out
def posterior_state(
*,
rows: Sequence[dict[str, Any]],
@@ -191,6 +284,7 @@ def remaining_times(state: dict[str, Any]) -> list[str]:
def evaluate_case(
case: dict[str, Any],
radius: int,
direction: str = "truth",
) -> dict[str, Any]:
true_time = str(case["birth"]["time"])[:5]
request = scoring_request_for(case, radius)
@@ -215,6 +309,7 @@ def evaluate_case(
return {
"case_id": case.get("case_id"),
"radius": radius,
"direction": direction,
"events_needed": 0,
"windows": 0,
"met": True,
@@ -227,10 +322,33 @@ def evaluate_case(
candidate_times=remaining or times,
today=TODAY,
)
boundaries = window_boundary_dates(
request,
built,
candidate_times=remaining or times,
today=TODAY,
)
pool = remaining or times
if direction == "truth":
source_time = true_time
else:
source_time = max(
pool,
key=lambda stamp: abs(_minutes(stamp) - _minutes(true_time)),
) if pool else true_time
extras: list[dict[str, Any]] = []
last_gate = gate
injected_from: list[str] = []
for index, window in enumerate(windows):
extras.append(synthetic_event(window, index))
key = (int(window["year"]), int(window["month_lo"]), int(window["month_hi"]))
per_time = boundaries.get(key) or {}
when = per_time.get(source_time)
if when is None and per_time:
when = per_time[min(per_time, key=lambda stamp: abs(_minutes(stamp) - _minutes(source_time)))]
if when is None:
when = date(int(window["year"]), int(window["month_lo"]), 1)
injected_from.append(f"{source_time}:{when.isoformat()}")
extras.append(synthetic_event(window, index, when=when))
injected = {**request, "events": list(request["events"]) + extras}
rebuilt = build_event_contribution_matrix(
scoreable_request(injected),
@@ -248,34 +366,55 @@ def evaluate_case(
return {
"case_id": case.get("case_id"),
"radius": radius,
"direction": direction,
"events_needed": len(extras),
"windows": len(windows),
"met": True,
"after_six": gate,
"final": last_gate,
"injected": injected_from,
}
return {
"case_id": case.get("case_id"),
"radius": radius,
"direction": direction,
"events_needed": None,
"windows": len(windows),
"met": False,
"after_six": gate,
"final": last_gate,
"injected": injected_from,
}
def summarize(rows: Sequence[dict[str, Any]], radius: int) -> dict[str, Any]:
subset = [row for row in rows if row.get("radius") == radius and not row.get("error")]
def summarize(rows: Sequence[dict[str, Any]], radius: int, direction: str) -> dict[str, Any]:
subset = [
row for row in rows
if row.get("radius") == radius and row.get("direction") == direction and not row.get("error")
]
needed = [int(row["events_needed"]) for row in subset if row.get("events_needed") is not None]
# Cases the six probes had not already settled: the only ones the guided
# windows can be credited for.
from_guided = [
int(row["events_needed"]) for row in subset
if row.get("events_needed") not in (None, 0)
]
met = sum(1 for row in subset if row.get("met"))
return {
"radius": radius,
"direction": direction,
"n": len(subset),
"met": met,
"met_after_six_only": sum(1 for row in subset if row.get("events_needed") == 0),
"met_via_guided": len(from_guided),
"median_events_to_gate": statistics.median(needed) if needed else None,
"median_events_via_guided": statistics.median(from_guided) if from_guided else None,
"median_windows": statistics.median([int(row.get("windows") or 0) for row in subset]) if subset else None,
"unmet": sum(1 for row in subset if not row.get("met")),
"errors": sum(1 for row in rows if row.get("radius") == radius and row.get("error")),
"errors": sum(
1 for row in rows
if row.get("radius") == radius and row.get("direction") == direction and row.get("error")
),
}
@@ -283,9 +422,11 @@ def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--limit", type=int, default=0)
parser.add_argument("--radii", default="10,30,60")
parser.add_argument("--directions", default="truth,opposite")
parser.add_argument("--json-out", default=str(REPORT_JSON))
args = parser.parse_args()
radii = tuple(int(item) for item in str(args.radii).split(",") if item.strip())
directions = tuple(item.strip() for item in str(args.directions).split(",") if item.strip())
cases = load_cases()
if args.limit:
cases = cases[: args.limit]
@@ -293,30 +434,38 @@ def main() -> int:
rows: list[dict[str, Any]] = []
for case in cases:
for radius in radii:
label = f"{case.get('case_id')} ±{radius}"
try:
result = evaluate_case(case, radius)
rows.append(result)
print(
f"{label} needed={result.get('events_needed')} "
f"met={result.get('met')} windows={result.get('windows')}",
flush=True,
)
except Exception as exc: # noqa: BLE001
rows.append({
"case_id": case.get("case_id"),
"radius": radius,
"error": f"{type(exc).__name__}: {exc}",
"trace": traceback.format_exc(limit=8),
"met": False,
"events_needed": None,
})
print(f"{label} ERROR {type(exc).__name__}: {exc}", flush=True)
summaries = [summarize(rows, radius) for radius in radii]
for direction in directions:
label = f"{case.get('case_id')} ±{radius} {direction}"
try:
result = evaluate_case(case, radius, direction)
rows.append(result)
print(
f"{label} needed={result.get('events_needed')} "
f"met={result.get('met')} windows={result.get('windows')}",
flush=True,
)
except Exception as exc: # noqa: BLE001
rows.append({
"case_id": case.get("case_id"),
"radius": radius,
"direction": direction,
"error": f"{type(exc).__name__}: {exc}",
"trace": traceback.format_exc(limit=8),
"met": False,
"events_needed": None,
})
print(f"{label} ERROR {type(exc).__name__}: {exc}", flush=True)
summaries = [summarize(rows, radius, direction) for radius in radii for direction in directions]
payload = {
"generated_at": TODAY.isoformat(),
"holdout": str(HOLDOUT.relative_to(ROOT)).replace("\\", "/"),
"ask_count": ASK_COUNT,
"guided_collect_limit": GUIDED_COLLECT_LIMIT,
"injection": (
"truth = the true candidate's own boundary date for that window's "
"track and index; opposite = the boundary of the remaining "
"candidate furthest from the true minute"
),
"delivery_max_width_minutes": DELIVERY_MAX_WIDTH_MINUTES,
"tie_percent": RANGE_DELIVERY_TIE_PERCENT,
"elapsed_s": round(time.perf_counter() - started, 1),