英文漏译改在渲染侧查表，测试不再放行 PyJHora 子进程。

冻结的 jaimini.py 恢复成 staging 原样。缺译单元格写成固定标记，不再留空。

Co-Authored-By: Grok (agent loop)
Co-authored-by: Cursor <cursoragent@cursor.com>

tests/report_english_fictional.py

@@ -83,8 +83,27 @@ def fictional_case(index: int) -> dict:
     return chosen
 
 
+_JHORA_BLOCKED = False
+_GUARD = {
+    "pyjhora_subprocess_runs": 0,
+    "subprocess_runs": 0,
+    "tajaka": None,
+    "annual": None,
+    "installed": False,
+}
+_PYJHORA_COMMAND_MARKERS = (
+    "_isolated_tajaka_worker",
+    "_isolated_annual_worker",
+    "from jhora",
+    "import jhora",
+)
+
+
 def block_jhora() -> None:
     """Fail closed if this process imports PyJHora. Call once per process."""
+    global _JHORA_BLOCKED
+    if _JHORA_BLOCKED:
+        return
     real_import = builtins.__import__
 
     def guarded(name, globals=None, locals=None, fromlist=(), level=0):
@@ -104,6 +123,116 @@ def block_jhora() -> None:
     for key in list(sys.modules):
         if key == "jhora" or key.startswith("jhora."):
             del sys.modules[key]
+    _JHORA_BLOCKED = True
+
+
+def _command_blob(args, kwargs) -> str:
+    command = args[0] if args else kwargs.get("args", ())
+    if isinstance(command, str):
+        return command
+    if isinstance(command, (list, tuple)):
+        return " ".join(str(part) for part in command)
+    return str(command)
+
+
+def _install_replay_stubs() -> None:
+    """Return the no-PyJHora blocked shape without starting the isolated children.
+
+    Production still calls the real ``_run_isolated_*`` functions. Only this
+    test process replaces them, including workers started by the process pool.
+    """
+    import subprocess
+
+    import scripts.annual_pyjhora_replay as replay
+    try:
+        import annual_pyjhora_replay as replay_bare
+    except ImportError:
+        replay_bare = replay
+
+    if not getattr(subprocess.run, "_english_sweep_counter", False):
+        real_run = subprocess.run
+
+        def counting_run(*args, **kwargs):
+            blob = _command_blob(args, kwargs)
+            _GUARD["subprocess_runs"] += 1
+            if any(marker in blob for marker in _PYJHORA_COMMAND_MARKERS):
+                _GUARD["pyjhora_subprocess_runs"] += 1
+            return real_run(*args, **kwargs)
+
+        counting_run._english_sweep_counter = True
+        subprocess.run = counting_run
+
+    if _GUARD["installed"]:
+        return
+
+    def blocked_tajaka(profile, birth_julian_day, place, age):
+        settings = replay._replay_settings(profile)
+        request = {
+            "birth_julian_day": birth_julian_day,
+            "place": replay._place_payload(place),
+            "age": age,
+            "settings": settings,
+        }
+        result = replay._no_jhora_tajaka_replay(settings)
+        result["request_hash"] = replay._request_hash(request)
+        _GUARD["tajaka"] = result
+        return result
+
+    def blocked_annual(profile, birth_julian_day, place, age):
+        settings = replay._replay_settings(profile)
+        request = {
+            "birth_julian_day": birth_julian_day,
+            "place": replay._place_payload(place),
+            "age": age,
+            "settings": settings,
+        }
+        result = replay._no_jhora_annual_replay(settings)
+        result["request_hash"] = replay._request_hash(request)
+        _GUARD["annual"] = result
+        return result
+
+    for module in {replay, replay_bare}:
+        module._run_isolated_tajaka_year_lord = blocked_tajaka
+        module._run_isolated_annual_replay = blocked_annual
+    _GUARD["installed"] = True
+
+
+def _reset_guard_counts() -> None:
+    _GUARD["pyjhora_subprocess_runs"] = 0
+    _GUARD["subprocess_runs"] = 0
+    _GUARD["tajaka"] = None
+    _GUARD["annual"] = None
+
+
+def _guard_snapshot() -> dict:
+    from scripts.annual_pyjhora_replay import _ANNUAL_SURFACE_KEYS
+
+    tajaka = _GUARD["tajaka"]
+    annual = _GUARD["annual"]
+    annual_reason = "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError"
+    annual_blocked = (
+        isinstance(annual, dict)
+        and isinstance(annual.get("request_hash"), str)
+        and bool(annual.get("request_hash"))
+        and all(
+            isinstance(annual.get(key), dict)
+            and annual[key].get("status") == "blocked"
+            and annual[key].get("reason") == annual_reason
+            for key in _ANNUAL_SURFACE_KEYS
+        )
+    )
+    tajaka_blocked = (
+        isinstance(tajaka, dict)
+        and tajaka.get("status") == "blocked"
+        and tajaka.get("reason") == "pyjhora_ayanamsa_configuration_failed:ModuleNotFoundError"
+        and isinstance(tajaka.get("request_hash"), str)
+        and bool(tajaka.get("request_hash"))
+    )
+    return {
+        "pyjhora_subprocess_runs": _GUARD["pyjhora_subprocess_runs"],
+        "tajaka_blocked": tajaka_blocked,
+        "annual_blocked": annual_blocked,
+    }
 
 
 def _prepare_process() -> None:
