让免费层限流的报告跟真实限流一致，并给少一块结果的静默失败留下函数名和错误类型。

Co-Authored-By: Grok (agent loop)
merge origin/staging into pl9 render split (issue #67 )
2026-10-15 23:12:56 +08:00 · 2026-10-15 21:23:23 +08:00 · 2026-10-15 20:59:11 +08:00
45 changed files with 1756 additions and 541 deletions
@@ -9,13 +9,18 @@ VEDASTRO_SELF_HOST_ENDPOINT=https://your-domain.example.com/vedastro
 # VEDASTRO_API_KEY=
 # VEDASTRO_ENABLE_NETWORK=0

-# Cache / TTL / free-tier queue. These keep ordinary users from calling upstream directly.
+# Cache / TTL. 网关离线排队与免费层限流是两件事，见下面两组变量。
 # 失败结果另有短 TTL（VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS，默认 300 秒，且不超过成功 TTL）。
 # 这些缓存含出生资料，7 天为上限。
 VEDASTRO_CACHE_TTL_SECONDS=604800
 VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS=604800
+# 网关离线排队作业。这不是免费层限流开关。
 VEDASTRO_GATEWAY_QUEUE_ENABLED=1
-VEDASTRO_FREE_TIER_QUEUE=1
+# 免费层限流由 API key 与下面两项决定。没有 key、且两项都大于 0 时，官方公共端点会限流。
+# VEDASTRO_FREE_TIER_QUEUE 已废弃，设置它不会打开或关闭限流。
+# VEDASTRO_FREE_TIER_MAX_REQUESTS=5
+# VEDASTRO_FREE_TIER_WINDOW_SECONDS=60
+# VEDASTRO_FREE_TIER_WAIT_BUDGET_SECONDS=
 VEDASTRO_FAIL_OPEN_LOCAL=1

 # Runtime guardrails.
@@ -1,5 +1,11 @@
 # 印度占星 Skill 更新日志

+## 2026-10-15 — 免费层限流状态与真实行为对齐
+
+- 诊断和审计里的「免费层是否在限流」改为按真实限流计算。旧的三个排队开关名字还在，但不再控制限流，诊断里会标明它们已废弃。
+- 限流本身没变：有没有 key、配额是多少，仍按原来的规则。
+- Skill 版本不变（BUG-1340）。
+
 ## 2026-10-15 — 内测容量：问得太密会停一下，人多时先排一小会儿

 - 同一个人短时间问得太密、校正答得太密，或一小时里报告生成太多，会先停一下，这次不扣点。
@@ -36303,4 +36303,20 @@
 - 复发自：BUG-933（源码合同没在同一轮跟上）
 - 修复版本：分支 `codex/beta-capacity-20261014`。

+## BUG-1340 | 免费层限流状态报的和实际相反
+
+- 状态：resolved
+- 首次发现：2026-10-13
+- 最近更新：2026-10-15
+- 影响面：VedAstro 免费层的诊断和审计字段 `queue_enabled` / `free_tier_queue_enabled`。限流等待和降级本身没改。
+- 用户现象：运维看诊断时，没配旧开关会显示「没开限流」，其实官方公共端点仍在限流；配了旧开关、又把配额设成 0，会显示「开了」，其实没有限流。
+- 触发条件：命中官方公共端点。旧变量 `VEDASTRO_FREE_TIER_QUEUE`、`VEDASTRO_FREE_TIER_QUEUE_ENABLED`、`VEDASTRO_ENABLE_FREE_TIER_QUEUE` 只被写进报告，不参与限流。
+- 根因：限流在适配器里只看端点、有没有 API key、以及配额窗口是否大于 0。三个旧名字是更早留下的报告开关，从未接到这套判断上。BUG-720 的防复发只写了「前台不得无上限排队」，没有要求报告字段和真实限流一致。
+- 修复：报告改为读同一套判断。三个旧名字保留，只在诊断里列为已废弃，不能用来关掉限流。
+- 验证：`tests/test_vedastro_free_tier_state.py`。没配旧开关时期望与 `free_tier_queue_active` 一致；旧开关设为开且配额为 0 时期望为关；有 key 或自建端点时期望为关。
+- 防复发：上述测试，以及源码扫描：除诊断脚本外，生产代码不再出现这三个变量名。不得把这三个名字做成真正的关闭开关。
+- 相关记录：BUG-720
+- 复发自：无
+- 修复版本：分支 `codex/python-hygiene-20261013`。本号是任务书预留的空号；本分支最大已用编号是 BUG-1357，1340 未被占用，因此不顺延。
+

@@ -29,8 +29,13 @@ VEDASTRO_RANGE_SCAN_NETWORK_ENABLED=0

 VEDASTRO_CACHE_TTL_SECONDS=604800
 VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS=604800
+# 网关离线排队作业，和下面的免费层限流不是一回事。
 VEDASTRO_GATEWAY_QUEUE_ENABLED=1
-VEDASTRO_FREE_TIER_QUEUE=1
+# 免费层限流由有没有 API key，以及 MAX_REQUESTS / WINDOW_SECONDS 是否大于 0 决定。
+# VEDASTRO_FREE_TIER_QUEUE 已废弃，留着也不会开关限流。
+# VEDASTRO_FREE_TIER_MAX_REQUESTS=5
+# VEDASTRO_FREE_TIER_WINDOW_SECONDS=60
+# VEDASTRO_FREE_TIER_WAIT_BUDGET_SECONDS=
 VEDASTRO_FAIL_OPEN_LOCAL=1
 VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT=0

@@ -78,7 +83,8 @@ npm run dev --prefix frontend

 - 不要让浏览器直连 VedAstro，也不要把 `VEDASTRO_API_KEY` 放进前端。
 - `VEDASTRO_CACHE_TTL_SECONDS` 和 `VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS` 用于缓存官方或自建服务结果。
- `VEDASTRO_GATEWAY_QUEUE_ENABLED=1` / `VEDASTRO_FREE_TIER_QUEUE=1` 用于把昂贵或被限流的外部请求排队。
+- `VEDASTRO_GATEWAY_QUEUE_ENABLED=1` 只打开网关自己的离线排队作业。
+- 免费层限流不看 `VEDASTRO_FREE_TIER_QUEUE`（这个名字已废弃，不控制行为）。没有 API key，并且 `VEDASTRO_FREE_TIER_MAX_REQUESTS` 与 `VEDASTRO_FREE_TIER_WINDOW_SECONDS` 都大于 0 时，官方公共端点会限流。等待上限是 `VEDASTRO_FREE_TIER_WAIT_BUDGET_SECONDS`。
 - 如果 VedAstro 官方或自建服务不可达，`VEDASTRO_FAIL_OPEN_LOCAL=1` 会保持本地 Jyotish 引擎继续输出，并在 Technique Audit Table 里降级标注。
 - Gateway 不会默认声称跑完 641 项；它只把 capability catalog、dynamic selection、cache/queue/fallback 状态作为证据边界交给 strict workflow。

@@ -187,7 +193,10 @@ VEDASTRO_ENABLE_NETWORK=1
 VEDASTRO_TIMEOUT_SECONDS=20
 VEDASTRO_CACHE_TTL_SECONDS=600
 VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS=600
-VEDASTRO_FREE_TIER_QUEUE=1
+# 免费层限流由 API key 与 MAX_REQUESTS / WINDOW_SECONDS 决定。
+# VEDASTRO_FREE_TIER_QUEUE 已废弃，不控制限流。
+# VEDASTRO_FREE_TIER_MAX_REQUESTS=5
+# VEDASTRO_FREE_TIER_WINDOW_SECONDS=60
 # 可选
 # VEDASTRO_API_KEY=sk_live_xxx
 ```
@@ -0,0 +1,48 @@
+# PROGRESS · 报告渲染函数拆分第一阶段
+
+- 日期：2026-10-15
+- 分支：`codex/pl9-render-split-20261013`
+- 基线：`origin/staging` `32b238cf8619cfab4b1107059ec997c974eb0d13`
+- 范围：只做任务书第一阶段。不改 BUG_HISTORY（开工最大号 BUG-1357，本单不占号）、CHANGELOG、前端、API 服务器。
+
+## 做了什么
+
+把 `render_pl9_markdown` 里不依赖外层数据的 34 个嵌套函数原样搬到 `scripts/report_render/`。函数体只把 `planet_label_cn` 改成模块常量 `PLANET_LABEL_CN`。`SIGNS` / `SIGNS_CN` 按原字面量抄到 `text.py`，避免新包顶层导入引擎。
+
+| 模块 | 函数 |
+| --- | --- |
+| envelope.py | `_unwrap` `_envelope_list` `_envelope_dict` `_bool_text` `_source_dict` `_has_executed_payload` |
+| text.py | `_humanize_reader_token` `_humanize_reader_group` `_humanize_tajika_candidate` `_humanize_tajika_motion` `_display_rashi_from_value` `_humanize_reader_summary_line` `_humanize_annual_highlight_line` `_humanize_annual_narrative_highlight` `_humanize_domain_title` `_workbook_readable_label` `_workbook_readable_sentence` `_workbook_reading_paragraph` `_md_cell` `_render_compact_payload` `_sign_text` `_value_text` `_compose_narrative_paragraph` `_append_unique_text` `_planet_list_text` |
+| sections_misc.py | `_render_boundary_packet_summary_line` `_status_from_report_pack_section` `_section_evidence_boundary` `_domi_pl9_current_gap_appendix` `_chart_planets_from_positions` `_kp_house_focus_text` `_high_value_theme_authority_sentence` `_kp_conflict_adjudication_sentence` `_prashna_exact_cusp_boundary_sentence` |
+
+## 排除
+
+- `_degree_text`：函数内定义两次，本阶段不搬。
+- `_sign_degree_text`、`_longitude_text`：任务书点名与度数簇一起排除。
+- `_profile_aware_benchmark_appendix`：要读引擎里的 `_blocked_planned_section` 和 `ROOT_DIR`。新包不能顶层导入引擎，否则成环，也过不了干净进程导入测试。
+- 依赖外层数据、或调用上述未搬函数的嵌套函数：留给下一阶段。
+
+## 前后
+
+| 项 | 搬之前 | 搬之后 |
+| --- | --- | --- |
+| `render_pl9_markdown` 行数 | 6917 | 6506 |
+| 嵌套函数 | 148 | 114 |
+
+行数少了 411，嵌套函数少了 34。没有改既有断言。
+
+## 检查
+
+| 检查 | 命令 | 结果数字 |
+| --- | --- | --- |
+| 未搬移，不屏蔽 | `report_render_split_diff.py` | 退出码 0，24 项 |
+| 未搬移，屏蔽 | 同上 `--mask-pyjhora` | 退出码 0，24 项 |
+| 搬移后，不屏蔽 | 同上 | 退出码 0，24 项 |
+| 搬移后，屏蔽 | 同上 `--mask-pyjhora` | 退出码 0，24 项 |
+| 指定 pytest + 隐私扫描 | `.venv\Scripts\python.exe -m pytest` 任务书列表与 `tests/test_repo_privacy_markers.py` | 退出码 0；236 收集，233 通过，3 跳过，0 失败 |
+
+同一轮里，基线工作树和新代码渲染同一份 pickle，24 个 sha 全部相同。完整数据版中英 8 个 sha 在四轮里不变。屏蔽那轮曾有一次基线对基线不一致：夜间盘接口响应里年主重放一会儿是超时、一会儿是导入失败。这是屏蔽后的外部重放抖动，不是这次搬函数引入的。原样重跑后退出码 0。
+
+## 偏离
+
+任务书预期行数下降约 600 到 900。实际 411，因为上面三个排除项里有一份 142 行的附录。没有做第二阶段及以后。
@@ -0,0 +1,95 @@
+# PROGRESS · 引擎卫生：限流状态、静默失败、功能吉凶兜底
+
+- 日期：2026-10-15
+- 分支：`codex/python-hygiene-20261013`
+- 基线：`origin/staging` `e4ff62c73856d1e10d44f42f998e94835b8acdf0`
+- 前置：`dbba127e715759f0e225fc57faf12508a01d2380` 是该基线的祖先
+- BUG：BUG-1340。本分支最大已用编号是 BUG-1357，1340 空着，按任务书使用，不顺延。
+
+## 做了什么
+
+1. 免费层「是否在限流」改成读适配器里的真实判断：官方公共端点、没有 API key、配额窗口大于 0。三个旧变量名还在，只在诊断里标成已废弃，不能拿来关掉限流。网关离线排队仍用原来的另一组变量。
+2. 功能吉凶兜底收成 `scripts/functional_layer.py` 一处。引擎、接口、MCP 都转发到它。正常成功的结果与原来逐字节一致。`functional_benefics.py` 未改。
+3. 会让结果少一块的静默失败改为记下模块名、函数名和错误类型。预期内的失败保持安静，并写明原因。裸 `except` 里任务点名的两处引擎、星宿大运和星宿命令已改成 `except Exception`。
+4. 快速门加入静默失败计数和 `scripts/` 根目录 `.py` 数量两条只许下降的守卫。
+
+## T3 分类
+
+写作时口径是 29 处。本基线多了 `scripts/api_scratch_retention.py::_loop`（前序单新增），收尾 30 处。A 类 14 处不计入守卫；B 类 16 处改为记日志。未收窄 A 类的异常类型：不能证明只会抛某一种。
+
+| 类 | 位置 | 处理 |
+| --- | --- | --- |
+| A | `annual_tajika_pack._iana_timezone_id` | 时区名解析失败保持安静，`# noqa: BLE001` |
+| A | `api_scratch_retention._loop` | 清理循环必须活着，保持安静 |
+| A | `jyotishganit_shadbala_surface_probe.clean` | 探针清理失败保持安静 |
+| A | `jyotish_engine.write_pl9_pdf` 字体循环 | 换下一个字体，最后仍可用 Helvetica |
+| A | `local_oss_observation_inventory.git_commit` | 读不到提交号保持安静 |
+| A | `muhurta` 导入 swisseph | Windows 上动态库失败不一定是 ImportError，保持安静 |
+| A | `nakshatra_dasha.calc_nakshatra_dasha_breakdown` | 日期文本解析失败不改变年龄明细；裸 `except` 已改为 `except Exception` |
+| A | `oracle_boundary_audit._date_delta_days` | 日期文本解析失败保持安静 |
+| A | `tajika._resolve_external_pyjhora_python` | 外部解释器探测失败保持安静 |
+| A | `vedastro_gateway._sdk_version` | 可选依赖版本读不到保持安静 |
+| A | `vedastro_python_bridge` 序列化兜底 | 对象转 JSON 失败再走原兜底 |
+| A | `yoga_engine` 自定义表达式 3 处 | 每条规则都可能失败，保持安静 |
+| B | `cmd_muhurta._get_sun_moon_lons` | 记下错误类型后仍走近似日月位置 |
+| B | `cmd_nakshatra_adv.cmd_nakshatra_dasha` | 过境位置失败记日志；裸 `except` 已改为 `except Exception` |
+| B | `daily_guidance_service._current_mahadasha` | 记日志，返回值不变 |
+| B | `jyotish_engine._conflict_arbitration` | 记日志；裸 `except` 已改为 `except Exception` |
+| B | `jyotish_engine.cmd_planetary_congregation` | 记日志 |
+| B | `jyotish_engine.cmd_full_reading` 过境位置 | 记日志；裸 `except` 已改为 `except Exception` |
+| B | `muhurta` 日出、儒略日、日月位置 3 处 | 记日志，返回值不变 |
+| B | `pl9_full_data_export` 中英 Sudarshana 各 1 处 | 记日志。报告正文不写「本段数据缺失」（需拍板） |
+| B | `reference_transparency_contract._case_chart` | 记日志 |
+| B | `solar_return._get_sun_lon_jd` | 记日志 |
+| B | `tajika._year_lord_is_daytime` | 记日志 |
+| B | `transit_trigger._get_transit_lon_precise` | 记日志 |
+| B | `mcp_server.strict_workflow` | 记日志 |
+
+引擎里另外三处裸 `except` 会赋值（记忆统计、两处年龄推算），不在本口径，未改。
+
+## 断言
+
+| 位置 | 原值 | 新值 | 原因 |
+| --- | --- | --- | --- |
+| `tests/test_api_server_security.py` 高严谨官方摘要 `queue_enabled` | False | True | 没设置已废弃开关时，官方免费层默认仍在限流 |
+| `tests/test_vedastro_user_entrypoint.py` 两处 `free_tier_queue_enabled` / `queue_enabled` | True | True | 该用例没有 key、端点是官方地址、配额为正。字段跟随真实限流，不再跟随废弃开关 |
+
+## 全量解读对照
+
+同一虚构盘、同一随机种子、关闭网络。退出码 0，标准错误 20 字节。标准输出 5236527 字节，摘要哈希与开工前基线不同。逐项对比只有 29 处差异：缓存创建和过期时间、耗时、最慢阶段的名次、结果哈希、一份大运的参照时钟。`modules.functional_benefic_malefic` 排序后逐字节相同，长度 1442。
+
+## 反向验证
+
+人为在 `scripts/functional_layer.py` 加一处 `except Exception: pass`，并在根目录加一个空 `.py`，然后立刻删掉。
+
+```
+FAILED tests/test_python_hygiene_budget.py::test_silent_except_budget_does_not_grow
+AssertionError: ['scripts/functional_layer.py:46']
+assert 1 <= 0
+FAILED tests/test_python_hygiene_budget.py::test_scripts_root_py_count_does_not_grow
+AssertionError: 447
+assert 447 <= 446
+```
+
+撤回后根目录仍是 446。守卫测试 2 通过。
+
+## 检查
+
+| 检查 | 命令 | 结果数字 |
+| --- | --- | --- |
+| 定向测试 | 任务书 Grok 自测清单，外加功能层快照 3 条、改过的安全断言 1 条 | 77 项，76 通过，1 失败 |
+| 失败项 | `test_gateway_completion_archives_official_raw_response` | `st_mode & 511` 实得 438，期望 384 |
+| 静默守卫 | `tests/test_python_hygiene_budget.py` | 2 通过；未标记静默 0；带 noqa 的 A 类 14 |
+| 功能层快照 | `tests/test_functional_layer_snapshot.py` | 3 通过 |
+| 根目录 `.py` | `scripts/*.py` | 446（含新建的 `functional_layer.py`） |
+
+失败的那条断言检查归档文件权限必须是 `0o600`。本单在 `vedastro_gateway.py` 只加了一行静默原因注释，`os.chmod(..., 0o600)` 未改。Windows 回报的权限位是 `0o666`。断言不放宽。
+
+## 偏离
+
+- T4 测试脚本归位未做。任务书让步顺序第一项，避免搬动被证据索引或冻结清单钉住的文件。
+- T5 第 2 条做了：根目录上限按收尾实测 446，只许下降。
+- A 类没有收窄异常类型。
+- 报告正文不增加「本段数据缺失」。
+- `skills/jyotish-vedic-astrology/scripts/` 里的旧排队开关解析未改。任务书文件清单不含它，也没有字节对照测试。
+- 不改前端、依赖、`.gitea/**`、生产配置、冻结文件。
@@ -31,6 +31,7 @@ import json
 import hashlib
 import subprocess
 import asyncio
+import logging
 from copy import deepcopy
 from datetime import datetime, timedelta
 from functools import lru_cache
@@ -43,12 +44,13 @@ sys.path.insert(0, os.path.join(SCRIPT_DIR, "scripts"))
 from local_env import load_local_env

 from mcp.server.fastmcp import FastMCP
-from functional_benefics import derive_functional_benefic_malefic
 from vedastro_priority import official_snapshot_evidence
 from unified_consultation_orchestrator import UnifiedConsultationOrchestrator
 from skill_experience import build_skill_doctor, build_skill_onboarding, summarize_execution_status
 from punarphoo import detect_punarphoo

+logger = logging.getLogger(__name__)
+
 load_local_env(SCRIPT_DIR)

 # ============================================================================
@@ -1838,19 +1840,8 @@ def _upagraha_gulika_maandi_audit(prashna_context: Any) -> List[Dict[str, Any]]:
 def _derive_functional_benefic_malefic(modules: Dict[str, Any]) -> Dict[str, Any]:
    chart = _safe_get(modules, "chart")
    ascendant = _safe_get(chart, "ascendant") if isinstance(chart, dict) else None
-    if not isinstance(ascendant, dict):
-        return {
-            "status": "blocked",
-            "ascendant": ascendant.get("sign") if isinstance(ascendant, dict) else None,
-            "functional_benefics": [],
-            "functional_malefics": [],
-            "functional_neutrals": [],
-            "yogakarakas": [],
-            "owned_houses": {},
-            "effect_on_confidence": "Missing chart.ascendant; functional layer blocked.",
-            "source": "strict_functional_benefic_malefic_v2",
-        }
-    return derive_functional_benefic_malefic(ascendant.get("sign"))
+    from functional_layer import functional_layer_for_ascendant
+    return functional_layer_for_ascendant(ascendant)


 def _derive_synastry_relationship_support(modules: Dict[str, Any]) -> Dict[str, Any]:
@@ -4788,8 +4779,9 @@ def strict_workflow(
                western_evidence_packet=western_packet,
                blind=False,
            )
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.warning("mcp_server.strict_workflow degraded: %s", type(exc).__name__)
+            logger.debug("mcp_server.strict_workflow degraded", exc_info=True)
    return result


@@ -89,7 +89,7 @@ def _iana_timezone_id(value: Any) -> str | None:
        return None
    try:
        ZoneInfo(value.strip())
-    except Exception:
+    except Exception:  # noqa: BLE001 - caller-supplied timezone text; an unknown IANA name is not a zone
        return None
    return value.strip()

@@ -603,7 +603,7 @@ def start_periodic_prune(interval_s: float | None = None):
            time.sleep(interval_s)
            try:
                prune_all()
-            except Exception:
+            except Exception:  # noqa: BLE001 - the periodic prune thread must stay alive; one failed cycle is not a missing reading
                continue

    thread = threading.Thread(target=_loop, name=_PRUNE_THREAD_NAME, daemon=True)
@@ -14,6 +14,9 @@ from typing import Dict, Optional, List
 from datetime import datetime, timedelta
 import sys
 import os
+import logging
+
+logger = logging.getLogger(__name__)

 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

@@ -40,8 +43,9 @@ def _get_sun_moon_lons(year: int, month: int, day: int,
        sun_res = swe.calc_ut(jd_ut, swe.SUN, flags)
        moon_res = swe.calc_ut(jd_ut, swe.MOON, flags)
        return _primary_longitude(sun_res[0]), _primary_longitude(moon_res[0]), True
-    except Exception:
-        pass
+    except Exception as exc:
+        logger.warning("cmd_muhurta._get_sun_moon_lons degraded: %s", type(exc).__name__)
+        logger.debug("cmd_muhurta._get_sun_moon_lons degraded", exc_info=True)
    # 近似算法
    from muhurta import _approx_sun_moon_lon
    sun_lon, moon_lon = _approx_sun_moon_lon(year, month, day)
@@ -12,9 +12,12 @@ Jyotish Vedic Astrology Skill — v6.0.22

 import sys
 import os
+import logging
 from typing import Dict, Any, Optional
 from datetime import datetime

+logger = logging.getLogger(__name__)
+

 def _compute_chart_from_args(args: Any) -> tuple:
    """
@@ -234,8 +237,9 @@ def cmd_nakshatra_dasha(args: Any) -> Dict[str, Any]:
        for pn, pd in transit_planets.items():
            if isinstance(pd, dict) and 'degree' in pd:
                transit_lons[pn] = pd['degree']
-    except:
-        pass
+    except Exception as exc:
+        logger.warning("cmd_nakshatra_adv.cmd_nakshatra_dasha degraded: %s", type(exc).__name__)
+        logger.debug("cmd_nakshatra_adv.cmd_nakshatra_dasha degraded", exc_info=True)

    return nakshatra_dasha_full_report(chart, birth_date_str, age, transit_lons)

@@ -4,6 +4,7 @@ from __future__ import annotations

 from datetime import datetime
 from pathlib import Path
+import logging
 import sys
 from typing import Any

@@ -25,6 +26,8 @@ SIGNS = [
    "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
 ]

+logger = logging.getLogger(__name__)
+
 HOUSE_THEMES = {
    1: ("自我", "整理状态、重启节奏"),
    2: ("财务", "记账、定价、整理资源"),
@@ -96,7 +99,9 @@ def _current_mahadasha(chart: dict[str, Any], body: dict[str, Any], reference_da
            moon_lon=float(moon_lon),
            current_date=datetime.strptime(reference_date[:10], "%Y-%m-%d"),
        )
-    except Exception:
+    except Exception as exc:
+        logger.warning("daily_guidance_service._current_mahadasha degraded: %s", type(exc).__name__)
+        logger.debug("daily_guidance_service._current_mahadasha degraded", exc_info=True)
        return None
    periods = timeline.get("periods") if isinstance(timeline, dict) else None
    if isinstance(periods, list):
@@ -18,8 +18,24 @@ ROOT = Path(__file__).resolve().parents[1]
 FAST_TIMEOUT_THRESHOLD_SECONDS = 5.0


-def _bool_env(name: str) -> bool:
-    return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
+_DEPRECATED_FREE_TIER_QUEUE_ENVS = (
+    "VEDASTRO_FREE_TIER_QUEUE",
+    "VEDASTRO_FREE_TIER_QUEUE_ENABLED",
+    "VEDASTRO_ENABLE_FREE_TIER_QUEUE",
+)
+
+
+def _deprecated_free_tier_queue_env() -> list[str]:
+    """Env names that used to be reported as the queue switch. They do not control it."""
+    return [name for name in _DEPRECATED_FREE_TIER_QUEUE_ENVS if os.environ.get(name, "").strip()]
+
+
+def _free_tier_queue_enabled() -> bool:
+    try:
+        from scripts.vedastro_service_adapter import free_tier_throttle_state
+    except ModuleNotFoundError:
+        from vedastro_service_adapter import free_tier_throttle_state
+    return bool(free_tier_throttle_state()["active"])


 def _timeout_seconds() -> float:
@@ -38,7 +54,8 @@ def build_report() -> dict:
    network_enabled = os.environ.get("VEDASTRO_ENABLE_NETWORK", "1").strip().lower() in {"1", "true", "yes", "on"}
    timeout_seconds = _timeout_seconds()
    has_api_key = bool(os.environ.get("VEDASTRO_API_KEY", "").strip())
-    free_tier_queue_enabled = _bool_env("VEDASTRO_FREE_TIER_QUEUE") or _bool_env("VEDASTRO_FREE_TIER_QUEUE_ENABLED") or _bool_env("VEDASTRO_ENABLE_FREE_TIER_QUEUE")
+    free_tier_queue_enabled = _free_tier_queue_enabled()
+    deprecated_env_present = _deprecated_free_tier_queue_env()
    missing = []
    readiness_blockers = []
    if not endpoint:
@@ -62,6 +79,7 @@ def build_report() -> dict:
        "timeout_seconds": timeout_seconds,
        "has_api_key": has_api_key,
        "free_tier_queue_enabled": free_tier_queue_enabled,
+        "deprecated_env_present": deprecated_env_present,
        "free_tier_possible_with_cache_queue": bool(endpoint and network_enabled and timeout_seconds > FAST_TIMEOUT_THRESHOLD_SECONDS),
        "missing": missing,
        "readiness_blockers": readiness_blockers,
@@ -110,6 +128,7 @@ def main() -> int:
    print(f"has_api_key: {str(report['has_api_key']).lower()}")
    print(f"free_tier_possible_with_cache_queue: {str(report['free_tier_possible_with_cache_queue']).lower()}")
    print(f"free_tier_queue_enabled: {str(report['free_tier_queue_enabled']).lower()}")
+    print(f"deprecated_env_present: {', '.join(report['deprecated_env_present']) or 'none'}")
    if report["readiness_blockers"]:
        print(f"readiness_blockers: {', '.join(report['readiness_blockers'])}")
    if report["missing"]:
@@ -0,0 +1,41 @@
+"""One functional benefic/malefic snapshot for the engine, API, and MCP.
+
+This module is not ``functional_benefics.py``. That file is part of the
+rectification freeze and must stay byte-for-byte unchanged.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def functional_layer_for_ascendant(ascendant: Any) -> dict[str, Any]:
+    """Return the functional layer for an ascendant payload.
+
+    A dict uses its ``sign``. Anything else is a missing sign and follows
+    the normal ``derive_functional_benefic_malefic`` blocked result.
+    Import or calculation failures return the engine's blocked shape.
+    The exception text stays inside ``effect_on_confidence`` because that
+    field already did; callers must not write this dict to a log.
+    """
+    try:
+        try:
+            from functional_benefics import derive_functional_benefic_malefic
+        except ModuleNotFoundError:
+            from scripts.functional_benefics import derive_functional_benefic_malefic
+
+        sign = ascendant.get("sign") if isinstance(ascendant, dict) else None
+        return derive_functional_benefic_malefic(sign)
+    except Exception as exc:
+        sign = ascendant.get("sign") if isinstance(ascendant, dict) else None
+        return {
+            "status": "blocked",
+            "ascendant": sign,
+            "functional_benefics": [],
+            "functional_malefics": [],
+            "functional_neutrals": [],
+            "yogakarakas": [],
+            "owned_houses": {},
+            "effect_on_confidence": f"未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}",
+            "source": "strict_functional_benefic_malefic_v2",
+        }
@@ -7110,22 +7110,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, VedastroEvidenceMixin, SynastryM
        }

    def _functional_benefic_malefic_snapshot(self, planets, ascendant):
-        try:
-            from functional_benefics import derive_functional_benefic_malefic
-            asc_sign = ascendant.get('sign')
-            return derive_functional_benefic_malefic(asc_sign)
-        except Exception as exc:
-            return {
-                'status': 'blocked',
-                'ascendant': ascendant.get('sign'),
-                'functional_benefics': [],
-                'functional_malefics': [],
-                'functional_neutrals': [],
-                'yogakarakas': [],
-                'owned_houses': {},
-                'effect_on_confidence': f'未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}',
-                'source': 'strict_functional_benefic_malefic_v2',
-            }
+        from functional_layer import functional_layer_for_ascendant
+        return functional_layer_for_ascendant(ascendant)

    def _detect_yogas(self, planets, asc_idx):
        yogas = []
@@ -70,6 +70,8 @@ from narayana_dasha import narayana_comparison_lines_zh as _narayana_comparison_
 from capability_evidence_pool import build_capability_evidence_pool_summary
 from guided_topic_discovery import build_guided_topics

+logger = logging.getLogger(__name__)
+

 def build_report_theme_catalog(report):
    try:
@@ -109,6 +111,94 @@ if ROOT_DIR not in sys.path:
    sys.path.insert(0, ROOT_DIR)
 if SCRIPT_DIR not in sys.path:
    sys.path.insert(0, SCRIPT_DIR)
+try:
+    from scripts.report_render.envelope import (
+        _bool_text,
+        _envelope_dict,
+        _envelope_list,
+        _has_executed_payload,
+        _source_dict,
+        _unwrap,
+    )
+except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution
+    from report_render.envelope import (
+        _bool_text,
+        _envelope_dict,
+        _envelope_list,
+        _has_executed_payload,
+        _source_dict,
+        _unwrap,
+    )
+try:
+    from scripts.report_render.sections_misc import (
+        _chart_planets_from_positions,
+        _domi_pl9_current_gap_appendix,
+        _high_value_theme_authority_sentence,
+        _kp_conflict_adjudication_sentence,
+        _kp_house_focus_text,
+        _prashna_exact_cusp_boundary_sentence,
+        _render_boundary_packet_summary_line,
+        _section_evidence_boundary,
+        _status_from_report_pack_section,
+    )
+except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution
+    from report_render.sections_misc import (
+        _chart_planets_from_positions,
+        _domi_pl9_current_gap_appendix,
+        _high_value_theme_authority_sentence,
+        _kp_conflict_adjudication_sentence,
+        _kp_house_focus_text,
+        _prashna_exact_cusp_boundary_sentence,
+        _render_boundary_packet_summary_line,
+        _section_evidence_boundary,
+        _status_from_report_pack_section,
+    )
+try:
+    from scripts.report_render.text import (
+        PLANET_LABEL_CN,
+        _append_unique_text,
+        _compose_narrative_paragraph,
+        _display_rashi_from_value,
+        _humanize_annual_highlight_line,
+        _humanize_annual_narrative_highlight,
+        _humanize_domain_title,
+        _humanize_reader_group,
+        _humanize_reader_summary_line,
+        _humanize_reader_token,
+        _humanize_tajika_candidate,
+        _humanize_tajika_motion,
+        _md_cell,
+        _planet_list_text,
+        _render_compact_payload,
+        _sign_text,
+        _value_text,
+        _workbook_readable_label,
+        _workbook_readable_sentence,
+        _workbook_reading_paragraph,
+    )
+except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution
+    from report_render.text import (
+        PLANET_LABEL_CN,
+        _append_unique_text,
+        _compose_narrative_paragraph,
+        _display_rashi_from_value,
+        _humanize_annual_highlight_line,
+        _humanize_annual_narrative_highlight,
+        _humanize_domain_title,
+        _humanize_reader_group,
+        _humanize_reader_summary_line,
+        _humanize_reader_token,
+        _humanize_tajika_candidate,
+        _humanize_tajika_motion,
+        _md_cell,
+        _planet_list_text,
+        _render_compact_payload,
+        _sign_text,
+        _value_text,
+        _workbook_readable_label,
+        _workbook_readable_sentence,
+        _workbook_reading_paragraph,
+    )

 try:
    from calculation_profile_contract import (
@@ -3501,173 +3591,7 @@ def render_pl9_markdown(packet: dict) -> str:
        else {}
    )

-    def _unwrap(raw_value):
-        # 兼容 packet 字段的 response_envelope 升级：
-        # 真实数据可能直接是 list/dict，也可能被包成 {"result": ..., "response_envelope": {...}}。
-        if (
-            isinstance(raw_value, dict)
-            and 'result' in raw_value
-            and 'response_envelope' in raw_value
-        ):
-            return raw_value.get('result')
-        return raw_value
-
-    def _envelope_list(raw_value):
-        raw_value = _unwrap(raw_value)
-        if isinstance(raw_value, list):
-            return raw_value
-        if isinstance(raw_value, tuple):
-            return list(raw_value)
-        # 期望 list 时，即使只有 result 键也解包，避免 nested planets 被空渲染成 '-'。
-        if isinstance(raw_value, dict) and 'result' in raw_value:
-            nested = raw_value.get('result')
-            if isinstance(nested, list):
-                return nested
-            if isinstance(nested, tuple):
-                return list(nested)
-        return []
-
-    def _envelope_dict(raw_value):
-        raw_value = _unwrap(raw_value)
-        return raw_value if isinstance(raw_value, dict) else {}
-
-    def _bool_text(value):
-        return '是' if value else '否'
-
-    planet_label_cn = {
-        'Sun': '太阳',
-        'Moon': '月亮',
-        'Mars': '火星',
-        'Mercury': '水星',
-        'Jupiter': '木星',
-        'Venus': '金星',
-        'Saturn': '土星',
-        'Rahu': '北交点',
-        'Ketu': '南交点',
-    }
-
-    def _humanize_reader_token(token: str) -> str:
-        value = str(token or '').strip()
-        if not value:
-            return value
-        if value in SIGNS_CN:
-            return SIGNS_CN[value]
-        if value in planet_label_cn:
-            return planet_label_cn[value]
-        return value
-
-    def _humanize_reader_group(raw_group: str) -> str:
-        seen = set()
-        parts = []
-        for item in str(raw_group or '').split(','):
-            humanized = _humanize_reader_token(item)
-            if humanized and humanized not in seen:
-                seen.add(humanized)
-                parts.append(humanized)
-        return ' / '.join(parts)
-
-    def _humanize_tajika_candidate(name: str) -> str:
-        return {
-            'Ithasala_candidate': '趋近相位候选',
-            'Easarapha_candidate': '分离相位候选',
-        }.get(str(name or ''), '年度相位候选')
-
-    def _humanize_tajika_motion(motion: str) -> str:
-        return {'applying': '趋近', 'separating': '分离'}.get(str(motion or ''), str(motion or '-'))
-
-    def _display_rashi_from_value(raw_value):
-        if isinstance(raw_value, int) and 0 <= raw_value < len(SIGNS):
-            return _humanize_reader_token(SIGNS[raw_value])
-        return _humanize_reader_token(raw_value)
-
-    def _render_boundary_packet_summary_line(packet: dict, summary_pairs: list[tuple[str, str]]) -> str:
-        summary = packet.get('summary') if isinstance(packet.get('summary'), dict) else {}
-        bits = [_md_cell(packet.get('packet_mode'))]
-        for key, suffix in summary_pairs:
-            value = summary.get(key)
-            if value not in (None, '', []):
-                bits.append(f"{_md_cell(value)}{suffix}")
-        return f"参考包摘要：{'；'.join(bits)}。"
-
-    def _humanize_reader_summary_line(summary_line: str) -> str:
-        summary_line = _unwrap(summary_line)
-        if isinstance(summary_line, dict):
-            summary_line = (
-                summary_line.get('brief')
-                or summary_line.get('text')
-                or summary_line.get('summary')
-                or ''
-            )
-        text = str(summary_line or '').strip()
-        if not text:
-            return text
-        if text == "年度层已经有可读壳层，但仍需要带着冲突标签阅读。":
-            return "这一年的年度主线已经形成。"
-        if text == "This pack converts Dasha trigger facts into report-ready interpretation seeds without final event claims.":
-            return "这层已经把 Dasha 触发事实压成可读提示，但仍不直接升级为具体事件断语。"
-        if text == "Source: dasha_master_pack.interpretation_triggers.":
-            return "当前解释仍然直接来自 dasha_master_pack.interpretation_triggers 的证据层。"
-        if text == "All claims remain evidence-labeled and parameter-sensitive where the source pack is unclosed.":
-            return "只要 source pack 还没闭环，相关结论就继续保持 evidence-labeled 与 parameter_sensitive。"
-        if text == "Tajika Yogas: annual candidate structures visible":
-            return "Tajika Yogas 当前已经能读到年度候选交互结构。"
-        if text == "Sahams: annual sensitive points visible":
-            return "Sahams 当前已经能读到年度敏感点。"
-        match = re.match(r'^([^:]+):\s*conflict between\s+(.+?)\s+vs\s+(.+)$', text)
-        if match:
-            topic = match.group(1).strip()
-            left = _humanize_reader_group(match.group(2))
-            right = _humanize_reader_group(match.group(3))
-            topic_label = {
-                'Muntha': 'Muntha',
-                'Year Lord': 'Year Lord',
-                'Annual Asc': '返照上升',
-            }.get(topic, topic)
-            if left == right:
-                return f"{topic_label} 当前结果指向{left}。"
-            return f"{topic_label} 当前存在候选分歧：一组结果指向{left}，另一组结果指向{right}。"
-        simple_match = re.match(r'^([^:]+):\s*(.+)$', text)
-        if simple_match:
-            topic = simple_match.group(1).strip()
-            value = _humanize_reader_group(simple_match.group(2))
-            if topic == 'Year Lord' and value:
-                return f"Year Lord 当前优先候选指向{value}。"
-            if topic == 'Muntha' and value:
-                return f"Muntha 当前优先候选指向{value}。"
-            if topic == 'Tajika Yogas' and 'annual candidate structures visible' in str(simple_match.group(2)).lower():
-                return "Tajika Yogas 当前已经能读到年度候选交互结构。"
-            if topic == 'Sahams' and 'annual sensitive points visible' in str(simple_match.group(2)).lower():
-                return "Sahams 当前已经能读到年度敏感点。"
-        return text
-
-    def _humanize_annual_highlight_line(highlight: str) -> str:
-        text = str(highlight or '').strip()
-        if not text:
-            return text
-        if text == "Solar Return / Tajika annual shell is available.":
-            return "本年度返照与 Tajika 时间结构已经生成。"
-        if text == "Muntha and Year Lord are visible, but Year Lord remains producer-disagree sensitive.":
-            return "Muntha 和 Year Lord 现在都已经可见，但 Year Lord 仍要带着分歧标签阅读。"
-        if text == "The report should read annual structure first, then the evidence appendix.":
-            return "这一层适合从年度结构展开，再回头核对证据附录。"
-        return text
-
-    def _humanize_annual_narrative_highlight(highlight: str) -> str:
-        return _humanize_annual_highlight_line(str(highlight)).rstrip('。')
-
-    def _humanize_domain_title(title: str) -> str:
-        text = str(title or '').strip()
-        if not text:
-            return text
-        mapping = {
-            "Career and public direction": "事业与公众方向",
-            "Wealth and growth opportunities": "财富与增长机会",
-            "Relationship and alliance themes": "关系与合作主题",
-            "Health and vitality caution": "健康与精力提醒",
-            "Risk, delay, and friction markers": "阻力、延迟与摩擦提示",
-            "Timing windows and cross-check needs": "时间窗口与交叉核对重点",
-        }
-        return mapping.get(text, text)
+    planet_label_cn = PLANET_LABEL_CN

    def _degree_text(value):
        if value in (None, '', [], {}):
@@ -3677,12 +3601,6 @@ def render_pl9_markdown(packet: dict) -> str:
        except (TypeError, ValueError):
            return str(value)

-    def _source_dict(*candidates):
-        for candidate in candidates:
-            if isinstance(candidate, dict) and candidate:
-                return candidate
-        return {}
-
    def _render_basic_tables() -> list[str]:
        birth_sheet = worksheets.get('birth_and_parameters') if isinstance(worksheets.get('birth_and_parameters'), dict) else {}
        d1_sheet = worksheets.get('d1_rasi_bhava') if isinstance(worksheets.get('d1_rasi_bhava'), dict) else {}
@@ -3907,113 +3825,11 @@ def render_pl9_markdown(packet: dict) -> str:
            readable = [_segment_sentence(segment_id).rstrip('。') for segment_id in segment_ids]
            return f"{group_name}包括：{'；'.join(readable)}。"

-        def _workbook_readable_label(worksheet_id: str) -> str:
-            if worksheet_id == 'birth_and_parameters':
-                return '出生与计算口径'
-            if worksheet_id == 'd1_rasi_bhava':
-                return '本命盘主结构'
-            if worksheet_id == 'strengths_and_scores':
-                return '力量与技术表'
-            if worksheet_id == 'divisional_and_special_charts':
-                return '分盘与特殊点位'
-            if worksheet_id == 'timing_and_predictive_systems':
-                return '时间系统与流年层'
-            if worksheet_id == 'advanced_systems':
-                return '进阶体系'
-            if worksheet_id == 'ai_and_audit':
-                return '审计与证据附录'
-            return '补充工作簿'
-
-        def _workbook_readable_sentence(worksheet_id: str) -> str:
-            if worksheet_id == 'birth_and_parameters':
-                return '出生参数、岁差、交点模式与计算口径已经固定，是整份报告的可信底座。'
-            if worksheet_id == 'd1_rasi_bhava':
-                return '本命盘主结构已经就位，适合先看性格主轴、房屋结构和基础趋势。'
-            if worksheet_id == 'strengths_and_scores':
-                return '力量、功能性吉凶与技术表可以直接用来判断支撑与压力来源。'
-            if worksheet_id == 'divisional_and_special_charts':
-                return '分盘与特殊点位可作为扩展阅读，适合补充看细节。'
-            if worksheet_id == 'timing_and_predictive_systems':
-                return '大运、行运与年度层已经可以形成时间主线。'
-            if worksheet_id == 'advanced_systems':
-                return 'Jaimini、KP、Yoga 等进阶体系可作为补充主轴。'
-            if worksheet_id == 'ai_and_audit':
-                return 'AI 读写、审计与证据层用于核对来源与边界，不直接替代正文。'
-            return f'{worksheet_id} 已纳入当前报告。'
-
        def _reader_group_paragraph(group_name: str, segment_ids: list[str]) -> str:
            if not segment_ids:
                return f'{group_name}当前暂无稳定内容。'
            return _segment_group_sentence(group_name, segment_ids)

-        def _workbook_reading_paragraph(worksheet_ids: list[str]) -> str:
-            parts = []
-            for worksheet_id in worksheet_ids:
-                sentence = _workbook_readable_sentence(worksheet_id).rstrip('。')
-                parts.append(f"先看{_workbook_readable_label(worksheet_id)}，这里会告诉你{sentence}")
-            if not parts:
-                return '当前暂无可用工作簿阅读顺序。'
-            return '；'.join(parts) + '。'
-
-        def _md_cell(value) -> str:
-            text = str(value if value not in (None, '') else '-').replace('\n', ' ').strip()
-            return text.replace('|', '/')
-
-        def _render_compact_payload(value, depth: int = 0) -> str:
-            if value in (None, '', [], {}):
-                return '-'
-            if depth >= 2:
-                return _md_cell(type(value).__name__)
-            if isinstance(value, dict):
-                parts = []
-                for key in sorted(value, key=str):
-                    if key in {
-                        'narrative_items',
-                        'domain_paragraphs',
-                        'response_envelope',
-                        'input_hash',
-                        'profile_hash',
-                        'raw_calculation',
-                        'shared_full_report_authority',
-                    }:
-                        continue
-                    parts.append(f"{key}={_render_compact_payload(value.get(key), depth + 1)}")
-                return '; '.join(parts) if parts else '-'
-            if isinstance(value, (list, tuple)):
-                items = [_render_compact_payload(item, depth + 1) for item in list(value)[:8]]
-                suffix = ' …' if len(value) > 8 else ''
-                return ', '.join(items) + suffix if items else '-'
-            return _md_cell(value)
-
-        def _status_from_report_pack_section(section: dict) -> str:
-            if not isinstance(section, dict):
-                return 'blocked'
-            audit = section.get('audit') if isinstance(section.get('audit'), dict) else {}
-            if section.get('status'):
-                return str(section.get('status'))
-            if audit.get('status'):
-                return str(audit.get('status'))
-            if section:
-                return 'partial_verified'
-            return 'blocked'
-
-        def _section_evidence_boundary(section: dict) -> str:
-            if not isinstance(section, dict):
-                return 'section_missing'
-            exports = section.get('exports') if isinstance(section.get('exports'), dict) else {}
-            ai_bundle = exports.get('ai_evidence_bundle') if isinstance(exports.get('ai_evidence_bundle'), dict) else {}
-            audit = section.get('audit') if isinstance(section.get('audit'), dict) else {}
-            boundaries = []
-            if ai_bundle.get('contains_private_pl9_text') is False:
-                boundaries.append('no_private_pl9_text')
-            if ai_bundle.get('claim_policy'):
-                boundaries.append(str(ai_bundle.get('claim_policy')))
-            if audit.get('must_not_claim'):
-                boundaries.append('must_not_claim=' + ','.join(map(str, audit.get('must_not_claim')[:3])))
-            if section.get('reason'):
-                boundaries.append(str(section.get('reason')))
-            return '; '.join(boundaries) if boundaries else 'evidence_labeled'
-
        def _full_report_pack_assembly_section() -> list[str]:
            if not isinstance(full_report_pack, dict) or not full_report_pack:
                return []
@@ -4071,9 +3887,6 @@ def render_pl9_markdown(packet: dict) -> str:
            out.append('')
            return out

-        def _domi_pl9_current_gap_appendix() -> list[str]:
-            return []
-
        def _profile_aware_benchmark_appendix() -> list[str]:
            dashboard_path = Path(ROOT_DIR) / 'references/oracle/profile_aware_benchmark_boundary_dashboard_2026_08_27.json'
            if not dashboard_path.exists():
@@ -4401,21 +4214,6 @@ def render_pl9_markdown(packet: dict) -> str:
            out.append('')
            return out

-        def _chart_planets_from_positions(positions: dict) -> dict:
-            chart_planets = {}
-            if not isinstance(positions, dict):
-                return chart_planets
-            for planet_name in ('Sun', 'Moon', 'Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn', 'Rahu', 'Ketu'):
-                planet = positions.get(planet_name)
-                if not isinstance(planet, dict):
-                    continue
-                sign = planet.get('sign') or planet.get('rashi') or planet.get('sign_name')
-                if not sign:
-                    continue
-                degree = planet.get('degree_in_sign', planet.get('degree', planet.get('longitude_in_sign', 0)))
-                chart_planets[planet_name] = {'sign': sign, 'degree': degree or 0}
-            return chart_planets
-
        def _render_south_chart(
            positions: dict,
            ascendant_row: dict,
@@ -5385,30 +5183,6 @@ def render_pl9_markdown(packet: dict) -> str:
                return "UL 在待确认位置、Pranapada 在待确认位置，但关系主题仍建议把制度承诺层与气质投射层分开看。"
            return None

-        def _kp_house_focus_text(value) -> str:
-            house_label_map = {
-                2: '收入与资源',
-                5: '机会与投入',
-                6: '职责与压力',
-                7: '合作与伴侣',
-                9: '运气与扩张',
-                10: '事业与位置',
-                11: '回报与支持',
-            }
-            parts = []
-            if isinstance(value, (list, tuple)):
-                raw_parts = [str(item).strip() for item in value if str(item).strip()]
-            else:
-                raw_parts = [part.strip() for part in str(value or '').replace('/', ',').split(',') if part.strip()]
-            for part in raw_parts:
-                try:
-                    house_no = int(part)
-                except ValueError:
-                    parts.append(part)
-                    continue
-                parts.append(f"{house_no}宫[{house_label_map.get(house_no, '-')}]")
-            return '、'.join(parts) if parts else '-'
-
        def _graha_padas_domain_bridge(domain: str) -> str | None:
            graha_padas = jaimini.get('graha_padas') if isinstance(jaimini.get('graha_padas'), dict) else {}
            graha_pada_rows = graha_padas.get('graha_padas') if isinstance(graha_padas.get('graha_padas'), dict) else {}
@@ -6230,41 +6004,6 @@ def render_pl9_markdown(packet: dict) -> str:
                segments.append('这些层当前仍只作为 ' + ' / '.join(statuses) + ' 的辅助阅读，不直接升级为主题主结论')
            return '主题支持补充：' + '；'.join(segments) + '。'

-        def _high_value_theme_authority_sentence(domain: str) -> str | None:
-            domain_label_map = {
-                'career': '事业',
-                'wealth': '财务',
-                'relationship': '关系',
-            }
-            action_map = {
-                'career': '优先排序事业推进顺序、外部舞台和月份节奏',
-                'wealth': '优先排序资源承接、现金流节奏和收益兑现顺序',
-                'relationship': '优先排序合作推进、联结温度和关系节奏',
-            }
-            return (
-                "主结论加厚：Narayana 第二时间轴、力量兑现结构、年度 highlights / field briefs、Patyayini 年内换挡、Graha Padas 这几层现在直接参与"
-                + f"{domain_label_map.get(domain, domain)}章主结论，"
-                + f"用来{action_map.get(domain, '补强主题排序')}；"
-                + "它们仍保留 parameter_sensitive / unclosed_divisional_chart 边界，但不再只停留在附录或登记表。"
-            )
-
-        def _kp_conflict_adjudication_sentence(domain: str) -> str:
-            domain_label_map = {
-                'career': '事业',
-                'wealth': '财务',
-                'relationship': '关系',
-            }
-            focus_map = {
-                'career': '外部舞台、职位变化和项目推进',
-                'wealth': '资源承接、现金流和回收节奏',
-                'relationship': '合作推进、联结温度和承诺节奏',
-            }
-            return (
-                f"如果 Vimshottari、Narayana 与年度专题同向，而 KP 月度支持也落在同一主题上，就把这个月份优先当作{domain_label_map.get(domain, domain)}线的可执行窗口。"
-                f"如果 KP 单独活跃、但本命结构或双大运没有接力，KP 只负责提示 {focus_map.get(domain, '主题变化')} 的月份排序，不改写主题事实。"
-                f"如果 KP 与年度专题或双大运出现冲突，先以本命结构 + Vimshottari + Narayana 定主轴，KP 退回到月度提醒与核对层，继续保持 parameter_sensitive。"
-            )
-
        def _patyayini_reader_bridge() -> list[str]:
            if not replay_patyayini_rows:
                return []
@@ -6305,12 +6044,6 @@ def render_pl9_markdown(packet: dict) -> str:
                )
            return out

-        def _prashna_exact_cusp_boundary_sentence() -> str | None:
-            return (
-                "如果下一步要把月份继续缩到更细日期，仍需要补 question_time / place / timezone / question category；"
-                "Prashna ruling planets 与 exact cusp oracle 在这份本命个人报告里继续保持 blocked，只作为后续缩窄入口，不冒充 timing truth。"
-            )
-
        def _reader_unexported_high_value_layers_section() -> list[str]:
            divisional_sheet = worksheets.get('divisional_and_special_charts') if isinstance(worksheets.get('divisional_and_special_charts'), dict) else {}
            points = divisional_sheet.get('sensitive_points') if isinstance(divisional_sheet.get('sensitive_points'), dict) else {}
@@ -6394,12 +6127,6 @@ def render_pl9_markdown(packet: dict) -> str:
            except (TypeError, ValueError):
                return _md_cell(value)

-        def _sign_text(row: dict, sign_key: str = 'sign', cn_key: str = 'sign_cn') -> str:
-            if not isinstance(row, dict):
-                return '-'
-            sign = row.get(cn_key) or _humanize_reader_token(row.get(sign_key)) or row.get(sign_key)
-            return _md_cell(sign)
-
        def _pl9_core_structure_tables() -> list[str]:
            birth_sheet = worksheets.get('birth_and_parameters') if isinstance(worksheets.get('birth_and_parameters'), dict) else {}
            d1_sheet = worksheets.get('d1_rasi_bhava') if isinstance(worksheets.get('d1_rasi_bhava'), dict) else {}
@@ -6508,12 +6235,6 @@ def render_pl9_markdown(packet: dict) -> str:
                out.append('')
            return out

-        def _value_text(*values):
-            for value in values:
-                if value not in (None, '', [], {}):
-                    return str(value)
-            return None
-
        ai_audit_sheet = worksheets.get('ai_and_audit') if isinstance(worksheets.get('ai_and_audit'), dict) else {}
        ai_summary = ai_audit_sheet.get('summary') if isinstance(ai_audit_sheet.get('summary'), dict) else {}
        guided_topics = ai_summary.get('guided_topics') if isinstance(ai_summary.get('guided_topics'), list) else []
@@ -6681,20 +6402,6 @@ def render_pl9_markdown(packet: dict) -> str:
                    return '；'.join(bits) + '。'
            return None

-        def _compose_narrative_paragraph(*parts: str | None) -> str | None:
-            cleaned = []
-            for part in parts:
-                text = str(part or '').strip()
-                if not text:
-                    continue
-                text = text.rstrip()
-                if text.endswith('。'):
-                    text = text[:-1]
-                cleaned.append(text)
-            if not cleaned:
-                return None
-            return '；'.join(cleaned) + '。'
-
        dasha_domain_rows = {}
        domain_paragraphs = dasha_interpretation_pack.get('domain_paragraphs') if isinstance(dasha_interpretation_pack.get('domain_paragraphs'), list) else []
        for row in domain_paragraphs:
@@ -6724,15 +6431,6 @@ def render_pl9_markdown(packet: dict) -> str:
        ]
        special_lagnas_layer = divisional_sheet.get('special_lagnas') if isinstance(divisional_sheet.get('special_lagnas'), dict) else {}

