Land relationship narrative and dynamic hook workflow

scripts/jyotish_api_server.py

@@ -183,6 +183,50 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
         if origin in allowed:
             self.send_header('Access-Control-Allow-Origin', origin)
 
+    def _vedastro_status(self):
+        adapter = _load_local_module('vedastro_service_adapter')
+        endpoint = os.environ.get('VEDASTRO_API_ENDPOINT', '').strip()
+        network_flag = os.environ.get('VEDASTRO_ENABLE_NETWORK', '').strip().lower()
+        network_enabled = network_flag in {'1', 'true', 'yes'}
+        parsed = urlparse(endpoint) if endpoint else None
+        configured = bool(endpoint)
+        if not configured:
+            status = 'service_endpoint_not_configured'
+        elif not network_enabled:
+            status = 'network_execution_disabled'
+        else:
+            status = 'live_ready'
+        artifact_dir = getattr(adapter, 'ARTIFACT_DIR', None)
+        latest_artifact = None
+        if artifact_dir and os.path.isdir(artifact_dir):
+            artifacts = sorted(
+                (os.path.join(artifact_dir, name) for name in os.listdir(artifact_dir) if name.endswith('.json')),
+                key=lambda path: os.path.getmtime(path),
+                reverse=True,
+            )
+            if artifacts:
+                latest_artifact = os.path.relpath(artifacts[0], REPO_ROOT)
+        return {
+            'adapter': 'vedastro_service_adapter',
+            'backend': 'vedastro_service_adapter_candidate',
+            'status': status,
+            'configured': configured,
+            'network_enabled': network_enabled,
+            'endpoint_host': parsed.netloc if parsed else None,
+            'required_env': {
+                'endpoint': 'VEDASTRO_API_ENDPOINT',
+                'network': 'VEDASTRO_ENABLE_NETWORK',
+                'api_key_optional': 'VEDASTRO_API_KEY',
+            },
+            'live_profile': 'vedastro-live',
+            'transport': 'http_json_service_boundary',
+            'range_scan_role': adapter.VEDASTRO_CALCULATION_COVERAGE['range_scan_role'],
+            'official_events_builder_methods': adapter.VEDASTRO_CALCULATION_COVERAGE['official_events_builder_methods'],
+            'artifact_dir': 'scratch/local/vedastro_adapter',
+            'latest_artifact': latest_artifact,
+            'boundary': 'VedAstro is optional external timing evidence; local Jyotish gates remain authoritative.',
+        }
+
     def do_OPTIONS(self):
         self._json({})
 
@@ -213,6 +257,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
                 self._json(self._capability_audit())
             elif path == '/api/technique_catalog':
                 self._json(self._technique_catalog())
+            elif path == '/api/vedastro/status':
+                self._json(self._vedastro_status())
             elif path == '/api/real_case_revalidation':
                 self._json(self._real_case_revalidation())
             else:
@@ -484,6 +530,65 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             return html[:body_close.start()] + summary + html[body_close.start():]
         return html + summary
 
