feat(report): render longform Markdown as the report and close gaps2 holes

New reports skip the writer, persist pl9 Markdown as the body, and settle zero-token usage on the catalog model. Planned longform sections now emit blocked rows instead of vanishing. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-06 21:46:16 +08:00
parent e428e9d056
commit cfcd369d4f
36 changed files with 2290 additions and 402 deletions
@@ -4788,6 +4788,11 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
            'declared_window_end': body.get('declared_window_end'),
            'uncertainty_before_minutes': body.get('uncertainty_before_minutes'),
            'uncertainty_after_minutes': body.get('uncertainty_after_minutes'),
+            'birthplace_label': body.get('birthplace_label'),
+            'coordinate_source': body.get('coordinate_source'),
+            'coordinate_precision': body.get('coordinate_precision') or 'unverified_user_coordinates',
+            'time_source': body.get('time_source'),
+            'uncertainty_minutes': body.get('uncertainty_minutes'),
        }

    def _high_rigor_requested_themes(self, body):
@@ -1988,27 +1988,22 @@ def _attach_annual_tajika_pack(packet: dict, args) -> dict:
        target_year = int(target_year)
    except (TypeError, ValueError):
        return packet
-    def _annual_series_surface(annual_pack: dict) -> dict:
-        """Keep independently calculated annual evidence without duplicating full replay payloads."""
-        surface = {
-            key: _json_safe_report_snapshot(annual_pack.get(key))
-            for key in ('schema', 'status', 'reason', 'profile', 'audit', 'report_sections', 'annual_chart', 'year_lord')
-            if key in annual_pack
-        }
-        surface['status'] = surface.get('status') or 'partial_verified'
-        return surface

    annual_years = {str(target_year): _annual_series_surface(timing['annual_tajika_pack'])}
    for year in range(target_year + 1, target_year + 3):
        annual_args = SimpleNamespace(**vars(args))
        annual_args.target_year = year
-        # The primary year keeps its full PyJHora replay. Future years are
-        # independently calculated annual packets, but do not repeat the costly
-        # external replay during a single full-report export.
        try:
-            from annual_tajika_pack import build_annual_tajika_pack
+            birth_year = int(getattr(args, 'year'))
+            annual_args.age = year - birth_year
+        except (TypeError, ValueError):
+            pass
+        try:
            annual_years[str(year)] = _annual_series_surface(
-                build_annual_tajika_pack(_annual_payload_from_args(annual_args, packet.get('calculation_profile') or {}))
+                _build_pl9_full_annual_section(
+                    annual_args,
+                    packet.get('calculation_profile') or {},
+                )
            )
        except Exception as exc:
            annual_years[str(year)] = {
@@ -2323,8 +2318,26 @@ def _build_timing_boundary_attribution(packet: dict) -> dict:
    }


-def _build_module_execution_audit(packet: dict) -> list[dict]:
+def _iter_audit_modules(packet: dict):
    raw_modules = ((packet.get('raw_full_reading') or {}).get('modules') or {})
+    if isinstance(raw_modules, dict) and raw_modules:
+        yield from raw_modules.items()
+        return
+    worksheets = packet.get('worksheets') if isinstance(packet.get('worksheets'), dict) else {}
+    seen = set()
+    for sheet in worksheets.values():
+        if not isinstance(sheet, dict):
+            continue
+        for key, value in sheet.items():
+            if key in {'summary_card', 'detail_blocks'} or str(key).endswith('_pack'):
+                continue
+            if not isinstance(value, dict) or not value or key in seen:
+                continue
+            seen.add(key)
+            yield key, value
+
+
+def _build_module_execution_audit(packet: dict) -> list[dict]:
    usage_map = packet.get('raw_module_usage_map') if isinstance(packet.get('raw_module_usage_map'), dict) else {}
    profile_id = packet.get('calculation_profile_id') or 'calculation_profile_id_missing'
    input_requirements = {
@@ -2332,8 +2345,7 @@ def _build_module_execution_audit(packet: dict) -> list[dict]:
        'muhurta': 'activity_goal, candidate_datetime_or_range, location, timezone',
    }
    rows = []
-    for module_id in sorted(raw_modules):
-        module = raw_modules[module_id]
+    for module_id, module in _iter_audit_modules(packet):
        if module in (None, '', [], {}):
            continue
        module_data = module if isinstance(module, dict) else {}