-        def _has_executed_payload(value) -> bool:
-            if value in (None, '', [], {}):
-                return False
-            if isinstance(value, dict):
-                status = str(value.get('status') or value.get('execution_status') or '').strip().lower()
-                if status in {'blocked', 'missing', 'not_found', 'parameter_sensitive'}:
-                    return False
-            return True
-
        def _has_varga_code(varga_code: str) -> bool:
            prefix = f'{varga_code}_'
            for layer in varga_layers:
@@ -6955,20 +6653,6 @@ def render_pl9_markdown(packet: dict) -> str:
                return None
            return '；'.join(summary_bits) + '。'

-        def _planet_list_text(items: list[str], limit: int = 3) -> str | None:
-            names = []
-            seen = set()
-            for item in items:
-                humanized = _humanize_reader_token(item)
-                if humanized and humanized not in seen:
-                    seen.add(humanized)
-                    names.append(humanized)
-                if len(names) >= limit:
-                    break
-            if not names:
-                return None
-            return '、'.join(names)
-
        def _functional_layer_sentence() -> str | None:
            if functional_layer.get('status') != 'used':
                return None
@@ -7368,11 +7052,6 @@ def render_pl9_markdown(packet: dict) -> str:
        annual_evidence_appendix = _envelope_dict(annual_report_sections.get('evidence_appendix'))
        annual_field_briefs = _envelope_dict(annual_evidence_appendix.get('field_briefs'))