+    def _inject_relationship_narrative_summary(self, html, narrative):
+        if not isinstance(narrative, dict):
+            return html
+        headline = narrative.get('headline')
+        if not headline:
+            return html
+
+        def _escape(value):
+            return html_lib.escape(str(value or ''))
+
+        def _list_html(items):
+            if not isinstance(items, list) or not items:
+                return '<li>暂无补充。</li>'
+            return ''.join(f'<li>{_escape(item)}</li>' for item in items[:6])
+
+        risks = narrative.get("risks")
+        boundaries = narrative.get("boundaries")
+        caution_block = ''
+        if (
+            isinstance(risks, list)
+            and any('不能误读成接近结婚' in str(item) for item in risks)
+        ) or (
+            isinstance(boundaries, list)
+            and any('不等于法律婚姻' in str(item) for item in boundaries)
+        ):
+            caution_block = (
+                '<div class="relationship-caution" '
+                'style="margin:12px 0 16px;padding:12px 14px;border:1px solid #f3d19c;'
+                'border-left:4px solid #c67a00;border-radius:8px;background:#fff8ed;color:#7a4b00;">'
+                '<strong style="display:block;margin:0 0 6px;">Caution</strong>'
+                '<span style="display:block;font-size:13px;line-height:1.6;">'
+                '当前公开化/关系可见度候选不能被误读成接近法律婚姻；若 core marriage promise、dual dasha 或 external timing 仍未收敛，'
+                '必须继续降置信度并保持 context-only 解释。'
+                '</span>'
+                '</div>'
+            )
+
+        summary = (
+            '<section data-relationship-strict-narrative="true" '
+            'style="margin:24px 0;padding:16px;border:1px solid #d9dde8;border-radius:8px;'
+            'background:#fbfcff;font-family:-apple-system,BlinkMacSystemFont,Segoe UI,sans-serif;">'
+            '<h2 style="margin:0 0 12px;font-size:20px;">Relationship Strict Narrative</h2>'
+            f'<p style="margin:0 0 12px;">{_escape(headline)}</p>'
+            f'{caution_block}'
+            '<div style="display:grid;grid-template-columns:repeat(3,minmax(0,1fr));gap:16px;">'
+            '<div><strong>Strengths</strong><ul style="margin:8px 0 0 18px;padding:0;">'
+            f'{_list_html(narrative.get("strengths"))}</ul></div>'
+            '<div><strong>Risks</strong><ul style="margin:8px 0 0 18px;padding:0;">'
+            f'{_list_html(narrative.get("risks"))}</ul></div>'
+            '<div><strong>Boundaries</strong><ul style="margin:8px 0 0 18px;padding:0;">'
+            f'{_list_html(narrative.get("boundaries"))}</ul></div>'
+            '</div>'
+            '</section>'
+        )
+        body_close = re.search(r'</body\s*>', html, re.IGNORECASE)
+        if body_close:
+            return html[:body_close.start()] + summary + html[body_close.start():]
+        return html + summary
+
     def _artifact_base64(self, path):
         size = os.path.getsize(path)
         if size > MAX_REPORT_BASE64_BYTES:
@@ -552,6 +657,10 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             html,
             body.get('functional_benefic_malefic'),
         )
+        html = self._inject_relationship_narrative_summary(
+            html,
+            body.get('relationship_narrative'),
+        )
         fmt = body.get('format', 'html')
         if fmt not in {'html', 'pdf'}:
             raise BadRequest('format must be html or pdf')
@@ -1050,6 +1159,23 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
                 source='full_reading.modules.vivah_saham',
                 details=vivah.get('vivah_saham') if isinstance(vivah.get('vivah_saham'), dict) else vivah,
             ))
+        strict_relationship = full_modules.get('relationship_strict_evidence') if isinstance(full_modules, dict) else {}
+        user_narrative = strict_relationship.get('user_narrative') if isinstance(strict_relationship, dict) else {}
+        if isinstance(user_narrative, dict) and user_narrative.get('markdown'):
+            items.append(self._theme_evidence(
+                'Relationship-strict-narrative',
+                'Strict',
+                user_narrative.get('markdown'),
+                'neutral',
+                'strong',
+                source='full_reading.modules.relationship_strict_evidence.user_narrative',
+                details={
+                    'headline': user_narrative.get('headline'),
+                    'strengths': user_narrative.get('strengths', [])[:3],
+                    'risks': user_narrative.get('risks', [])[:3],
+                    'boundaries': user_narrative.get('boundaries', [])[:3],
+                },
+            ))
         return items
 
     def _derived_career_evidence(self, chart_data, context):

scripts/jyotish_engine.py

@@ -42,9 +42,11 @@ import os
 import csv
 import math
 import sqlite3
+import importlib.util
 from datetime import datetime, timedelta
 from typing import Dict, List
 from tabulate import tabulate