@@ -112,6 +241,12 @@ def _prepare_process() -> None:
         if entry not in sys.path:
             sys.path.insert(0, entry)
     block_jhora()
+    _install_replay_stubs()
+
+
+def init_english_worker() -> None:
+    """Process-pool startup. ``render_english`` installs the same guards again."""
+    _prepare_process()
 
 
 def _residue(text: str) -> dict:
@@ -120,16 +255,24 @@ def _residue(text: str) -> dict:
     return {"han": len(found), "han_chars": "".join(sorted(set(found)))[:40], "glue": glue}
 
 
-def render_english(index: int) -> dict:
-    """Build one fictional packet and run ``_english_edition``. No birth data in the result."""
+def _load_packet(index: int) -> dict:
     _prepare_process()
+    _reset_guard_counts()
     from scripts.jyotish_engine import build_professional_report_reference_packet, cmd_full_reading
-    from scripts.professional_report_reference import _english_edition, _export_args
+    from scripts.professional_report_reference import _export_args
 
     birth = fictional_case(index)
     reading = cmd_full_reading(type("Args", (), dict(birth))())
     packet = dict(build_professional_report_reference_packet(reading, _export_args(dict(birth)), []))
     packet["report_version"] = "pl9_personal_long_report.v3"
+    return packet
+
+
+def render_english(index: int) -> dict:
+    """Build one fictional packet and run ``_english_edition``. No birth data in the result."""
+    from scripts.professional_report_reference import _english_edition
+
+    packet = _load_packet(index)
     edition = _english_edition(packet, True)
     markdown = edition.get("markdown_en") or ""
     flags = _flags_from_packet(packet)
@@ -141,12 +284,15 @@ def render_english(index: int) -> dict:
         path = Path(dump_dir)
         path.mkdir(parents=True, exist_ok=True)
         (path / f"chart-{index:03d}.md").write_text(markdown, encoding="utf-8")
+    rendered = f"{markdown}\n{applicability}"
     return {
         "i": index,
         "status": "ok" if markdown and not edition.get("english_unavailable") else edition.get("english_unavailable") or "empty",
         "han": body["han"] + extra["han"],
         "han_chars": (body["han_chars"] + extra["han_chars"])[:40],
         "glue": sorted(set(body["glue"] + extra["glue"])),
+        "missing": rendered.count("[translation missing:"),
+        **_guard_snapshot(),
         **flags,
     }
 
@@ -173,33 +319,27 @@ def _flags_from_packet(packet: dict) -> dict:
     }
 
 
+def row_problem(row: dict) -> str:
+    """Empty when this chart's English edition matches the no-PyJHora production report."""
+    clean = (
+        row["status"] == "ok"
+        and not row["han"]
+        and not row["glue"]
+        and not row["missing"]
+        and row["pyjhora_subprocess_runs"] == 0
+        and row["tajaka_blocked"]
+        and row["annual_blocked"]
+    )
+    if clean:
+        return ""
+    return (
+        f"#{row['i']} status={row['status']} han={row['han']} chars={row['han_chars']} "
+        f"glue={row['glue']} missing={row['missing']} "
+        f"pyjhora_subprocess_runs={row['pyjhora_subprocess_runs']} "
+        f"tajaka_blocked={row['tajaka_blocked']} annual_blocked={row['annual_blocked']}"
+    )
+
+
 def chart_flags(index: int) -> dict:
     """Class flags used to prove the smoke set covers L1–L6. No birth data."""