@@ -2345,7 +2357,7 @@ def _build_module_execution_audit(packet: dict) -> list[dict]:
        rows.append({
            'module_id': str(module_id),
            'status': status,
-            'producer': module_data.get('source') or module_data.get('method') or 'jyotish_engine.cmd_full_reading',
+            'producer': module_data.get('source') or module_data.get('method') or 'cmd_full_reading',
            'input_profile': profile_id,
            'external_reference_status': 'not_asserted_by_module',
            'report_sections': usage.get('report_sections') or ['原始模块索引与未下沉字段附录'],
@@ -2357,7 +2369,6 @@ def _build_module_execution_audit(packet: dict) -> list[dict]:


 def _raw_module_usage_map(packet: dict) -> dict:
-    raw_modules = ((packet.get('raw_full_reading') or {}).get('modules') or {})
    section_map = {
        'dasha': '大运与时间主线', 'narayana_dasha': '大运与时间主线',
        'bhrigu_pada_dasha': '大运与时间主线', 'transit_multi_reference': '大运与时间主线',
@@ -2376,7 +2387,7 @@ def _raw_module_usage_map(packet: dict) -> dict:
            'report_sections': [section_map.get(str(module_id), '原始模块索引与未下沉字段附录')],
            'usage_status': 'thematic_or_audit_surface' if str(module_id) in section_map else 'raw_appendix_only',
        }
-        for module_id, module in raw_modules.items()
+        for module_id, module in _iter_audit_modules(packet)
        if module not in (None, '', [], {})
    }

@@ -2406,7 +2417,10 @@ def _attach_report_governance_contracts(packet: dict, args) -> dict:
    packet['timing_boundary_attribution'] = _build_timing_boundary_attribution(packet)
    packet['module_execution_audit'] = _build_module_execution_audit(packet)
    ai_pack = ((packet.get('raw_full_reading') or {}).get('ai_prompt_pack') or {})
-    packet['technique_audit_table'] = ai_pack.get('evidence_snapshot', {}).get('technique_audit_table', [])
+    snapshot = ai_pack.get('evidence_snapshot') if isinstance(ai_pack.get('evidence_snapshot'), dict) else {}
+    technique_table = snapshot.get('technique_audit_table') if isinstance(snapshot.get('technique_audit_table'), list) else []
+    packet['technique_audit_table'] = technique_table or _rebuild_technique_audit_table(packet)
+    packet['raw_module_index'] = _build_compact_raw_module_index(packet)
    return packet


@@ -2557,13 +2571,220 @@ def _native_dasha_family_status(module) -> dict:

 def _native_dasha_master_families(modules: dict) -> dict:
    modules = modules if isinstance(modules, dict) else {}
-    return {
-        'vimshottari': _native_dasha_family_status(modules.get('dasha')),
-        'narayana': _native_dasha_family_status(modules.get('narayana_dasha')),
-        'yogini': _native_dasha_family_status(modules.get('yogini_dasha')),
-        'ashtottari': _native_dasha_family_status(modules.get('ashtottari_dasha')),
-        'kala_chakra': _native_dasha_family_status(modules.get('kalachakra_dasha')),
+    mapping = {
+        'vimshottari': modules.get('dasha'),
+        'narayana': modules.get('narayana_dasha'),
+        'yogini': modules.get('yogini_dasha'),
+        'ashtottari': modules.get('ashtottari_dasha'),
+        'kala_chakra': modules.get('kalachakra_dasha'),
    }
+    families = {}
+    for key, module in mapping.items():
+        family = _native_dasha_family_status(module)
+        periods = _normalize_dasha_family_periods(module)
+        if periods:
+            family['periods'] = periods
+        families[key] = family
+    return families
+
+
+def _blocked_planned_section(heading: str, reason_code: str, detail: str = "") -> list[str]:
+    """Emit a planned heading with an explicit blocked reason instead of vanishing."""
+    lines = [heading, '', f"`blocked` / `{reason_code}`"]
+    if detail:
+        lines.append(str(detail).strip())
+    lines.append('')
+    return lines
+
+
+PLANNED_LONGFORM_SECTION_HEADINGS = (
+    '### KP Lord / Sub 原始表（本地计算）',
+    '#### KP 事业结构小表',
+    '#### KP 事业宫位核对（2 / 6 / 10 / 11 宫）',
+    '#### KP 财务结构小表',
+    '#### KP 财务宫位核对（2 / 5 / 9 / 11 宫）',
+    '#### KP 关系结构小表',
+    '#### KP 关系宫位核对（2 / 7 / 11 宫）',
+    '## 逐模块执行与结论追踪审计',
+    '## 完整 Technique Audit Table',
+    '#### Bhava Bala（本地三分量评分）',
+    '#### Yogini Dasha 周期（本地计算）',
+    '#### Narayana Rashi Dasha',
+    '#### Kala Chakra Dasha（本地原始周期，冲突保留）',
+    '#### Kala Chakra 深度差异诊断',
+    '#### Kala Chakra 参考对齐',
+    '## 原始模块索引与未下沉字段附录',
+    '### Profile-aware Benchmark Boundary',
+)
+
+
+def _normalize_dasha_family_periods(module) -> list:
+    if not isinstance(module, dict):
+        return []
+    periods = module.get('periods')
+    if isinstance(periods, list) and periods:
+        return [row for row in periods if isinstance(row, dict)]
+    major = module.get('major')
+    if isinstance(major, list) and major:
+        normalized = []
+        for row in major:
+            if not isinstance(row, dict):
+                continue
+            normalized.append({
+                **row,
+                'lord': row.get('planet') or row.get('lord'),
+                'start': row.get('start_date') or row.get('start'),
+                'end': row.get('end_date') or row.get('end'),
+                'years': row.get('years'),
+                'raw_period': row,
+            })
+        return normalized
+    timeline = module.get('timeline')
+    if isinstance(timeline, list) and timeline:
+        return [row for row in timeline if isinstance(row, dict)]
+    return []
+
+
+def _annual_series_surface(annual_pack: dict) -> dict:
+    """Keep independently calculated annual evidence, including replay surfaces."""
+    annual_pack = annual_pack if isinstance(annual_pack, dict) else {}
+    keep_keys = (
+        'schema', 'status', 'reason', 'profile', 'audit', 'report_sections',
+        'annual_chart', 'year_lord', 'external_engine_comparison',
+        'patyayini_dasha', 'sahams', 'tajika_yogas', 'muntha',
+    )
+    surface = {
+        key: _json_safe_report_snapshot(annual_pack.get(key))
+        for key in keep_keys
+        if key in annual_pack
+    }
+    surface['status'] = surface.get('status') or 'partial_verified'
+    return surface
+
+
+def _rebuild_technique_audit_table(packet: dict) -> list[dict]:
+    existing = packet.get('technique_audit_table')
+    if isinstance(existing, list) and existing:
+        return existing
+    worksheets = packet.get('worksheets') if isinstance(packet.get('worksheets'), dict) else {}
+    timing = worksheets.get('timing_and_predictive_systems') if isinstance(worksheets.get('timing_and_predictive_systems'), dict) else {}
+    advanced = worksheets.get('advanced_systems') if isinstance(worksheets.get('advanced_systems'), dict) else {}
+    strengths = worksheets.get('strengths_and_scores') if isinstance(worksheets.get('strengths_and_scores'), dict) else {}
+    raw_modules = ((packet.get('raw_full_reading') or {}).get('modules') or {})
+
+    def _row_status(value) -> str:
+        if isinstance(value, dict) and value and not value.get('error'):
+            status = str(value.get('status') or value.get('execution_status') or '').strip().lower()
+            if status in {'blocked', 'error'}:
+                return 'blocked'
+            return 'parameter_sensitive'
+        return 'blocked'
+
+    return [
+        {
+            'technique': 'Vimshottari Dasha',
+            'status': _row_status(timing.get('dasha') or raw_modules.get('dasha')),
+            'source': 'local_module',
+            'note': 'Local timeline only; not event timing truth.',
+        },
+        {
+            'technique': 'Narayana Dasha',
+            'status': _row_status(timing.get('narayana_dasha') or raw_modules.get('narayana_dasha')),
+            'source': 'local_module',
+            'note': 'Age-axis sequence; date-boundary parity unclosed.',
+        },
+        {
+            'technique': 'KP Lord/Sub',
+            'status': _row_status(advanced.get('kp') or raw_modules.get('kp')),
+            'source': 'local_module',
+            'note': 'KP display evidence; exact-cusp promotion remains closed.',
+        },
+        {
+            'technique': 'Functional Benefic/Malefic',
+            'status': _row_status(strengths.get('functional_benefic_malefic') or raw_modules.get('functional_benefic_malefic')),
+            'source': 'local_module',
+            'note': 'Lagna-conditioned classification for this chart.',
+        },
+        {
+            'technique': 'Annual Tajika',
+            'status': _row_status(timing.get('annual_tajika_pack') or timing.get('annual_tajika_series')),
+            'source': 'local_module',
+            'note': 'Independent annual packs; replay remains parameter-sensitive.',
+        },
+        {
+            'technique': 'MEVG / Global Web Evidence',
+            'status': 'blocked',
+            'source': 'not_executed_in_export',
+            'note': 'Deterministic export does not run live web evidence.',
+        },
+        {
+            'technique': 'Real Case Calibration',
+            'status': 'blocked',
+            'source': 'not_executed_in_export',
+            'note': 'No user event file supplied for blind replay.',
+        },
+    ]
+
+
+def _normalize_bhava_bala_module(bhava) -> dict:
+    """Accept house-keyed Shadbala output or a houses list used by the renderer."""
+    if not isinstance(bhava, dict) or not bhava:
+        return {}
+    if isinstance(bhava.get('houses'), list):
+        houses = [row for row in bhava['houses'] if isinstance(row, dict)]
+        return {'houses': houses} if houses else {}
+    houses = []
+    for key, row in bhava.items():
+        if not isinstance(row, dict):
+            continue
+        try:
+            house_no = int(key)
+        except (TypeError, ValueError):
+            house_no = row.get('house')
+        if house_no in (None, ''):
+            continue
+        houses.append({
+            'house': house_no,
+            'sign': row.get('sign'),
+            'lord': row.get('lord'),
+            'score': row.get('score'),
+            'components': row.get('components') if isinstance(row.get('components'), dict) else {
+                'lord_position': row.get('lord_position'),
+                'occupant_influence': row.get('occupant_influence'),
+                'aspect_influence': row.get('aspect_influence'),
+            },
+            'factors': row.get('factors'),
+            'strength': row.get('strength'),
+        })
+    houses.sort(key=lambda item: int(item.get('house') or 0))
+    return {'houses': houses} if houses else {}
+
+
+def _build_compact_raw_module_index(packet: dict) -> list[dict]:
+    usage_map = packet.get('raw_module_usage_map') if isinstance(packet.get('raw_module_usage_map'), dict) else {}
+    rows = []
+    for module_id, module in _iter_audit_modules(packet):
+        if module in (None, '', [], {}):
+            continue
+        module_data = module if isinstance(module, dict) else {}
+        usage = usage_map.get(str(module_id)) if isinstance(usage_map.get(str(module_id)), dict) else {}
+        if isinstance(module, dict):
+            field_names = ', '.join(str(key) for key in sorted(module, key=str))
+            status = module_data.get('status') or module_data.get('execution_status') or 'available'
+        elif isinstance(module, list):
+            field_names = f'list[{len(module)}]'
+            status = 'available'
+        else:
+            field_names = type(module).__name__
+            status = 'available'
+        rows.append({
+            'module_id': str(module_id),
+            'status': status,
+            'usage_status': usage.get('usage_status') or 'raw_appendix_only',
+            'report_sections': usage.get('report_sections') or ['原始模块索引与未下沉字段附录'],
+            'field_names': field_names,
+        })
+    return rows


 def _build_pl9_full_dasha_section(packet: dict, modules: dict) -> dict:
@@ -3573,11 +3794,17 @@ def render_pl9_markdown(packet: dict) -> str:
        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():
-                return []
+                return _blocked_planned_section(
+                    '### Profile-aware Benchmark Boundary',
+                    'profile_aware_benchmark_dashboard_missing',
+                )
            try:
                dashboard = json.loads(dashboard_path.read_text(encoding='utf-8'))
            except Exception:
-                return []
+                return _blocked_planned_section(
+                    '### Profile-aware Benchmark Boundary',
+                    'profile_aware_benchmark_dashboard_missing',
+                )
            production = dashboard.get('production_readiness') if isinstance(dashboard.get('production_readiness'), dict) else {}
            profiles = dashboard.get('profiles') if isinstance(dashboard.get('profiles'), list) else []
            capability_rows = dashboard.get('capability_rows') if isinstance(dashboard.get('capability_rows'), list) else []
@@ -3585,7 +3812,10 @@ def render_pl9_markdown(packet: dict) -> str:
            jhora_pending = dashboard.get('jhora_pending_values') if isinstance(dashboard.get('jhora_pending_values'), dict) else {}
            reader_view = dashboard.get('reader_view') if isinstance(dashboard.get('reader_view'), dict) else {}
            if not profiles and not capability_rows:
-                return []
+                return _blocked_planned_section(
+                    '### Profile-aware Benchmark Boundary',
+                    'profile_aware_benchmark_empty',
+                )
            out = [
                '### Profile-aware Benchmark Boundary',
                '',
@@ -3718,9 +3948,9 @@ def render_pl9_markdown(packet: dict) -> str:
                families = _native_dasha_master_families({
                    'dasha': timing_sheet.get('dasha') or modules.get('dasha'),
                    'narayana_dasha': timing_sheet.get('narayana_dasha') or modules.get('narayana_dasha'),
-                    'yogini_dasha': modules.get('yogini_dasha'),
-                    'ashtottari_dasha': modules.get('ashtottari_dasha'),
-                    'kalachakra_dasha': modules.get('kalachakra_dasha'),
+                    'yogini_dasha': timing_sheet.get('yogini_dasha') or modules.get('yogini_dasha'),
+                    'ashtottari_dasha': timing_sheet.get('ashtottari_dasha') or modules.get('ashtottari_dasha'),
+                    'kalachakra_dasha': timing_sheet.get('kalachakra_dasha') or modules.get('kalachakra_dasha'),
                })

            def _family_status_row(label: str, key: str, confidence_when_executed: str, blocked_text: str) -> None:
@@ -4049,7 +4279,7 @@ def render_pl9_markdown(packet: dict) -> str:
                if svg:
                    rendered.extend([f'#### {title}', '', svg, ''])
            if not rendered:
-                return []
+                return _blocked_planned_section('### Vargas I', 'varga_atlas_unavailable')
            varga_full = _varga_full_sheet()
            core_chart = packet.get('core_chart') if isinstance(packet.get('core_chart'), dict) else {}
            d1_planets = core_chart.get('planets') if isinstance(core_chart.get('planets'), dict) else {}
@@ -4327,7 +4557,11 @@ def render_pl9_markdown(packet: dict) -> str:
            calculation_profile = kp_pack.get('calculation_profile') if isinstance(kp_pack.get('calculation_profile'), dict) else {}
            effective_settings = calculation_profile.get('effective_settings') if isinstance(calculation_profile.get('effective_settings'), dict) else {}
            if not planets and not houses and not ruling_planets:
-                return []
+                return _blocked_planned_section(
+                    '### KP Lord / Sub 原始表（本地计算）',
+                    'kp_lord_sub_unavailable',
+                    str(kp_pack.get('error') or kp_pack.get('reason') or ''),
+                )

            explicit_kp_cusps = kp_pack.get('house_basis') == 'explicit_cusps' or any(
                isinstance(row, dict) and row.get('cusp_longitude') is not None
@@ -4613,8 +4847,13 @@ def render_pl9_markdown(packet: dict) -> str:
                significators = row.get('significators') if isinstance(row, dict) and isinstance(row.get('significators'), dict) else {}
                if significators:
                    rows.append((house, significators))
+            suffix = ' / '.join(str(house) for house in house_numbers)
            if not rows:
-                return []
+                return _blocked_planned_section(
+                    f'#### KP {label}宫位核对（{suffix} 宫）',
+                    'kp_domain_evidence_unavailable',
+                    domain,
+                )

            def _cell(value) -> str:
                if isinstance(value, (list, tuple, set)):
@@ -4874,7 +5113,12 @@ def render_pl9_markdown(packet: dict) -> str:
            ruling_planets = kp_pack.get('ruling_planets') if isinstance(kp_pack.get('ruling_planets'), dict) else {}
            note = _kp_domain_promise_note(domain)
            if not note:
-                return []
+                domain_label = {'career': '事业', 'wealth': '财务', 'relationship': '关系'}.get(domain, '专项')
+                return _blocked_planned_section(
+                    f'#### KP {domain_label}结构小表',
+                    'kp_domain_summary_unavailable',
+                    domain,
+                )
            specs = {
                'career': ('事业', (2, 6, 10, 11)),
                'wealth': ('财务', (2, 5, 9, 11)),
@@ -5394,12 +5638,12 @@ def render_pl9_markdown(packet: dict) -> str:
            monthly_pack = _unwrap(advanced_sheet.get('kp_monthly_report')) or {}
            monthly_pack = monthly_pack if isinstance(monthly_pack, dict) else {}
            yearly_highlights = _envelope_list(monthly_pack.get('yearly_highlights'))
-            if not yearly_highlights:
-                return []
            annual_series = _unwrap(timing_sheet.get('annual_tajika_series')) or {}
            annual_series = annual_series if isinstance(annual_series, dict) else {}
            annual_years = _unwrap(annual_series.get('years')) or {}
            annual_years = annual_years if isinstance(annual_years, dict) else {}
+            if not yearly_highlights and not annual_years:
+                return []
            target_year_text = str(annual_target_year) if annual_target_year not in (None, "", [], {}) else ""
            shared_anchor_bits = [
                _humanize_reader_summary_line(str(annual_field_briefs.get("muntha", ""))).rstrip("。") if isinstance(annual_field_briefs, dict) else "",
@@ -5432,16 +5676,91 @@ def render_pl9_markdown(packet: dict) -> str:
                            labels.append(label)
                return " / ".join(labels) if labels else "事业 / 财务 / 关系"

-            out: list[str] = []
+            def _year_replay_subblocks(annual_pack: dict, year: str) -> list[str]:
+                annual_external = _envelope_dict(annual_pack.get('external_engine_comparison'))
+                pyjhora_replay = _envelope_dict(annual_external.get('pyjhora'))
+                replay_patyayini = _envelope_dict(pyjhora_replay.get('patyayini_dasha'))
+                replay_rows = _envelope_list(replay_patyayini.get('normalized_rows'))
+                replay_sahams = _envelope_dict(pyjhora_replay.get('sahams'))
+                replay_sahams_raw = _envelope_dict(replay_sahams.get('raw'))
+                replay_saham_values = _envelope_dict(replay_sahams_raw.get('sahams'))
+                blocks: list[str] = []
+                if replay_rows:
+                    blocks.extend([
+                        '',
+                        '#### Patyayini Dasha 外部回放（PyJHora/JHora）',
+                        '',
+                        f'{year} 年外部回放只保留原始 tuple 层级；未完成 PL9 p134 对照，所有行保持 pyjhora_behavior_only / not_multiengine_parity。',
+                        '',
+                        '| 序号 | 主层原始代号 | 子层原始代号 | 边界时间（外部 tuple；时区未返回） | 原始时长 | 状态 |',
+                        '|------|--------------|--------------|-------------------------------------|----------|------|',
+                    ])
+                    for row in replay_rows[:72]:
+                        if not isinstance(row, dict):
+                            continue
+                        blocks.append(
+                            f"| {_md_cell(row.get('order'))} | {_md_cell(row.get('main_code'))} | {_md_cell(row.get('sub_code'))} | "
+                            f"{_md_cell(row.get('boundary_display'))} ({_md_cell(row.get('boundary_semantics'))}) | "
+                            f"{_md_cell(row.get('duration_raw'))} | pyjhora_behavior_only / not_multiengine_parity |"
+                        )
+                    blocks.append('')
+                else:
+                    blocks.extend(_blocked_planned_section(
+                        '#### Patyayini Dasha 外部回放（PyJHora/JHora）',
+                        'annual_replay_unavailable',
+                        f'year={year}',
+                    ))
+                if replay_saham_values:
+                    blocks.extend([
+                        '',
+                        '#### 年度 Saham 外部回放（PyJHora/JHora）',
+                        '',
+                        f'{year} 年 PyJHora/JHora 年度返照 Saham 只用于数值交叉核对，不证明年度事件。',
+                        '',
+                        '| Saham（PyJHora 原始 callable） | 星座 | 星座内度数 | 状态 |',
+                        '|------------------------------|------|------------|------|',
+                    ])
+                    for name, longitude in sorted(replay_saham_values.items(), key=lambda item: str(item[0])):
+                        if not isinstance(longitude, (int, float)):
+                            continue
+                        sign = SIGNS[int(float(longitude) // 30) % 12]
+                        label = str(name).removesuffix('_saham').replace('_', ' ').title()
+                        blocks.append(
+                            f"| {_md_cell(label)} | {_md_cell(SIGNS_CN.get(sign) or sign)} | {_degree_text(float(longitude) % 30)} | pyjhora_behavior_only / not_multiengine_parity |"
+                        )
+                    blocks.append('')
+                else:
+                    blocks.extend(_blocked_planned_section(
+                        '#### 年度 Saham 外部回放（PyJHora/JHora）',
+                        'annual_replay_unavailable',
+                        f'year={year}',
+                    ))
+                return blocks
+
+            year_entries: list[tuple[str, list]] = []
+            seen_years: set[str] = set()
            for yearly in yearly_highlights:
+                if not isinstance(yearly, dict):
+                    continue
                year = str(yearly.get("year") or "").strip()
                rows = yearly.get("months") if isinstance(yearly.get("months"), list) else []
-                if not year or not rows or year == target_year_text:
+                if not year or year == target_year_text or year in seen_years:
                    continue
+                year_entries.append((year, rows))
+                seen_years.add(year)
+            for year in annual_years:
+                year_text = str(year)
+                if not year_text or year_text == target_year_text or year_text in seen_years:
+                    continue
+                year_entries.append((year_text, []))
+                seen_years.add(year_text)
+            if not year_entries:
+                return []
+
+            out: list[str] = []
+            for year, rows in year_entries:
                annual_pack = _unwrap(annual_years.get(year)) or {}
                annual_pack = annual_pack if isinstance(annual_pack, dict) else {}
-                annual_profile = _unwrap(annual_pack.get('profile')) or {}
-                annual_profile = annual_profile if isinstance(annual_profile, dict) else {}
                annual_sections = _unwrap(annual_pack.get('report_sections')) or {}
                annual_sections = annual_sections if isinstance(annual_sections, dict) else {}
                annual_exec = _unwrap(annual_sections.get('executive_summary')) or {}
@@ -5451,13 +5770,15 @@ def render_pl9_markdown(packet: dict) -> str:
                    _humanize_reader_summary_line(str(line)).rstrip('。')
                    for line in annual_lines if line
                ]
-                focus_months = [str(row.get("month")) for row in rows if row.get("month")]
-                reasons = [str(row.get("reason")).strip() for row in rows if row.get("reason")]
+                focus_months = [str(row.get("month")) for row in rows if isinstance(row, dict) and row.get("month")]
+                reasons = [str(row.get("reason")).strip() for row in rows if isinstance(row, dict) and row.get("reason")]
                out.extend([f"### {year} 年度重点", ""])
                out.append(
                    f"这一章沿用 {target_year_text or '当前'} 年度重点模板，但这里真正下沉的是 `annual_tajika_pack` 已给出的年度结构锚点，"
                    f"再叠加 `kp_monthly_report_packet` 的 {year} 年重点月份排序，因此主要服务下一年度的主题优先级安排，继续保持 parameter_sensitive。"
                )
+                if annual_pack.get('status') == 'blocked':
+                    out.append(f"`blocked` / `{_md_cell(annual_pack.get('reason') or 'independent_annual_packet_failed')}`")
                if annual_bits:
                    out.append(f"独立年度锚点：{'；'.join(annual_bits[:3])}。")
                if focus_months:
@@ -5470,6 +5791,7 @@ def render_pl9_markdown(packet: dict) -> str:
                    out.append(
                        "年度边界提醒：" + "、".join(flagged_labels) + " 继续带着 parameter_sensitive / blocked 标签阅读，不把这些锚点直接升级成下一年度已确定事件。"
                    )
+                out.extend(_year_replay_subblocks(annual_pack, year))
                out.append("")
            return out

@@ -6838,7 +7160,11 @@ def render_pl9_markdown(packet: dict) -> str:
            functional = strength_sheet.get('functional_benefic_malefic') if isinstance(strength_sheet.get('functional_benefic_malefic'), dict) else {}
            friendship = strength_sheet.get('planetary_friendship') if isinstance(strength_sheet.get('planetary_friendship'), dict) else {}
            if not shadbala and not bhava_bala and not ashtakavarga and not vimsopaka and not functional and not friendship:
-                return []
+                return [
+                    '### 力量、Ashtakavarga 与功能性吉凶',
+                    '',
+                    *_blocked_planned_section('#### Bhava Bala（本地三分量评分）', 'bhava_bala_unavailable'),
+                ]
            out = [
                '### 力量、Ashtakavarga 与功能性吉凶',
                '',
@@ -7050,7 +7376,10 @@ def render_pl9_markdown(packet: dict) -> str:
                            f"{_md_cell(row.get('score'))} | parameter_sensitive |"
                        )
                    out.append('')
-            if vimsopaka:
+                else:
+                    out.extend(_blocked_planned_section('#### Bhava Bala（本地三分量评分）', 'bhava_bala_unavailable'))
+            else:
+                out.extend(_blocked_planned_section('#### Bhava Bala（本地三分量评分）', 'bhava_bala_unavailable'))
                out.extend([
                    '#### Vimsopaka 十六分盘力量摘要',
                    '',
@@ -7146,9 +7475,22 @@ def render_pl9_markdown(packet: dict) -> str:

        def _auxiliary_dasha_section() -> list[str]:
            dasha_master_pack = timing_sheet.get('dasha_master_pack') if isinstance(timing_sheet.get('dasha_master_pack'), dict) else {}
-            families = dasha_master_pack.get('families') if isinstance(dasha_master_pack.get('families'), dict) else {}
-            if not families:
-                return []
+            families = dict(dasha_master_pack.get('families') or {}) if isinstance(dasha_master_pack.get('families'), dict) else {}
+
+            def _overlay_family(family_key: str, native_key: str) -> None:
+                native = timing_sheet.get(native_key) if isinstance(timing_sheet.get(native_key), dict) else {}
+                family = families.get(family_key) if isinstance(families.get(family_key), dict) else {}
+                periods = family.get('periods') if isinstance(family.get('periods'), list) else []
+                if not periods:
+                    periods = _normalize_dasha_family_periods(native)
+                    if periods:
+                        family = {**family, 'periods': periods}
+                        family.setdefault('execution_status', 'executed')
+                families[family_key] = family
+
+            _overlay_family('yogini', 'yogini_dasha')
+            _overlay_family('narayana', 'narayana_dasha')
+            _overlay_family('kala_chakra', 'kalachakra_dasha')

            def _period_start(period: dict) -> str:
                raw = period.get('raw_period') if isinstance(period.get('raw_period'), dict) else {}
@@ -7262,6 +7604,8 @@ def render_pl9_markdown(packet: dict) -> str:
                        f"| {idx} | {_md_cell(yogini_name)} | {_md_cell(_humanize_reader_token(planet))} | {_period_start(period)} | {_period_end(period)} | {_period_years(period)} | parameter_sensitive |"
                    )
                out.append('')
+            else:
+                out.extend(_blocked_planned_section('#### Yogini Dasha 周期（本地计算）', 'yogini_dasha_unavailable'))

            vimshottari = timing_sheet.get('dasha') if isinstance(timing_sheet.get('dasha'), dict) else {}
            vimshottari_timeline = vimshottari.get('timeline') if isinstance(vimshottari.get('timeline'), list) else []
@@ -7880,6 +8224,15 @@ def render_pl9_markdown(packet: dict) -> str:
                            ),
                        ])
                out.append('')
+            markdown_so_far = '\n'.join(out)
+            for heading, reason in (
+                ('#### Narayana Rashi Dasha', 'narayana_dasha_unavailable'),
+                ('#### Kala Chakra Dasha（本地原始周期，冲突保留）', 'kala_chakra_unavailable'),
+                ('#### Kala Chakra 深度差异诊断', 'kala_chakra_divergence_unavailable'),
+                ('#### Kala Chakra 参考对齐', 'kala_chakra_alignment_unavailable'),
+            ):
+                if heading not in markdown_so_far:
+                    out.extend(_blocked_planned_section(heading, reason))
            return out

        def _jaimini_special_section() -> list[str]:
@@ -9535,38 +9888,40 @@ def render_pl9_markdown(packet: dict) -> str:

        # Keep every non-empty calculation module discoverable even when it has
        # not yet earned a dedicated thematic section.
-        raw_modules = raw_full_reading.get('modules') if isinstance(raw_full_reading, dict) else {}
-        if isinstance(raw_modules, dict) and raw_modules:
+        raw_module_index = packet.get('raw_module_index') if isinstance(packet.get('raw_module_index'), list) else []
+        if not raw_module_index:
+            raw_modules = raw_full_reading.get('modules') if isinstance(raw_full_reading, dict) else {}
+            if isinstance(raw_modules, dict) and raw_modules:
+                raw_module_index = _build_compact_raw_module_index({
+                    'raw_full_reading': raw_full_reading,
+                    'raw_module_usage_map': raw_module_usage_map,
+                })
+        if raw_module_index:
            lines.extend([
                '',
                '## 原始模块索引与未下沉字段附录',
                '',
-                '本附录逐项登记当前 packet 中的非空原始模块。独立章节已经展开的模块仍保留这里的路径；尚未下沉到正文或专门技术页的模块，明确标为“仅原始附录”，不代表缺失或已升级为主判断证据。完整原始数值保留在同一导出的 JSON packet 的 raw_full_reading.modules 下；Markdown 不重复嵌入 JSON，避免正文与原始负载双重膨胀。',
+                '本附录逐项登记当前 packet 中的非空原始模块。独立章节已经展开的模块仍保留这里的登记；尚未下沉到正文或专门技术页的模块，明确标为“仅原始附录”，不代表缺失或已升级为主判断证据。完整原始数值不嵌入 Markdown，避免正文与原始负载双重膨胀。',
                '',
-                '| 模块 | 状态 | 原始路径 | 报告使用/结论入口 | 未下沉字段 |',
-                '|------|------|----------|---------------------|------------|',
+                '| 模块 | 状态 | 登记键 | 报告使用/结论入口 | 未下沉字段 |',
+                '|------|------|--------|---------------------|------------|',
            ])
-            for module_id in sorted(raw_modules):
-                module = raw_modules.get(module_id)
-                if module in (None, '', [], {}):
+            for row in raw_module_index:
+                if not isinstance(row, dict):
                    continue
-                if isinstance(module, dict):
-                    status = module.get('status') or module.get('execution_status') or 'available'
-                    field_names = ', '.join(str(key) for key in sorted(module, key=str))
-                elif isinstance(module, list):
-                    status = 'available'
-                    field_names = f'list[{len(module)}]'
-                else:
-                    status = 'available'
-                    field_names = type(module).__name__
-                usage = raw_module_usage_map.get(str(module_id)) if isinstance(raw_module_usage_map.get(str(module_id)), dict) else {}
-                usage_sections = usage.get('report_sections') if isinstance(usage.get('report_sections'), list) else ['原始模块索引与未下沉字段附录']
+                module_id = row.get('module_id')
+                usage_sections = row.get('report_sections') if isinstance(row.get('report_sections'), list) else ['原始模块索引与未下沉字段附录']
                lines.append(
-                    f"| `{_md_cell(module_id)}` | {_md_cell(status)} | "
-                    f"`raw_full_reading.modules.{_md_cell(module_id)}` | "
-                    f"{_md_cell('；'.join(usage_sections))} | {_md_cell(field_names)} |"
+                    f"| `{_md_cell(module_id)}` | {_md_cell(row.get('status'))} | "
+                    f"`{_md_cell(module_id)}` | "
+                    f"{_md_cell('；'.join(usage_sections))} | {_md_cell(row.get('field_names'))} |"
                )
            lines.append('')
+        else:
+            lines.extend(_blocked_planned_section(
+                '## 原始模块索引与未下沉字段附录',
+                'raw_module_index_unavailable',
+            ))

        lines.extend([
            '',
@@ -9670,7 +10025,13 @@ def render_pl9_markdown(packet: dict) -> str:
            f"- Timing Precision Contract: {_bool_text(bool(ai_audit.get('timing_precision_contract')))}",
        ])

-    lines.extend(['', render_pl9_continuation_prompt()])
+        lines.extend(['', render_pl9_continuation_prompt()])
+
+    markdown_so_far = '\n'.join(lines)
+    for heading in PLANNED_LONGFORM_SECTION_HEADINGS:
+        if heading not in markdown_so_far:
+            lines.extend(_blocked_planned_section(heading, 'planned_section_unavailable'))
+            markdown_so_far = '\n'.join(lines)

    markdown = '\n'.join(lines).strip() + '\n'
    if isinstance(packet.get('reader_engine_boundary_notice'), dict):
@@ -15714,6 +16075,8 @@ def cmd_full_reading(args):
                    planet.update(get_kp_lords(float(planet.get('degree_raw', 0.0))))
                    planet['r_c'] = ''
        report['modules']['shadbala'] = shadbala_result
+        if isinstance(shadbala_result, dict) and shadbala_result.get('bhava_bala'):
+            report['modules']['bhava_bala'] = _normalize_bhava_bala_module(shadbala_result.get('bhava_bala'))
    except Exception as e:
        report['errors'].append(f"shadbala: {e}")

@@ -16604,7 +16967,9 @@ def build_pl9_style_export_packet(full_reading: dict, include_raw: bool = False)
            'shadbala_component_status': shadbala_component_status,
            'pl9_p33_declination_contract': pl9_p33_declination_contract,
            'pl9_p43_44_shadbala_contract': pl9_p43_44_shadbala_contract,
-            'bhava_bala': modules.get('bhava_bala'),
+            'bhava_bala': _normalize_bhava_bala_module(
+                modules.get('bhava_bala') or ((modules.get('shadbala') or {}).get('bhava_bala') if isinstance(modules.get('shadbala'), dict) else {})
+            ),
            'ashtakavarga': modules.get('ashtakavarga'),
            'vimsopaka': modules.get('vimsopaka'),
            'pushkara': modules.get('pushkara'),
@@ -16670,6 +17035,9 @@ def build_pl9_style_export_packet(full_reading: dict, include_raw: bool = False)
            'dasha_sandhi': modules.get('dasha_sandhi'),
            'bhrigu_pada_dasha': modules.get('bhrigu_pada_dasha'),
            'narayana_dasha': modules.get('narayana_dasha'),
+            'yogini_dasha': modules.get('yogini_dasha'),
+            'ashtottari_dasha': modules.get('ashtottari_dasha'),
+            'kalachakra_dasha': modules.get('kalachakra_dasha'),
            'transit_multi_reference': modules.get('transit_multi_reference'),
            'double_transit_pac': modules.get('double_transit_pac'),
            'transit_ll7l': modules.get('transit_ll7l'),
@@ -16874,7 +17242,9 @@ def build_pl9_style_export_packet(full_reading: dict, include_raw: bool = False)
        'strengths_and_scores': {
            'shadbala': modules.get('shadbala'),
            'shadbala_component_status': shadbala_component_status,
-            'bhava_bala': modules.get('bhava_bala'),
+            'bhava_bala': _normalize_bhava_bala_module(
+                modules.get('bhava_bala') or ((modules.get('shadbala') or {}).get('bhava_bala') if isinstance(modules.get('shadbala'), dict) else {})
+            ),
            'ashtakavarga': modules.get('ashtakavarga'),
            'vimsopaka': modules.get('vimsopaka'),
            'pushkara': modules.get('pushkara'),
@@ -16900,6 +17270,9 @@ def build_pl9_style_export_packet(full_reading: dict, include_raw: bool = False)
            'dasha_sandhi': modules.get('dasha_sandhi'),
            'bhrigu_pada_dasha': modules.get('bhrigu_pada_dasha'),
            'narayana_dasha': modules.get('narayana_dasha'),
+            'yogini_dasha': modules.get('yogini_dasha'),
+            'ashtottari_dasha': modules.get('ashtottari_dasha'),
+            'kalachakra_dasha': modules.get('kalachakra_dasha'),
            'transit_multi_reference': modules.get('transit_multi_reference'),
            'double_transit_pac': modules.get('double_transit_pac'),
            'transit_ll7l': modules.get('transit_ll7l'),
@@ -17141,7 +17514,7 @@ def build_professional_report_reference_packet(
    selected_pack_ids: list[str] | None = None,
 ) -> dict:
    """Build the public, non-writer professional reference packet for CLI and API."""
-    packet = build_pl9_style_export_packet(full_reading, include_raw=False)
+    packet = build_pl9_style_export_packet(full_reading, include_raw=True)
    packet = attach_calculation_profile(packet, args)
    packet = _attach_annual_tajika_pack(packet, args)
    packet = _attach_report_governance_contracts(packet, args)
@@ -1,16 +1,46 @@
 #!/usr/bin/env python3
-"""Honest three-layer timing contract for uncalibrated day/month rankings."""
+"""Honest four-tier timing contract for uncalibrated day/month/year claims."""

 from __future__ import annotations

 from typing import Any


+def _tier(
+    *,
+    status: str,
+    allowed_claims: list[str],
+    prohibited_claims: list[str],
+    note: str,
+) -> dict[str, Any]:
+    return {
+        "status": status,
+        "allowed_claims": allowed_claims,
+        "prohibited_claims": prohibited_claims,
+        "note": note,
+    }
+
+
+def _has_annual_series(source: dict[str, Any]) -> bool:
+    annual_series = source.get("annual_series")
+    if isinstance(annual_series, dict):
+        years = annual_series.get("years")
+        if isinstance(years, dict) and years:
+            return True
+        if isinstance(years, list) and years:
+            return True
+    if source.get("annual_tajika_series") or source.get("annual_chart") or source.get("solar_return"):
+        return True
+    return False
+
+
 def build_timing_precision_contract(payload: dict[str, Any] | None = None) -> dict[str, Any]:
    source = payload if isinstance(payload, dict) else {}
    candidates = source.get("candidate_windows") if isinstance(source.get("candidate_windows"), list) else []
    triggers = source.get("exact_triggers") if isinstance(source.get("exact_triggers"), list) else []
    verified_window = source.get("verified_window") or source.get("broad_window")
+    has_annual = _has_annual_series(source)
+    year_status = "partial_verified" if has_annual else "blocked"
    return {
        "timing_precision": "candidate_day_window" if candidates else "broad_window_only",
        "claim_status": "exploratory_unvalidated",
@@ -22,5 +52,35 @@ def build_timing_precision_contract(payload: dict[str, Any] | None = None) -> di
            "required": "new_independently_labeled_day_level_holdout",
            "current_negative_controls_reusable_for_tuning": False,
        },
+        "tiers": {
+            "day": _tier(
+                status="blocked",
+                allowed_claims=[],
+                prohibited_claims=["exact_day_event", "exact_clock_event", "dated_outcome_promise"],
+                note="日级 holdout 未通过，日结论恒 blocked。",
+            ),
+            "week": _tier(
+                status="parameter_sensitive",
+                allowed_claims=["candidate_observation_window"],
+                prohibited_claims=["exact_day_event", "guaranteed_weekly_outcome"],
+                note="周级只可作为候选观察窗，不能升级为确定事件。",
+            ),
+            "month": _tier(
+                status="parameter_sensitive",
+                allowed_claims=["candidate_observation_window"],
+                prohibited_claims=["exact_day_event", "guaranteed_monthly_outcome"],
+                note="月级只可作为候选观察窗，不能升级为确定事件。",
+            ),
+            "year": _tier(
+                status=year_status,
+                allowed_claims=["annual_solar_return_structure"] if has_annual else [],
+                prohibited_claims=["exact_day_event", "year_as_verified_event_calendar"],
+                note=(
+                    "年度返照系列存在时，年结构可标 partial_verified；仍禁止把年层写成已验证事件日历。"
+                    if has_annual
+                    else "缺少年度返照系列，年结论保持 blocked。"
+                ),
+            ),
+        },
        "boundary": "候选日期未通过独立日级 holdout 验证，不能作为确定事件承诺；精确时间仅表示技术触发点。",
    }