feat: add Railway-ready Jyotish chat product

scripts/deployment_preflight.py

@@ -27,8 +27,8 @@ def delivery_matrix() -> list[dict]:
             "label": "Local dev",
             "user_url": "http://127.0.0.1:5173",
             "commands": [
+                ".venv/bin/python scripts/jyotish_api_server.py --host 127.0.0.1 --port 5200",
                 "cd jyotish-app && npm run dev -- --host 127.0.0.1 --port 5173",
-                "python3 scripts/jyotish_api_server.py --host 127.0.0.1 --port 5200",
             ],
             "api_required": True,
             "scope": "Full web/app user experience with local API.",

scripts/jyotish_api_server.py

@@ -22,6 +22,17 @@ from http.server import HTTPServer, BaseHTTPRequestHandler
 from pathlib import Path
 from urllib.parse import urlparse
 
+# This file is intentionally runnable as ``python scripts/jyotish_api_server.py``.
+# In that mode Python adds ``scripts/`` (not the repository root) to sys.path,
+# so later lazy imports such as ``from scripts.vedastro_gateway import ...``
+# otherwise fail during a real consultation request.
+SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
+REPO_ROOT = os.path.abspath(os.path.join(SCRIPTS_DIR, '..'))
+if REPO_ROOT not in sys.path:
+    sys.path.insert(0, REPO_ROOT)
+if SCRIPTS_DIR not in sys.path:
+    sys.path.insert(0, SCRIPTS_DIR)
+
 try:
     from scripts.local_env import load_local_env
 except ModuleNotFoundError:  # pragma: no cover - script execution path
@@ -35,9 +46,6 @@ try:
 except ModuleNotFoundError:  # pragma: no cover - script execution path
     from western_oracle_adapter import build_packet_from_oracle_payload
 
-SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
-REPO_ROOT = os.path.abspath(os.path.join(SCRIPTS_DIR, '..'))
-sys.path.insert(0, SCRIPTS_DIR)
 load_local_env(REPO_ROOT)
 _LOCAL_MODULE_CACHE = {}
 _API_CHART_CACHE_SCOPE = 'api_chart_response'
@@ -66,6 +74,209 @@ def _western_evidence_packet_from_body(body: dict, route_packet: dict) -> dict |
         }
 
 