-    _prepare_process()
-    from scripts.jyotish_engine import build_professional_report_reference_packet, cmd_full_reading
-    from scripts.professional_report_reference import _export_args
-
-    birth = fictional_case(index)
-    reading = cmd_full_reading(type("Args", (), dict(birth))())
-    packet = dict(build_professional_report_reference_packet(reading, _export_args(dict(birth)), []))
-    rath = (((packet.get("worksheets") or {}).get("timing_and_predictive_systems") or {}).get("narayana_dasha") or {}).get("rath") or {}
-    rule = rath.get("current_md_antardasha_rule") or {}
-    direction = rule.get("direction") if isinstance(rule, dict) else {}
-    advanced = (packet.get("worksheets") or {}).get("advanced_systems") or {}
-    doshas = advanced.get("yogas_doshas") or {}
-    kaal = doshas.get("kaal_sarp_dosha") or {}
-    birth_sheet = (packet.get("worksheets") or {}).get("birth_and_parameters") or {}
-    pack = birth_sheet.get("panchanga") or {}
-    panchanga = pack.get("raw") if isinstance(pack, dict) and isinstance(pack.get("raw"), dict) else pack
-    yoga = advanced.get("yoga") if isinstance(advanced.get("yoga"), dict) else {}
-    names = [str(row.get("name") or "") for row in (yoga.get("detected_yogas") or []) if isinstance(row, dict)]
-    strengths = [str(row.get("strength") or "") for row in (yoga.get("detected_yogas") or []) if isinstance(row, dict)]
-    return {
-        "i": index,
-        "level": rath.get("strength_level_decided"),
-        "direction": rath.get("direction"),
-        "ad_rule": direction.get("rule") if isinstance(direction, dict) else None,
-        "overall": panchanga.get("overall_quality") if isinstance(panchanga, dict) else None,
-        "kaal": kaal.get("type") if isinstance(kaal, dict) else None,
-        "punarphoo": "观察" in strengths or any("Punarphoo" in name for name in names),
-    }
+    return {"i": index, **_flags_from_packet(_load_packet(index))}

tests/test_narayana_legacy_label.py

@@ -94,6 +94,9 @@ def test_reference_edition_shows_both_and_english_terms_cover_the_rath_lines() -
     from scripts.narayana_legacy_label import rath_report_lines_en
 
     packet = _packet(CASES["day"])
+    # 原值：reference 版同时有旧算法句和 Rath 表
+    # 新值：full_data 版同时有这两段
+    # 原因：用户看到的是完整数据版；Rath 并列表必须跟着走（BUG-1218）
     # 原值：Rath 中文行经 PHRASES_EN 可译干净
     # 新值：完整数据版英文渲染的 Rath 段无汉字且用英文函数
     # 原因：完整数据版不走 PHRASES_EN，旧断言测的不是生产路径

tests/test_report_english_han_sources.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 
 import re
 