-        def _append_unique_text(items: list[str], value: str) -> None:
-            text = str(value or '').strip().rstrip('。')
-            if text and text not in items:
-                items.append(text)
-
        def _reader_executive_summary() -> dict:
            ai_audit = packet.get('ai_and_audit') if isinstance(packet.get('ai_and_audit'), dict) else {}
            worksheet_ai = worksheets.get('ai_and_audit') if isinstance(worksheets.get('ai_and_audit'), dict) else {}
@@ -10422,7 +10101,7 @@ def write_pl9_pdf(packet: dict, output_path: str | None = None) -> dict:
            pdfmetrics.registerFont(TTFont(candidate_name, candidate_path))
            font_name = candidate_name
            break
-        except Exception:
+        except Exception:  # noqa: BLE001 - optional CJK font; the next file is tried, and Helvetica remains if none load
            continue

    styles = getSampleStyleSheet()
@@ -10876,22 +10555,8 @@ def is_combust(planet_name, longitude, sun_longitude, retrograde=False):


 def _functional_benefic_malefic_snapshot(planets, ascendant):
-    try:
-        from functional_benefics import derive_functional_benefic_malefic
-        asc_sign = ascendant.get('sign') if isinstance(ascendant, dict) else None
-        return derive_functional_benefic_malefic(asc_sign)
-    except Exception as exc:
-        return {
-            'status': 'blocked',
-            'ascendant': ascendant.get('sign') if isinstance(ascendant, dict) else None,
-            'functional_benefics': [],
-            'functional_malefics': [],
-            'functional_neutrals': [],
-            'yogakarakas': [],
-            'owned_houses': {},
-            'effect_on_confidence': f'未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}',
-            'source': 'strict_functional_benefic_malefic_v2',
-        }
+    from functional_layer import functional_layer_for_ascendant
+    return functional_layer_for_ascendant(ascendant)


 def _build_technique_audit_table(functional_layer, oracle_progress, modules):
