fix(rectification): dedupe same-domain probe years and drop unanchored style cards (BUG-559)

Pass asked probe keys into the engine without changing result fingerprints, block nearby years already asked, and require a dated same-domain ledger event before rendering varga_style cards.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/rectification/event_probes.py

@@ -8,6 +8,7 @@ the case cap is respected. Unanchored quality stays clarification-only.
 
 from __future__ import annotations
 
+import re
 from datetime import date, datetime, timedelta
 from math import log2
 from typing import Any, Sequence
@@ -220,6 +221,20 @@ EXISTENCE_NEARBY_YEARS = {
     "career": 1,
     "relocation": 1,
 }
+_SEMANTIC_YEAR = re.compile(r"^(?P<domain>[a-z_]+)\.(?P<year>(?:19|20)\d{2})(?:\.|$)")
+
+
+def asked_years_for_domain(asked_probe_keys: Sequence[str] | None, domain: str) -> set[int]:
+    years: set[int] = set()
+    prefix = f"{domain}."
+    for raw in asked_probe_keys or []:
+        key = str(raw or "").strip()
+        if not key.startswith(prefix):
+            continue
+        match = _SEMANTIC_YEAR.match(key)
+        if match and match.group("domain") == domain:
+            years.add(int(match.group("year")))
+    return years
 
 
 def _clock(value: str) -> int:
@@ -770,23 +785,38 @@ def _agent_brief(
     family: str,
     quality: bool = False,
     exam: bool = False,
+    nearby_note: str = "",
 ) -> str:
+    nearby = nearby_note.strip()
     if exam:
         return (
             f"时间范围锁定 {year_label}；领域锁定 {domain}。"
             "语义目标是那次考试的实际体验。结合最近对话，只选一个容易回答的口语入口，"
             "问是否明显失常或压力很大；不要堆叠例子，不得改时间范围。"
+            + (f"{nearby}" if nearby else "")
         )
     if quality:
         return (
             f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
             "结合最近对话，只选一个容易回答的口语入口来核对体验；"
             "不要逐字复述语义目标，不要堆叠例子，不得改时间范围。"
+            + (f"{nearby}" if nearby else "")
+        )
+    lead = (
+        f"{nearby}时间范围锁定 {year_label}；请问用户那段时间身上发生了什么变化；"
+        if nearby
+        else f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
+        "结合最近对话，只选一个容易回答的口语入口，写一句自然的是/否题；"
+    )
+    if nearby:
+        return (
+            lead
+            + f"领域锁定 {domain}；语义目标是 {family}。"
+            "选项由服务端给出；不要发明年份，不得改时间范围。"
         )
     return (
-        f"时间范围锁定 {year_label}；领域锁定 {domain}；语义目标是 {family}。"
-        "结合最近对话，只选一个容易回答的口语入口，写一句自然的是/否题；"
-        "不要逐字复述语义目标，不要把所有例子堆进一句，不得改时间范围。"
+        lead
+        + "不要逐字复述语义目标，不要把所有例子堆进一句，不得改时间范围。"
     )
 
 
@@ -889,6 +919,79 @@ def _display_date_label(event: dict[str, Any]) -> str:
     return f"{year} 年"
 
 
+def _event_month_index(year: int, month: int | None) -> int | None:
+    if month is None or not 1 <= month <= 12:
+        return None
+    return year * 12 + month
+
+
+def nearby_ledger_note(
+    events: Sequence[dict[str, Any]],
+    *,
+    domain: str,
+    year: int,
+    month: int | None,
+) -> str:
+    if year <= 0:
+        return ""
+    probe_index = _event_month_index(year, month)
+    best: dict[str, Any] | None = None
+    best_delta = 99
+    for event in events:
+        if not isinstance(event, dict):
+            continue
+        other_domain = str(event.get("domain") or "")
+        if other_domain == domain or other_domain not in DOMAIN_CATALOG:
+            continue
+        other_year = _event_year(event)
+        if other_year is None:
+            continue
+        other_month = _event_month(event)
+        other_index = _event_month_index(other_year, other_month)
+        if probe_index is not None and other_index is not None:
+            delta = abs(probe_index - other_index)
+            if delta > 2:
+                continue
+        elif other_year != year:
+            continue
+        else:
+            delta = 2 if probe_index is not None or other_index is not None else 0
+        if delta < best_delta:
+            best_delta = delta
+            best = event
+    if best is None:
+        return ""
+    family = str(DOMAIN_CATALOG[str(best.get("domain") or "")]["event_family"])
+    return f"账本里 { _display_date_label(best) } 有{family}；题干先提那件事再问。"
+
+
+def _annotate_nearby_ledger(
+    probes: Sequence[dict[str, Any]],
+    events: Sequence[dict[str, Any]],
+) -> None:
+    for probe in probes:
+        if not isinstance(probe, dict):
+            continue
+        if str(probe.get("choice_kind") or "existence") != "existence":
+            continue
+        if str(probe.get("source") or "") not in {"dasha_boundary", "dasha_activation"}:
+            continue
+        year = probe.get("year")
+        if not isinstance(year, int) or year <= 0:
+            continue
+        note = nearby_ledger_note(
+            events,
+            domain=str(probe.get("domain") or ""),
+            year=year,
+            month=int(probe["month"]) if isinstance(probe.get("month"), int) else None,
+        )
+        if not note:
+            continue
+        meaning = str(probe.get("user_meaning") or "")
+        if note not in meaning:
+            probe["user_meaning"] = f"{note}{meaning}"
+
+
 def _event_kind_name(event: dict[str, Any]) -> str:
     return str(event.get("event_kind") or event.get("kind") or "")
 
@@ -1402,6 +1505,11 @@ def _discriminating_event_probe_lists(
     except ValueError:
         return empty
     events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
+    asked_probe_keys = [
+        str(item).strip()
+        for item in (request.get("asked_probe_keys") or [])
+        if isinstance(item, str) and str(item).strip()
+    ]
     if not discriminator_gate_open(events):
         return empty
     holdout_keys = holdout_domain_years(events)
@@ -1438,7 +1546,7 @@ def _discriminating_event_probe_lists(
     for domain in domains:
         if domain not in DOMAIN_CATALOG:
             continue
-        known_years = _event_years(events, domain)
+        known_years = _event_years(events, domain) | asked_years_for_domain(asked_probe_keys, domain)
         blocked_years = _existence_blocked_years(domain, known_years)
         domain_lo = _domain_year_floor(birth_year, domain, lo)
         eligible = [
@@ -1506,6 +1614,7 @@ def _discriminating_event_probe_lists(
         holdout_ids=set(holdout_event_ids(events)),
         holdout_keys=set(holdout_keys),
     ))
+    _annotate_nearby_ledger(probes, events)
     probes.sort(key=_probe_sort_key)
     public, dropped = _partition_ranked_probes(probes)
     assert_distinguish_contract(public)