+from life_stage_hook import generate_life_stage_hooks
 
 from ayanamsa_utils import (
     AYANAMSA_DISPLAY_NAMES,
@@ -58,6 +60,7 @@ from ayanamsa_utils import (
 # 路径常量
 # ============================================================================
 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+ROOT_DIR = os.path.dirname(SCRIPT_DIR)
 HOME_DIR = os.path.expanduser('~')
 CLAW_DIR = os.path.join(HOME_DIR, 'WorkBuddy', 'Claw')
 DB_PATH = os.path.join(CLAW_DIR, 'vedic_astrology_validation.db')
@@ -378,7 +381,7 @@ def _get_temporary_relationship(planet1, planet2, planets_data):
     """Calculate Temporary Friendship (Tatkalika Maitri)."""
     if not planets_data or planet1 not in planets_data or planet2 not in planets_data:
         return 'NEUTRAL'
-    
+
     p1_sign = planets_data[planet1].get('sign')
     p2_sign = planets_data[planet2].get('sign')
     if not p1_sign or not p2_sign:
@@ -387,7 +390,7 @@ def _get_temporary_relationship(planet1, planet2, planets_data):
     idx1 = SIGNS.index(p1_sign)
     idx2 = SIGNS.index(p2_sign)
     distance = (idx2 - idx1) % 12 + 1
-    
+
     # 2, 3, 4, 10, 11, 12 from planet are temporary friends
     if distance in [2, 3, 4, 10, 11, 12]:
         return 'FRIEND'
@@ -884,6 +887,7 @@ def _build_technique_audit_table(functional_layer, oracle_progress, modules):
     ashtakavarga = modules.get('ashtakavarga') if isinstance(modules, dict) else {}
     vimsopaka = modules.get('vimsopaka') if isinstance(modules, dict) else {}
     dasa_convergence = modules.get('dasa_convergence') if isinstance(modules, dict) else {}
+    relationship = modules.get('relationship_strict_evidence') if isinstance(modules, dict) else {}
 
     rows = [
         {
@@ -939,9 +943,131 @@ def _build_technique_audit_table(functional_layer, oracle_progress, modules):
             f"Vimsopaka status={vimsopaka.get('status') if isinstance(vimsopaka, dict) else None}。"
         ),
     })
+
+    event_judgement = relationship.get('event_judgement') if isinstance(relationship, dict) else {}
+    secondary_context = event_judgement.get('secondary_context') if isinstance(event_judgement, dict) else []
+    synastry_context = [item for item in (secondary_context or []) if isinstance(item, str) and item.startswith('synastry_')]
+    rows.append({
+        'technique': 'Relationship Synastry Taxonomy',
+        'status': 'used' if synastry_context else 'blocked',
+        'source': 'relationship strict workflow + synastry_relationship_bridge_v1',
+        'note': (
+            f"relationship secondary_context 中的 synastry 语义={synastry_context}; "
+            "compatibility support 表示匹配/延续性支持；"
+            "protective kuta support 表示防护型 Kuta 清洁度支持；"
+            "若 dual dasha / external timing / marriage convergence 冲突，"
+            "不得把这些支持越权解释成 legal marriage 的高置信度落地。"
+        ),
+    })
     return rows
 
 
+def _build_relationship_narrative_payload(relationship_strict):
+    if not isinstance(relationship_strict, dict) or not relationship_strict:
+        return {
+            'headline': '婚恋严格裁决证据尚未完成，当前不能生成高严谨关系叙事。',
+            'strengths': [],
+            'risks': ['缺少 relationship strict workflow 的核心证据，婚恋正文需降级。'],
+            'boundaries': [
+                '未完成 D1 + D9 + UL + dual dasha 交叉前，不得把单一关系信号写成高置信度婚姻结论。',
+            ],
+            'markdown': (
+                "### 婚恋严格裁决\n"
+                "- 当前缺少 relationship strict evidence，无法生成高严谨婚恋 narrative。\n"
+                "- 在 D1、D9、UL、Vimshottari 与 Narayana 未齐备前，应标记为 blocked 或降低置信度。"
+            ),
+        }
+
+    event_judgement = relationship_strict.get('event_judgement') if isinstance(relationship_strict, dict) else {}
+    present = relationship_strict.get('present_evidence') if isinstance(relationship_strict, dict) else {}
+    missing = relationship_strict.get('missing_evidence') or []
+    secondary_context = event_judgement.get('secondary_context') if isinstance(event_judgement, dict) else []
+    secondary_context = secondary_context if isinstance(secondary_context, list) else []
+    confidence_cap = relationship_strict.get('confidence_cap') or event_judgement.get('confidence_cap') or 'unknown'
+    dominant_label = event_judgement.get('dominant_label') if isinstance(event_judgement, dict) else None
+    synastry = present.get('synastry_relationship_support') if isinstance(present, dict) else {}
+    synastry_signals = synastry.get('signals') if isinstance(synastry, dict) else []
+    synastry_signals = synastry_signals if isinstance(synastry_signals, list) else []
+
+    strengths = []
+    risks = []
+    boundaries = []
+
+    if dominant_label == 'legal_marriage':
+        strengths.append('本轮严格裁决已把婚恋主标签抬到 legal_marriage，但仍需尊重时机与现实承诺层。')
+    elif dominant_label == 'public_formalization':
+        strengths.append('当前更偏向 public_formalization，表示关系可见度/公开化支持强于法律婚姻落地。')
+    elif 'public_formalization_candidate' in secondary_context:
+        strengths.append('当前更接近 public_formalization_candidate，表示公开化/关系可见度候选正在增强，但仍未达到法律婚姻落地。')
+
+    if 'jaimini_support' in secondary_context:
+        strengths.append('Jaimini 桥接已提供配偶征象支持，DK/UL 线索可用于补强婚恋叙事。')
+    if 'ul_support' in secondary_context:
+        strengths.append('Upapada Lagna 已进入严格证据，可作为关系承诺与婚姻叙事的辅助锚点。')
+    if 'synastry_support' in secondary_context:
+        strengths.append('合盘支持已进入婚恋主链，但它只说明关系兼容度有帮助，不能单独决定婚姻落地。')
+    if 'synastry_compatibility_support' in secondary_context:
+        strengths.append('protective kuta / compatibility support 说明部分 Kuta 与关系延续性维度较干净。')
+    if 'synastry_protective_kuta_support' in secondary_context:
+        strengths.append('protective kuta support 已被识别，可作为关系稳定性的次级支持语义。')
+    if 'synastry_exception_mitigated' in secondary_context:
+        strengths.append('存在 exception mitigation，说明部分 Dosha/不利匹配在传统规则里有缓解条件。')
+
+    if confidence_cap in {'low', 'blocked'}:
+        risks.append('当前 confidence cap 偏低，dual dasha / external timing / marriage convergence 至少有一层存在冲突或不足。')
+        if 'public_formalization_candidate' in secondary_context:
+            risks.append('当前虽更接近 public_formalization_candidate，但在 timing conflict 未解除前，不能误读成接近结婚。')
+    if missing:
+        risks.append(f"仍缺少关键层：{', '.join(str(item) for item in missing[:4])}。")
+    if 'virodhargala_obstruction' in secondary_context:
+        risks.append('第七宫 Argala 出现阻滞，关系推进可能伴随现实阻力或时间延后。')
+    if 'dignity_high_friction' in secondary_context:
+        risks.append('相关婚恋行星尊贵度摩擦较高，关系推进时更容易出现磨损与反复确认。')
+    if 'shadbala_component_gap' in secondary_context:
+        risks.append('Shadbala 六分量还存在缺口，关系强弱结论需继续保守处理。')
+
+    boundaries.append('婚恋高严谨模式至少需要 D1、D9、UL、Vimshottari 与 Narayana dual dasha 同时在场。')
+    boundaries.append('protective kuta support、Mahendra、Stree Deergha 等合盘细信号只能辅助，不得越权抬升 legal_marriage。')
+    boundaries.append('若 dual dasha、external timing 或 marriage convergence 冲突，必须明确降置信度，而不是把关系窗口包装成婚姻必然落地。')
+    if 'public_formalization_candidate' in secondary_context:
+        boundaries.append('public_formalization_candidate 只表示公开化候选，不等于法律婚姻，不能越权替代 legal_marriage。')
+    if synastry_signals:
+        boundaries.append(f"当前 synastry taxonomy 已命中 {', '.join(synastry_signals[:5])}，但这些信号仍从属于 secondary-context。")
+
+    if not strengths:
+        strengths.append('当前婚恋 strict workflow 主要提供边界与缺口提示，尚未形成足够稳定的正向落地支持。')
+    if not risks:
+        risks.append('未见强烈负面冲突，但仍需用现实事件、D9 和 dual dasha 做最后复核。')
+
+    headline = (
+        '婚恋严格裁决已接入 synastry taxonomy，可把合盘支持翻译成次级关系语义。'
+        if 'synastry_support' in secondary_context
+        else '婚恋严格裁决已接入主链，但当前更依赖本命、D9 与时机层，而非合盘辅助。'
+    )
+
+    markdown_lines = [
+        '### 婚恋严格裁决',
+        f"- headline: {headline}",
+        f"- dominant_label: {dominant_label or 'none'}",
+        f"- confidence_cap: {confidence_cap}",
+        f"- secondary_context: {secondary_context}",
+        '- strengths:',
+        *[f"  - {item}" for item in strengths],
+        '- risks:',
+        *[f"  - {item}" for item in risks],
+        '- boundaries:',
+        *[f"  - {item}" for item in boundaries],
+    ]
+
+    return {
+        'headline': headline,
+        'strengths': strengths,
+        'risks': risks,
+        'boundaries': boundaries,
+        'markdown': "\n".join(markdown_lines),
+    }
+
+
 def _build_ai_prompt_pack(report):
     """Build a compact, evidence-first prompt pack for downstream AI/RAG reading."""
     modules = report.get('modules', {}) if isinstance(report, dict) else {}
@@ -960,6 +1086,7 @@ def _build_ai_prompt_pack(report):
     functional_layer = _functional_benefic_malefic_snapshot(planets, chart.get('ascendant', {}))
     oracle_progress = _oracle_progress_snapshot()
     technique_audit_table = _build_technique_audit_table(functional_layer, oracle_progress, modules)
+    relationship_narrative = _build_relationship_narrative_payload(modules.get('relationship_strict_evidence'))
 
     shadbala_ranking = []
     for planet_name, pdata in sorted(
@@ -1031,6 +1158,7 @@ def _build_ai_prompt_pack(report):
         'oracle_progress': oracle_progress,
         'functional_benefic_malefic': functional_layer,
         'technique_audit_table': technique_audit_table,
+        'relationship_narrative': relationship_narrative,
     }
 
     prompt_lines = [
@@ -1068,6 +1196,25 @@ def _build_ai_prompt_pack(report):
     }
 
 
+def _load_relationship_strict_collector():
+    try:
+        from mcp_server import _collect_strict_evidence as collector
+        return collector
+    except Exception:
+        mcp_path = os.path.join(ROOT_DIR, 'mcp_server.py')
+        if not os.path.exists(mcp_path):
+            raise
+        spec = importlib.util.spec_from_file_location("jyotish_root_mcp_server", mcp_path)
+        if spec is None or spec.loader is None:
+            raise ImportError(f"Unable to load mcp_server from {mcp_path}")
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        collector = getattr(module, "_collect_strict_evidence", None)
+        if collector is None:
+            raise ImportError("mcp_server._collect_strict_evidence not found")
+        return collector
+
+
 # ============================================================================
 # 公共星盘计算（供 chart/shadbala/ashtakavarga 共用，v3.4提取）
 # ============================================================================
@@ -4739,6 +4886,20 @@ def cmd_full_reading(args):
     elapsed = round(time.time() - t0, 2)
     module_count = len(report['modules'])
     error_count = len(report['errors'])
+
+    # ── 生成动态引导 (Dynamic Hooks) ──
+    try:
+        report['dynamic_hooks'] = generate_life_stage_hooks(
+            planets=report['modules'].get('chart', {}).get('planets', {}),
+            asc_sign=report['modules'].get('chart', {}).get('ascendant', {}).get('sign', ''),
+            asc_idx=report['modules'].get('chart', {}).get('ascendant', {}).get('sign_idx', 0),
+            current_dasha=report['modules'].get('dasha', {}).get('current_dasha', {}),
+            narayana_dasha=report['modules'].get('narayana_dasha', {})
+        )
+    except Exception as e:
+        report['dynamic_hooks'] = []
+        report['errors'].append(f"hook_engine: {e}")
+
     report['summary'] = {
         'elapsed_seconds': elapsed,
         'modules_computed': module_count,
@@ -4746,6 +4907,16 @@ def cmd_full_reading(args):
         'status': 'complete' if error_count == 0 else f'{error_count} errors',
         'next_step': '⭐ v6.1.6: full-reading 已输出 transit_multi_reference(四参考点) + dasa_convergence(五系统交叉) + yogini_dasha + ashtottari_dasha + kalachakra_dasha + d9_navamsa_expanded。AI必须使用四参考点分析Transit，Dasa预测必须标注多系统收敛等级。',
     }
+
+    try:
+        relationship_strict_collector = _load_relationship_strict_collector()
+        report['modules']['relationship_strict_evidence'] = relationship_strict_collector('relationship', report)
+        report['modules']['relationship_strict_evidence']['user_narrative'] = _build_relationship_narrative_payload(
+            report['modules']['relationship_strict_evidence']
+        )
+    except Exception as e:
+        report['errors'].append(f"relationship-strict-evidence: {e}")
+
     report['ai_prompt_pack'] = _build_ai_prompt_pack(report)
 
     return report

scripts/life_stage_hook.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""
+Dynamic Life-Stage Hook Engine
+Generates context-aware prompt suggestions based on the user's current Dasha and Transits.
+"""
+
+from typing import Any, Dict, List
+
+# Core mapping for planets/houses to life themes
+PLANET_THEMES = {
+    "Sun": ["career", "status", "authority", "father"],
+    "Moon": ["mind", "mother", "property", "emotional well-being"],
+    "Mars": ["property", "energy", "courage", "siblings", "conflicts"],
+    "Mercury": ["business", "communication", "intellect", "skills"],
+    "Jupiter": ["wealth", "children", "expansion", "luck", "wisdom"],
+    "Venus": ["marriage", "relationships", "luxury", "comfort"],
+    "Saturn": ["career", "delays", "discipline", "hard work", "karma"],
+    "Rahu": ["foreign affairs", "sudden events", "technology", "obsession"],
+    "Ketu": ["spirituality", "losses", "detachment", "isolation"]
+}
+
+HOUSE_THEMES = {
+    1: ["health", "self-development", "life path"],
+    2: ["wealth", "savings", "family"],
+    3: ["courage", "short trips", "communication"],
+    4: ["property", "mother", "home", "peace"],
+    5: ["investments", "children", "creativity"],
+    6: ["health issues", "debts", "enemies", "daily work"],
+    7: ["marriage", "business partnerships", "public relations"],
+    8: ["hidden matters", "sudden gains/losses", "research", "crises"],
+    9: ["fortune", "higher education", "long travel", "spirituality"],
+    10: ["career advancement", "public status", "leadership"],
+    11: ["gains", "networks", "profits", "elder siblings"],
+    12: ["foreign lands", "expenses", "isolation", "spirituality"]
+}
+
+def _get_lord_houses(planet: str, asc_idx: int) -> List[int]:
+    """Return the houses owned by the given planet for the given ascendant index."""
+    SIGNS = ["Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
+             "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces"]
+    SIGN_LORDS = {
+        "Aries": "Mars", "Taurus": "Venus", "Gemini": "Mercury", "Cancer": "Moon",
+        "Leo": "Sun", "Virgo": "Mercury", "Libra": "Venus", "Scorpio": "Mars",
+        "Sagittarius": "Jupiter", "Capricorn": "Saturn", "Aquarius": "Saturn", "Pisces": "Jupiter"
+    }
+    owned = []
+    for h in range(1, 13):
+        sign = SIGNS[(asc_idx + h - 1) % 12]
+        if SIGN_LORDS.get(sign) == planet:
+            owned.append(h)
+    return owned
+
+def generate_life_stage_hooks(
+    planets: Dict[str, Any],
+    asc_sign: str,
+    asc_idx: int,
+    current_dasha: Dict[str, Any],
+    narayana_dasha: Dict[str, Any]
+) -> List[Dict[str, str]]:
+    """
+    Generate top 3 prompt hooks based on the current life stage.
+    """
+    hooks = []
+
+    # 1. Antardasha (Sub-period) Hook
+    ad_lord = None
+    if isinstance(current_dasha, dict) and "antardasha" in current_dasha:
+        ad = current_dasha.get("antardasha")
+        if isinstance(ad, dict):
+            ad_lord = ad.get("lord")
+    elif isinstance(current_dasha, dict):
+        ad_lord = current_dasha.get("lord")  # Fallback to MD lord
+
+    if ad_lord:
+        owned_houses = _get_lord_houses(ad_lord, asc_idx)
+        core_themes = []
+        for h in owned_houses:
+            core_themes.extend(HOUSE_THEMES.get(h, []))
+
+        if not core_themes:
+            core_themes = PLANET_THEMES.get(ad_lord, [])
+
+        if "career advancement" in core_themes or "wealth" in core_themes or ad_lord in ["Sun", "Jupiter"]:
+            hooks.append({
+                "type": "career_wealth",
+                "trigger": f"Current Antardasha Lord ({ad_lord})",
+                "question": f"🔮 测一测我在当前「{ad_lord}运」期间，是否有实质性的财务跃迁或事业升职机会？",
+                "rationale": f"系统检测到你正处于 {ad_lord} 主导的运势周期，该星体掌管你的核心资源/事业宫位，近期极易触发财务或职级变动。"
+            })
+        elif "marriage" in core_themes or "business partnerships" in core_themes or ad_lord in ["Venus", "Moon"]:
+            hooks.append({
+                "type": "relationship",
+                "trigger": f"Current Antardasha Lord ({ad_lord})",
+                "question": f"🔮 我的正缘大概会在什么时间点出现？当前运势对我的婚恋/合伙关系有什么影响？",
+                "rationale": f"系统检测到你正处于 {ad_lord} 运势周期，极易触发亲密关系或重要合伙人的变动，建议深度扫描婚恋应期。"
+            })
+        elif "foreign lands" in core_themes or "hidden matters" in core_themes or "health issues" in core_themes or ad_lord in ["Ketu", "Rahu"]:
+            hooks.append({
+                "type": "transition_healing",
+                "trigger": f"Current Antardasha Lord ({ad_lord})",
+                "question": f"🔮 我近期感到强烈的内耗/变动倾向，是否适合出国、换环境或进行重大人生断舍离？",
+                "rationale": f"当前 {ad_lord} 运势激活了隐秘/变动宫位，容易带来精神内耗或海外发展的契机，需要诊断当前的卡点。"
+            })
+        else:
+            # Fallback hook for Antardasha
+            hooks.append({
+                "type": "general_ad",
+                "trigger": f"Current Antardasha Lord ({ad_lord})",
+                "question": f"🔮 当前「{ad_lord}运」对我接下来的 1-2 年有什么本质性的影响？",
+                "rationale": f"你目前处于 {ad_lord} 掌管的次级运势中，深度解读该星体能帮你把握近期的核心节奏。"
+            })
+
+    # 2. Narayana Dasha Hook
+    if isinstance(narayana_dasha, dict):
+        nd_obj = narayana_dasha.get("current_dasha")
+        nd_sign = None
+        if isinstance(nd_obj, dict):
+            nd_sign = nd_obj.get("md", {}).get("sign")
+        elif isinstance(nd_obj, str):
+            nd_sign = nd_obj
+
+        if nd_sign:
+            hooks.append({
+                "type": "macro_trend",
+                "trigger": f"Narayana Dasha ({nd_sign})",
+                "question": f"🔮 我在当前的「{nd_sign}星座大运」中，最应该把精力聚焦在哪个领域才能利益最大化？",
+                "rationale": f"Jaimini 系统的 {nd_sign} 大运主轴已确认。顺应星座大运的能量流动，能帮你找到未来几年的阻力最小路径。"
+            })
+
+    # 3. Upcoming Transit (Gochar) Mock Hook (Requires actual ephemeris in full implementation)
+    # For now, we inject a generic but highly actionable transit hook.
+    hooks.append({
+        "type": "transit_alert",
+        "trigger": "Upcoming Major Transit",
+        "question": "🔮 未来半年内，木星或土星的换座/过宫，会给我带来哪些具体的机遇或危机？",
+        "rationale": "流年大星（木/土）的轨迹往往是触发本命盘事件的最后一把钥匙，提前观测能帮你避坑或抓红利。"
+    })
+
+    # Ensure we return exactly 3 top hooks
+    # Remove duplicates by type
+    seen_types = set()
+    unique_hooks = []
+    for h in hooks:
+        if h["type"] not in seen_types:
+            unique_hooks.append(h)
+            seen_types.add(h["type"])
+
+    # Always provide a fallback general reading option
+    unique_hooks.append({
+        "type": "general_reading",
+        "trigger": "User Preference",
+        "question": "🔮 跳过引导，我想查看完整的十年人生起伏与本命深度体检图谱。",
+        "rationale": "生成最全面的静态命运基调与大运概览。"
+    })
+
+    return unique_hooks[:3]