@@ -13752,8 +13417,9 @@ def cmd_planetary_congregation(args):
                    ksi = int(klon / 30)
                    khouse = ((ksi - asc_idx) % 12) + 1
                    result['transit'][str(khouse)].append('Ketu')
-            except Exception:
-                pass
+            except Exception as exc:
+                logger.warning("jyotish_engine.cmd_planetary_congregation degraded: %s", type(exc).__name__)
+                logger.debug("jyotish_engine.cmd_planetary_congregation degraded", exc_info=True)

    # 判定
    flags = []
@@ -14193,8 +13859,9 @@ def _conflict_arbitration(report):
            from ashtakavarga import calc_ashtakavarga
            asht_result = calc_ashtakavarga(planets, asc_idx_local)
            house_scores = asht_result.get('house_scores', {})
-        except:
-            pass
+        except Exception as exc:
+            logger.warning("jyotish_engine._conflict_arbitration degraded: %s", type(exc).__name__)
+            logger.debug("jyotish_engine._conflict_arbitration degraded", exc_info=True)

    for h in [6, 8, 12]:
        hs = house_scores.get(f'house_{h}', {})
@@ -16550,8 +16217,9 @@ def cmd_full_reading(args):
                for pn, pd in transit_pl_nak.items():
                    if isinstance(pd, dict) and 'degree' in pd:
                        transit_lons_for_nak[pn] = pd['degree']
-            except:
-                pass
+            except Exception as exc:
+                logger.warning("jyotish_engine.cmd_full_reading degraded: %s", type(exc).__name__)
+                logger.debug("jyotish_engine.cmd_full_reading degraded", exc_info=True)

            nak_dasha = nakshatra_dasha_full_report(
                chart, birth_date_str, nak_age, transit_lons_for_nak)
@@ -24,7 +24,7 @@ def clean(v: Any) -> Any:
    if hasattr(v, "item"):
        try:
            return v.item()
-        except Exception:
+        except Exception:  # noqa: BLE001 - numpy scalar conversion is optional; a non-scalar value is kept as-is
            pass
    return v

@@ -24,7 +24,7 @@ def sha256_file(path: Path) -> str:
 def git_commit(path: Path) -> str | None:
    try:
        return subprocess.check_output(["git", "-C", str(path), "rev-parse", "HEAD"], text=True, stderr=subprocess.DEVNULL).strip()
-    except Exception:
+    except Exception:  # noqa: BLE001 - git metadata is optional; a checkout without HEAD leaves the field empty
        return None


@@ -16,10 +16,13 @@ from __future__ import annotations
 from typing import Dict, List, Optional, Tuple
 from datetime import datetime, timedelta
 import calendar
+import logging
 import math

 from ayanamsa_utils import DEFAULT_AYANAMSA_NAME, sidereal_flags

