ClaudeandClaude Opus 5.5
9ca1a9bdd4
The brief's own case (both candidates carry e1-e5, only 12:01's e5 MD/AD is
indivisible) and the unchanged can_apply/confidence check were not written
into tests. Values measured against staging c9f2f76f.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DbBXoyn7Hj3LCW3BUs5bZ8
412 lines
19 KiB
Python
412 lines
19 KiB
Python
"""BUG-1827 · 占星师第七轮 P1 第 2 条：把「候选分钟数 / 连续区间数 / 是否唯一时刻」分开。
|
||
|
||
她的例子：共同分数 8:8、候选区间 12:00–12:01、``comparable_tied=false``——这是一个**连续领先的
|
||
区间**，不是唯一的分钟。代表分钟不能冒充唯一解，所以结果里三个计数分开写。
|
||
|
||
本文件先只覆盖这三个计数（P1 第 1、3 条要动计分粒度与缺失字段形状，等任务书拆分后再补）。
|
||
合成匿名行，不含任何个人资料。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
for path in (str(ROOT), str(ROOT / 'scripts')):
|
||
if path not in sys.path:
|
||
sys.path.insert(0, path)
|
||
|
||
from scripts.active_rectification_events import adjudicate_candidate_rows # noqa: E402
|
||
from scripts.active_rectification_event_engine import compute_event_candidate_rows # noqa: E402
|
||
from tests.neutral_subject import NEUTRAL_BIRTH_PAYLOAD # noqa: E402
|
||
|
||
DOMAINS = ["career", "relationship", "finance", "career", "relationship"]
|
||
|
||
|
||
def _row(time: str, points: dict[str, float], missing: list[str] | None = None) -> dict:
|
||
return {
|
||
"time": time,
|
||
"score": round(sum(points.values()), 4),
|
||
"evidence": [
|
||
{"event_id": event_id, "domain": DOMAINS[int(event_id[1:]) - 1], "candidate_time": time,
|
||
"rule_ids": ["synthetic"], "points": value}
|
||
for event_id, value in points.items()
|
||
],
|
||
"missing_layers": missing or [],
|
||
}
|
||
|
||
|
||
def _adjudicate(rows: list[dict]) -> dict:
|
||
return adjudicate_candidate_rows(
|
||
rows, event_count=5, domain_count=3, request_fingerprint="synthetic-bug1827",
|
||
leave_one_event_out={"status": "pass", "runs": []},
|
||
)
|
||
|
||
|
||
FIVE = {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2}
|
||
|
||
|
||
def test_a_two_minute_lead_is_one_segment_but_not_one_unique_time() -> None:
|
||
rows = [_row("12:00", FIVE), _row("12:01", FIVE), _row("12:02", {"e1": 2, "e2": 2})]
|
||
result = _adjudicate(rows)
|
||
# 12:00 与 12:01 同分且相邻：两个候选分钟，但只有一个连续区间。
|
||
assert result["top_candidate_count"] == 2
|
||
assert result["top_segment_count"] == 1
|
||
assert result["unique_time"] is False
|
||
|
||
|
||
def test_a_single_leader_is_one_candidate_and_one_segment() -> None:
|
||
rows = [_row("12:00", FIVE), _row("12:01", {"e1": 2, "e2": 2, "e3": 2})]
|
||
result = _adjudicate(rows)
|
||
assert result["top_candidate_count"] == 1
|
||
assert result["top_segment_count"] == 1
|
||
assert result["unique_time"] is True
|
||
|
||
|
||
def test_equal_scores_on_two_separate_minutes_are_two_segments() -> None:
|
||
rows = [_row("12:00", FIVE), _row("12:05", FIVE), _row("12:10", {"e1": 2})]
|
||
result = _adjudicate(rows)
|
||
assert result["top_candidate_count"] == 2
|
||
assert result["top_segment_count"] == 2
|
||
assert result["unique_time"] is False
|
||
assert "tied_leader" in result["reasons"]
|
||
|
||
|
||
def test_the_representative_time_never_claims_to_be_the_only_minute() -> None:
|
||
rows = [_row("12:00", FIVE), _row("12:01", FIVE)]
|
||
result = _adjudicate(rows)
|
||
# 两分钟连续领先：winning_segment 是一段（宽 2 分钟），不是一个唯一时刻。
|
||
segment = result["winning_segment"]
|
||
assert segment["width_minutes"] == 2
|
||
assert segment["start_time"] == "12:00" and segment["end_time"] == "12:01"
|
||
assert result["unique_time"] is False
|
||
assert result["top_candidate_count"] == 2
|
||
|
||
|
||
def test_counts_are_zero_when_there_are_no_candidates() -> None:
|
||
result = adjudicate_candidate_rows(
|
||
[], event_count=5, domain_count=3, request_fingerprint="synthetic-bug1827-empty",
|
||
)
|
||
assert result["top_candidate_count"] == 0
|
||
assert result["top_segment_count"] == 0
|
||
assert result["unique_time"] is False
|
||
|
||
|
||
# --- P1 第 1 条：逐项贡献 ---
|
||
|
||
def _scored_row(time: str, points: dict[str, float], contributions: list[dict]) -> dict:
|
||
row = _row(time, points)
|
||
row["contributions"] = contributions
|
||
return row
|
||
|
||
|
||
def _contribution(event_id: str, layer: str, points, decomposable=True, reason=None) -> dict:
|
||
return {
|
||
"event_id": event_id, "layer": layer,
|
||
"dependency": ["Vimshottari_MD_AD_PD", "Narayana_MD", "Narayana_AD"],
|
||
"profile": {"functional_profile": "pvr_table30_v1"},
|
||
"points": points, "decomposable": decomposable, "missing_reason": reason,
|
||
}
|
||
|
||
|
||
def test_contribution_points_add_up_to_the_row_score() -> None:
|
||
rows = [_scored_row("12:00", {"e1": 2, "e2": 2}, [
|
||
_contribution("e1", "dasha_varga_core", 1.5),
|
||
_contribution("e1", "controlled_transit", 0.5),
|
||
_contribution("e2", "dasha_varga_core", 2.0),
|
||
])]
|
||
result = _adjudicate(rows)
|
||
assert result["top_score"] == 4.0
|
||
total = round(sum(item["points"] for item in rows[0]["contributions"]), 4)
|
||
assert total == rows[0]["score"] == 4.0
|
||
|
||
|
||
def test_an_event_that_is_not_scored_is_recorded_as_a_contribution() -> None:
|
||
rows = [_scored_row("12:00", {"e1": 2}, [
|
||
_contribution("e1", "dasha_varga_core", 2.0),
|
||
_contribution("e2", "Narayana_MD_AD", None, decomposable=False,
|
||
reason="event_not_scored:Narayana_MD_AD"),
|
||
])]
|
||
# 整件不计分的那件事在贡献里留了痕：points 为 null、原因点名门控层。
|
||
blocked = [item for item in rows[0]["contributions"]
|
||
if item["layer"] == "Narayana_MD_AD" and item["points"] is None]
|
||
assert blocked
|
||
assert blocked[0]["missing_reason"] == "event_not_scored:Narayana_MD_AD"
|
||
assert blocked[0]["decomposable"] is False
|
||
# 它不是行上的 missing_layers，所以缺失清单里不会凭空多出一层。
|
||
assert _adjudicate(rows)["missing_mandatory_layers"] == []
|
||
|
||
|
||
# --- P1 第 1 条：拆不开的 MD/AD 合分对所有候选一致排除 ---
|
||
|
||
def test_indivisible_dasha_contributions_are_excluded_for_every_candidate() -> None:
|
||
left = _row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2})
|
||
right = _row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_AD"])
|
||
left["contributions"] = [_contribution("e5", "dasha_varga_core", 2.0, decomposable=True)]
|
||
right["contributions"] = [_contribution("e5", "dasha_varga_core", 2.0, decomposable=False,
|
||
reason="Narayana_AD_unresolved")]
|
||
result = _adjudicate([left, right])
|
||
comparison = result["coverage_comparison"]
|
||
assert comparison["excluded_contributions"] == [
|
||
{"event_id": "e5", "layer": "dasha_varga_core", "reason": "md_ad_not_decomposable"},
|
||
]
|
||
# 两边一起拿掉之后，共同覆盖里的分数相同才是可比的。
|
||
assert comparison["comparison"] == "common_coverage"
|
||
|
||
|
||
def test_legacy_rows_without_contributions_never_exclude_anything() -> None:
|
||
left = _row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2})
|
||
right = _row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_AD"])
|
||
result = _adjudicate([left, right])
|
||
# 没有贡献明细的行不会被当成「拆不开」，所以不会排除任何单元。
|
||
assert "excluded_contributions" not in result["coverage_comparison"]
|
||
# 层名对照照常生效（那是输入适配，与 contribution 无关）。
|
||
assert result["coverage_comparison"]["comparison"] == "common_coverage"
|
||
|
||
|
||
# --- P1 第 3 条：missing_mandatory_layers ---
|
||
|
||
def test_missing_mandatory_layers_lists_candidate_layer_and_events() -> None:
|
||
left = _row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2})
|
||
right = _row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_AD"])
|
||
right["contributions"] = [_contribution("e5", "dasha_varga_core", 2.0, decomposable=False,
|
||
reason="Narayana_AD_unresolved")]
|
||
result = _adjudicate([left, right])
|
||
assert result["missing_mandatory_layers"] == [
|
||
{"candidate_time": "12:01", "layer": "Narayana_AD", "event_ids": ["e5"]},
|
||
]
|
||
assert "missing_mandatory_layers" in result["reasons"]
|
||
|
||
|
||
def test_legacy_rows_mark_the_event_ids_as_unknown() -> None:
|
||
result = _adjudicate([_row("12:00", {"e1": 2}), _row("12:01", {"e1": 2}, ["D9"])])
|
||
entry = next(item for item in result["missing_mandatory_layers"] if item["layer"] == "D9")
|
||
assert entry["event_ids"] is None
|
||
assert entry["event_ids_source"] == "legacy_row_without_contributions"
|
||
|
||
|
||
def test_the_missing_list_is_never_null() -> None:
|
||
clean = _adjudicate([_row("12:00", {"e1": 2}), _row("12:01", {"e1": 2})])
|
||
assert clean["missing_mandatory_layers"] == []
|
||
assert "missing_mandatory_layers" not in clean["reasons"]
|
||
empty = adjudicate_candidate_rows([], event_count=5, domain_count=3, request_fingerprint="synthetic-empty")
|
||
assert empty["missing_mandatory_layers"] == []
|
||
|
||
|
||
# --- 层名对照（输入适配） ---
|
||
|
||
def _alias_row(time: str, points: dict[str, float], missing: list[str]) -> dict:
|
||
return _row(time, points, missing)
|
||
|
||
|
||
def test_the_astrologer_spelling_is_explained_and_recorded() -> None:
|
||
full = _alias_row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2}, [])
|
||
short_her = _alias_row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_AD"])
|
||
her = _adjudicate([full, short_her])
|
||
assert her["coverage_comparison"]["comparison"] == "common_coverage"
|
||
assert her["coverage_comparison"]["layer_aliases_applied"] == {"Narayana_AD": "Narayana_MD_AD"}
|
||
|
||
full2 = _alias_row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2}, [])
|
||
short_product = _alias_row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_MD_AD"])
|
||
product = _adjudicate([full2, short_product])
|
||
assert product["coverage_comparison"]["comparison"] == "common_coverage"
|
||
assert "layer_aliases_applied" not in product["coverage_comparison"]
|
||
|
||
|
||
def test_a_second_gating_layer_blocks_the_alias() -> None:
|
||
full = _alias_row("12:00", {"e1": 2, "e2": 2, "e3": 2, "e4": 2, "e5": 2}, [])
|
||
short = _alias_row("12:01", {"e1": 2, "e2": 2, "e3": 2, "e4": 2}, ["Narayana_AD", "D9"])
|
||
result = _adjudicate([full, short])
|
||
assert result["coverage_comparison"]["comparison"] == "not_reducible"
|
||
assert "layer_aliases_applied" not in result["coverage_comparison"]
|
||
|
||
|
||
def test_the_product_spelling_for_only_the_antardasha_stays_unreducible() -> None:
|
||
left = _alias_row("12:00", {"e1": 2, "e2": 2}, [])
|
||
right = _alias_row("12:01", {"e1": 2, "e2": 1}, ["Narayana_AD"])
|
||
result = _adjudicate([left, right])
|
||
# 产品的 Narayana_AD 意思是「这件事仍计分、只缺小运那一项」：不能套对照表。
|
||
assert result["coverage_comparison"]["comparison"] == "not_reducible" or (
|
||
result["coverage_comparison"].get("unreducible_layers") == ["Narayana_AD"]
|
||
)
|
||
|
||
|
||
# --- 真实形状：引擎行上确实带 contributions，且与非空分值对得上 ---
|
||
|
||
def test_real_engine_rows_carry_contributions_that_add_up_to_the_score() -> None:
|
||
request = {
|
||
**NEUTRAL_BIRTH_PAYLOAD,
|
||
"birth_date": "1990-01-01",
|
||
"start_time": "12:00",
|
||
"end_time": "12:03",
|
||
"events": [
|
||
{"id": "e1", "domain": "career", "date": "2015-06", "precision": "month", "description": "synthetic"},
|
||
{"id": "e2", "domain": "relationship", "date": "2017-09", "precision": "month", "description": "synthetic"},
|
||
{"id": "e3", "domain": "finance", "date": "2019-02", "precision": "month", "description": "synthetic"},
|
||
],
|
||
}
|
||
rows = compute_event_candidate_rows(request)
|
||
assert rows
|
||
for row in rows:
|
||
contributions = row["contributions"]
|
||
assert contributions, "every candidate row records its per-layer contributions"
|
||
scored = round(sum(item["points"] for item in contributions if item["points"] is not None), 4)
|
||
assert scored == row["score"], (scored, row["score"])
|
||
# layer_points 与 contributions 一一对应（同一件事、同一层）。
|
||
from_points = {(event_id, layer) for event_id, layers in row["layer_points"].items() for layer in layers}
|
||
from_contributions = {
|
||
(item["event_id"], item["layer"]) for item in contributions if item["points"] is not None
|
||
}
|
||
assert from_points == from_contributions, (sorted(from_points), sorted(from_contributions))
|
||
|
||
|
||
def test_real_engine_rows_expose_the_planned_layer_set() -> None:
|
||
request = {
|
||
**NEUTRAL_BIRTH_PAYLOAD,
|
||
"birth_date": "1990-01-01",
|
||
"start_time": "12:00",
|
||
"end_time": "12:01",
|
||
"events": [
|
||
{"id": "e1", "domain": "career", "date": "2015-06", "precision": "month", "description": "synthetic"},
|
||
],
|
||
}
|
||
row = compute_event_candidate_rows(request)[0]
|
||
layers = {item["layer"] for item in row["contributions"]}
|
||
assert "dasha_varga_core" in layers
|
||
core = next(item for item in row["contributions"] if item["layer"] == "dasha_varga_core")
|
||
assert core["dependency"][:3] == ["Vimshottari_MD_AD_PD", "Narayana_MD", "Narayana_AD"]
|
||
assert isinstance(core["decomposable"], bool)
|
||
|
||
|
||
# --- FIX-3：排除之后要按剩下的共同覆盖比较（真实形状行） ---
|
||
|
||
def _units_row(time: str, units: dict[str, dict[str, float]], *,
|
||
missing: list[str] | None = None,
|
||
indecomposable: tuple[str, str] | None = None) -> dict:
|
||
"""真实形状行：layer_points 与 contributions 一一对应；`indecomposable=(event, reason)` 标拆不开的那一项。"""
|
||
score = round(sum(sum(layers.values()) for layers in units.values()), 4)
|
||
evidence = [
|
||
{"event_id": event_id, "domain": DOMAINS[int(event_id[1:]) - 1], "candidate_time": time,
|
||
"rule_ids": ["synthetic"], "points": round(sum(layers.values()), 4)}
|
||
for event_id, layers in units.items()
|
||
]
|
||
contributions = []
|
||
for event_id, layers in units.items():
|
||
for layer, points in layers.items():
|
||
flag = indecomposable is not None and layer == "dasha_varga_core" and event_id == "e5"
|
||
contributions.append(_contribution(
|
||
event_id, layer, points,
|
||
decomposable=not flag,
|
||
reason=indecomposable[1] if flag else None,
|
||
))
|
||
return {
|
||
"time": time, "score": score, "evidence": evidence,
|
||
"missing_layers": list(missing or []),
|
||
"layer_points": {event_id: dict(layers) for event_id, layers in units.items()},
|
||
"contributions": contributions,
|
||
}
|
||
|
||
|
||
CORE = {"dasha_varga_core": 2.0, "controlled_transit": 0.5}
|
||
|
||
|
||
def test_indivisible_dasha_is_excluded_and_the_rest_is_compared() -> None:
|
||
full = _units_row("12:00", {"e1": CORE, "e2": CORE})
|
||
short = _units_row("12:01", {"e1": CORE, "e2": CORE, "e5": CORE},
|
||
missing=["Narayana_AD"], indecomposable=("e5", "Narayana_AD_unresolved"))
|
||
result = _adjudicate([full, short])
|
||
comparison = result["coverage_comparison"]
|
||
assert comparison["comparison"] == "common_coverage", comparison
|
||
assert comparison["excluded_contributions"] == [
|
||
{"event_id": "e5", "layer": "dasha_varga_core", "reason": "md_ad_not_decomposable"},
|
||
]
|
||
assert "unreducible_layers" not in comparison
|
||
# 共同覆盖里 e5 的那一层已被拿掉，两侧都按 e1、e2 的剩余单元计分。
|
||
assert comparison["common_scores"] == [
|
||
{"time": "12:00", "score": 5.0},
|
||
{"time": "12:01", "score": 5.0},
|
||
]
|
||
assert comparison.get("comparable_tied") is True or result["comparable_leader"] is not None
|
||
|
||
|
||
def test_a_second_missing_layer_keeps_it_unreducible_but_reports_the_exclusion() -> None:
|
||
full = _units_row("12:00", {"e1": CORE, "e2": CORE})
|
||
short = _units_row("12:01", {"e1": CORE, "e2": CORE, "e5": CORE},
|
||
missing=["Narayana_AD", "D9"], indecomposable=("e5", "Narayana_AD_unresolved"))
|
||
result = _adjudicate([full, short])
|
||
comparison = result["coverage_comparison"]
|
||
assert comparison["comparison"] == "not_reducible"
|
||
# 第二个缺失层让这一处差异说不清，所以仍是 not_reducible；但排除清单照常输出。
|
||
assert comparison["unreducible_layers"]
|
||
assert comparison["excluded_contributions"] == [
|
||
{"event_id": "e5", "layer": "dasha_varga_core", "reason": "md_ad_not_decomposable"},
|
||
]
|
||
|
||
|
||
def test_legacy_rows_never_report_exclusions() -> None:
|
||
left = _row("12:00", {"e1": 2, "e2": 2}, [])
|
||
right = _row("12:01", {"e1": 2, "e2": 2}, ["Narayana_AD"])
|
||
result = _adjudicate([left, right])
|
||
assert "excluded_contributions" not in result["coverage_comparison"]
|
||
|
||
|
||
def test_negative_contributions_still_count_after_exclusion() -> None:
|
||
negative = {"dasha_varga_core": -1.0, "controlled_transit": 0.5}
|
||
full = _units_row("12:00", {"e1": CORE, "e2": negative})
|
||
short = _units_row("12:01", {"e1": CORE, "e2": negative, "e5": CORE},
|
||
missing=["Narayana_AD"], indecomposable=("e5", "Narayana_AD_unresolved"))
|
||
result = _adjudicate([full, short])
|
||
comparison = result["coverage_comparison"]
|
||
assert comparison["comparison"] == "common_coverage"
|
||
# e1: 2.5 + e2: -0.5 = 2.0；负分没有被抹掉，也没有被当成 0。
|
||
assert [item["score"] for item in comparison["common_scores"]] == [2.0, 2.0]
|
||
|
||
|
||
# --- 验收补测（Claude 2026-11-11，FIX-3 第 3 条第 1、5 项）---
|
||
|
||
def test_brief_example_both_candidates_have_e1_to_e5_and_only_e5_core_is_excluded() -> None:
|
||
"""FIX-3 原表：两个候选都有 e1–e5，只有 12:01 的 e5 MD/AD 合分拆不开。"""
|
||
events = {f"e{index}": CORE for index in range(1, 6)}
|
||
full = _units_row("12:00", events)
|
||
short = _units_row("12:01", events, missing=["Narayana_AD"], indecomposable=("e5", "Narayana_AD_unresolved"))
|
||
result = _adjudicate([full, short])
|
||
comparison = result["coverage_comparison"]
|
||
assert comparison["comparison"] == "common_coverage", comparison
|
||
assert "unreducible_layers" not in comparison
|
||
assert comparison["excluded_contributions"] == [
|
||
{"event_id": "e5", "layer": "dasha_varga_core", "reason": "md_ad_not_decomposable"},
|
||
]
|
||
# e5 的 dasha_varga_core 对两个候选一起拿掉：4 × 2.0 + 5 × 0.5 = 10.5，两侧相同。
|
||
assert comparison["common_scores"] == [
|
||
{"time": "12:00", "score": 10.5},
|
||
{"time": "12:01", "score": 10.5},
|
||
]
|
||
assert result["missing_mandatory_layers"] == [
|
||
{"candidate_time": "12:01", "layer": "Narayana_AD", "event_ids": ["e5"]},
|
||
]
|
||
|
||
|
||
def test_fix3_examples_keep_can_apply_and_confidence_unchanged() -> None:
|
||
"""FIX-3 第 3 条最后一项：上述各例的 can_apply / confidence 与改前（staging c9f2f76f）相同。
|
||
|
||
改前值由 Claude 用 c9f2f76f 的 adjudicate_candidate_rows 对同一组行逐例实测：全部 can_apply=False、confidence=low。
|
||
"""
|
||
events = {f"e{index}": CORE for index in range(1, 6)}
|
||
negative = {"dasha_varga_core": -1.0, "controlled_transit": 0.5}
|
||
examples = [
|
||
[_units_row("12:00", events),
|
||
_units_row("12:01", events, missing=["Narayana_AD"], indecomposable=("e5", "Narayana_AD_unresolved"))],
|
||
[_units_row("12:00", {"e1": CORE, "e2": CORE}),
|
||
_units_row("12:01", {"e1": CORE, "e2": CORE, "e5": CORE}, missing=["Narayana_AD", "D9"],
|
||
indecomposable=("e5", "Narayana_AD_unresolved"))],
|
||
[_units_row("12:00", {"e1": CORE, "e2": negative}),
|
||
_units_row("12:01", {"e1": CORE, "e2": negative, "e5": CORE}, missing=["Narayana_AD"],
|
||
indecomposable=("e5", "Narayana_AD_unresolved"))],
|
||
[_row("12:00", {"e1": 2, "e2": 2}, []), _row("12:01", {"e1": 2, "e2": 2}, ["Narayana_AD"])],
|
||
]
|
||
for rows in examples:
|
||
result = _adjudicate(rows)
|
||
assert result["can_apply"] is False
|
||
assert result["confidence"] == "low"
|