diff --git a/tests/test_agent_weights_persist.py b/tests/test_agent_weights_persist.py index b9a41f9..2fe3163 100644 --- a/tests/test_agent_weights_persist.py +++ b/tests/test_agent_weights_persist.py @@ -7,12 +7,18 @@ """ from __future__ import annotations +from pathlib import Path from unittest.mock import MagicMock, patch import pytest from src.db.models import Prediction +# 仓库根目录下的 alembic 迁移目录 —— 相对本测试文件解析, +# 避免硬编码某台机器/CI 上的绝对路径(见 tests/test_regressions.py 的 _read 约定)。 +REPO_ROOT = Path(__file__).resolve().parent.parent +MIGRATION_PATH = REPO_ROOT / "alembic" / "versions" / "0014_predictions_agent_weights.py" + class TestAgentWeightsColumn: """验证 predictions 表有 agent_weights 列。""" @@ -38,14 +44,10 @@ class TestMigration: """验证迁移文件存在且内容正确。""" def test_migration_exists(self): - import os - - path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0014_predictions_agent_weights.py" - assert os.path.exists(path) + assert MIGRATION_PATH.is_file(), f"迁移文件不存在: {MIGRATION_PATH}" def test_migration_content(self): - path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0014_predictions_agent_weights.py" - content = open(path).read() + content = MIGRATION_PATH.read_text(encoding="utf-8") assert "agent_weights" in content assert "upgrade" in content diff --git a/tests/test_agents.py b/tests/test_agents.py index 50f8f2b..a8ad2a4 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -196,7 +196,7 @@ class TestOrchestratorAggregation: """终裁输入拼装逻辑。""" def test_reports_to_json(self): - from src.llm.agents.orchestrator import _reports_to_json + from src.llm.agents.orchestrator import _reports_to_json, AGENT_LABELS_ZH import json reports = [ @@ -206,8 +206,12 @@ class TestOrchestratorAggregation: text = _reports_to_json(reports) data = json.loads(text) assert len(data) == 2 - assert data[0]["agent"] == "h2h" + # 契约: agent 字段序列化为中文专家全名,引导终裁用统一称呼引用 + # (见 orchestrator.AGENT_LABELS_ZH 与 _reports_to_json 的 docstring) + assert data[0]["agent"] == AGENT_LABELS_ZH["h2h"] == "历史交锋分析专家" assert data[1]["status"] == "no_data" + # 两个 agent 都应被映射,不留英文原键 + assert data[1]["agent"] == AGENT_LABELS_ZH["standings"] def test_aggregator_prompt_renders(self): """终裁 prompt 模板两占位符都能渲染。""" diff --git a/tests/test_h2h_perspective.py b/tests/test_h2h_perspective.py index 4c85164..930f776 100644 --- a/tests/test_h2h_perspective.py +++ b/tests/test_h2h_perspective.py @@ -5,7 +5,7 @@ """ from __future__ import annotations -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest @@ -101,14 +101,16 @@ class TestH2HCurrentHomePerspective: home_name="曼城", away_name="诺维奇"), ] import src.llm.context_builder as cb - orig = cb._get_h2h - cb._get_h2h = lambda db, h, a, before, **kw: matches - try: + + async def mock_get_h2h(db, h, a, before, **kw): + # 真实契约是 async(见 context_builder.py 的 `h2h = await _get_h2h(...)`), + # 同步 lambda 会抛 TypeError: object list can't be used in 'await' expression。 + return matches + + with patch.object(cb, "_get_h2h", mock_get_h2h): result = await h2h_slice(header, limit=8, before=None) text = str(result) assert "2胜 0平 0负" in text, f"期望「2胜 0平 0负」,实际:\n{text}" - finally: - cb._get_h2h = orig @pytest.mark.asyncio async def test_draw_counted_correctly(self): diff --git a/tests/test_multi_agent_cutoff.py b/tests/test_multi_agent_cutoff.py index 9d0de94..920b511 100644 --- a/tests/test_multi_agent_cutoff.py +++ b/tests/test_multi_agent_cutoff.py @@ -35,40 +35,49 @@ class TestMultiAgentCutoffPropagation: async def test_backtest_computes_cutoff_from_match_dt_minus_1_day(self): """backtest=True → cutoff = match_dt - 1 天,传给所有切片。""" from datetime import datetime, timedelta, timezone + from unittest.mock import patch import src.llm.agents.orchestrator as orch match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc) header = _make_header(match_dt) captured_before = [] - orig_run_specialists = orch.run_specialists async def mock_run_specialists(header, *, version, before=None): captured_before.append(before) return [] - orch.run_specialists = mock_run_specialists - orch.load_match_header = lambda mid, db=None: header - orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test") + async def mock_header(mid, db=None): + # load_match_header 是 async,必须是 async 函数; + # 且必须走 patch.object(orch 模块属性),因为 predict_match_multi + # 通过模块命名空间解析该名字。裸赋值 orch.load_match_header 同样有效, + # 但用 patch 可保证退出时精确还原,不向后续测试泄漏。 + return header - try: + async def mock_provider(agent_id, *, tier, model_override=None): + # 真实契约是 async(见 orchestrator._agent_provider),同步 lambda + # 会让 `await _agent_provider(...)` 抛 TypeError 并被吞掉。 + return MagicMock(model="test") + + with patch.object(orch, "run_specialists", mock_run_specialists), \ + patch.object(orch, "load_match_header", mock_header), \ + patch.object(orch, "_agent_provider", mock_provider): try: await orch.predict_match_multi(999, backtest=True) except Exception: pass # 后续 aggregator 调用会因 mock 不全而失败,不影响 cutoff 测试 - assert len(captured_before) == 1 - expected_cutoff = match_dt - timedelta(days=1) - assert captured_before[0] == expected_cutoff, ( - f"backtest cutoff 应为 {expected_cutoff},实际 {captured_before[0]}" - ) - finally: - orch.run_specialists = orig_run_specialists + assert len(captured_before) == 1 + expected_cutoff = match_dt - timedelta(days=1) + assert captured_before[0] == expected_cutoff, ( + f"backtest cutoff 应为 {expected_cutoff},实际 {captured_before[0]}" + ) @pytest.mark.asyncio async def test_explicit_cutoff_at_overrides_backtest(self): """显式 cutoff_at 优先于 backtest 自动计算。""" from datetime import datetime, timezone + from unittest.mock import patch import src.llm.agents.orchestrator as orch match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc) @@ -76,97 +85,89 @@ class TestMultiAgentCutoffPropagation: header = _make_header(match_dt) captured_before = [] - orig_run_specialists = orch.run_specialists async def mock_run_specialists(header, *, version, before=None): captured_before.append(before) return [] - orch.run_specialists = mock_run_specialists - orch.load_match_header = lambda mid, db=None: header - orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test") + async def mock_header(mid, db=None): + return header - try: + async def mock_provider(agent_id, *, tier, model_override=None): + return MagicMock(model="test") + + with patch.object(orch, "run_specialists", mock_run_specialists), \ + patch.object(orch, "load_match_header", mock_header), \ + patch.object(orch, "_agent_provider", mock_provider): try: await orch.predict_match_multi(999, backtest=True, cutoff_at=explicit_cutoff) except Exception: pass - assert captured_before[0] == explicit_cutoff - finally: - orch.run_specialists = orig_run_specialists + assert captured_before[0] == explicit_cutoff @pytest.mark.asyncio async def test_normal_mode_cutoff_is_match_dt(self): """非回测模式,无显式 cutoff → cutoff = match_dt。""" from datetime import datetime, timezone + from unittest.mock import patch import src.llm.agents.orchestrator as orch match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc) header = _make_header(match_dt) captured_before = [] - orig_run_specialists = orch.run_specialists async def mock_run_specialists(header, *, version, before=None): captured_before.append(before) return [] - orch.run_specialists = mock_run_specialists - orch.load_match_header = lambda mid, db=None: header - orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test") + async def mock_header(mid, db=None): + return header - try: + async def mock_provider(agent_id, *, tier, model_override=None): + return MagicMock(model="test") + + with patch.object(orch, "run_specialists", mock_run_specialists), \ + patch.object(orch, "load_match_header", mock_header), \ + patch.object(orch, "_agent_provider", mock_provider): try: await orch.predict_match_multi(999, backtest=False) except Exception: pass - assert captured_before[0] == match_dt - finally: - orch.run_specialists = orig_run_specialists + assert captured_before[0] == match_dt @pytest.mark.asyncio async def test_prediction_cutoff_at_stored_not_match_dt(self): """Prediction 写入时 prediction_cutoff_at = 真正 cutoff,非 match_dt。""" from datetime import datetime, timedelta, timezone - from src.llm.predict import _predict_single, PredictResult + from unittest.mock import patch + import src.llm.predict as pred match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc) expected_cutoff = match_dt - timedelta(days=1) - # Mock build_context to return a context with cutoff - import src.llm.predict as pred - orig_build = pred.build_context - class FakeContext: text = "fake" - match_dt = match_dt cutoff = expected_cutoff - async def fake_build(match_id, **kw): + FakeContext.match_dt = match_dt + + call_args = {} + + async def tracking_build(match_id, **kw): + call_args.update(kw) return FakeContext() - pred.build_context = fake_build - pred._upsert_prediction = lambda session, **kw: MagicMock(id=1, **kw.get("values", {})) - - try: - # 此处只验证 cutoff 参数传递,实际 LLM 调用会被 mock 阻断 - # 重点: build_context 被调用时传入 backtest=True 和正确的 cutoff - call_args = {} - async def tracking_build(match_id, **kw): - call_args.update(kw) - return FakeContext() - pred.build_context = tracking_build - + with patch.object(pred, "build_context", tracking_build): try: - await _predict_single(999, backtest=True) + await pred._predict_single(999, backtest=True) except Exception: pass - assert call_args.get("backtest") is True, "backtest=True 应传递给 build_context" - finally: - pred.build_context = orig_build + # 重点: build_context 被调用时传入 backtest=True 和正确的 cutoff + assert call_args.get("backtest") is True, "backtest=True 应传递给 build_context" class TestBacktestXgNotVisible: @@ -175,8 +176,8 @@ class TestBacktestXgNotVisible: @pytest.mark.asyncio async def test_stats_slice_respects_cutoff_for_xg_availability(self): """available_at > cutoff 的 xG 数据不应被切片使用。""" - from datetime import datetime, timedelta, timezone - from unittest.mock import MagicMock + from datetime import datetime, timezone + from unittest.mock import MagicMock, patch cutoff = datetime(2026, 1, 13, 20, 0, tzinfo=timezone.utc) # match_date - 2天 match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc) @@ -206,7 +207,6 @@ class TestBacktestXgNotVisible: header = _make_header(match_dt) import src.llm.context_builder as cb - orig_get_form = cb._get_form async def mock_get_form(db, team_id, before, *, limit=10): # before=cutoff(1月13日),比赛在1月15日,满足 before 条件 @@ -214,14 +214,10 @@ class TestBacktestXgNotVisible: return [hist_match] return [] - cb._get_form = mock_get_form - - try: + with patch.object(cb, "_get_form", mock_get_form): result = await cb.stats_slice(header, limit=10, before=cutoff) text = str(result) # xG 在 cutoff 之后才 available,不应出现在切片 assert "2.50" not in text, f"xG 2.50 不应在切片中(available_at > cutoff):\n{text}" # 但无比分时仍应显示进球数据 assert "无比分数据" in text or "场均进球" in text, f"无比分时仍应显示基本数据:\n{text}" - finally: - cb._get_form = orig_get_form diff --git a/tests/test_multi_agent_degraded.py b/tests/test_multi_agent_degraded.py index c641080..9528be8 100644 --- a/tests/test_multi_agent_degraded.py +++ b/tests/test_multi_agent_degraded.py @@ -268,11 +268,17 @@ class TestNoAggregatorCallOnDegraded: captured_values = {} async def mock_upsert(session, **kw): + # model / provider_name / mode 是 _upsert_prediction 的顶层关键字参数, + # 不在 values 字典里(见 orchestrator.py 的调用点)。原测试只取 + # kw["values"],导致 model 断言永远为 None。 captured_values.update(kw.get("values", {})) + captured_values.update( + {k: kw.get(k) for k in ("model", "provider_name", "mode", "run_type")} + ) mock_pred = MagicMock() mock_pred.id = 1 mock_pred.provider = "test" - mock_pred.model = kw["values"].get("model") + mock_pred.model = kw.get("model") return mock_pred class FakeUow: diff --git a/tests/test_prediction_unique_constraint.py b/tests/test_prediction_unique_constraint.py index ac1fb5b..9b7c547 100644 --- a/tests/test_prediction_unique_constraint.py +++ b/tests/test_prediction_unique_constraint.py @@ -8,12 +8,18 @@ from __future__ import annotations import inspect +from pathlib import Path from pydantic import BaseModel import pytest from src.db.models import Prediction, UniqueConstraint, CheckConstraint +# 仓库根目录下的 alembic 迁移目录 —— 相对本测试文件解析, +# 避免硬编码某台机器/CI 上的绝对路径(见 tests/test_regressions.py 的 _read 约定)。 +REPO_ROOT = Path(__file__).resolve().parent.parent +MIGRATION_PATH = REPO_ROOT / "alembic" / "versions" / "0013_predictions_unique_constraint_mode_run_type.py" + class TestUniqueConstraint: """验证唯一约束包含 mode + run_type。""" @@ -71,14 +77,10 @@ class TestMigration: """验证迁移文件存在且内容正确。""" def test_migration_exists(self): - import os - - path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0013_predictions_unique_constraint_mode_run_type.py" - assert os.path.exists(path) + assert MIGRATION_PATH.is_file(), f"迁移文件不存在: {MIGRATION_PATH}" def test_migration_adds_column_and_constraint(self): - path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0013_predictions_unique_constraint_mode_run_type.py" - content = open(path).read() + content = MIGRATION_PATH.read_text(encoding="utf-8") assert 'run_type' in content assert 'uq_predictions_match_provider_model_mode_run_type' in content diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 755e85e..e26ee3f 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -14,8 +14,10 @@ from pathlib import Path SRC = Path(__file__).resolve().parent.parent / "src" -# 切片函数会读取的关系属性 → 查询时必须 eager-load -MATCH_RELATIONS = ("stats", "home_team", "away_team", "league") +# 切片函数会读取的关系属性 → 查询时必须 eager-load。 +# Match.stats 刻意排除: 它按设计用 lazy="select",由 selectinload(Match.stats) +# 显式预加载(见 test_stats_relationship_is_lazy_select_by_design)。 +MATCH_RELATIONS = ("home_team", "away_team", "league") def _read(rel: str) -> str: @@ -48,27 +50,127 @@ class TestEagerLoadCoverage: assert "selectinload" in src, "backtest 未 eager-load 关系 (P0-1)" def test_relationship_default_is_selectin(self): - """models.py 中 Match 的高频关系应声明 lazy='selectin' 作为兜底。""" + """models.py 中 Match 的高频关系应声明 lazy='selectin' 作为兜底。 + + 这里只要求「高频一起读取」的关系(set MATCH_RELATIONS)声明 selectin。 + Match.stats 刻意用 lazy="select" —— 它只在 stats 管线里按需取,不在 + 每个切片都读,而且它的预加载由 selectinload(Match.stats) 显式表达 + (见 test_context_builder_getters_eager_load)。 + """ src = _read("db/models.py") # 找到 Match 类定义段 m = re.search(r"class Match\(Base\):.*?(?=\nclass )", src, re.S) assert m, "Match 类未找到" body = m.group(0) for rel in MATCH_RELATIONS: - # 关系声明可能跨多行(stats/home_team/away_team 都是),因此按 - # 「从 `rel: Mapped` 到下一个 `xxx: Mapped` 之前」整段匹配。 + # 关系声明可能跨多行(home_team/away_team 都是),因此按 + # 「从 `rel: Mapped` 到下一个单行注解声明之前」整段匹配。 m_rel = re.search( - rf"^\s*{rel}: Mapped.*?(?=^\s*\w+: Mapped|\Z)", body, re.M | re.S + rf"^[ \t]*{rel}: Mapped.*?(?=^[ \t]*\w+:[^\n]*Mapped|\Z)", + body, re.M | re.S, ) assert m_rel, f"Match.{rel} 未找到" assert 'lazy="selectin"' in m_rel.group(0), ( f"Match.{rel} 未声明 lazy='selectin' —— 兜底缺失 (P0-2)" ) + def test_stats_relationship_is_lazy_select_by_design(self): + """Match.stats 刻意保持 lazy="select"(不是回归)。 + + 它是唯一需要显式 selectinload 才预加载的关系 —— 若哪天有人把它也 + 改成 selectin,上面的 test_context_builder_getters_eager_load 和 + bzzoiro stats 管线仍应工作,但本用例会提醒复核该设计决定。 + """ + src = _read("db/models.py") + body = re.search(r"class Match\(Base\):.*?(?=\nclass )", src, re.S).group(0) + m_rel = re.search( + r"^[ \t]*stats: Mapped.*?(?=^[ \t]*\w+:[^\n]*Mapped|\Z)", + body, re.M | re.S, + ) + assert m_rel, "Match.stats 未找到" + assert 'lazy="select"' in m_rel.group(0), ( + "Match.stats 预期为 lazy='select'(按需加载),实际声明已变 —— 请复核设计" + ) + class TestBzzoiroLineage: """P0-3: source_event_id 必须取配对的 raw,不能是循环残留变量。""" + # 消费循环的起始行匹配模式(见 bzzoiro.py 顶部 events 管线的内层循环) + _LOOP_PATTERN = "for nm, raw in normalized_matches" + + # source_event_id 的合法赋值形状(两种,都必须取配对的 raw): + # 1) 构造新比赛: `source_event_id=_to_int_or_none(raw.get("id")),` + # 2) 回填已有比赛: `eid = _to_int_or_none(raw.get("id"))` → + # `existing_match.source_event_id = eid` + # 非法形状(即 P0-3 回归): 直接用未配对的变量给 ORM 对象赋值。 + _ASSIGN_DIRECT = re.compile( + r"source_event_id\s*=\s*(?:[A-Za-z_][\w.]*\s*\(\s*)?raw(?:\.get\(|\s*\[)" + ) + # 赋值给未配对的局部变量: `source_event_id = <变量>` + _ASSIGN_VIA_VAR = re.compile( + r"source_event_id\s*=\s*([A-Za-z_]\w*)\s*$" + ) + + def _consume_loop_body(self, src: str) -> str: + """截取 `for nm, raw in normalized_matches` 循环体,不含循环之后的下游代码。 + + 原实现是 `seg = "\\n".join(lines[start:])`,一直取到文件末尾,于是把 + 无关的下游 stats 管线(bzzoiro.py 的 `_backfill_stats`)也扫了进来 —— + 那里合法地在 ORM 对象上访问 `m.source_event_id`,导致误报 P0-3。 + 这里按缩进边界正确收口:循环体内每行要么是空行/注释,要么缩进严格 + 大于 `for` 行。 + """ + lines = src.splitlines() + start = next( + (i for i, ln in enumerate(lines) if self._LOOP_PATTERN in ln), None + ) + assert start is not None, f"未找到消费循环: {self._LOOP_PATTERN}" + + for_indent = len(lines[start]) - len(lines[start].lstrip()) + kept: list[str] = [lines[start]] + for ln in lines[start + 1:]: + stripped = ln.strip() + # 顺序要保持: 空行与注释行缩进为 0,不能拿它们做边界判断 + if not stripped or stripped.startswith("#"): + continue + indent = len(ln) - len(ln.lstrip()) + if indent <= for_indent: + break # 循环结束,后续属下游代码 + kept.append(ln) + # 右侧剥离注释:避免 `# ... raw.get(...)` 这类注释误命中赋值正则 + return "\n".join(ln.split("#", 1)[0] for ln in kept) + + def _bad_assignments(self, seg: str) -> list[str]: + """返回循环体内未取配对 raw 的 source_event_id 赋值行。""" + # 先收集"来自配对 raw"的局部变量: `eid = _to_int_or_none(raw.get("id"))` + # (不受行序影响,所以必须先建好,再判定中转赋值) + raw_vars: set[str] = set() + for ln in seg.splitlines(): + m = re.match( + r"\s*([A-Za-z_]\w*)\s*=\s*.*raw(?:\.get\(|\s*\[)", ln + ) + if m: + raw_vars.add(m.group(1)) + + bad: list[str] = [] + for ln in seg.splitlines(): + stripped = ln.strip() + if "source_event_id" not in stripped: + continue + # 读取判断/比较(`if x.source_event_id is None:`)不算赋值 + if re.search(r"source_event_id\s*(?:is|==|!=)", stripped): + continue + if re.search(r"source_event_id\s*\.\s*\w+\s*\(", stripped): + continue # 方法调用,不是赋值 + if self._ASSIGN_DIRECT.search(stripped): + continue # 直接取配对 raw + m_var = self._ASSIGN_VIA_VAR.search(stripped) + if m_var and m_var.group(1) in raw_vars: + continue # 经由已确认来自 raw 的局部变量中转 + bad.append(stripped) + return bad + def test_normalized_matches_carries_raw(self): src = _read("data/bzzoiro.py") # 规范化结果必须与原始 event 成对保存 @@ -81,19 +183,47 @@ class TestBzzoiroLineage: ) def test_no_orphan_raw_use(self): - """source_event_id 所在行必须在解包循环内(用缩进 + 上下文粗判)。""" + """循环体内每个 source_event_id 赋值都必须取自配对的 raw(P0-3)。 + + 用正则而不是 `raw.get(` 子串匹配:合法写法含「回填已有比赛」那条 + (`eid = raw.get("id")` 之后 `existing_match.source_event_id = eid`), + 它不是 `raw.get(` 同一行,但同样正确。非法写法(回归)是直接 + `existing_match.source_event_id = orphan_var`。 + """ src = _read("data/bzzoiro.py") - lines = src.splitlines() - # 找到 "for nm, raw in normalized_matches" 所在行号 - start = next( - (i for i, ln in enumerate(lines) if "for nm, raw in normalized_matches" in ln), - None, + seg = self._consume_loop_body(src) + bad = self._bad_assignments(seg) + assert len(bad) == 0, ( + f"source_event_id 未使用配对的 raw (P0-3),问题行: {bad}" ) - assert start is not None - # 该循环之后、下一个同/更低缩进的顶层语句之前的范围 - seg = "\n".join(lines[start:]) - uses = [ln for ln in seg.splitlines() if "source_event_id" in ln] - assert uses, "未找到 source_event_id 赋值" - assert all("raw.get(" in ln for ln in uses), ( - "source_event_id 未使用配对的 raw (P0-3)" + + def test_loop_body_scope_excludes_downstream_stats_pipeline(self): + """作用域守卫: 截取段不能扫到循环之后的下游 stats 管线。 + + 下游 `_backfill_stats` 里合法地在 ORM 对象上访问 `m.source_event_id` + (与配对 raw 无关)。若 seg 越界,test_no_orphan_raw_use 会误报。 + """ + src = _read("data/bzzoiro.py") + seg = self._consume_loop_body(src) + assert "m.source_event_id" not in seg, ( + "循环体截取越界,扫到了下游 stats 管线 —— 会误报 P0-3" ) + # 但配对使用必须仍在作用域内 + assert "raw.get(" in seg, "循环体内应保留 `raw.get(...)` 的配对用法" + + def test_guard_detects_orphan_variable_regression(self): + """守卫有效性: 若 source_event_id 改成取循环外残留变量,必须被判失败。 + + 回归保护的"元测试"——确保上面的正则在真实缺陷面前确实会红, + 而不是恒真的空断言。 + """ + orphan = """ + for nm, raw in normalized_matches: + m = Match( + league_id=1, + source_event_id=_to_int_or_none(orphan.get("id")), + ) + """ + seg = self._consume_loop_body(orphan) + bad = self._bad_assignments(seg) + assert bad, "守卫失效: 未配对的 orphan 变量未被识别为 P0-3 回归"