perf(consultation): 健康探测加 TTL、yoga 表达式只编译一次

BUG-734: probe_official_rest_health 不是 ping —— 它把一份虚构 smoke 排盘
POST 给官方 HoroscopePredictions（实测单次 1,034 ms），还经 call() 吃掉对方
每分钟 5 个限流令牌之一，而结论零 TTL。连续 6 轮前台请求下，第 6 次探测会被
本地限流器挡下、反过来报出假的 official_blocked，真正的业务调用还要和它抢令牌。

加进程级 TTL 缓存（成功 60 s / 失败 10 s，配置变即失效，缓存键只记 key 配没配
的布尔）。force_refresh 默认 True：诊断端点调的是裸 gateway_status() 且该文件
本轮不得改，默认实时才不会让运维看到旧结论；run_gateway_packet 显式走 False。
探测在锁外执行。6 轮实测：探测 6→1、握手 5→1、限流令牌 6→1。

连接复用（任务书 6.2）按让步顺序砍掉：实测 urllib.request 没有连接池，模块级
共享 opener 在 6 次请求下仍开 6 条 TCP 连接，按任务书写法只能骗过「同一个
opener 实例」的断言而省不掉任何握手。另立单。

BUG-735: yoga_engine._eval_custom 对静态规则表的表达式每请求重新编译。实测每次
检测 386 次源码编译（182 次 eval 字符串 + 204 次 ast.parse/compile）。拆出
_build_custom_exec_globals（每次新建，绑定当前盘 ctx）与 lru_cache 的
_compiled_eval_code / _compiled_exec_code，只缓存 code object，绝不缓存求值
结果。求值语义逐条对齐改前。第二次检测起编译 0 次，单次 27.4 ms → 3.3 ms。

等价用同进程差分（BUG-733 的做法）：192 条去重表达式 × 3 张公开虚构示例盘，
与逐字复刻的改前实现同值，另有反向验证证明差分非恒真。

未改 scripts/jyotish_api_server.py。既有断言一条未改。新增 23 条测试。
快速门唯一红的 test_chat_page_uses_authenticated_cloud_persistence 经干净
origin/staging 检出复跑确认为基线红，记入 BLOCKED.md BLK-002，不顺手修。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JUei7K13cYxLHE3Axe4A45

tests/test_vedastro_health_probe_cache.py

@@ -0,0 +1,271 @@
+#!/usr/bin/env python3
+"""BUG-734: 官方 REST 健康探测的 TTL 缓存。
+
+探测不是 ping——它 POST 一份虚构 smoke 排盘到官方 HoroscopePredictions，
+并消耗对方每分钟 5 个限流令牌之一。改前每一轮前台请求都做一次。
+
+两条边界同时成立：
+- 前台路径（`run_gateway_packet`）读缓存；
+- 诊断端点仍然实时，运维不得看到一个 60 秒前的假象。
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from scripts import vedastro_gateway
+
+
+@pytest.fixture(autouse=True)
+def _clean_cache(monkeypatch):
+    """每个用例都从空缓存、确定的配置起步。"""
+    monkeypatch.delenv("JYOTISH_SKIP_LOCAL_ENV", raising=False)
+    monkeypatch.delenv("VEDASTRO_API_KEY", raising=False)
+    monkeypatch.delenv("VEDASTRO_SELF_HOST_ENDPOINT", raising=False)
+    monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
+    monkeypatch.delenv("VEDASTRO_GATEWAY_MODE", raising=False)
+    monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1")
+    vedastro_gateway.reset_rest_health_cache()
+    yield
+    vedastro_gateway.reset_rest_health_cache()
+
+
+class _Clock:
+    """可推进的单调时钟。"""
+
+    def __init__(self) -> None:
+        self.now = 1000.0
+
+    def __call__(self) -> float:
+        return self.now
+
+    def advance(self, seconds: float) -> None:
+        self.now += seconds
+
+
+@pytest.fixture
+def clock(monkeypatch) -> _Clock:
+    c = _Clock()
+    monkeypatch.setattr(vedastro_gateway, "_monotonic", c)
+    return c
+
+
+class _ProbeCounter:
+    """替掉真正的外网探测，统计它被调了几次。"""
+
+    def __init__(self, status: str = "official_verified") -> None:
+        self.calls = 0
+        self.status = status
+
+    def __call__(self, *args, **kwargs) -> dict:
+        self.calls += 1
+        reason = "status_pass" if self.status == "official_verified" else "URLError"
+        return {
+            "status": self.status,
+            "reason": reason,
+            "endpoint": vedastro_gateway.OFFICIAL_CALCULATE_ENDPOINT,
+            "transport": "rest",
+        }
+
+
+@pytest.fixture
+def probe(monkeypatch) -> _ProbeCounter:
+    counter = _ProbeCounter()
+    from scripts import vedastro_rest_bridge
+
+    monkeypatch.setattr(vedastro_rest_bridge, "probe_calculate_health", counter)
+    return counter
+
+
+# ---------------------------------------------------------------------------
+# TTL 本身
+# ---------------------------------------------------------------------------
+def test_three_cached_probes_hit_the_network_once(probe, clock) -> None:
+    results = [vedastro_gateway.probe_official_rest_health(force_refresh=False) for _ in range(3)]
+
+    assert probe.calls == 1, f"连调 3 次只应探测 1 次，实际 {probe.calls} 次"
+    assert all(r["status"] == "official_verified" for r in results)
+    assert results[0] == results[1] == results[2]
+
+
+def test_success_result_is_re_probed_after_the_success_ttl(probe, clock) -> None:
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 1
+
+    clock.advance(vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS - 1)
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 1, "TTL 之内不得重新探测"
+
+    clock.advance(2)
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 2, "推过 TTL 后必须重新探测"
+
+
+def test_failure_is_not_cached_for_the_success_ttl(probe, clock) -> None:
+    """红线 §5.4：对方恢复了我们不能还要再瞎等一分钟。"""
+    probe.status = "official_blocked"
+    assert vedastro_gateway.probe_official_rest_health(force_refresh=False)["status"] == "official_blocked"
+    assert probe.calls == 1
+
+    # 失败 TTL 之内：不重探。
+    clock.advance(vedastro_gateway._REST_HEALTH_FAILURE_TTL_SECONDS - 1)
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 1
+
+    # 过了失败 TTL、但还远在成功 TTL 之内：必须重探。
+    clock.advance(2)
+    assert clock.now < 1000.0 + vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS
+    probe.status = "official_verified"
+    recovered = vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 2
+    assert recovered["status"] == "official_verified", "服务恢复后必须能在失败 TTL 内被看见"
+
+
+def test_failure_ttl_is_clearly_shorter_than_success_ttl() -> None:
+    assert vedastro_gateway._REST_HEALTH_FAILURE_TTL_SECONDS < vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS
+    assert vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS <= 60.0
+
+
+# ---------------------------------------------------------------------------
+# 配置一变立刻失效
+# ---------------------------------------------------------------------------
+def test_configuration_change_invalidates_the_cache(probe, clock, monkeypatch) -> None:
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 1
+
+    monkeypatch.setenv("VEDASTRO_API_KEY", "fictional-test-key")
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 2, "配了 API key 之后必须重新探测"
+
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 2, "同一份新配置应当重新命中缓存"
+
+    monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://example.invalid/api")
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 3, "换 endpoint 必须重新探测"
+
+
+def test_cache_key_never_carries_the_secret_value(monkeypatch) -> None:
+    """AGENTS §8：缓存键只记「配没配」，不记 key 本身。"""
+    monkeypatch.setenv("VEDASTRO_API_KEY", "fictional-test-key")
+    key_with = vedastro_gateway._rest_health_cache_key()
+    monkeypatch.delenv("VEDASTRO_API_KEY")
+    key_without = vedastro_gateway._rest_health_cache_key()
+
+    assert "fictional-test-key" not in repr(key_with), "缓存键里出现了 key 的值"
+    assert key_with != key_without, "配没配 key 必须能区分开，否则缓存会跨配置串"
+
+
+def test_network_disabled_is_not_probed_at_all(monkeypatch, clock) -> None:
+    from scripts import vedastro_rest_bridge
+
+    def explode(*args, **kwargs):  # pragma: no cover - must never run
+        raise AssertionError("network disabled 时不得触碰外网")
+
+    monkeypatch.setattr(vedastro_rest_bridge, "probe_calculate_health", explode)
+    monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "0")
+    result = vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert result["status"] == "official_blocked"
+    assert result["reason"] == "network_disabled"
+
+
+# ---------------------------------------------------------------------------
+# 诊断端点必须实时
+# ---------------------------------------------------------------------------
+def test_force_refresh_always_probes(probe, clock) -> None:
+    for _ in range(3):
+        vedastro_gateway.probe_official_rest_health(force_refresh=True)
+    assert probe.calls == 3, "force_refresh 必须每次都打真探测"
+
+
+def test_force_refresh_defaults_to_true_so_the_diagnostic_endpoint_stays_live(probe, clock) -> None:
+    """`_compute_vedastro_gateway_status` 调的是裸 `gateway_status()`（该文件本单不得改）。
+
+    因此默认值必须是实时探测，否则运维会看到一个 60 秒前的假象。
+    """
+    for fn in (vedastro_gateway.probe_official_rest_health, vedastro_gateway.gateway_status):
+        assert inspect.signature(fn).parameters["force_refresh"].default is True
+
+    vedastro_gateway.gateway_status()
+    vedastro_gateway.gateway_status()
+    assert probe.calls == 2, "诊断端点默认路径不得走缓存"
+
+
+def test_forced_probe_sees_the_current_state_even_while_a_stale_entry_exists(probe, clock) -> None:
+    """前台缓存里躺着一条旧结论时，诊断端点仍必须看到此刻的真实状态。"""
+    probe.status = "official_blocked"
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert probe.calls == 1
+
+    probe.status = "official_verified"
+    forced = vedastro_gateway.probe_official_rest_health(force_refresh=True)
+    assert forced["status"] == "official_verified"
+    assert probe.calls == 2
+
+
+# ---------------------------------------------------------------------------
+# 前台路径
+# ---------------------------------------------------------------------------
+def test_gateway_status_passes_force_refresh_through(probe, clock) -> None:
+    vedastro_gateway.gateway_status(force_refresh=False)
+    vedastro_gateway.gateway_status(force_refresh=False)
+    vedastro_gateway.gateway_status(force_refresh=False)
+    assert probe.calls == 1
+
+
+def test_run_gateway_packet_uses_the_cached_probe(monkeypatch, probe, clock) -> None:
+    """每一轮前台请求不得各拿一个对方的限流令牌去做健康探测。"""
+    seen: list[bool] = []
+    real_status = vedastro_gateway.gateway_status
+
+    def spy(*, force_refresh: bool = True):
+        seen.append(force_refresh)
+        return real_status(force_refresh=force_refresh)
+
+    monkeypatch.setattr(vedastro_gateway, "gateway_status", spy)
+    monkeypatch.setattr(
+        vedastro_gateway,
+        "_entrypoint_args",
+        lambda *a, **k: __import__("types").SimpleNamespace(),
+    )
+    import sys
+    import types as _types
+
+    stub = _types.ModuleType("scripts.vedastro_user_entrypoint")
+    stub.build_report = lambda args: {"input": {}, "runtime_mode": {}}
+    monkeypatch.setitem(sys.modules, "scripts.vedastro_user_entrypoint", stub)
+
+    for _ in range(4):
+        vedastro_gateway.run_gateway_packet({"year": 1955}, question="x")
+
+    assert seen == [False, False, False, False], "前台路径必须显式走缓存"
+    assert probe.calls == 1, f"4 轮前台请求只应探测 1 次，实际 {probe.calls} 次"
+
+
+def test_cached_result_cannot_be_mutated_by_a_caller(probe, clock) -> None:
+    first = vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    first["status"] = "tampered"
+    second = vedastro_gateway.probe_official_rest_health(force_refresh=False)
+    assert second["status"] == "official_verified"
+    assert probe.calls == 1
+
+
+def test_probe_does_not_hold_the_lock_while_on_the_network(monkeypatch, clock) -> None:
+    """不得在等外网时把进程级锁按住（BUG-718 的教训）。"""
+    observed: list[bool] = []
+
+    def probing(*args, **kwargs):
+        acquired = vedastro_gateway._REST_HEALTH_LOCK.acquire(blocking=False)
+        observed.append(acquired)
+        if acquired:
+            vedastro_gateway._REST_HEALTH_LOCK.release()
+        return {"status": "official_verified", "reason": "status_pass", "transport": "rest"}
+
+    from scripts import vedastro_rest_bridge
+
+    monkeypatch.setattr(vedastro_rest_bridge, "probe_calculate_health", probing)
+    vedastro_gateway.probe_official_rest_health(force_refresh=False)
+
+    assert observed == [True], "探测期间锁必须是可获取的"

tests/test_yoga_expr_compile_cache.py

@@ -0,0 +1,328 @@
+#!/usr/bin/env python3
+"""BUG-735: custom yoga 表达式的编译缓存。
+
+三件事各自独立证明：
+
+1. **等价**：对规则表里全部去重表达式，缓存版本与改前实现逐条同值。
+   证明方式是同进程差分（BUG-733 立下的做法）：参照实现在本文件里逐字复刻改前代码，
+   与生产实现在同一进程、同一批 exec_globals 上对跑，不写跨机 golden。
+2. **只缓存 code object**：同一条表达式配不同的盘必须得到不同答案。
+   这条是红线——缓存求值结果会让所有人拿到同一张盘的 yoga 判定。
+3. **编译次数**：第二次检测的源码编译次数为 0，不随请求数线性增长。
+"""
+
+from __future__ import annotations
+
+import ast
+import builtins
+import json
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+RULES_PATH = ROOT / "references" / "yoga_rules.json"
+
+import yoga_engine  # noqa: E402
+from yoga_engine import YogaContext, YogaEngine  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# 公开虚构示例盘。不是任何真实用户的出生资料。
+# ---------------------------------------------------------------------------
+SAMPLE_CHARTS = [
+    (
+        "Aries",
+        {
+            "Sun": {"house": 10, "sign": "Capricorn", "degree": 16.4},
+            "Moon": {"house": 4, "sign": "Cancer", "degree": 2.1},
+            "Mars": {"house": 12, "sign": "Pisces", "degree": 23.8},
+            "Mercury": {"house": 9, "sign": "Sagittarius", "degree": 27.3},
+            "Jupiter": {"house": 4, "sign": "Cancer", "degree": 8.9},
+            "Venus": {"house": 9, "sign": "Sagittarius", "degree": 13.6},
+            "Saturn": {"house": 10, "sign": "Capricorn", "degree": 20.2},
+            "Rahu": {"house": 8, "sign": "Scorpio", "degree": 1.5},
+            "Ketu": {"house": 2, "sign": "Taurus", "degree": 1.5},
+        },
+    ),
+    (
+        "Leo",
+        {
+            "Sun": {"house": 5, "sign": "Gemini", "degree": 10.0},
+            "Moon": {"house": 5, "sign": "Gemini", "degree": 15.0},
+            "Mars": {"house": 8, "sign": "Pisces", "degree": 10.0},
+            "Mercury": {"house": 10, "sign": "Capricorn", "degree": 10.0},
+            "Jupiter": {"house": 11, "sign": "Sagittarius", "degree": 5.0},
+            "Venus": {"house": 2, "sign": "Libra", "degree": 10.0},
+            "Saturn": {"house": 7, "sign": "Aquarius", "degree": 10.0},
+            "Rahu": {"house": 6, "sign": "Capricorn", "degree": 10.0},
+            "Ketu": {"house": 12, "sign": "Cancer", "degree": 10.0},
+        },
+    ),
+    (
+        "Libra",
+        {
+            "Sun": {"house": 1, "sign": "Libra", "degree": 22.0},
+            "Moon": {"house": 7, "sign": "Aries", "degree": 4.0},
+            "Mars": {"house": 4, "sign": "Capricorn", "degree": 28.0},
+            "Mercury": {"house": 2, "sign": "Scorpio", "degree": 3.0},
+            "Jupiter": {"house": 10, "sign": "Cancer", "degree": 17.0},
+            "Venus": {"house": 12, "sign": "Virgo", "degree": 9.0},
+            "Saturn": {"house": 1, "sign": "Libra", "degree": 6.0},
+            "Rahu": {"house": 3, "sign": "Sagittarius", "degree": 19.0},
+            "Ketu": {"house": 9, "sign": "Gemini", "degree": 19.0},
+        },
+    ),
+]
+
+
+def _all_rule_expressions() -> list[str]:
+    """规则表里全部 `expr`（去重后按文本排序，保证用例顺序稳定）。"""
+    data = json.loads(RULES_PATH.read_text(encoding="utf-8"))
+    found: list[str] = []
+
+    def walk(node: object) -> None:
+        if isinstance(node, dict):
+            value = node.get("expr")
+            if isinstance(value, str) and value.strip():
+                found.append(value)
+            for child in node.values():
+                walk(child)
+        elif isinstance(node, list):
+            for child in node:
+                walk(child)
+
+    walk(data)
+    return sorted(set(found))
+
+
+RULE_EXPRESSIONS = _all_rule_expressions()
+
+
+# ---------------------------------------------------------------------------
+# 参照实现：BUG-735 改前的 `_eval_custom` 尾部，逐字复刻。
+# ---------------------------------------------------------------------------
+def _reference_capture_tail_expr(block):
+    if not block:
+        return block
+    last = block[-1]
+    if isinstance(last, ast.Expr):
+        block[-1] = ast.Assign(targets=[ast.Name(id="__result__", ctx=ast.Store())], value=last.value)
+    elif isinstance(last, ast.If):
+        last.body = _reference_capture_tail_expr(last.body)
+        last.orelse = _reference_capture_tail_expr(last.orelse)
+    elif isinstance(last, (ast.For, ast.While)):
+        last.body = _reference_capture_tail_expr(last.body)
+        last.orelse = _reference_capture_tail_expr(last.orelse)
+    elif isinstance(last, ast.Try):
+        last.body = _reference_capture_tail_expr(last.body)
+        last.orelse = _reference_capture_tail_expr(last.orelse)
+        last.finalbody = _reference_capture_tail_expr(last.finalbody)
+        for handler in last.handlers:
+            handler.body = _reference_capture_tail_expr(handler.body)
+    return block
+
+
+def _reference_run(expr, exec_globals):
+    """改前行为：每次重新编译；先 eval，SyntaxError 退到改写过的多语句 exec。"""
+    result = None
+    try:
+        result = eval(expr, exec_globals, exec_globals)
+    except SyntaxError:
+        try:
+            tree = ast.parse(expr.strip(), mode="exec")
+            tree.body = _reference_capture_tail_expr(tree.body)
+            ast.fix_missing_locations(tree)
+            exec(compile(tree, "<yoga_custom>", "exec"), exec_globals, exec_globals)
+            result = exec_globals.get("__result__")
+        except Exception:
+            pass
+    except Exception:
+        pass
+    return result
+
+
+@pytest.fixture(scope="module")
+def engine() -> YogaEngine:
+    return YogaEngine(str(RULES_PATH))
+
+
+def _globals_for(engine: YogaEngine, chart_index: int) -> dict:
+    ascendant, planets = SAMPLE_CHARTS[chart_index]
+    ctx = YogaContext(planets, ascendant)
+    return engine._build_custom_exec_globals(ctx, {}, {})
+
+
+# ---------------------------------------------------------------------------
+# 1. 等价：同进程差分
+# ---------------------------------------------------------------------------
+def test_cached_compile_matches_pre_change_behaviour_for_every_rule_expression(engine) -> None:
+    """全部 192 条去重表达式，缓存版与参照版在三张示例盘上逐条同值。"""
+    assert len(RULE_EXPRESSIONS) >= 190, "规则表表达式数量异常，等价证明的覆盖面不足"
+
+    mismatches = []
+    for chart_index in range(len(SAMPLE_CHARTS)):
+        for expr in RULE_EXPRESSIONS:
+            # 两侧各用一份全新命名空间：exec 会往里写 __result__，不能共用。
+            live = yoga_engine._run_custom_expr(expr, _globals_for(engine, chart_index))
+            reference = _reference_run(expr, _globals_for(engine, chart_index))
+            if live != reference or bool(live) != bool(reference):
+                mismatches.append((chart_index, expr[:80], reference, live))
+
+    assert not mismatches, f"{len(mismatches)} 条表达式与改前行为不同: {mismatches[:3]}"
+
+
+def test_reference_differential_can_actually_fail(engine) -> None:
+    """反向验证：差分不是恒真——喂一个被改坏的表达式两侧就该分叉。"""
+    exec_globals_a = _globals_for(engine, 0)
+    exec_globals_b = _globals_for(engine, 0)
+    assert yoga_engine._run_custom_expr("house_of('Sun')", exec_globals_a) == _reference_run(
+        "house_of('Sun')", exec_globals_b
+    )
+    assert yoga_engine._run_custom_expr("house_of('Sun')", _globals_for(engine, 0)) != _reference_run(
+        "house_of('Moon')", _globals_for(engine, 0)
+    )
+
+
+# ---------------------------------------------------------------------------
+# 2. 红线：只缓存 code object，绝不缓存求值结果
+# ---------------------------------------------------------------------------
+def test_cache_stores_code_objects_not_results() -> None:
+    expr = "house_of('Sun')"
+    yoga_engine._compiled_eval_code.cache_clear()
+    compiled = yoga_engine._compiled_eval_code(expr)
+    assert isinstance(compiled, types.CodeType), "eval 分支缓存里必须是 code object"
+
+    multi = "x = house_of('Sun')\nx"
+    yoga_engine._compiled_exec_code.cache_clear()
+    assert yoga_engine._compiled_eval_code(multi) is None, "多语句表达式在 eval 模式下必须仍然是 None"
+    assert isinstance(yoga_engine._compiled_exec_code(multi), types.CodeType)
+
+
+def test_same_expression_on_different_charts_never_reuses_a_result(engine) -> None:
+    """红线：同一条表达式配不同的盘必须给出不同答案（不得跨盘串结果）。"""
+    expr = "house_of('Sun')"
+    first = yoga_engine._run_custom_expr(expr, _globals_for(engine, 0))
+    second = yoga_engine._run_custom_expr(expr, _globals_for(engine, 1))
+    third = yoga_engine._run_custom_expr(expr, _globals_for(engine, 0))
+
+    assert first == 10 and second == 5, "示例盘的 Sun 宫位应当不同，否则这条断言没有鉴别力"
+    assert first != second, "求值结果被缓存了：同一表达式在不同盘上返回了同一个答案"
+    assert first == third, "同一张盘重复求值应当稳定"
+
+    multi = "h = house_of('Sun')\nh + 100"
+    assert yoga_engine._run_custom_expr(multi, _globals_for(engine, 0)) == 110
+    assert yoga_engine._run_custom_expr(multi, _globals_for(engine, 1)) == 105
+
+
+def test_run_custom_expr_is_not_memoized() -> None:
+    """求值入口本身不得带缓存装饰器。"""
+    assert not hasattr(yoga_engine._run_custom_expr, "cache_info")
+    assert hasattr(yoga_engine._compiled_eval_code, "cache_info")
+    assert hasattr(yoga_engine._compiled_exec_code, "cache_info")
+
+
+def test_compile_cache_is_bounded() -> None:
+    """编译缓存必须有界（红线 §5.3）。"""
+    for cached in (yoga_engine._compiled_eval_code, yoga_engine._compiled_exec_code):
+        maxsize = cached.cache_info().maxsize
+        assert maxsize is not None, "编译缓存不得无界"
+        assert maxsize >= 256, "maxsize 必须容纳规则表去重后的表达式数量"
+    assert yoga_engine._CUSTOM_EXPR_CACHE_MAXSIZE >= len(RULE_EXPRESSIONS)
+
+
+# ---------------------------------------------------------------------------
+# 3. 编译次数
+# ---------------------------------------------------------------------------
+class _CompileCounter:
+    """统计一次检测里发生了多少次「源码 → 字节码」编译。
+
+    `eval(str)` 的编译发生在 CPython 内部、不经过 builtins.compile，
+    所以两者都要数：`eval(str)` 调用数 + `builtins.compile` 调用数。
+    """
+
+    def __init__(self) -> None:
+        self.eval_str = 0
+        self.eval_code = 0
+        self.compile = 0
+        self._orig_eval = builtins.eval
+        self._orig_compile = builtins.compile
+
+    def __enter__(self) -> "_CompileCounter":
+        def counting_eval(source, *args, **kwargs):
+            if isinstance(source, str):
+                self.eval_str += 1
+            else:
+                self.eval_code += 1
+            return self._orig_eval(source, *args, **kwargs)
+
+        def counting_compile(source, *args, **kwargs):
+            self.compile += 1
+            return self._orig_compile(source, *args, **kwargs)
+
+        builtins.eval = counting_eval
+        builtins.compile = counting_compile
+        return self
+
+    def __exit__(self, *exc) -> None:
+        builtins.eval = self._orig_eval
+        builtins.compile = self._orig_compile
+
+    @property
+    def source_compiles(self) -> int:
+        return self.eval_str + self.compile
+
+
+def test_second_detection_compiles_nothing(engine) -> None:
+    ascendant, planets = SAMPLE_CHARTS[0]
+    yoga_engine._compiled_eval_code.cache_clear()
+    yoga_engine._compiled_exec_code.cache_clear()
+
+    with _CompileCounter() as first:
+        first_result = engine.detect(planets, ascendant)
+    with _CompileCounter() as second:
+        second_result = engine.detect(planets, ascendant)
+    with _CompileCounter() as third:
+        third_result = engine.detect(planets, ascendant)
+
+    assert first.source_compiles > 0, "冷缓存的第一次检测必须真的编译"
+    assert second.source_compiles == 0, f"第二次检测仍编译了 {second.source_compiles} 次"
+    assert third.source_compiles == 0, "编译次数不得随请求数增长"
+    assert second.eval_code > 0, "第二次检测应当在复用 code object"
+    assert second.eval_str == 0, "不得再有字符串 eval"
+    assert first_result == second_result == third_result
+
+
+def test_detection_output_is_unchanged_across_charts(engine) -> None:
+    """缓存不得改变任何一张盘的判定输出。"""
+    for ascendant, planets in SAMPLE_CHARTS:
+        yoga_engine._compiled_eval_code.cache_clear()
+        yoga_engine._compiled_exec_code.cache_clear()
+        cold = engine.detect(planets, ascendant)
+        warm = engine.detect(planets, ascendant)
+        assert cold == warm
+
+
+# ---------------------------------------------------------------------------
+# 4. 来源合同：expr 只来自仓库内的静态规则表
+# ---------------------------------------------------------------------------
+def test_rule_expressions_come_only_from_the_static_rules_table() -> None:
+    """红线 §5.3：若有用户输入能进入 `expr`，立刻停手。"""
+    assert RULES_PATH.is_file()
+    assert RULES_PATH.resolve().is_relative_to(ROOT.resolve())
+    for expr in RULE_EXPRESSIONS:
+        assert isinstance(expr, str)
+
+    source = (ROOT / "scripts" / "yoga_engine.py").read_text(encoding="utf-8")
+    # 表达式的唯一读取点。
+    assert source.count('cond.get("expr"') == 1
+
+    # 生产入口不得把 rules_path 交给调用方：默认路径是仓库内的规则表。
+    for entry in ("scripts/jyotish_engine.py", "scripts/jyotish_api_server.py"):
+        text = (ROOT / entry).read_text(encoding="utf-8")
+        for line in text.splitlines():
+            if "detect_yogas(" in line and "def " not in line:
+                assert "rules_path" not in line, f"{entry} 给 yoga 引擎传了自定义规则路径: {line.strip()}"