+from pathlib import Path
+
 from scripts.muhurta import calc_panchanga
 from scripts.narayana_legacy_label import (
     DIRECTION_EN,
@@ -146,3 +148,50 @@ def test_punarphoo_strength_column_uses_english() -> None:
     assert "| Punarphoo Yoga | relationship_observation | observational | Moon conjunct Saturn |" in text
     assert "观察" not in text
     assert STRENGTH_EN["观察"] == "Watch"
+    assert "[translation missing:" not in text
+
+
+def test_missing_english_twin_is_marked_instead_of_blank() -> None:
+    packet = {
+        "report_language": "en",
+        "worksheets": {"advanced_systems": {"yoga": {"detected_yogas": [{
+            "name": "Example Yoga",
+            "category": "special",
+            "strength": "强",
+            "combination": "月亮与土星合相",
+            "effects": ["中文作用"],
+        }]}}},
+    }
+    text = render_pl9_parity_markdown(packet)
+    assert "[translation missing: combination]" in text
+    assert "[translation missing: effects]" in text
+    assert "月亮" not in text
+    assert "中文" not in text
+    assert "| Strong |" in text or "| strong |" in text or "Strong" in text
+
+
+def test_jaimini_domain_is_looked_up_by_karaka_name() -> None:
+    source = (Path(__file__).resolve().parents[1] / "scripts" / "jaimini.py").read_text(encoding="utf-8")
+    assert "domain_en" not in source
+    assert "KARAKA_DOMAINS_EN" not in source
+    packet = {
+        "report_language": "en",
+        "worksheets": {
+            "strengths_and_scores": {"scope": "jaimini-domain-fixture"},
+            "advanced_systems": {"jaimini": {"chara_karaka_7": {"karaka_table": {
+                "Amatyakaraka": {
+                    "planet": "Saturn",
+                    "degree_in_sign": 1.25,
+                    "effective_degree_in_sign": 1.25,
+                    "domain": "事业方向、主要谋士、权力代理",
+                    "cn_name": "事业星AmK",
+                },
+            }}}},
+        },
+    }
+    text = render_pl9_parity_markdown(packet)
+    assert "career direction, main advisor, delegated authority" in text
+    assert "事业" not in text
+    assert "careerdirection" not in text
+    assert "careerplanetAmK" not in text
+    assert "[translation missing:" not in text

tests/test_report_english_sweep.py

@@ -10,18 +10,17 @@ from __future__ import annotations
 import os
 from concurrent.futures import ProcessPoolExecutor, as_completed
 
-from tests.report_english_fictional import CHART_COUNT, render_english
+from tests.report_english_fictional import CHART_COUNT, init_english_worker, render_english, row_problem
 
 
 def test_two_hundred_fictional_charts_have_clean_english() -> None:
     workers = min(4, os.cpu_count() or 1, CHART_COUNT)
     failures = []
-    with ProcessPoolExecutor(max_workers=workers) as pool:
+    with ProcessPoolExecutor(max_workers=workers, initializer=init_english_worker) as pool:
         futures = [pool.submit(render_english, index) for index in range(CHART_COUNT)]
         for future in as_completed(futures):
             row = future.result()
-            if row["status"] != "ok" or row["han"] or row["glue"]:
-                failures.append(
-                    f"#{row['i']} status={row['status']} han={row['han']} chars={row['han_chars']} glue={row['glue']}"
-                )
+            problem = row_problem(row)
+            if problem:
+                failures.append(problem)
     assert failures == []

tests/test_report_english_sweep_smoke.py

@@ -1,4 +1,4 @@
-"""Smoke slice of the English sweep, on the quick gate (issue #89 / BUG-1410).
+"""Smoke slice of the English sweep, on the quick gate (issue #89 / BUG-1413).
 
 Indices are from tests.report_english_fictional.fictional_case. Together they
 hit Rath level 5, an inauspicious Panchanga, Ketu in the start sign, a Table 14
@@ -9,7 +9,7 @@ from __future__ import annotations
 
 from concurrent.futures import ProcessPoolExecutor, as_completed
 
-from tests.report_english_fictional import render_english
+from tests.report_english_fictional import init_english_worker, render_english, row_problem
 
 # Filled once the flag scan names a chart for every class. The test checks the
 # flags, so a stale index fails instead of silently dropping a class.
@@ -23,15 +23,14 @@ SMOKE_INDEXES = (1, 4, 5, 6, 9, 10, 11, 12, 14, 15, 32, 34)
 def test_smoke_charts_cover_every_leak_class_and_render_clean_english() -> None:
     rows = []
     failures = []
-    with ProcessPoolExecutor(max_workers=min(4, len(SMOKE_INDEXES))) as pool:
+    with ProcessPoolExecutor(max_workers=min(4, len(SMOKE_INDEXES)), initializer=init_english_worker) as pool:
         futures = [pool.submit(render_english, index) for index in SMOKE_INDEXES]
         for future in as_completed(futures):
             row = future.result()
             rows.append(row)
-            if row["status"] != "ok" or row["han"] or row["glue"]:
-                failures.append(
-                    f"#{row['i']} status={row['status']} han={row['han']} chars={row['han_chars']} glue={row['glue']}"
-                )
+            problem = row_problem(row)
+            if problem:
+                failures.append(problem)
     levels = {row["level"] for row in rows}
     directions = {row["direction"] for row in rows}
     ad_rules = {row["ad_rule"] for row in rows}