+logger = logging.getLogger(__name__)
+
 # ── Vara（周日行星）──────────────────────────────────────────────────
 VARA_LORDS = {
    0: ('Sunday',    'Sun',     'asubha'),  # 周日
@@ -870,7 +873,7 @@ def _swisseph_solar_times_local(
    """Return precise local sunrise/sunset via pyswisseph when available."""
    try:
        import swisseph as swe
-    except Exception:
+    except Exception:  # noqa: BLE001 - optional pyswisseph; a missing or unloadable binary uses the approximate sunrise
        return None

    try:
@@ -879,7 +882,9 @@ def _swisseph_solar_times_local(
        flags = getattr(swe, 'FLG_SWIEPH', 2)
        rise_res, rise_tret = swe.rise_trans(local_midnight_jd, swe.SUN, swe.CALC_RISE, geopos, flags=flags)
        set_res, set_tret = swe.rise_trans(local_midnight_jd, swe.SUN, swe.CALC_SET, geopos, flags=flags)
-    except Exception:
+    except Exception as exc:
+        logger.warning("muhurta._swisseph_solar_times_local degraded: %s", type(exc).__name__)
+        logger.debug("muhurta._swisseph_solar_times_local degraded", exc_info=True)
        return None

    if rise_res != 0 or set_res != 0:
@@ -2129,7 +2134,9 @@ def _jd_for_local_time(year: int, month: int, day: int, local_hours: float, tz:
    try:
        import swisseph as swe
        return swe.julday(year, month, day, local_hours - tz)
-    except Exception:
+    except Exception as exc:
+        logger.warning("muhurta._jd_for_local_time degraded: %s", type(exc).__name__)
+        logger.debug("muhurta._jd_for_local_time degraded", exc_info=True)
        return None


@@ -2145,7 +2152,9 @@ def _swisseph_sun_moon_lon(jd: float, ayanamsa_name: str = DEFAULT_AYANAMSA_NAME
        sun = swe.calc_ut(jd, swe.SUN, flags)[0][0] % 360
        moon = swe.calc_ut(jd, swe.MOON, flags)[0][0] % 360
        return sun, moon
-    except Exception:
+    except Exception as exc:
+        logger.warning("muhurta._swisseph_sun_moon_lon degraded: %s", type(exc).__name__)
+        logger.debug("muhurta._swisseph_sun_moon_lon degraded", exc_info=True)
        return None


@@ -463,7 +463,7 @@ def calc_nakshatra_dasha_breakdown(
    birth_date = None
    try:
        birth_date = datetime.strptime(birth_date_str, '%Y-%m-%d')
-    except:
+    except Exception:  # noqa: BLE001 - caller-supplied date text; an unparseable string leaves the breakdown ages unchanged
        pass

    return {
@@ -81,12 +81,11 @@ def _build_vedastro_official_full_snapshot_payload_from_chart(chart):
    }

 def _free_tier_queue_enabled_env() -> bool:
-    raw_values = [
-        str(os.environ.get("VEDASTRO_FREE_TIER_QUEUE", "")).strip().lower(),
-        str(os.environ.get("VEDASTRO_FREE_TIER_QUEUE_ENABLED", "")).strip().lower(),
-        str(os.environ.get("VEDASTRO_ENABLE_FREE_TIER_QUEUE", "")).strip().lower(),
-    ]
-    return any(value in {"1", "true", "yes", "on"} for value in raw_values)
+    try:
+        from scripts.vedastro_service_adapter import free_tier_throttle_state
+    except ModuleNotFoundError:
+        from vedastro_service_adapter import free_tier_throttle_state
+    return bool(free_tier_throttle_state()["active"])

 def _preferred_strict_contract(strict_workflow_contracts, primary_route=None):
    if not isinstance(strict_workflow_contracts, dict) or not strict_workflow_contracts:
@@ -283,7 +283,7 @@ def _date_delta_days(engine_date: str | None, target_date: str | None) -> int |
        engine_parts = [int(part) for part in engine_date.split("-")[:3]]
        target_parts = [int(part) for part in target_date.split("-")[:3]]
        return (date(*engine_parts) - date(*target_parts)).days
-    except Exception:
+    except Exception:  # noqa: BLE001 - compared date strings are caller-supplied; an unparseable value is an absent delta
        return None


@@ -11,6 +11,7 @@ that profile, so it is not called.
 from __future__ import annotations

 import copy
+import logging
 import re
 from collections.abc import Mapping
 from typing import Any
@@ -34,6 +35,8 @@ except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution
        render_pl9_parity_markdown,
    )

+logger = logging.getLogger(__name__)
+
 FULL_DATA_EDITION = "full_data"
 REPORT_VERSION_FULL = "pl9_personal_long_report.v4"
 FULL_DATA_DOMAINS = {
@@ -1698,8 +1701,12 @@ def _render_pl9_ai_density_raw_data_markdown_en(packet: dict) -> str:
                            chart_rows.append([chart_name, planet, item.get("sign"), item.get("house")])
                if chart_rows:
                    lines.extend(["### Sudarshana Chakra Three-Reference Placements", "", table(["Reference chart", "Planet", "Sign", "House"], chart_rows), ""])
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.warning(
+                "pl9_full_data_export._render_pl9_ai_density_raw_data_markdown_en degraded: %s",
+                type(exc).__name__,
+            )
+            logger.debug("pl9_full_data_export._render_pl9_ai_density_raw_data_markdown_en degraded", exc_info=True)

    d1_sheet = dict_get(("worksheets", "d1_rasi_bhava"), {})
    d1_houses = d1_sheet.get("houses") if isinstance(d1_sheet, dict) else {}
@@ -2191,8 +2198,12 @@ def _render_pl9_ai_density_raw_data_markdown_zh(packet: dict) -> str:
                        table(["参考盘", "行星", "星座", "宫位"], chart_rows),
                        "",
                    ])
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.warning(
+                "pl9_full_data_export._render_pl9_ai_density_raw_data_markdown_zh degraded: %s",
+                type(exc).__name__,
+            )
+            logger.debug("pl9_full_data_export._render_pl9_ai_density_raw_data_markdown_zh degraded", exc_info=True)

    text = "\n".join(line for line in lines if line is not None)
    text = re.sub(r"\n{3,}", "\n\n", text).strip() + "\n"
@@ -4,6 +4,7 @@
 from __future__ import annotations

 import json
+import logging
 import sys
 from datetime import datetime
 from functools import lru_cache
@@ -14,6 +15,8 @@ SCRIPT_DIR = Path(__file__).resolve().parent
 if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

+logger = logging.getLogger(__name__)
+
 try:
    from scripts.domain_calculation_service import compute_chart, compute_transit_longitude, compute_vimshottari_timeline
    from scripts.timing_precision_contract import build_timing_precision_contract
@@ -253,9 +256,11 @@ def _case_chart(case_id: str, year: int, month: int, day: int, hour: int, minute
    try:
        return compute_chart({
            "year": year, "month": month, "day": day, "hour": hour, "minute": minute,
-            "lat": lat, "lon": lon, "tz": tz, "ayanamsa": "lahiri", "node_mode": node_mode,
+            "lat": lat, "lon": lon, "tz": tz, "ayanamsa": "lahiri",             "node_mode": node_mode,
        })
-    except Exception:
+    except Exception as exc:
+        logger.warning("reference_transparency_contract._case_chart degraded: %s", type(exc).__name__)
+        logger.debug("reference_transparency_contract._case_chart degraded", exc_info=True)
        return None


@@ -0,0 +1 @@
+"""Helpers moved unchanged out of render_pl9_markdown (phase 1)."""
@@ -0,0 +1,49 @@
+"""Helpers moved unchanged out of render_pl9_markdown (phase 1)."""
+
+def _unwrap(raw_value):
+    # 兼容 packet 字段的 response_envelope 升级：
+    # 真实数据可能直接是 list/dict，也可能被包成 {"result": ..., "response_envelope": {...}}。
+    if (
+        isinstance(raw_value, dict)
+        and 'result' in raw_value
+        and 'response_envelope' in raw_value
+    ):
+        return raw_value.get('result')
+    return raw_value
+
+def _envelope_list(raw_value):
+    raw_value = _unwrap(raw_value)
+    if isinstance(raw_value, list):
+        return raw_value
+    if isinstance(raw_value, tuple):
+        return list(raw_value)
+    # 期望 list 时，即使只有 result 键也解包，避免 nested planets 被空渲染成 '-'。
+    if isinstance(raw_value, dict) and 'result' in raw_value:
+        nested = raw_value.get('result')
+        if isinstance(nested, list):
+            return nested
+        if isinstance(nested, tuple):
+            return list(nested)
+    return []
+
+def _envelope_dict(raw_value):
+    raw_value = _unwrap(raw_value)
+    return raw_value if isinstance(raw_value, dict) else {}
+
+def _bool_text(value):
+    return '是' if value else '否'
+
+def _source_dict(*candidates):
+    for candidate in candidates:
+        if isinstance(candidate, dict) and candidate:
+            return candidate
+    return {}
+
+def _has_executed_payload(value) -> bool:
+    if value in (None, '', [], {}):
+        return False
+    if isinstance(value, dict):
+        status = str(value.get('status') or value.get('execution_status') or '').strip().lower()
+        if status in {'blocked', 'missing', 'not_found', 'parameter_sensitive'}:
+            return False
+    return True
@@ -0,0 +1,127 @@
+"""Helpers moved unchanged out of render_pl9_markdown (phase 1)."""
+
+try:
+    from scripts.report_render.text import _md_cell
+except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution
+    from report_render.text import _md_cell
+
+def _render_boundary_packet_summary_line(packet: dict, summary_pairs: list[tuple[str, str]]) -> str:
+    summary = packet.get('summary') if isinstance(packet.get('summary'), dict) else {}
+    bits = [_md_cell(packet.get('packet_mode'))]
+    for key, suffix in summary_pairs:
+        value = summary.get(key)
+        if value not in (None, '', []):
+            bits.append(f"{_md_cell(value)}{suffix}")
+    return f"参考包摘要：{'；'.join(bits)}。"
+
+def _status_from_report_pack_section(section: dict) -> str:
+    if not isinstance(section, dict):
+        return 'blocked'
+    audit = section.get('audit') if isinstance(section.get('audit'), dict) else {}
+    if section.get('status'):
+        return str(section.get('status'))
+    if audit.get('status'):
+        return str(audit.get('status'))
+    if section:
+        return 'partial_verified'
+    return 'blocked'
+
+def _section_evidence_boundary(section: dict) -> str:
+    if not isinstance(section, dict):
+        return 'section_missing'
+    exports = section.get('exports') if isinstance(section.get('exports'), dict) else {}
+    ai_bundle = exports.get('ai_evidence_bundle') if isinstance(exports.get('ai_evidence_bundle'), dict) else {}
+    audit = section.get('audit') if isinstance(section.get('audit'), dict) else {}
+    boundaries = []
+    if ai_bundle.get('contains_private_pl9_text') is False:
+        boundaries.append('no_private_pl9_text')
+    if ai_bundle.get('claim_policy'):
+        boundaries.append(str(ai_bundle.get('claim_policy')))
+    if audit.get('must_not_claim'):
+        boundaries.append('must_not_claim=' + ','.join(map(str, audit.get('must_not_claim')[:3])))
+    if section.get('reason'):
+        boundaries.append(str(section.get('reason')))
+    return '; '.join(boundaries) if boundaries else 'evidence_labeled'
+
+def _domi_pl9_current_gap_appendix() -> list[str]:
+    return []
+
+def _chart_planets_from_positions(positions: dict) -> dict:
+    chart_planets = {}
+    if not isinstance(positions, dict):
+        return chart_planets
+    for planet_name in ('Sun', 'Moon', 'Mars', 'Mercury', 'Jupiter', 'Venus', 'Saturn', 'Rahu', 'Ketu'):
+        planet = positions.get(planet_name)
+        if not isinstance(planet, dict):
+            continue
+        sign = planet.get('sign') or planet.get('rashi') or planet.get('sign_name')
+        if not sign:
+            continue
+        degree = planet.get('degree_in_sign', planet.get('degree', planet.get('longitude_in_sign', 0)))
+        chart_planets[planet_name] = {'sign': sign, 'degree': degree or 0}
+    return chart_planets
+
+def _kp_house_focus_text(value) -> str:
+    house_label_map = {
+        2: '收入与资源',
+        5: '机会与投入',
+        6: '职责与压力',
+        7: '合作与伴侣',
+        9: '运气与扩张',
+        10: '事业与位置',
+        11: '回报与支持',
+    }
+    parts = []
+    if isinstance(value, (list, tuple)):
+        raw_parts = [str(item).strip() for item in value if str(item).strip()]
+    else:
+        raw_parts = [part.strip() for part in str(value or '').replace('/', ',').split(',') if part.strip()]
+    for part in raw_parts:
+        try:
+            house_no = int(part)
+        except ValueError:
+            parts.append(part)
+            continue
+        parts.append(f"{house_no}宫[{house_label_map.get(house_no, '-')}]")
+    return '、'.join(parts) if parts else '-'
+
+def _high_value_theme_authority_sentence(domain: str) -> str | None:
+    domain_label_map = {
+        'career': '事业',
+        'wealth': '财务',
+        'relationship': '关系',
+    }
+    action_map = {
+        'career': '优先排序事业推进顺序、外部舞台和月份节奏',
+        'wealth': '优先排序资源承接、现金流节奏和收益兑现顺序',
+        'relationship': '优先排序合作推进、联结温度和关系节奏',
+    }
+    return (
+        "主结论加厚：Narayana 第二时间轴、力量兑现结构、年度 highlights / field briefs、Patyayini 年内换挡、Graha Padas 这几层现在直接参与"
+        + f"{domain_label_map.get(domain, domain)}章主结论，"
+        + f"用来{action_map.get(domain, '补强主题排序')}；"
+        + "它们仍保留 parameter_sensitive / unclosed_divisional_chart 边界，但不再只停留在附录或登记表。"
+    )
+
+def _kp_conflict_adjudication_sentence(domain: str) -> str:
+    domain_label_map = {
+        'career': '事业',
+        'wealth': '财务',
+        'relationship': '关系',
+    }
+    focus_map = {
+        'career': '外部舞台、职位变化和项目推进',
+        'wealth': '资源承接、现金流和回收节奏',
+        'relationship': '合作推进、联结温度和承诺节奏',
+    }
+    return (
+        f"如果 Vimshottari、Narayana 与年度专题同向，而 KP 月度支持也落在同一主题上，就把这个月份优先当作{domain_label_map.get(domain, domain)}线的可执行窗口。"
+        f"如果 KP 单独活跃、但本命结构或双大运没有接力，KP 只负责提示 {focus_map.get(domain, '主题变化')} 的月份排序，不改写主题事实。"
+        f"如果 KP 与年度专题或双大运出现冲突，先以本命结构 + Vimshottari + Narayana 定主轴，KP 退回到月度提醒与核对层，继续保持 parameter_sensitive。"
+    )
+
+def _prashna_exact_cusp_boundary_sentence() -> str | None:
+    return (
+        "如果下一步要把月份继续缩到更细日期，仍需要补 question_time / place / timezone / question category；"
+        "Prashna ruling planets 与 exact cusp oracle 在这份本命个人报告里继续保持 blocked，只作为后续缩窄入口，不冒充 timing truth。"
+    )
@@ -0,0 +1,255 @@
+"""Helpers moved unchanged out of render_pl9_markdown (phase 1)."""
+
+import re
+
+try:
+    from scripts.report_render.envelope import _unwrap
+except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution
+    from report_render.envelope import _unwrap
+
+SIGNS = ['Aries', 'Taurus', 'Gemini', 'Cancer', 'Leo', 'Virgo', 'Libra', 'Scorpio', 'Sagittarius', 'Capricorn', 'Aquarius', 'Pisces']
+SIGNS_CN = {'Aries': '白羊座', 'Taurus': '金牛座', 'Gemini': '双子座', 'Cancer': '巨蟹座', 'Leo': '狮子座', 'Virgo': '处女座', 'Libra': '天秤座', 'Scorpio': '天蝎座', 'Sagittarius': '射手座', 'Capricorn': '摩羯座', 'Aquarius': '水瓶座', 'Pisces': '双鱼座'}
+
+PLANET_LABEL_CN = {
+    'Sun': '太阳',
+    'Moon': '月亮',
+    'Mars': '火星',
+    'Mercury': '水星',
+    'Jupiter': '木星',
+    'Venus': '金星',
+    'Saturn': '土星',
+    'Rahu': '北交点',
+    'Ketu': '南交点',
+}
+
+def _humanize_reader_token(token: str) -> str:
+    value = str(token or '').strip()
+    if not value:
+        return value
+    if value in SIGNS_CN:
+        return SIGNS_CN[value]
+    if value in PLANET_LABEL_CN:
+        return PLANET_LABEL_CN[value]
+    return value
+
+def _humanize_reader_group(raw_group: str) -> str:
+    seen = set()
+    parts = []
+    for item in str(raw_group or '').split(','):
+        humanized = _humanize_reader_token(item)
+        if humanized and humanized not in seen:
+            seen.add(humanized)
+            parts.append(humanized)
+    return ' / '.join(parts)
+
+def _humanize_tajika_candidate(name: str) -> str:
+    return {
+        'Ithasala_candidate': '趋近相位候选',
+        'Easarapha_candidate': '分离相位候选',
+    }.get(str(name or ''), '年度相位候选')
+
+def _humanize_tajika_motion(motion: str) -> str:
+    return {'applying': '趋近', 'separating': '分离'}.get(str(motion or ''), str(motion or '-'))
+
+def _display_rashi_from_value(raw_value):
+    if isinstance(raw_value, int) and 0 <= raw_value < len(SIGNS):
+        return _humanize_reader_token(SIGNS[raw_value])
+    return _humanize_reader_token(raw_value)
+
+def _humanize_reader_summary_line(summary_line: str) -> str:
+    summary_line = _unwrap(summary_line)
+    if isinstance(summary_line, dict):
+        summary_line = (
+            summary_line.get('brief')
+            or summary_line.get('text')
+            or summary_line.get('summary')
+            or ''
+        )
+    text = str(summary_line or '').strip()
+    if not text:
+        return text
+    if text == "年度层已经有可读壳层，但仍需要带着冲突标签阅读。":
+        return "这一年的年度主线已经形成。"
+    if text == "This pack converts Dasha trigger facts into report-ready interpretation seeds without final event claims.":
+        return "这层已经把 Dasha 触发事实压成可读提示，但仍不直接升级为具体事件断语。"
+    if text == "Source: dasha_master_pack.interpretation_triggers.":
+        return "当前解释仍然直接来自 dasha_master_pack.interpretation_triggers 的证据层。"
+    if text == "All claims remain evidence-labeled and parameter-sensitive where the source pack is unclosed.":
+        return "只要 source pack 还没闭环，相关结论就继续保持 evidence-labeled 与 parameter_sensitive。"
+    if text == "Tajika Yogas: annual candidate structures visible":
+        return "Tajika Yogas 当前已经能读到年度候选交互结构。"
+    if text == "Sahams: annual sensitive points visible":
+        return "Sahams 当前已经能读到年度敏感点。"
+    match = re.match(r'^([^:]+):\s*conflict between\s+(.+?)\s+vs\s+(.+)$', text)
+    if match:
+        topic = match.group(1).strip()
+        left = _humanize_reader_group(match.group(2))
+        right = _humanize_reader_group(match.group(3))
+        topic_label = {
+            'Muntha': 'Muntha',
+            'Year Lord': 'Year Lord',
+            'Annual Asc': '返照上升',
+        }.get(topic, topic)
+        if left == right:
+            return f"{topic_label} 当前结果指向{left}。"
+        return f"{topic_label} 当前存在候选分歧：一组结果指向{left}，另一组结果指向{right}。"
+    simple_match = re.match(r'^([^:]+):\s*(.+)$', text)
+    if simple_match:
+        topic = simple_match.group(1).strip()
+        value = _humanize_reader_group(simple_match.group(2))
+        if topic == 'Year Lord' and value:
+            return f"Year Lord 当前优先候选指向{value}。"
+        if topic == 'Muntha' and value:
+            return f"Muntha 当前优先候选指向{value}。"
+        if topic == 'Tajika Yogas' and 'annual candidate structures visible' in str(simple_match.group(2)).lower():
+            return "Tajika Yogas 当前已经能读到年度候选交互结构。"
+        if topic == 'Sahams' and 'annual sensitive points visible' in str(simple_match.group(2)).lower():
+            return "Sahams 当前已经能读到年度敏感点。"
+    return text
+
+def _humanize_annual_highlight_line(highlight: str) -> str:
+    text = str(highlight or '').strip()
+    if not text:
+        return text
+    if text == "Solar Return / Tajika annual shell is available.":
+        return "本年度返照与 Tajika 时间结构已经生成。"
+    if text == "Muntha and Year Lord are visible, but Year Lord remains producer-disagree sensitive.":
+        return "Muntha 和 Year Lord 现在都已经可见，但 Year Lord 仍要带着分歧标签阅读。"
+    if text == "The report should read annual structure first, then the evidence appendix.":
+        return "这一层适合从年度结构展开，再回头核对证据附录。"
+    return text
+
+def _humanize_annual_narrative_highlight(highlight: str) -> str:
+    return _humanize_annual_highlight_line(str(highlight)).rstrip('。')
+
+def _humanize_domain_title(title: str) -> str:
+    text = str(title or '').strip()
+    if not text:
+        return text
+    mapping = {
+        "Career and public direction": "事业与公众方向",
+        "Wealth and growth opportunities": "财富与增长机会",
+        "Relationship and alliance themes": "关系与合作主题",
+        "Health and vitality caution": "健康与精力提醒",
+        "Risk, delay, and friction markers": "阻力、延迟与摩擦提示",
+        "Timing windows and cross-check needs": "时间窗口与交叉核对重点",
+    }
+    return mapping.get(text, text)
+
+def _workbook_readable_label(worksheet_id: str) -> str:
+    if worksheet_id == 'birth_and_parameters':
+        return '出生与计算口径'
+    if worksheet_id == 'd1_rasi_bhava':
+        return '本命盘主结构'
+    if worksheet_id == 'strengths_and_scores':
+        return '力量与技术表'
+    if worksheet_id == 'divisional_and_special_charts':
+        return '分盘与特殊点位'
+    if worksheet_id == 'timing_and_predictive_systems':
+        return '时间系统与流年层'
+    if worksheet_id == 'advanced_systems':
+        return '进阶体系'
+    if worksheet_id == 'ai_and_audit':
+        return '审计与证据附录'
+    return '补充工作簿'
+
+def _workbook_readable_sentence(worksheet_id: str) -> str:
+    if worksheet_id == 'birth_and_parameters':
+        return '出生参数、岁差、交点模式与计算口径已经固定，是整份报告的可信底座。'
+    if worksheet_id == 'd1_rasi_bhava':
+        return '本命盘主结构已经就位，适合先看性格主轴、房屋结构和基础趋势。'
+    if worksheet_id == 'strengths_and_scores':
+        return '力量、功能性吉凶与技术表可以直接用来判断支撑与压力来源。'
+    if worksheet_id == 'divisional_and_special_charts':
+        return '分盘与特殊点位可作为扩展阅读，适合补充看细节。'
+    if worksheet_id == 'timing_and_predictive_systems':
+        return '大运、行运与年度层已经可以形成时间主线。'
+    if worksheet_id == 'advanced_systems':
+        return 'Jaimini、KP、Yoga 等进阶体系可作为补充主轴。'
+    if worksheet_id == 'ai_and_audit':
+        return 'AI 读写、审计与证据层用于核对来源与边界，不直接替代正文。'
+    return f'{worksheet_id} 已纳入当前报告。'
+
+def _workbook_reading_paragraph(worksheet_ids: list[str]) -> str:
+    parts = []
+    for worksheet_id in worksheet_ids:
+        sentence = _workbook_readable_sentence(worksheet_id).rstrip('。')
+        parts.append(f"先看{_workbook_readable_label(worksheet_id)}，这里会告诉你{sentence}")
+    if not parts:
+        return '当前暂无可用工作簿阅读顺序。'
+    return '；'.join(parts) + '。'
+
+def _md_cell(value) -> str:
+    text = str(value if value not in (None, '') else '-').replace('\n', ' ').strip()
+    return text.replace('|', '/')
+
+def _render_compact_payload(value, depth: int = 0) -> str:
+    if value in (None, '', [], {}):
+        return '-'
+    if depth >= 2:
+        return _md_cell(type(value).__name__)
+    if isinstance(value, dict):
+        parts = []
+        for key in sorted(value, key=str):
+            if key in {
+                'narrative_items',
+                'domain_paragraphs',
+                'response_envelope',
+                'input_hash',
+                'profile_hash',
+                'raw_calculation',
+                'shared_full_report_authority',
+            }:
+                continue
+            parts.append(f"{key}={_render_compact_payload(value.get(key), depth + 1)}")
+        return '; '.join(parts) if parts else '-'
+    if isinstance(value, (list, tuple)):
+        items = [_render_compact_payload(item, depth + 1) for item in list(value)[:8]]
+        suffix = ' …' if len(value) > 8 else ''
+        return ', '.join(items) + suffix if items else '-'
+    return _md_cell(value)
+
+def _sign_text(row: dict, sign_key: str = 'sign', cn_key: str = 'sign_cn') -> str:
+    if not isinstance(row, dict):
+        return '-'
+    sign = row.get(cn_key) or _humanize_reader_token(row.get(sign_key)) or row.get(sign_key)
+    return _md_cell(sign)
+
+def _value_text(*values):
+    for value in values:
+        if value not in (None, '', [], {}):
+            return str(value)
+    return None
+
+def _compose_narrative_paragraph(*parts: str | None) -> str | None:
+    cleaned = []
+    for part in parts:
+        text = str(part or '').strip()
+        if not text:
+            continue
+        text = text.rstrip()
+        if text.endswith('。'):
+            text = text[:-1]
+        cleaned.append(text)
+    if not cleaned:
+        return None
+    return '；'.join(cleaned) + '。'
+
+def _planet_list_text(items: list[str], limit: int = 3) -> str | None:
+    names = []
+    seen = set()
+    for item in items:
+        humanized = _humanize_reader_token(item)
+        if humanized and humanized not in seen:
+            seen.add(humanized)
+            names.append(humanized)
+        if len(names) >= limit:
+            break
+    if not names:
+        return None
+    return '、'.join(names)
+
+def _append_unique_text(items: list[str], value: str) -> None:
+    text = str(value or '').strip().rstrip('。')
+    if text and text not in items:
+        items.append(text)
@@ -0,0 +1,367 @@
+#!/usr/bin/env python3
+"""Byte-compare phase-1 report rendering against a baseline worktree.
+
+The current tree builds real-engine packets and pickles them outside the repo.
+A baseline checkout renders those same packets. Baseline-vs-baseline runs first.
+"""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import hashlib
+import json
+import os
+import pickle
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+
+
+ROOT = Path(__file__).resolve().parents[2]
+PRODUCTION_BODY = {
+    "format": "markdown",
+    "edition": "full_data",
+    "include_fact_tables": True,
+    "languages": ["zh", "en"],
+}
+FOURTH = {
+    "name": "虚构丁",
+    "year": 1990,
+    "month": 1,
+    "day": 1,
+    "hour": 12,
+    "minute": 0,
+    "second": 0,
+    "lat": 28.6139,
+    "lon": 77.2090,
+    "tz": 5.5,
+    "node_mode": "mean",
+    "ayanamsa": "lahiri",
+    "today": "2026-10-13",
+    "transit_date": "2026-10-13",
+    "target_year": 2026,
+    "age": 36,
+    "birth_time_accuracy": "confirmed",
+}
+
+
+def _ensure_hashseed() -> None:
+    if os.environ.get("PYTHONHASHSEED") == "0":
+        return
+    env = dict(os.environ)
+    env["PYTHONHASHSEED"] = "0"
+    raise SystemExit(subprocess.call([sys.executable, *sys.argv], env=env))
+
+
+def _activate(root: Path) -> None:
+    root = root.resolve()
+    script_dir = str(Path(__file__).resolve().parent)
+    while script_dir in sys.path:
+        sys.path.remove(script_dir)
+    folders = [str(root / "jyotish_vedic"), str(root), str(root / "scripts")]
+    for folder in folders:
+        if not Path(folder).exists():
+            continue
+        while folder in sys.path:
+            sys.path.remove(folder)
+        sys.path.insert(0, folder)
+
+
+def _mask_pyjhora() -> None:
+    import importlib
+
+    root = Path(tempfile.mkdtemp(prefix="nojhora-"))
+    package = root / "jhora"
+    package.mkdir()
+    (package / "__init__.py").write_text('raise ImportError("masked")\n', encoding="utf-8")
+    sys.path.insert(0, str(root))
+    prior = os.environ.get("PYTHONPATH", "")
+    os.environ["PYTHONPATH"] = str(root) + (os.pathsep + prior if prior else "")
+    for name in list(sys.modules):
+        if name == "jhora" or name.startswith("jhora."):
+            del sys.modules[name]
+    importlib.invalidate_caches()
+
+
+def _sha(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def _freeze_clocks() -> None:
+    import datetime as dt_mod
+
+    real = dt_mod.datetime
+    fixed_utc = real(2026, 10, 13, 0, 0, 0)
+    fixed_local = real(2026, 10, 13, 8, 0, 0)
+
+    class Frozen(real):
+        @classmethod
+        def now(cls, tz=None):
+            if tz is not None:
+                return fixed_utc.replace(tzinfo=tz)
+            return fixed_local
+
+        @classmethod
+        def utcnow(cls):
+            return fixed_utc
+
+    for module in list(sys.modules.values()):
+        if module is None:
+            continue
+        current = getattr(module, "datetime", None)
+        if isinstance(current, type) and current.__name__ == "datetime":
+            module.datetime = Frozen
+
+
+def _cases() -> list[tuple[str, dict]]:
+    from tests.test_report_english_edition import CASES
+
+    ordered = [(name, dict(CASES[name])) for name in ("day", "night", "south")]
+    ordered.append(("fourth", dict(FOURTH)))
+    return ordered
+
+
+def _subset_pack_id(packet: dict) -> str | None:
+    manifest = packet.get("report_pack_manifest") if isinstance(packet, dict) else None
+    packs = manifest.get("packs") if isinstance(manifest, dict) else None
+    if not isinstance(packs, list):
+        return None
+    for item in packs:
+        if isinstance(item, dict) and item.get("id") and item.get("id") != "full":
+            return str(item["id"])
+    return None
+
+
+def _build_records() -> list[dict]:
+    from scripts.jyotish_engine import (
+        _apply_pl9_pack_selection,
+        build_professional_report_reference_packet,
+        cmd_full_reading,
+    )
+    from scripts.professional_report_reference import _export_args
+
+    records = []
+    for name, case in _cases():
+        print(f"build {name}", file=sys.stderr, flush=True)
+        reading = cmd_full_reading(SimpleNamespace(**case))
+        packet = build_professional_report_reference_packet(reading, _export_args(case), [])
+        records.append({"id": f"{name}/reference", "kind": "pl9", "payload": packet})
+        stripped = copy.deepcopy(packet)
+        if isinstance(stripped, dict):
+            stripped.pop("personal_report_producer", None)
+        records.append({"id": f"{name}/no-producer", "kind": "pl9", "payload": stripped})
+        pack_id = _subset_pack_id(packet if isinstance(packet, dict) else {})
+        if pack_id:
+            selected = _apply_pl9_pack_selection(copy.deepcopy(packet), [pack_id])
+            records.append({"id": f"{name}/pack/{pack_id}", "kind": "pl9", "payload": selected})
+        else:
+            print(f"skip pack {name}: no non-full pack", file=sys.stderr)
+        records.append({"id": f"{name}/full-data-zh", "kind": "full", "payload": packet})
+        english = copy.deepcopy(packet)
+        if isinstance(english, dict):
+            english["report_language"] = "en"
+        records.append({"id": f"{name}/full-data-en", "kind": "full", "payload": english})
+        records.append({
+            "id": f"{name}/api-production",
+            "kind": "api",
+            "payload": {"birth": case, "reading": reading},
+        })
+    return records
+
+
+def _render_text(kind: str, payload) -> str:
+    if kind == "pl9":
+        from scripts.jyotish_engine import render_pl9_markdown
+
+        return render_pl9_markdown(copy.deepcopy(payload))
+    if kind == "full":
+        from scripts.pl9_full_data_export import render_full_data_markdown
+
+        return render_full_data_markdown(copy.deepcopy(payload))
+    if kind == "api":
+        from scripts.professional_report_reference import build_professional_report_reference
+
+        birth = payload["birth"]
+        reading = copy.deepcopy(payload["reading"])
+
+        class Handler:
+            def _high_rigor_birth_payload(self, _body):
+                return dict(birth)
+
+            def _compute_full_reading_for_thematic(self, _birth):
+                return reading
+
+        response = build_professional_report_reference(Handler(), dict(PRODUCTION_BODY))
+        return json.dumps(response, ensure_ascii=False, default=str)
+    raise RuntimeError(f"unknown kind {kind}")
+
+
+def _render_records(records: list[dict]) -> list[dict]:
+    import scripts.jyotish_engine  # noqa: F401
+    import scripts.pl9_full_data_export  # noqa: F401
+    import scripts.professional_report_reference  # noqa: F401
+
+    _freeze_clocks()
+    rendered = []
+    for record in records:
+        text = _render_text(record["kind"], record["payload"])
+        if not isinstance(text, str):
+            raise RuntimeError(f"{record['id']} did not render to text")
+        rendered.append({"id": record["id"], "sha": _sha(text), "text": text})
+    return rendered
+
+
+def _diff_context(left: str, right: str) -> str:
+    left_lines = left.splitlines() or [left]
+    right_lines = right.splitlines() or [right]
+    limit = min(len(left_lines), len(right_lines))
+    index = next((i for i in range(limit) if left_lines[i] != right_lines[i]), limit)
+    start = max(0, index - 2)
+    rows = []
+    for line_no in range(start, start + 5):
+        sample = left_lines[line_no] if line_no < len(left_lines) else ""
+        other = right_lines[line_no] if line_no < len(right_lines) else ""
+        if sample == other and line_no != index:
+            rows.append(f"  {line_no + 1}: {sample}")
+            continue
+        cut = _window(sample, other)
+        rows.append(f"- {line_no + 1}: {cut[0]}")
+        rows.append(f"+ {line_no + 1}: {cut[1]}")
+    return "\n".join(rows)
+
+
+def _window(left: str, right: str) -> tuple[str, str]:
+    if len(left) <= 240 and len(right) <= 240:
+        return left, right
+    limit = min(len(left), len(right))
+    index = next((i for i in range(limit) if left[i] != right[i]), limit)
+    start = max(0, index - 80)
+    stop = index + 80
+    return left[start:stop], right[start:stop]
+
+
+def _invoke_baseline(baseline: Path, pickle_path: Path, result_path: Path, mask: bool) -> subprocess.CompletedProcess[str]:
+    command = [
+        sys.executable,
+        str(Path(__file__).resolve()),
+        "--render-only",
+        "--pickle",
+        str(pickle_path),
+        "--engine-root",
+        str(baseline),
+        "--result",
+        str(result_path),
+    ]
+    if mask:
+        command.append("--mask-pyjhora")
+    env = dict(os.environ)
+    env["PYTHONHASHSEED"] = "0"
+    env["JYOTISH_REPORT_REFERENCE_CACHE"] = "0"
+    env["JYOTISH_REPORT_RENDER_FAST"] = "0"
+    return subprocess.run(
+        command,
+        cwd=str(baseline),
+        env=env,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def _load_result(path: Path) -> list[dict]:
+    return pickle.loads(path.read_bytes())
+
+
+def _print_shas(rows: list[dict]) -> None:
+    for row in rows:
+        print(f"{row['id']}\t{row['sha']}")
+
+
+def _compare(label: str, left: list[dict], right: list[dict]) -> int:
+    if [row["id"] for row in left] != [row["id"] for row in right]:
+        print(f"{label}: record ids differ", file=sys.stderr)
+        return 1
+    failed = 0
+    for lhs, rhs in zip(left, right):
+        if lhs["sha"] == rhs["sha"]:
+            continue
+        failed += 1
+        print(f"{label} mismatch {lhs['id']} {lhs['sha']} vs {rhs['sha']}", file=sys.stderr)
+        print(_diff_context(lhs["text"], rhs["text"]), file=sys.stderr)
+    return 1 if failed else 0
+
+
+def _render_only(args: argparse.Namespace) -> int:
+    if args.mask_pyjhora:
+        _mask_pyjhora()
+    _activate(Path(args.engine_root))
+    os.environ["JYOTISH_REPORT_REFERENCE_CACHE"] = "0"
+    os.environ["JYOTISH_REPORT_RENDER_FAST"] = "0"
+    records = pickle.loads(Path(args.pickle).read_bytes())
+    rendered = _render_records(records)
+    Path(args.result).write_bytes(pickle.dumps(rendered, protocol=pickle.HIGHEST_PROTOCOL))
+    return 0
+
+
+def main() -> int:
+    _ensure_hashseed()
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--baseline", type=Path)
+    parser.add_argument("--mask-pyjhora", action="store_true")
+    parser.add_argument("--render-only", action="store_true")
+    parser.add_argument("--pickle", type=Path)
+    parser.add_argument("--engine-root", type=Path)
+    parser.add_argument("--result", type=Path)
+    args = parser.parse_args()
+    if args.render_only:
+        return _render_only(args)
+    if args.baseline is None:
+        print("missing --baseline", file=sys.stderr)
+        return 2
+    if args.mask_pyjhora:
+        _mask_pyjhora()
+    _activate(ROOT)
+    os.environ["JYOTISH_REPORT_REFERENCE_CACHE"] = "0"
+    os.environ["JYOTISH_REPORT_RENDER_FAST"] = "0"
+    records = _build_records()
+    temp_dir = Path(tempfile.mkdtemp(prefix="pl9-render-split-"))
+    try:
+        pickle_path = temp_dir / "packets.pkl"
+        pickle_path.write_bytes(pickle.dumps(records, protocol=pickle.HIGHEST_PROTOCOL))
+        first_path = temp_dir / "baseline-a.pkl"
+        second_path = temp_dir / "baseline-b.pkl"
+        first = _invoke_baseline(args.baseline, pickle_path, first_path, args.mask_pyjhora)
+        second = _invoke_baseline(args.baseline, pickle_path, second_path, args.mask_pyjhora)
+        if first.returncode != 0 or second.returncode != 0:
+            print(first.stderr[-4000:], file=sys.stderr)
+            print(second.stderr[-4000:], file=sys.stderr)
+            print(first.stdout[-1000:], file=sys.stderr)
+            return first.returncode or second.returncode or 1
+        baseline_a = _load_result(first_path)
+        baseline_b = _load_result(second_path)
+        if _compare("baseline-vs-baseline", baseline_a, baseline_b) != 0:
+            print("基线对基线两次 sha 不同，存在非确定性，停止", file=sys.stderr)
+            return 2
+        current = _render_records(pickle.loads(pickle_path.read_bytes()))
+        if _compare("current-vs-baseline", current, baseline_a) != 0:
+            return 1
+        _print_shas(current)
+        print(f"ok items={len(current)} masked={int(args.mask_pyjhora)}")
+        return 0
+    finally:
+        for child in temp_dir.glob("*"):
+            try:
+                child.unlink()
+            except OSError:
+                pass
+        try:
+            temp_dir.rmdir()
+        except OSError:
+            pass
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -79,6 +79,8 @@ CORE_PYTEST_TARGETS = [
    "tests/test_birth_time_journey_contract.py",
    # Freeze scripts/jyotish_api_server.py growth; new features must be modules.
    "tests/test_api_server_growth_contract.py",
+    # Silent-except and scripts-root budgets. New research scripts go under scripts/audit or scripts/research.
+    "tests/test_python_hygiene_budget.py",
    # Pins the two closed __new__ forgeries: MCP/report scripts must keep reaching the
    # compute mixins without constructing an HTTP handler (TASK-api-server-backdoor-close).
    "tests/test_offline_compute_mixins.py",
@@ -18,6 +18,7 @@ from typing import Dict, Optional, Tuple, List
 from datetime import datetime, timedelta, timezone
 from zoneinfo import ZoneInfo
 import math
+import logging
 import sys
 import os

@@ -30,6 +31,8 @@ try:
 except ImportError:
    HAS_SWE = False

+logger = logging.getLogger(__name__)
+

 def _gregorian_calendar_flag() -> int:
    """Return the Gregorian calendar flag across swisseph builds."""
@@ -140,7 +143,9 @@ def _get_sun_lon_jd(jd_ut: float, ayanamsa_name: str = 'lahiri') -> Optional[flo
    try:
        res = swe.calc_ut(jd_ut, swe.SUN, sidereal_flags(swe, ayanamsa_name))
        return float(res[0][0] % 360.0)
-    except Exception:
+    except Exception as exc:
+        logger.warning("solar_return._get_sun_lon_jd degraded: %s", type(exc).__name__)
+        logger.debug("solar_return._get_sun_lon_jd degraded", exc_info=True)
        return None


@@ -22,6 +22,9 @@ import os
 import shutil
 import subprocess
 import sys
+import logging
+
+logger = logging.getLogger(__name__)

 SIGNS = ['Aries','Taurus','Gemini','Cancer','Leo','Virgo',
         'Libra','Scorpio','Sagittarius','Capricorn','Aquarius','Pisces']
@@ -1324,8 +1327,9 @@ def _year_lord_is_daytime(
        try:
            from saham_daynight import determine_daytime
            return bool(determine_daytime(annual_dt, lat=float(lat), lon=float(lon), tz=float(tz))['is_daytime'])
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.warning("tajika._year_lord_is_daytime degraded: %s", type(exc).__name__)
+            logger.debug("tajika._year_lord_is_daytime degraded", exc_info=True)
    if annual_dt is None:
        annual_dt = datetime.utcnow()
    return _is_daytime(annual_dt, sun_lon, asc_lon)
@@ -1548,7 +1552,7 @@ def _lookup_external_pyjhora_python() -> Optional[str]:
                text=True,
                timeout=8,
            )
-        except Exception:
+        except Exception:  # noqa: BLE001 - probing an optional external interpreter; a missing binary is expected
            continue
        if completed.returncode == 0:
            return candidate
@@ -11,10 +11,13 @@ Transit精确触发搜索 (v6.9.0)
 """
 from datetime import datetime, timedelta
 from typing import Dict, List, Optional, Tuple
+import logging
 import math

 from ayanamsa_utils import ayanamsa_display_name, normalize_ayanamsa_name, sidereal_flags

+logger = logging.getLogger(__name__)
+
 # 行星每日运动速度（°/天）- 用于步长优化
 PLANET_SPEED = {
    'Sun': 0.9856, 'Moon': 13.176, 'Mars': 0.524, 'Mercury': 1.383,
@@ -88,8 +91,9 @@ def _get_transit_lon_precise(
        lon = _get_planet_lon_swe(planet, _datetime_to_jd(dt), ayanamsa_name=ayanamsa)
        if lon is not None:
            return lon, f'swiss_ephemeris_{ayanamsa}'
-    except Exception:
-        pass
+    except Exception as exc:
+        logger.warning("transit_trigger._get_transit_lon_precise degraded: %s", type(exc).__name__)
+        logger.debug("transit_trigger._get_transit_lon_precise degraded", exc_info=True)
    return _get_transit_lon(planet, base_date, (dt - base_date).total_seconds() / 86400.0), 'mean_speed_fallback'


@@ -35,7 +35,7 @@ def _sdk_version() -> str | None:
    try:
        import importlib.metadata
        return importlib.metadata.version("vedastro")
-    except Exception:
+    except Exception:  # noqa: BLE001 - the vedastro package version is optional metadata; a missing install returns None
        return None


@@ -426,7 +426,7 @@ def _serialize_runtime_value(value: Any) -> Any:
                except json.JSONDecodeError:
                    return payload
            return _serialize_runtime_value(payload)
-        except Exception:
+        except Exception:  # noqa: BLE001 - to_json is an optional serializer; failure falls through to __dict__ or str
            pass
    if hasattr(value, "__dict__"):
        return {key: _serialize_runtime_value(item) for key, item in vars(value).items() if not key.startswith("_")}
@@ -1931,36 +1931,57 @@ def _store_cached_payload(cache_key: str, payload: dict[str, Any]) -> dict[str,
    return payload_copy


-def _acquire_free_tier_slot(request_url: str) -> dict[str, Any]:
-    if not _is_official_public_endpoint(request_url):
-        return {
-            "mode": "non_public_or_self_host_endpoint",
-            "queue_active": False,
-            "waited_seconds": 0.0,
-            "window_seconds": _free_tier_window_seconds(),
-            "max_requests": _free_tier_max_requests(),
-        }
-    if os.environ.get("VEDASTRO_API_KEY", "").strip():
-        return {
-            "mode": "api_key_present",
-            "queue_active": False,
-            "waited_seconds": 0.0,
-            "window_seconds": _free_tier_window_seconds(),
-            "max_requests": _free_tier_max_requests(),
-        }
+_OFFICIAL_PUBLIC_CALCULATE_URL = "https://api.vedastro.org/api/Calculate"
+
+
+def free_tier_throttle_state(request_url: str | None = None) -> dict[str, Any]:
+    """Whether the free-tier throttle would apply, without waiting or taking a slot.
+
+    The gates match ``_acquire_free_tier_slot`` before its wait loop: official
+    public host, no API key, and a positive request budget. ``request_url``
+    defaults to the official public Calculate endpoint.
+    """
+    if request_url is None:
+        request_url = _OFFICIAL_PUBLIC_CALCULATE_URL
    max_requests = _free_tier_max_requests()
    window_seconds = _free_tier_window_seconds()
-    if max_requests <= 0 or window_seconds <= 0:
+    wait_budget_seconds = _free_tier_wait_budget_seconds()
+    if not _is_official_public_endpoint(request_url):
+        reason = "non_public_or_self_host_endpoint"
+        active = False
+    elif os.environ.get("VEDASTRO_API_KEY", "").strip():
+        reason = "api_key_present"
+        active = False
+    elif max_requests <= 0 or window_seconds <= 0:
+        reason = "free_tier_queue_disabled"
+        active = False
+    else:
+        reason = "free_tier_queue_active"
+        active = True
+    return {
+        "active": active,
+        "reason": reason,
+        "max_requests": max_requests,
+        "window_seconds": window_seconds,
+        "wait_budget_seconds": wait_budget_seconds,
+    }
+
+
+def _acquire_free_tier_slot(request_url: str) -> dict[str, Any]:
+    state = free_tier_throttle_state(request_url)
+    if not state["active"]:
        return {
-            "mode": "free_tier_queue_disabled",
+            "mode": state["reason"],
            "queue_active": False,
            "waited_seconds": 0.0,
-            "window_seconds": window_seconds,
-            "max_requests": max_requests,
+            "window_seconds": state["window_seconds"],
+            "max_requests": state["max_requests"],
        }
+    max_requests = state["max_requests"]
+    window_seconds = state["window_seconds"]

    waited_seconds = 0.0
-    budget = _free_tier_wait_budget_seconds()
+    budget = state["wait_budget_seconds"]
    with _FREE_TIER_REQUEST_LOCK:
        while True:
            now = time.monotonic()
@@ -220,11 +220,11 @@ def _strict_workflow_summary(route: str, catalog: dict[str, Any]) -> dict[str, A


 def _cache_and_queue_report() -> dict[str, Any]:
-    queue_enabled = (
-        _bool_env("VEDASTRO_FREE_TIER_QUEUE")
-        or _bool_env("VEDASTRO_FREE_TIER_QUEUE_ENABLED")
-        or _bool_env("VEDASTRO_ENABLE_FREE_TIER_QUEUE")
-    )
+    try:
+        from scripts.vedastro_service_adapter import free_tier_throttle_state
+    except ModuleNotFoundError:
+        from vedastro_service_adapter import free_tier_throttle_state
+    queue_enabled = bool(free_tier_throttle_state()["active"])
    using_free_tier = not bool(os.environ.get("VEDASTRO_API_KEY", "").strip())
    return {
        "official_full_snapshot_cache_scope": "official_full_snapshot_semantic_cache",
@@ -683,7 +683,7 @@ def _compiled_exec_code(expr):
        tree.body = _capture_tail_expr(tree.body)
        ast.fix_missing_locations(tree)
        return compile(tree, "<yoga_custom>", "exec")
-    except Exception:
+    except Exception:  # noqa: BLE001 - a custom rule expression is expected to fail; that rule is skipped
        return None


@@ -700,7 +700,7 @@ def _run_custom_expr(expr, exec_globals):
            result = eval(code, exec_globals, exec_globals)
        except SyntaxError:
            code = None  # 运行期 SyntaxError：与改前一样退到多语句分支
-        except Exception:
+        except Exception:  # noqa: BLE001 - a custom rule expression is expected to fail; that rule is skipped
            return None
    if code is None:
        exec_code = _compiled_exec_code(expr)
@@ -708,7 +708,7 @@ def _run_custom_expr(expr, exec_globals):
            try:
                exec(exec_code, exec_globals, exec_globals)
                result = exec_globals.get("__result__")
-            except Exception:
+            except Exception:  # noqa: BLE001 - a custom rule expression is expected to fail; that rule is skipped
                return None
    return result

@@ -516,6 +516,9 @@ def test_high_rigor_vedastro_official_summary_passes_through_contract_fields(mon
    monkeypatch.delenv('VEDASTRO_FREE_TIER_QUEUE', raising=False)
    monkeypatch.delenv('VEDASTRO_FREE_TIER_QUEUE_ENABLED', raising=False)
    monkeypatch.delenv('VEDASTRO_ENABLE_FREE_TIER_QUEUE', raising=False)
+    monkeypatch.delenv('VEDASTRO_API_KEY', raising=False)
+    monkeypatch.delenv('VEDASTRO_FREE_TIER_MAX_REQUESTS', raising=False)
+    monkeypatch.delenv('VEDASTRO_FREE_TIER_WINDOW_SECONDS', raising=False)
    handler = _handler()

    chart = {
@@ -564,7 +567,10 @@ def test_high_rigor_vedastro_official_summary_passes_through_contract_fields(mon
    assert runtime_truth['fallback_active'] is True
    assert runtime_truth['blocked_items'] == ['official_event_radar_partial']
    assert runtime_truth['free_tier_strategy']['using_free_tier'] is True
-    assert runtime_truth['free_tier_strategy']['queue_enabled'] is False
+    # 原值：False
+    # 新值：True
+    # 原因：没设置已废弃的排队开关时，官方免费层默认仍在限流。报告改为跟随真实限流。
+    assert runtime_truth['free_tier_strategy']['queue_enabled'] is True
    assert runtime_truth['free_tier_strategy']['cache_hit'] is False
    assert runtime_truth['free_tier_strategy']['guard_status'] == 'degraded_or_partial'

@@ -0,0 +1,89 @@
+"""Functional benefic/malefic snapshot is one helper on three call paths."""
+
+from __future__ import annotations
+
+import builtins
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from functional_benefics import SIGNS, derive_functional_benefic_malefic  # noqa: E402
+from functional_layer import functional_layer_for_ascendant  # noqa: E402
+from jyotish_api_server import JyotishAPIHandler  # noqa: E402
+import jyotish_engine  # noqa: E402
+import mcp_server  # noqa: E402
+
+_BLOCKED_KEYS = (
+    "status",
+    "ascendant",
+    "functional_benefics",
+    "functional_malefics",
+    "functional_neutrals",
+    "yogakarakas",
+    "owned_houses",
+    "effect_on_confidence",
+    "source",
+)
+
+
+def _dump(payload: dict) -> str:
+    return json.dumps(payload, sort_keys=True)
+
+
+def _paths(ascendant):
+    handler = object.__new__(JyotishAPIHandler)
+    return (
+        jyotish_engine._functional_benefic_malefic_snapshot({}, ascendant),
+        handler._functional_benefic_malefic_snapshot({}, ascendant),
+        mcp_server._derive_functional_benefic_malefic({"chart": {"ascendant": ascendant}}),
+    )
+
+
+def test_twelve_signs_match_the_core_derivation() -> None:
+    for sign in SIGNS:
+        expected = _dump(derive_functional_benefic_malefic(sign))
+        for actual in _paths({"sign": sign}):
+            assert _dump(actual) == expected
+
+
+def test_missing_and_unknown_signs_agree_across_paths() -> None:
+    # 原值：MCP 在上升不是 dict 时返回 "Missing chart.ascendant; functional layer blocked."
+    # 新值：三条路径都等于 derive_functional_benefic_malefic 对缺失或非法星座的结果
+    # 原因：三份兜底合成一份。缺失上升走统一的正常路径，不再单独写英文句子。
+    cases = (None, {}, {"sign": "NotASign"})
+    for ascendant in cases:
+        sign = ascendant.get("sign") if isinstance(ascendant, dict) else None
+        expected = _dump(derive_functional_benefic_malefic(sign))
+        dumped = [_dump(item) for item in _paths(ascendant)]
+        assert dumped[0] == dumped[1] == dumped[2] == expected
+        assert "Missing chart.ascendant" not in dumped[0]
+
+
+def test_import_failure_returns_the_engine_blocked_shape(monkeypatch) -> None:
+    real_import = builtins.__import__
+
+    def boom(name, globals=None, locals=None, fromlist=(), level=0):  # noqa: A002
+        if name == "functional_benefics" or name.endswith(".functional_benefics"):
+            raise ImportError("functional_benefics unavailable")
+        return real_import(name, globals, locals, fromlist, level)
+
+    monkeypatch.setattr(builtins, "__import__", boom)
+    blocked = functional_layer_for_ascendant({"sign": "Aries"})
+    assert blocked["status"] == "blocked"
+    assert blocked["ascendant"] == "Aries"
+    assert blocked["functional_benefics"] == []
+    assert blocked["functional_malefics"] == []
+    assert blocked["functional_neutrals"] == []
+    assert blocked["yogakarakas"] == []
+    assert blocked["owned_houses"] == {}
+    assert blocked["source"] == "strict_functional_benefic_malefic_v2"
+    assert "未完成功能性吉凶星判定" in blocked["effect_on_confidence"]
+    for key in _BLOCKED_KEYS:
+        assert key in blocked
@@ -0,0 +1,53 @@
+"""Budgets that only allow the silent-except and scripts-root counts to fall."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+
+# Measured at the end of TASK-python-hygiene-20261013. New silent handlers
+# must log or carry ``# noqa: BLE001`` with a reason. New research scripts
+# belong in scripts/audit or scripts/research.
+SILENT_EXCEPT_BUDGET = 0
+ROOT_PY_BUDGET = 446
+
+
+def _silent_handlers() -> list[str]:
+    hits: list[str] = []
+    paths = list(SCRIPTS.glob("*.py")) + [ROOT / "mcp_server.py"]
+    for path in paths:
+        source = path.read_text(encoding="utf-8")
+        lines = source.splitlines()
+        tree = ast.parse(source)
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.ExceptHandler):
+                continue
+            typ = node.type
+            broad = typ is None or (isinstance(typ, ast.Name) and typ.id in {"Exception", "BaseException"})
+            if not broad or len(node.body) != 1:
+                continue
+            stmt = node.body[0]
+            silent = isinstance(stmt, (ast.Pass, ast.Continue)) or (
+                isinstance(stmt, ast.Return)
+                and (stmt.value is None or (isinstance(stmt.value, ast.Constant) and stmt.value.value is None))
+            )
+            if not silent:
+                continue
+            segment = "\n".join(lines[node.lineno - 1: node.end_lineno])
+            if "noqa: BLE001" in segment:
+                continue
+            hits.append(f"{path.relative_to(ROOT).as_posix()}:{node.lineno}")
+    return hits
+
+
+def test_silent_except_budget_does_not_grow() -> None:
+    hits = _silent_handlers()
+    assert len(hits) <= SILENT_EXCEPT_BUDGET, hits
+
+
+def test_scripts_root_py_count_does_not_grow() -> None:
+    count = len(list(SCRIPTS.glob("*.py")))
+    assert count <= ROOT_PY_BUDGET, count
@@ -0,0 +1,114 @@
+"""Phase-1 report_render package: import boundary and moved-helper parity."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ORIGINAL_PLANET_LABEL_CN = {
+    "Sun": "太阳",
+    "Moon": "月亮",
+    "Mars": "火星",
+    "Mercury": "水星",
+    "Jupiter": "木星",
+    "Venus": "金星",
+    "Saturn": "土星",
+    "Rahu": "北交点",
+    "Ketu": "南交点",
+}
+
+
+def test_report_render_imports_without_engine() -> None:
+    code = "\n".join([
+        "import sys",
+        "import scripts.report_render.envelope",
+        "import scripts.report_render.text",
+        "import scripts.report_render.sections_misc",
+        "banned = [name for name in sys.modules if name == 'jyotish_engine' or name.startswith('jyotish_engine.') or name == 'scripts.jyotish_engine' or name.startswith('scripts.jyotish_engine.')]",
+        "assert banned == []",
+    ])
+    env = dict(os.environ)
+    env["PYTHONPATH"] = str(ROOT)
+    env["PYTHONHASHSEED"] = "0"
+    completed = subprocess.run(
+        [sys.executable, "-c", code],
+        cwd=ROOT,
+        env=env,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    assert completed.returncode == 0, completed.stderr
+
+
+def test_planet_label_and_signs_match_engine_literals() -> None:
+    from scripts.jyotish_engine import SIGNS, SIGNS_CN
+    from scripts.report_render import text as moved
+
+    assert list(moved.PLANET_LABEL_CN) == list(ORIGINAL_PLANET_LABEL_CN)
+    for key, value in ORIGINAL_PLANET_LABEL_CN.items():
+        assert moved.PLANET_LABEL_CN[key] == value
+    assert moved.SIGNS == SIGNS
+    assert moved.SIGNS_CN == SIGNS_CN
+
+
+def _first(packet, kind: str):
+    stack = [packet]
+    while stack:
+        current = stack.pop()
+        if kind == "bool" and isinstance(current, bool):
+            return current
+        if kind == "str" and isinstance(current, str) and current.strip():
+            return current
+        if kind == "list" and isinstance(current, list) and current:
+            return current
+        if isinstance(current, dict):
+            stack.extend(current.values())
+        elif isinstance(current, (list, tuple)):
+            stack.extend(list(current))
+    raise AssertionError(kind)
+
+
+def test_five_moved_helpers_match_engine_on_real_packet() -> None:
+    from scripts.jyotish_engine import (
+        _bool_text,
+        _envelope_list,
+        _humanize_reader_token,
+        _md_cell,
+        _unwrap,
+        build_professional_report_reference_packet,
+        cmd_full_reading,
+        render_pl9_markdown,
+    )
+    from scripts.professional_report_reference import _export_args
+    from scripts.report_render.envelope import _bool_text as moved_bool
+    from scripts.report_render.envelope import _envelope_list as moved_list
+    from scripts.report_render.envelope import _unwrap as moved_unwrap
+    from scripts.report_render.text import _humanize_reader_token as moved_token
+    from scripts.report_render.text import _md_cell as moved_cell
+    from tests.test_report_reader_main import FICTIONAL
+
+    reading = cmd_full_reading(SimpleNamespace(**FICTIONAL))
+    packet = build_professional_report_reference_packet(reading, _export_args(dict(FICTIONAL)), [])
+    rendered = render_pl9_markdown(packet)
+    assert isinstance(rendered, str) and len(rendered) > 100
+    flag = _first(packet, "bool")
+    label = _first(packet, "str")
+    rows = _first(packet, "list")
+    birth = packet.get("birth_info")
+    assert _bool_text(flag) == moved_bool(flag)
+    assert _md_cell(label) == moved_cell(label)
+    assert _unwrap(birth) == moved_unwrap(birth)
+    assert _humanize_reader_token("Sun") == moved_token("Sun")
+    assert _humanize_reader_token("Aries") == moved_token("Aries")
+    assert _envelope_list(rows) == moved_list(rows)
+    assert _bool_text is moved_bool
+    assert _md_cell is moved_cell
+    assert _unwrap is moved_unwrap
+    assert _humanize_reader_token is moved_token
+    assert _envelope_list is moved_list
@@ -0,0 +1,89 @@
+"""Degraded calculations leave a warning that names the function, not the chart."""
+
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import daily_guidance_service  # noqa: E402
+import jyotish_engine  # noqa: E402
+import pl9_full_data_export  # noqa: E402
+
+_SENTINEL_DATE = "1990-01-01"
+_SENTINEL_LAT = "28.6139"
+_SENTINEL_LON = "77.2090"
+
+
+def _warnings(caplog):
+    return [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING]
+
+
+def test_sudarshana_failure_logs_the_function_not_the_chart(monkeypatch, caplog) -> None:
+    def boom(*_args, **_kwargs):
+        raise RuntimeError(f"{_SENTINEL_DATE} {_SENTINEL_LAT} {_SENTINEL_LON}")
+
+    monkeypatch.setattr(pl9_full_data_export, "_load_sudarshana", lambda: boom)
+    packet = {
+        "core_chart": {
+            "ascendant": {"longitude": 10.0, "sign": "Aries"},
+            "planets": {"Sun": {"longitude": 20.0, "sign": "Aries"}},
+        }
+    }
+    with caplog.at_level(logging.WARNING):
+        text = pl9_full_data_export._render_pl9_ai_density_raw_data_markdown_en(packet)
+    assert "Sudarshana" not in text
+    warnings = _warnings(caplog)
+    assert any(
+        "pl9_full_data_export._render_pl9_ai_density_raw_data_markdown_en degraded: RuntimeError" == item
+        for item in warnings
+    )
+    assert all(_SENTINEL_DATE not in item and _SENTINEL_LAT not in item and _SENTINEL_LON not in item for item in warnings)
+
+
+def test_current_mahadasha_failure_logs_the_function_not_the_chart(monkeypatch, caplog) -> None:
+    def boom(**_kwargs):
+        raise ValueError(f"{_SENTINEL_DATE} {_SENTINEL_LAT}")
+
+    monkeypatch.setattr(daily_guidance_service, "compute_vimshottari_timeline", boom)
+    with caplog.at_level(logging.WARNING):
+        result = daily_guidance_service._current_mahadasha(
+            {"birth_info": {}, "planets": {"Moon": {"lon": 10.0}}},
+            {"year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0},
+            "2026-01-01",
+        )
+    assert result is None
+    warnings = _warnings(caplog)
+    assert any(
+        "daily_guidance_service._current_mahadasha degraded: ValueError" == item
+        for item in warnings
+    )
+    assert all(_SENTINEL_DATE not in item and _SENTINEL_LAT not in item for item in warnings)
+
+
+def test_conflict_arbitration_failure_logs_the_function_not_the_chart(monkeypatch, caplog) -> None:
+    import ashtakavarga
+
+    def boom(*_args, **_kwargs):
+        raise RuntimeError(f"{_SENTINEL_DATE} {_SENTINEL_LON}")
+
+    monkeypatch.setattr(ashtakavarga, "calc_ashtakavarga", boom)
+    report = {
+        "planets": {},
+        "audit": {"P1_identity": {"asc_sign": "Aries"}},
+        "ashtakavarga": {"present": True},
+    }
+    with caplog.at_level(logging.WARNING):
+        conflicts = jyotish_engine._conflict_arbitration(report)
+    assert isinstance(conflicts, list)
+    warnings = _warnings(caplog)
+    assert any(
+        "jyotish_engine._conflict_arbitration degraded: RuntimeError" == item
+        for item in warnings
+    )
+    assert all(_SENTINEL_DATE not in item and _SENTINEL_LON not in item for item in warnings)
@@ -0,0 +1,98 @@
+"""BUG-1340: free-tier reports follow the real throttle, not the old env names."""
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import vedastro_service_adapter as adapter  # noqa: E402
+from diagnose_vedastro_mode import build_report  # noqa: E402
+from offline_compute_mixins import _free_tier_queue_enabled_env  # noqa: E402
+from vedastro_user_entrypoint import _cache_and_queue_report  # noqa: E402
+
+OFFICIAL = "https://api.vedastro.org/api/Calculate/x"
+DEPRECATED = (
+    "VEDASTRO_FREE_TIER_QUEUE",
+    "VEDASTRO_FREE_TIER_QUEUE_ENABLED",
+    "VEDASTRO_ENABLE_FREE_TIER_QUEUE",
+)
+_NAME = re.compile(
+    r"VEDASTRO_(?:ENABLE_FREE_TIER_QUEUE|FREE_TIER_QUEUE_ENABLED|FREE_TIER_QUEUE(?!_ENABLED))"
+)
+
+
+def _clear_throttle_env(monkeypatch) -> None:
+    for name in (
+        "VEDASTRO_API_KEY",
+        "VEDASTRO_FREE_TIER_MAX_REQUESTS",
+        "VEDASTRO_FREE_TIER_WINDOW_SECONDS",
+        *DEPRECATED,
+    ):
+        monkeypatch.delenv(name, raising=False)
+    adapter._FREE_TIER_REQUEST_TIMESTAMPS.clear()
+
+
+def test_unset_switch_still_reports_the_real_active_throttle(monkeypatch) -> None:
+    _clear_throttle_env(monkeypatch)
+    state = adapter.free_tier_throttle_state(OFFICIAL)
+    slot = adapter._acquire_free_tier_slot(OFFICIAL)
+    assert state["active"] is True
+    assert state["reason"] == "free_tier_queue_active"
+    assert slot["mode"] == "free_tier_queue_active"
+    assert _free_tier_queue_enabled_env() is (slot["mode"] == "free_tier_queue_active")
+    assert _cache_and_queue_report()["free_tier_queue_enabled"] is True
+
+
+def test_deprecated_switch_cannot_turn_a_zero_budget_into_active(monkeypatch) -> None:
+    _clear_throttle_env(monkeypatch)
+    monkeypatch.setenv("VEDASTRO_FREE_TIER_QUEUE", "1")
+    monkeypatch.setenv("VEDASTRO_FREE_TIER_MAX_REQUESTS", "0")
+    state = adapter.free_tier_throttle_state(OFFICIAL)
+    slot = adapter._acquire_free_tier_slot(OFFICIAL)
+    assert state["active"] is False
+    assert slot["mode"] == "free_tier_queue_disabled"
+    assert _free_tier_queue_enabled_env() is False
+    assert _cache_and_queue_report()["free_tier_strategy"]["queue_enabled"] is False
+
+
+def test_api_key_and_self_host_are_inactive(monkeypatch) -> None:
+    _clear_throttle_env(monkeypatch)
+    monkeypatch.setenv("VEDASTRO_API_KEY", "sample-key")
+    assert adapter.free_tier_throttle_state()["active"] is False
+    assert adapter._acquire_free_tier_slot(OFFICIAL)["mode"] == "api_key_present"
+
+    monkeypatch.delenv("VEDASTRO_API_KEY", raising=False)
+    local = "http://127.0.0.1:9/api/Calculate"
+    assert adapter.free_tier_throttle_state(local)["active"] is False
+    assert adapter._acquire_free_tier_slot(local)["mode"] == "non_public_or_self_host_endpoint"
+
+
+def test_diagnose_lists_deprecated_names_without_letting_them_control_the_flag(monkeypatch) -> None:
+    _clear_throttle_env(monkeypatch)
+    monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1")
+    monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api")
+    monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "0")
+    monkeypatch.setenv("VEDASTRO_FREE_TIER_QUEUE_ENABLED", "yes")
+    monkeypatch.setenv("VEDASTRO_FREE_TIER_MAX_REQUESTS", "0")
+    report = build_report()
+    assert report["free_tier_queue_enabled"] is False
+    assert report["deprecated_env_present"] == ["VEDASTRO_FREE_TIER_QUEUE_ENABLED"]
+
+
+def test_deprecated_env_names_remain_only_in_diagnose() -> None:
+    offenders: list[str] = []
+    paths = list(SCRIPTS.rglob("*.py")) + [ROOT / "mcp_server.py"]
+    for path in paths:
+        if path.name == "diagnose_vedastro_mode.py":
+            continue
+        text = path.read_text(encoding="utf-8")
+        if _NAME.search(text):
+            offenders.append(path.relative_to(ROOT).as_posix())
+    assert offenders == []
@@ -115,6 +115,9 @@ def test_user_entrypoint_runs_catalog_and_strict_workflow_contract() -> None:
            "VEDASTRO_TIMEOUT_SECONDS": "20",
            "VEDASTRO_CACHE_TTL_SECONDS": "600",
            "VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS": "600",
+            "VEDASTRO_API_KEY": "",
+            "VEDASTRO_FREE_TIER_MAX_REQUESTS": "5",
+            "VEDASTRO_FREE_TIER_WINDOW_SECONDS": "60",
            "VEDASTRO_FREE_TIER_QUEUE": "1",
            "VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT": "8",
            "VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB": json.dumps(catalog_stub),
@@ -134,10 +137,16 @@ def test_user_entrypoint_runs_catalog_and_strict_workflow_contract() -> None:
    assert "health" in report["official_capability_catalog"]["domain_routing"]
    assert report["cache_and_queue"]["official_full_snapshot_cache_ttl_seconds"] == 600
    assert report["cache_and_queue"]["range_scan_cache_ttl_seconds"] == 600
+    # 原值：True（因为设置了 VEDASTRO_FREE_TIER_QUEUE=1）
+    # 新值：True
+    # 原因：该字段跟随真实限流。本用例没有 API key、端点是官方地址、配额用默认正数，所以仍然开启。废弃开关不再决定它。
    assert report["cache_and_queue"]["free_tier_queue_enabled"] is True
    assert report["runtime_mode"]["free_tier_possible_with_cache_queue"] is True
    assert report["runtime_mode"]["readiness_blockers"] == ["premium_key_missing"]
    assert report["cache_and_queue"]["free_tier_strategy"]["using_free_tier"] is True
+    # 原值：True（因为设置了 VEDASTRO_FREE_TIER_QUEUE=1）
+    # 新值：True
+    # 原因：同上，跟随真实限流，不是跟随废弃开关。
    assert report["cache_and_queue"]["free_tier_strategy"]["queue_enabled"] is True
    assert report["cache_and_queue"]["free_tier_strategy"]["guard_status"] == "within_free_tier_strategy"
    assert report["strict_workflow"]["triggered"] is True
Author	SHA1	Message	Date
jesse-uxandGrok (agent loop)	a7162db271	让免费层限流的报告跟真实限流一致，并给少一块结果的静默失败留下函数名和错误类型。 Co-Authored-By: Grok (agent loop)	2026-10-15 23:12:56 +08:00
Claude	e4ff62c738	merge origin/staging into pl9 render split (issue #67 ) Independent Staging Quality Gate / validate (push) Successful in 14m16s Details Independent Staging Quality Gate / publish (push) Successful in 4m9s Details Docs-only catch-up; agent-loop rule A. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LutgzyJqNhPpZy1nFsJtN3	2026-10-15 21:23:23 +08:00
jesse-uxandGrok (agent loop)	dbba127e71	拆分报告渲染函数的第一阶段，把 34 个无外层依赖的辅助函数搬到 report_render。同一份真实盘包在屏蔽与不屏蔽 PyJHora 下新旧渲染 sha 都一致，各 24 项且退出码 0。指定 pytest 233 通过、3 跳过。 Co-Authored-By: Grok (agent loop)	2026-10-15 20:59:11 +08:00
				`@@ -0,0 +1 @@`
				`"""Helpers moved unchanged out of render_pl9_markdown (phase 1)."""`