+def _consultation_reference_date(body: dict) -> datetime:
+    raw = (
+        body.get('reference_date')
+        or body.get('transit_date')
+        or body.get('today')
+        or body.get('current_date')
+    )
+    if isinstance(raw, str) and raw.strip():
+        try:
+            return datetime.fromisoformat(raw.strip().replace('Z', '+00:00'))
+        except ValueError:
+            pass
+    return datetime.now()
+
+
+def _consultation_current_age(birth_payload: dict, body: dict) -> float:
+    try:
+        born = datetime(
+            int(birth_payload['year']),
+            int(birth_payload['month']),
+            int(birth_payload['day']),
+            int(float(birth_payload.get('hour', 0))),
+            int(float(birth_payload.get('minute', 0))),
+        )
+    except (KeyError, TypeError, ValueError):
+        return 0.0
+    reference = _consultation_reference_date(body)
+    return max(0.0, (reference.replace(tzinfo=None) - born).total_seconds() / (365.2425 * 86400))
+
+
+def _attach_local_consultation_layers(handler, chart: dict, birth_payload: dict, body: dict) -> dict:
+    """Attach locally-computable consultation layers before reports/evidence are assembled.
+
+    VedAstro is an optional external cross-check for the web chat. D9/D10, Arudha
+    padas (including A10/UL), and Narayana Dasha are available in this repository
+    and must not be reported as missing merely because the external provider is
+    not configured.
+    """
+    if not isinstance(chart, dict) or not chart.get('success', True):
+        return chart
+
+    modules = chart.get('modules') if isinstance(chart.get('modules'), dict) else {}
+    chart['modules'] = modules
+    planets = chart.get('planets') if isinstance(chart.get('planets'), dict) else {}
+    ascendant = chart.get('ascendant') if isinstance(chart.get('ascendant'), dict) else {}
+    diagnostics = []
+
+    if planets and ascendant and not isinstance(modules.get('varga_full'), dict):
+        try:
+            varga_response = handler._compute_varga_full({
+                'planets': planets,
+                'ascendant': ascendant,
+                'divisions': ['D2', 'D4', 'D9', 'D10', 'D11'],
+            })
+            varga_result = varga_response.get('result') if isinstance(varga_response, dict) else None
+            if isinstance(varga_result, dict):
+                modules['varga_full'] = varga_result
+        except Exception as exc:  # optional local layer; preserve core D1 result
+            diagnostics.append({'layer': 'varga_full', 'status': 'unavailable', 'reason': exc.__class__.__name__})
+
+    if planets and ascendant and not isinstance(modules.get('arudha_padas'), dict):
+        try:
+            jaimini = _load_local_module('jaimini')
+            planet_lons = {
+                name: float(data.get('lon'))
+                for name, data in planets.items()
+                if isinstance(data, dict) and data.get('lon') is not None
+            }
+            asc_sign_idx = int(ascendant.get('sign_idx', int(float(ascendant.get('lon', 0))) // 30)) % 12
+            arudha_padas = jaimini.calc_arudha_padas(asc_sign_idx, planet_lons)
+            if isinstance(arudha_padas, dict):
+                modules['arudha_padas'] = arudha_padas
+                modules.setdefault('jaimini', {})['arudha_padas'] = arudha_padas
+                chart['arudha_padas'] = arudha_padas
+        except Exception as exc:
+            diagnostics.append({'layer': 'arudha_padas', 'status': 'unavailable', 'reason': exc.__class__.__name__})
+
+    if planets and ascendant and not isinstance(modules.get('narayana_dasha'), dict):
+        try:
+            narayana = _load_local_module('narayana_dasha')
+            planet_lons = {
+                name: float(data.get('lon'))
+                for name, data in planets.items()
+                if isinstance(data, dict) and data.get('lon') is not None
+            }
+            asc_sign_idx = int(ascendant.get('sign_idx', int(float(ascendant.get('lon', 0))) // 30)) % 12
+            narayana_result = narayana.narayana_dasha_full_report(
+                lagna_sign_idx=asc_sign_idx,
+                planet_lons=planet_lons,
+                current_age=_consultation_current_age(birth_payload, body),
+                birth_year=int(birth_payload.get('year', 0) or 0),
+            )
+            if isinstance(narayana_result, dict):
+                modules['narayana_dasha'] = narayana_result
+        except Exception as exc:
+            diagnostics.append({'layer': 'narayana_dasha', 'status': 'unavailable', 'reason': exc.__class__.__name__})
+
+    chart['local_consultation_layers'] = {
+        'status': 'ready' if not diagnostics else 'partial',
+        'source': 'repository_local_engines',
+        'available': [
+            name
+            for name in ('varga_full', 'arudha_padas', 'narayana_dasha')
+            if isinstance(modules.get(name), dict) and modules.get(name)
+        ],
+        'diagnostics': diagnostics,
+    }
+    return chart
+
+
+def _build_consumer_context(
+    *,
+    question: str,
+    route_packet: dict,
+    chart: dict,
+    rectification: dict,
+    machine_evidence_packet: dict,
+    vedastro_official: dict,
+) -> dict:
+    """Build a chat-facing truth contract without exposing provider noise as a fatal error."""
+    sections = (
+        machine_evidence_packet.get('sections')
+        if isinstance(machine_evidence_packet.get('sections'), dict)
+        else {}
+    )
+    route = str(route_packet.get('question_type') or route_packet.get('primary_theme') or 'general')
+    route_requirements = {
+        'career': ['D1', 'D10', 'A10', 'dasha_boundaries', 'narayana_dasha'],
+        'relationship': ['D1', 'D9', 'UL', 'dasha_boundaries'],
+        'finance': ['D1', 'D2', 'dasha_boundaries'],
+        'timing': ['D1', 'dasha_boundaries', 'narayana_dasha'],
+        'general': ['D1', 'D9', 'dasha_boundaries'],
+    }
+    required = route_requirements.get(route, route_requirements['general'])
+    available_layers = sorted(
+        name for name, section in sections.items()
+        if isinstance(section, dict) and section.get('status') == 'used'
+    )
+    missing_route_layers = [
+        name for name in required
+        if not isinstance(sections.get(name), dict) or sections[name].get('status') != 'used'
+    ]
+    d1_ready = 'D1' in available_layers and bool(chart.get('success', True))
+    hard_blockers = [] if d1_ready else ['core_chart_unavailable']
+
+    official_state = (
+        sections.get('external_oracle_status', {}).get('status')
+        if isinstance(sections.get('external_oracle_status'), dict)
+        else 'official_blocked'
+    )
+    optional_unavailable = []
+    if official_state != 'official_verified':
+        optional_unavailable.append({
+            'layer': 'vedastro_official_cross_check',
+            'status': official_state,
+            'impact': 'external_cross_validation_only',
+        })
+
+    rect_summary = rectification.get('summary') if isinstance(rectification.get('summary'), dict) else {}
+    warned_vargas = rect_summary.get('warned') if isinstance(rect_summary.get('warned'), list) else []
+    disabled_vargas = rect_summary.get('disabled') if isinstance(rect_summary.get('disabled'), list) else []
+    relevant_warned_vargas = [name for name in warned_vargas if name in required]
+    precise_timing_requested = route == 'timing' or bool(re.search(r'(具体|精确|哪一|几月|月份|日期|何时|什么时候|时间点|年份)', question or ''))
+    timing_layers_ready = all(name not in missing_route_layers for name in ('dasha_boundaries', 'narayana_dasha'))
+    precision_allows_timing = not any(name in disabled_vargas for name in ('D9', 'D10'))
+    can_answer_precise_timing = d1_ready and timing_layers_ready and precision_allows_timing and not missing_route_layers
+
+    limitation_parts = []
+    if missing_route_layers:
+        limitation_parts.append(f"当前问题仍缺少 {', '.join(missing_route_layers)}")
+    if relevant_warned_vargas:
+        limitation_parts.append(f"{', '.join(relevant_warned_vargas)} 对出生时间精度敏感")
+
+    return {
+        'core_status': 'blocked' if hard_blockers else ('degraded' if missing_route_layers else 'ready'),
+        'calculation_source': 'repository_local_engine',
+        'route': route,
+        'available_layers': available_layers,
+        'missing_route_layers': missing_route_layers,
+        'optional_unavailable_layers': optional_unavailable,
+        'hard_blockers': hard_blockers,
+        'precision': {
+            'warned_layers': warned_vargas,
+            'disabled_layers': disabled_vargas,
+            'summary': rect_summary.get('headline'),
+        },
+        'answer_policy': {
+            'can_answer_direction': d1_ready,
+            'can_answer_chart_interpretation': d1_ready,
+            'can_answer_precise_timing': can_answer_precise_timing,
+            'precise_timing_requested': precise_timing_requested,
+            'should_lead_with_limitations': bool(hard_blockers) or (precise_timing_requested and not can_answer_precise_timing),
+            'provider_unavailable_is_fatal': False,
+        },
+        'user_facing_limitation': '；'.join(limitation_parts) if limitation_parts else None,
+        'provider_status': {
+            'vedastro': official_state,
+            'role': 'optional_external_cross_check',
+            'runtime_status': vedastro_official.get('status') if isinstance(vedastro_official, dict) else None,
+        },
+    }
+
+
 def execute_consultation_workflow(
     handler,
     *,
@@ -175,6 +386,8 @@ def execute_consultation_workflow(
                 computed_chart = True
             executed_steps.append('compute_chart')
 
+    chart = _attach_local_consultation_layers(handler, chart, birth_payload, body)
+
     historical_backtest = {}
     if 'run_historical_event_backtest' in runtime_planner.get('sync_steps', []):
         historical_backtest = handler._run_high_rigor_historical_backtest(birth_payload, events)
@@ -222,6 +435,14 @@ def execute_consultation_workflow(
         vedastro_official=vedastro_official,
         vedastro_archive_manifest=vedastro_archive_manifest,
     )
+    consumer_context = _build_consumer_context(
+        question=question,
+        route_packet=route_packet,
+        chart=chart,
+        rectification=rectification,
+        machine_evidence_packet=machine_evidence_packet,
+        vedastro_official=vedastro_official,
+    )
     real_case_calibration = _UNIFIED_CONSULTATION_ORCHESTRATOR.real_case_calibration_catalog(
         route_packet=route_packet,
         machine_evidence_packet=machine_evidence_packet,
@@ -283,6 +504,7 @@ def execute_consultation_workflow(
         'runtime_truth': runtime_truth,
         'interpretation_source_runtime_coverage': interpretation_source_runtime_coverage,
         'machine_evidence_packet': machine_evidence_packet,
+        'consumer_context': consumer_context,
         'western_evidence_packet': western_evidence_packet or {},
         'real_case_calibration': real_case_calibration,
         'runtime_evidence_log': runtime_evidence_log,
@@ -2623,7 +2845,15 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
         )
         if skip_full_reading:
             module_status['full_reading'] = 'skipped_reuse_chart_data'
-        full_modules = full_reading.get('modules', {}) if isinstance(full_reading, dict) else {}
+        # Consultation workflow already computed and attached the local modules to
+        # ``chart_data`` before asking for a thematic report.  When the caller
+        # explicitly skips a second full-reading pass, reuse those modules rather
+        # than silently discarding D9/D10, Arudha padas and Narayana Dasha.
+        chart_modules = raw.get('modules') if isinstance(raw.get('modules'), dict) else {}
+        computed_modules = full_reading.get('modules', {}) if isinstance(full_reading, dict) else {}
+        full_modules = {**chart_modules, **computed_modules}
+        if chart_modules:
+            module_status['chart_modules'] = 'reused'
         chart = None
         if full_reading:
             chart = self._chart_from_full_reading(full_reading)
@@ -2671,6 +2901,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             'periods': (dasha or {}).get('periods') or (dasha or {}).get('timeline') or [],
             'yogas': chart.get('yogas') or [],
             'ashtakavarga': ashtakavarga or {},
+            'modules': full_modules,
         }
         if yogas and isinstance(yogas.get('result'), dict):
             enriched['yogas'] = list(enriched['yogas']) + (yogas['result'].get('extended_yogas') or [])
@@ -2697,6 +2928,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             'warnings': warnings,
             'evidence_counts': {theme: len(items) for theme, items in evidence.items()},
             'full_reading_used': bool(full_reading),
+            'chart_modules_reused': bool(chart_modules),
             'full_reading_summary': full_reading.get('summary', {}) if isinstance(full_reading, dict) else {},
             'full_reading_module_count': len(full_modules) if isinstance(full_modules, dict) else 0,
         }
@@ -2704,11 +2936,19 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
     def _thematic_evidence_source(self, mode, derived_context, has_custom_evidence):
         if mode == 'derived_chart_evidence' and derived_context:
             full_reading_used = bool(derived_context.get('full_reading_used'))
+            chart_modules_reused = bool(derived_context.get('chart_modules_reused'))
             return {
                 'mode': mode,
-                'source': 'full_reading_modules' if full_reading_used else 'birth_or_chart_payload',
+                'source': (
+                    'full_reading_modules'
+                    if full_reading_used
+                    else 'reused_chart_modules'
+                    if chart_modules_reused
+                    else 'birth_or_chart_payload'
+                ),
                 'sample_fallback': False,
                 'full_reading_used': full_reading_used,
+                'chart_modules_reused': chart_modules_reused,
                 'full_reading_module_count': derived_context.get('full_reading_module_count', 0),
                 'full_reading_summary': derived_context.get('full_reading_summary', {}),
                 'module_status': derived_context.get('module_status', {}),
@@ -2856,6 +3096,66 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
                 source='shadbala',
                 details=top_strength,
             ))
+
+        varga = full_modules.get('varga_full') if isinstance(full_modules.get('varga_full'), dict) else {}
+        d10 = varga.get('D10_Dasamsa') or varga.get('D10')
+        if isinstance(d10, dict) and d10:
+            ascendant = d10.get('ascendant') or d10.get('Ascendant') or {}
+            d10_planets = d10.get('planets') if isinstance(d10.get('planets'), dict) else {}
+            house_chart = d10.get('house_chart') if isinstance(d10.get('house_chart'), list) else []
+            tenth_house = house_chart[9] if len(house_chart) >= 10 and isinstance(house_chart[9], list) else []
+            asc_sign = ascendant.get('sign_cn') or ascendant.get('sign') or '未知'
+            tenth_label = '、'.join(str(name) for name in tenth_house) if tenth_house else '无行星直接落入'
+            items.append(self._theme_evidence(
+                'D10-Dashamsha-local',
+                'D10',
+                f'D10 事业分盘已完成：上升落 {asc_sign}，第10宫为{tenth_label}；用于校验职业角色、执行方式与社会位置。',
+                'neutral',
+                'strong',
+                source='chart.modules.varga_full.D10_Dasamsa',
+                details={
+                    'ascendant': ascendant,
+                    'tenth_house_planets': tenth_house,
+                    'planet_count': len(d10_planets),
+                },
+            ))
+
+        arudha = full_modules.get('arudha_padas') if isinstance(full_modules.get('arudha_padas'), dict) else {}
+        if not arudha and isinstance(full_modules.get('jaimini'), dict):
+            candidate = full_modules['jaimini'].get('arudha_padas')
+            arudha = candidate if isinstance(candidate, dict) else {}
+        padas = arudha.get('padas') if isinstance(arudha.get('padas'), dict) else arudha
+        a10 = padas.get('A10') if isinstance(padas, dict) else None
+        if isinstance(a10, dict) and a10:
+            sign = a10.get('sign_cn') or a10.get('sign') or '未知'
+            lord = a10.get('lord') or '未知'
+            items.append(self._theme_evidence(
+                'A10-Karma-Pada-local',
+                'A10',
+                f'A10（事业形象点）落 {sign}，主星为 {lord}；用于观察职业品牌、可见度与外界如何识别你的事业角色。',
+                'neutral',
+                'strong',
+                source='chart.modules.arudha_padas.A10',
+                details=a10,
+            ))
+
+        narayana = full_modules.get('narayana_dasha') if isinstance(full_modules.get('narayana_dasha'), dict) else {}
+        current_narayana = narayana.get('current_dasha') if isinstance(narayana.get('current_dasha'), dict) else {}
+        current_md = current_narayana.get('md') if isinstance(current_narayana.get('md'), dict) else {}
+        current_ad = current_narayana.get('ad') if isinstance(current_narayana.get('ad'), dict) else {}
+        if current_md:
+            md_sign = current_md.get('sign_cn') or current_md.get('sign') or '未知'
+            ad_sign = current_ad.get('sign_cn') or current_ad.get('sign') or '未知'
+            items.append(self._theme_evidence(
+                'Narayana-Dasha-local',
+                'Narayana',
+                f'Narayana Dasha 已完成：当前主周期为 {md_sign}，子周期为 {ad_sign}；可与行星大运交叉判断事业阶段。',
+                'neutral',
+                'moderate',
+                source='chart.modules.narayana_dasha.current_dasha',
+                details={'current_dasha': current_narayana},
+            ))
+
         convergence = full_modules.get('dasa_convergence') if isinstance(full_modules, dict) else {}
         if not isinstance(convergence, dict):
             convergence = {}

scripts/jyotish_engine.py

@@ -1188,9 +1188,13 @@ def _base_strict_narrative_payload(route_label, strict, *, fallback_headline, st
     monthly_frame = strict.get('monthly_adjudication_summary') if isinstance(strict, dict) else {}
     monthly_frame = monthly_frame if isinstance(monthly_frame, dict) else {}
     event_judgement = strict.get('event_judgement') if isinstance(strict, dict) else {}
+    event_judgement = event_judgement if isinstance(event_judgement, dict) else {}
     adjudication = strict.get('adjudication_stages') if isinstance(strict, dict) else {}
+    adjudication = adjudication if isinstance(adjudication, dict) else {}
     boundary_contract = strict.get('prediction_boundary_contract') if isinstance(strict, dict) else {}
+    boundary_contract = boundary_contract if isinstance(boundary_contract, dict) else {}
     confidence_boundary = boundary_contract.get('confidence_boundary') if isinstance(boundary_contract, dict) else {}
+    confidence_boundary = confidence_boundary if isinstance(confidence_boundary, dict) else {}
     confidence_cap = 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
 

scripts/run_quality_gate.py

@@ -299,8 +299,8 @@ def format_failure_summary(
         lines.append(f"stderr tail:\n{tail_text(stderr)}")
     lines.extend([
         "普通用户启动路径:",
-        "1. 网页服务：cd jyotish-app && npm run dev -- --host 127.0.0.1 --port 5173",
-        "2. 本地 API 服务：python3 scripts/jyotish_api_server.py --host 127.0.0.1 --port 5200",
+        "1. 本地 API 服务：.venv/bin/python scripts/jyotish_api_server.py --host 127.0.0.1 --port 5200",
+        "2. 网页服务：cd jyotish-app && npm run dev -- --host 127.0.0.1 --port 5173",
         "3. Open http://127.0.0.1:5173, then open Trust Center and run the health check.",
         "4. PWA 安装壳只包装网页服务，本地 API 服务仍需单独启动。",
         "Next action: Run the focused command above, add --keep-logs for browser click smoke, then compare the app state with the startup path above.",

scripts/unified_consultation_orchestrator.py

@@ -391,6 +391,15 @@ class UnifiedConsultationOrchestrator:
             if isinstance(chart_data.get("special_lagnas"), dict)
             else modules.get("special_lagnas") if isinstance(modules.get("special_lagnas"), dict) else {}
         )
+        arudha_padas = (
+            chart_data.get("arudha_padas")
+            if isinstance(chart_data.get("arudha_padas"), dict)
+            else modules.get("arudha_padas") if isinstance(modules.get("arudha_padas"), dict) else {}
+        )
+        if not arudha_padas and isinstance(modules.get("jaimini"), dict):
+            jaimini_arudha = modules["jaimini"].get("arudha_padas")
+            arudha_padas = jaimini_arudha if isinstance(jaimini_arudha, dict) else {}
+        pada_map = arudha_padas.get("padas") if isinstance(arudha_padas.get("padas"), dict) else arudha_padas
         ascendant = base_chart.get("ascendant") if isinstance(base_chart.get("ascendant"), dict) else {}
         ascendant_sign = ascendant.get("sign") if isinstance(ascendant, dict) else None
         functional_layer = derive_functional_benefic_malefic(ascendant_sign)
@@ -402,6 +411,15 @@ class UnifiedConsultationOrchestrator:
             or official.get("raw_payload")
             or official.get("raw")
         )
+        official_state = self._vedastro_cloud_state(vedastro_official)
+        raw_response_section = (
+            self._section(raw_response, "vedastro_official.raw_response")
+            if official_state == "official_verified"
+            else {
+                "status": "received_unverified" if raw_response else "missing",
+                "source_path": "vedastro_official.raw_response",
+            }
+        )
         sections = {
             "D1": self._section(
                 base_chart.get("planets") and base_chart.get("ascendant"),
@@ -414,22 +432,35 @@ class UnifiedConsultationOrchestrator:
             "planet_degrees": self._section(base_chart.get("planets"), "chart.planets"),
             "house_degrees": self._section(base_chart.get("houses") or chart_data.get("houses"), "chart.houses"),
             "dasha_boundaries": self._section(modules.get("dasha") or chart_data.get("dasha"), "modules.dasha"),
+            "narayana_dasha": self._section(modules.get("narayana_dasha"), "modules.narayana_dasha"),
             "shadbala": self._section(modules.get("shadbala") or chart_data.get("shadbala"), "modules.shadbala"),
             "ashtakavarga": self._section(modules.get("ashtakavarga") or chart_data.get("ashtakavarga"), "modules.ashtakavarga"),
             "yogas": self._section(modules.get("yogas") or chart_data.get("yogas"), "modules.yogas"),
-            "UL": self._section(special_lagnas.get("UL") or special_lagnas.get("Upapada_Lagna"), "special_lagnas.UL"),
-            "A7": self._section(special_lagnas.get("A7") or special_lagnas.get("Darapada"), "special_lagnas.A7"),
-            "A10": self._section(special_lagnas.get("A10") or special_lagnas.get("A10_Karma_Pada"), "special_lagnas.A10"),
+            "UL": self._section(
+                pada_map.get("UL")
+                or arudha_padas.get("upapada")
+                or special_lagnas.get("UL")
+                or special_lagnas.get("Upapada_Lagna"),
+                "modules.arudha_padas.UL",
+            ),
+            "A7": self._section(
+                pada_map.get("A7") or special_lagnas.get("A7") or special_lagnas.get("Darapada"),
+                "modules.arudha_padas.A7",
+            ),
+            "A10": self._section(
+                pada_map.get("A10") or special_lagnas.get("A10") or special_lagnas.get("A10_Karma_Pada"),
+                "modules.arudha_padas.A10",
+            ),
             "KP_cusp": self._section(modules.get("kp") or modules.get("kp_cusps") or chart_data.get("kp_cusps"), "modules.kp_cusps"),
             "functional_benefic_malefic": self._section(
                 functional_layer if functional_layer.get("status") == "used" else None,
                 "chart.ascendant.sign -> scripts.functional_benefics",
             ),
             "external_oracle_status": {
-                "status": self._vedastro_cloud_state(vedastro_official),
+                "status": official_state,
                 "source_path": "vedastro_official.runtime_truth",
             },
-            "vedastro_official_raw_response": self._section(raw_response, "vedastro_official.raw_response"),
+            "vedastro_official_raw_response": raw_response_section,
             "vedastro_official_raw_archive_manifest": self._section(
                 archive_manifest if archive_manifest.get("archive_count") else None,
                 "vedastro_gateway.archives",

scripts/vedastro_official_capability_runner.py

@@ -13,6 +13,13 @@ from datetime import datetime
 from pathlib import Path
 from typing import Any
 
+try:
+    from scripts.vedastro_python_bridge import call_method as call_bridge_method
+    from scripts.vedastro_python_bridge import list_capabilities as list_bridge_capabilities
+except ModuleNotFoundError:  # pragma: no cover - direct script execution
+    from vedastro_python_bridge import call_method as call_bridge_method
+    from vedastro_python_bridge import list_capabilities as list_bridge_capabilities
+
 
 ROOT = Path(__file__).resolve().parents[1]
 PYTHON = sys.executable
@@ -22,6 +29,7 @@ CATALOG_STUB_ENV = "VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB"
 PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu", "Ascendant"]
 HOUSES = [f"House{i}" for i in range(1, 13)]
 DEFAULT_SIGIL_SAMPLE_LIMIT = int(os.environ.get("VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT", "0") or 0)
+SNAPSHOT_FANOUT_ENABLED = os.environ.get("VEDASTRO_FULL_SNAPSHOT_FANOUT_ENABLED", "1").strip().lower() in {"1", "true", "yes", "on"}
 DOMAIN_ORDER = [
     "career",
     "marriage",
@@ -434,18 +442,7 @@ def _method_payload_for_instance(method: str, case: dict[str, Any], identity: st
 
 
 def _call_bridge(method: str, payload: dict[str, Any]) -> dict[str, Any]:
-    completed = subprocess.run(
-        [PYTHON, str(BRIDGE), "--method", method, "--params-json", json.dumps(payload, ensure_ascii=False)],
-        cwd=ROOT,
-        text=True,
-        capture_output=True,
-        timeout=240,
-        check=False,
-        env=os.environ.copy(),
-    )
-    if completed.returncode != 0:
-        return {"available": False, "status": "bridge_runtime_error", "stderr": (completed.stderr or "").strip()}
-    return json.loads(completed.stdout)
+    return call_bridge_method(method, payload)
 
 
 def _list_official_capabilities() -> dict[str, Any]:
@@ -454,36 +451,7 @@ def _list_official_capabilities() -> dict[str, Any]:
         payload = json.loads(stub_raw)
         payload.setdefault("source", "stubbed_official_capability_catalog")
         return payload
-
-    completed = subprocess.run(
-        [PYTHON, str(BRIDGE), "--list-capabilities"],
-        cwd=ROOT,
-        text=True,
-        capture_output=True,
-        timeout=240,
-        check=False,
-        env=os.environ.copy(),
-    )
-    if completed.returncode != 0:
-        return {
-            "available": False,
-            "status": "bridge_runtime_error",
-            "capabilities": [],
-            "buckets": {},
-            "stderr": (completed.stderr or "").strip(),
-            "source": "vedastro_official_capability_runner",
-        }
-    try:
-        return json.loads(completed.stdout)
-    except json.JSONDecodeError:
-        return {
-            "available": False,
-            "status": "bridge_invalid_json",
-            "capabilities": [],
-            "buckets": {},
-            "stdout_excerpt": (completed.stdout or "").strip()[:500],
-            "source": "vedastro_official_capability_runner",
-        }
+    return list_bridge_capabilities()
 
 
 def run_selected_methods(methods: list[str], birth_payload: dict[str, Any]) -> dict[str, Any]:
@@ -769,7 +737,7 @@ def run_snapshot_bundle(bundle: str, birth_payload: dict[str, Any]) -> dict[str,
 
     chart_core: dict[str, Any] = {}
     planet_statuses: dict[str, str] = {}
-    for planet in PLANETS:
+    for planet in PLANETS if SNAPSHOT_FANOUT_ENABLED else []:
         stub_key = f"AllPlanetData:{planet}"
         if stub_key in stub_map:
             report = stub_map[stub_key]
@@ -797,7 +765,7 @@ def run_snapshot_bundle(bundle: str, birth_payload: dict[str, Any]) -> dict[str,
 
     house_core: dict[str, Any] = {}
     house_statuses: dict[str, str] = {}
-    for house in HOUSES:
+    for house in HOUSES if SNAPSHOT_FANOUT_ENABLED else []:
         stub_key = f"AllHouseData:{house}"
         if stub_key in stub_map:
             report = stub_map[stub_key]
@@ -861,10 +829,11 @@ def run_snapshot_bundle(bundle: str, birth_payload: dict[str, Any]) -> dict[str,
         "available": bool(coverage_sections),
         "status": overall_status,
         "summary": {
-            "requested_method_count": len(PLANETS) + len(HOUSES) + len(scalar_methods),
-            "executed_method_count": len(PLANETS) + len(HOUSES) + len(scalar_methods),
+            "requested_method_count": (len(PLANETS) + len(HOUSES) if SNAPSHOT_FANOUT_ENABLED else 0) + len(scalar_methods),
+            "executed_method_count": (len(PLANETS) + len(HOUSES) if SNAPSHOT_FANOUT_ENABLED else 0) + len(scalar_methods),
             "ok_count": ok_count,
             "skipped_count": skipped_count,
+            "fanout_enabled": SNAPSHOT_FANOUT_ENABLED,
         },
         "result": {
             "snapshot_sections": snapshot_sections,
@@ -874,6 +843,7 @@ def run_snapshot_bundle(bundle: str, birth_payload: dict[str, Any]) -> dict[str,
                 "filled_sections": coverage_sections,
                 "planet_count": len(chart_core),
                 "house_count": len(house_core),
+                "fanout_enabled": SNAPSHOT_FANOUT_ENABLED,
             },
         },
     }

scripts/vedastro_python_bridge.py

@@ -14,8 +14,12 @@ separate. It stays deliberately thin:
 from __future__ import annotations
 
 import argparse
+import contextlib
+import enum
 import inspect
+import importlib
 import importlib.util
+import io
 import json
 import os
 import subprocess
@@ -339,6 +343,148 @@ if __name__ == "__main__":
 """
 
 
+def _offset_to_string(offset: Any) -> str:
+    if isinstance(offset, str):
+        raw = offset.strip()
+        if not raw:
+            return "+00:00"
+        if raw[0] in "+-" and ":" in raw:
+            return raw
+        try:
+            offset = float(raw)
+        except ValueError:
+            return raw
+    if isinstance(offset, (int, float)):
+        sign = "+" if offset >= 0 else "-"
+        absolute = abs(float(offset))
+        hours = int(absolute)
+        minutes = int(round((absolute - hours) * 60))
+        if minutes == 60:
+            hours += 1
+            minutes = 0
+        return f"{sign}{hours:02d}:{minutes:02d}"
+    return "+00:00"
+
+
+def _import_runtime_module() -> tuple[str | None, Any | None]:
+    for name in MODULE_CANDIDATES:
+        with contextlib.redirect_stdout(io.StringIO()):
+            try:
+                return name, importlib.import_module(name)
+            except ModuleNotFoundError:
+                continue
+    return None, None
+
+
+def _coerce_runtime_value(module: Any, value: Any) -> Any:
+    if isinstance(value, list):
+        return [_coerce_runtime_value(module, item) for item in value]
+    if isinstance(value, dict):
+        enum_name = value.get("__vedastro_enum__")
+        if enum_name:
+            return getattr(getattr(module, enum_name), value["value"])
+        type_name = value.get("__vedastro_type__")
+        if type_name == "GeoLocation":
+            return module.GeoLocation(value["location_name"], value["longitude"], value["latitude"])
+        if type_name == "Time":
+            geolocation = _coerce_runtime_value(module, value["geolocation"]) if value.get("geolocation") else None
+            if value.get("time_string") is not None:
+                return module.Time(value["time_string"], geolocation)
+            time_string = (
+                f"{int(value['hour']):02d}:{int(value['minute']):02d} "
+                f"{int(value['day']):02d}/{int(value['month']):02d}/{int(value['year'])} "
+                f"{_offset_to_string(value.get('offset', '+00:00'))}"
+            )
+            return module.Time(time_string, geolocation)
+        if type_name:
+            target = getattr(module, type_name)
+            args = [_coerce_runtime_value(module, item) for item in value.get("args", [])]
+            kwargs = {key: _coerce_runtime_value(module, item) for key, item in value.get("kwargs", {}).items()}
+            if args or kwargs:
+                return target(*args, **kwargs)
+            payload = {key: _coerce_runtime_value(module, item) for key, item in value.items() if not key.startswith("__")}
+            return target(**payload)
+        return {key: _coerce_runtime_value(module, item) for key, item in value.items()}
+    return value
+
+
+def _serialize_runtime_value(value: Any) -> Any:
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    if isinstance(value, enum.Enum):
+        return {"type": type(value).__name__, "name": value.name, "value": value.value}
+    if isinstance(value, (list, tuple, set)):
+        return [_serialize_runtime_value(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _serialize_runtime_value(item) for key, item in value.items()}
+    if hasattr(value, "to_json"):
+        try:
+            payload = value.to_json()
+            if isinstance(payload, str):
+                try:
+                    return json.loads(payload)
+                except json.JSONDecodeError:
+                    return payload
+            return _serialize_runtime_value(payload)
+        except Exception:
+            pass
+    if hasattr(value, "__dict__"):
+        return {key: _serialize_runtime_value(item) for key, item in vars(value).items() if not key.startswith("_")}
+    return str(value)
+
+
+def _resolve_runtime_callable(module: Any, method_name: str) -> Any:
+    parts = [part for part in method_name.split(".") if part]
+    if not parts:
+        raise AttributeError("empty_method_name")
+    if len(parts) == 1:
+        return getattr(module.Calculate, parts[0])
+    current = module
+    for part in parts:
+        current = getattr(current, part)
+    return current
+
+
+def _call_in_current_python(method: str, params: dict[str, Any]) -> dict[str, Any]:
+    module_name, module = _import_runtime_module()
+    if module is None:
+        return _missing_package_result(method)
+    try:
+        target = _resolve_runtime_callable(module, method)
+        args: list[Any] = []
+        kwargs: dict[str, Any] = {}
+        if isinstance(params, dict) and ("args" in params or "kwargs" in params):
+            args = [_coerce_runtime_value(module, item) for item in params.get("args", [])]
+            kwargs = {key: _coerce_runtime_value(module, item) for key, item in params.get("kwargs", {}).items()}
+        elif isinstance(params, dict):
+            kwargs = {key: _coerce_runtime_value(module, item) for key, item in params.items()}
+        elif isinstance(params, list):
+            args = [_coerce_runtime_value(module, item) for item in params]
+        elif params is not None:
+            args = [_coerce_runtime_value(module, params)]
+        with contextlib.redirect_stdout(io.StringIO()):
+            result = target(*args, **kwargs)
+        return {
+            "available": True,
+            "status": "ok",
+            "method": method,
+            "module_name": module_name,
+            "result": _serialize_runtime_value(result),
+            "python_bin": sys.executable,
+            "source": "vedastro_python_bridge",
+        }
+    except Exception as exc:
+        return {
+            "available": False,
+            "status": "bridge_runtime_error",
+            "method": method,
+            "module_name": module_name,
+            "python_bin": sys.executable,
+            "error": {"type": type(exc).__name__, "message": str(exc)},
+            "source": "vedastro_python_bridge",
+        }
+
+
 def _package_available() -> bool:
     if os.environ.get(FORCE_UNAVAILABLE_ENV, "").strip().lower() in {"1", "true", "yes"}:
         return False
@@ -537,6 +683,9 @@ def call_method(method: str, params: dict[str, Any]) -> dict[str, Any]:
     if os.environ.get(FORCE_UNAVAILABLE_ENV, "").strip().lower() in {"1", "true", "yes"}:
         return _missing_package_result(method)
 
+    if _package_available():
+        return _call_in_current_python(method, params)
+
     python_bin = _select_python_bin()
     if not python_bin:
         return _missing_package_result(method)

scripts/vedastro_service_adapter.py

@@ -15,6 +15,7 @@ import json
 import os
 import socket
 import subprocess
+import sys
 import threading
 import time
 from datetime import datetime
@@ -36,6 +37,15 @@ VEDASTRO_PYTHON_BRIDGE = ROOT / "scripts" / "vedastro_python_bridge.py"
 VEDASTRO_OFFICIAL_CAPABILITY_RUNNER = ROOT / "scripts" / "vedastro_official_capability_runner.py"
 
 
+def _vedastro_python_bin() -> str:
+    """Run VedAstro child processes in the active backend interpreter by default."""
+    return (
+        os.environ.get("VEDASTRO_PYTHON_BIN", "").strip()
+        or os.environ.get("PYTHON_BIN", "").strip()
+        or sys.executable
+    )
+
+
 PARITY_CASES = {
     "steve_jobs_public_aa": {
         "year": 1955,
@@ -927,7 +937,7 @@ def _call_vedastro_python_bridge_high_value(method_key: str, payload: dict[str,
     try:
         completed = subprocess.run(
             [
-                os.environ.get("PYTHON_BIN", "") or "python3",
+                _vedastro_python_bin(),
                 str(VEDASTRO_PYTHON_BRIDGE),
                 "--high-value",
                 method_key,
@@ -1025,7 +1035,7 @@ def _try_official_capability_runner_snapshot_bundle(case: dict[str, Any]) -> dic
     try:
         completed = subprocess.run(
             [
-                os.environ.get("PYTHON_BIN", "") or "python3",
+                _vedastro_python_bin(),
                 str(VEDASTRO_OFFICIAL_CAPABILITY_RUNNER),
                 "--bundle",
                 "official_full_snapshot",
@@ -1102,7 +1112,7 @@ def _try_official_full_capability_catalog_bundle(case: dict[str, Any]) -> dict[s
     try:
         completed = subprocess.run(
             [
-                os.environ.get("PYTHON_BIN", "") or "python3",
+                _vedastro_python_bin(),
                 str(VEDASTRO_OFFICIAL_CAPABILITY_RUNNER),
                 "--bundle",
                 "official_full_capability_catalog",
@@ -2479,6 +2489,18 @@ def _run_official_full_snapshot_case(case: dict[str, Any], case_id: str = "user_
     )
     endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
     network_enabled = os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() in {"1", "true", "yes"}
+    fanout_enabled = os.environ.get("VEDASTRO_FULL_SNAPSHOT_FANOUT_ENABLED", "1").strip().lower() in {"1", "true", "yes", "on"}
+    if bridge_sections and not fanout_enabled:
+        return _normalize_official_full_snapshot_success(
+            endpoint,
+            manifest,
+            bridge_sections,
+            bridge_section_statuses,
+            1,
+            [],
+            official_python_bundle=official_python_bundle,
+            official_full_capability_catalog=official_full_capability_catalog,
+        )
     if budget_exhausted and endpoint and network_enabled and _is_official_public_endpoint(endpoint):
         result = {
             "backend": "vedastro_service_adapter_candidate",
@@ -2726,6 +2748,17 @@ def _run_range_scan_case(case: dict[str, Any], domain: str, start_date: str, end
             "source_metadata": _source_metadata(endpoint),
         }
 
+    range_scan_enabled = os.environ.get("VEDASTRO_RANGE_SCAN_NETWORK_ENABLED", "1").strip().lower() in {"1", "true", "yes", "on"}
+    if not range_scan_enabled:
+        return {
+            "backend": "vedastro_service_adapter_candidate",
+            "available": False,
+            "status": "network_execution_disabled",
+            "reason": "VEDASTRO_RANGE_SCAN_NETWORK_ENABLED is disabled for the interactive chat path.",
+            "request_preview": request_preview,
+            "source_metadata": _source_metadata(endpoint),
+        }
+
     sample_dates = _iter_sample_dates(start_date, end_date)
     reports: list[dict[str, Any]] = []
     for sample_date in sample_dates: