fix(consult): race a bounded VedAstro gateway on the chat path

Foreground still skips the heavy overview stack, but overlaps a timed
gateway so official evidence can merge when it finishes in budget.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/jyotish_api_server.py

@@ -22,7 +22,7 @@ import threading
 import time
 import tempfile
 import uuid
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
 from datetime import datetime, timedelta
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
@@ -105,6 +105,11 @@ _ASYNC_JOB_EXECUTOR = ThreadPoolExecutor(
     max_workers=_ASYNC_JOB_WORKERS,
     thread_name_prefix='jyotish-job',
 )
+_FOREGROUND_VEDASTRO_WORKERS = max(int(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_WORKERS', '2')), 1)
+_FOREGROUND_VEDASTRO_EXECUTOR = ThreadPoolExecutor(
+    max_workers=_FOREGROUND_VEDASTRO_WORKERS,
+    thread_name_prefix='jyotish-vedastro-fg',
+)
 _ASYNC_JOB_CAPACITY = threading.BoundedSemaphore(_ASYNC_JOB_WORKERS + _ASYNC_JOB_QUEUE_SIZE)
 _RATE_LIMIT_LOCK = threading.Lock()
 _RATE_LIMIT_BUCKETS: dict[str, tuple[float, int]] = {}
@@ -1586,6 +1591,123 @@ def _build_consumer_context(
     }
 
 
+_VEDASTRO_NATAL_SIGN_KEYS = ('sun', 'moon', 'ascendant')
+_VEDASTRO_COMPACT_DENY_KEYS = {
+    'lat', 'lon', 'lng', 'latitude', 'longitude', 'hour', 'minute', 'second',
+    'date', 'time', 'birth', 'tz', 'timezone', 'city', 'address', 'reported_time',
+}
+
+
+def _foreground_vedastro_budget_seconds() -> float:
+    raw = str(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS', '8')).strip()
+    try:
+        return min(max(float(raw), 2.0), 12.0)
+    except ValueError:
+        return 8.0
+
+
+def _foreground_vedastro_join_seconds() -> float:
+    raw = str(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS', '1.5')).strip()
+    try:
+        return min(max(float(raw), 0.0), 3.0)
+    except ValueError:
+        return 1.5
+
+
+def _vedastro_natal_signs(raw) -> dict | None:
+    if not isinstance(raw, dict):
+        return None
+    natal = raw.get('natal') if isinstance(raw.get('natal'), dict) else raw
+    if not isinstance(natal, dict):
+        return None
+    compact = {}
+    sign_names = set(SIGNS)
+    for key in _VEDASTRO_NATAL_SIGN_KEYS:
+        value = natal.get(key) or natal.get(key.title()) or natal.get(key.capitalize())
+        sign = None
+        if isinstance(value, str) and value in sign_names:
+            sign = value
+        elif isinstance(value, dict):
+            candidate = value.get('sign') or value.get('rasi')
+            if isinstance(candidate, str) and candidate in sign_names:
+                sign = candidate
+        if sign:
+            compact[key] = sign
+    return compact or None
+
+
+def _compact_vedastro_cross_check(vedastro_official: dict, vedastro_gateway: dict | None = None) -> dict:
+    official = vedastro_official if isinstance(vedastro_official, dict) else {}
+    gateway = vedastro_gateway if isinstance(vedastro_gateway, dict) else {}
+    state = str(
+        official.get('official_closure_state')
+        or gateway.get('official_closure_state')
+        or official.get('status')
+        or gateway.get('status')
+        or 'official_blocked'
+    )
+    reason = official.get('official_closure_reason') or gateway.get('official_closure_reason')
+    compact = {
+        'status': 'executed' if state == 'official_verified' else 'blocked',
+        'official_closure_state': state,
+    }
+    if reason:
+        compact['official_closure_reason'] = str(reason)
+    raw = (
+        official.get('raw_response')
+        or official.get('official_raw_response')
+        or gateway.get('official_raw_response')
+        or gateway.get('raw_response')
+    )
+    natal = _vedastro_natal_signs(raw)
+    if natal:
+        compact['natal'] = natal
+    for denied in _VEDASTRO_COMPACT_DENY_KEYS:
+        compact.pop(denied, None)
+        if isinstance(compact.get('natal'), dict):
+            compact['natal'].pop(denied, None)
+    return compact
+
+
+def _blocked_foreground_vedastro(*, reason: str, error_type: str | None = None) -> dict:
+    packet = {
+        'scope': 'vedastro_gateway_run',
+        'status': 'official_blocked',
+        'official_closure_state': 'official_blocked',
+        'official_closure_reason': reason,
+    }
+    if error_type:
+        packet['error_type'] = error_type
+    return packet
+
+
+def _run_foreground_vedastro_gateway(handler, body: dict) -> dict:
+    try:
+        with temporary_timeout_seconds(_foreground_vedastro_budget_seconds()):
+            result = handler._compute_vedastro_gateway_run(body)
+    except Exception as exc:
+        return _blocked_foreground_vedastro(
+            reason='gateway_invocation_error',
+            error_type=type(exc).__name__,
+        )
+    return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason='gateway_invocation_error')
+
+
+def _join_foreground_vedastro(future, *, timeout: float) -> dict:
+    if future is None:
+        return _blocked_foreground_vedastro(reason='foreground_optional_evidence_timeout')
+    try:
+        result = future.result(timeout=timeout)
+    except FuturesTimeoutError:
+        return _blocked_foreground_vedastro(reason='foreground_optional_evidence_timeout')
+    except Exception as exc:
+        return _blocked_foreground_vedastro(
+            reason='gateway_invocation_error',
+            error_type=type(exc).__name__,
+        )
+    return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason='gateway_invocation_error')
+
+
 def execute_consultation_workflow(
     handler,
     *,
@@ -1709,6 +1831,13 @@ def execute_consultation_workflow(
     rectification = {}
     muhurta_panchanga = {}
     computed_chart = bool(chart)
+    foreground_vedastro_future = None
+    if defer_optional_external_evidence:
+        # Overlap a bounded official gateway with local compute. Main-entry
+        # overview / range scan stay skipped so foreground cannot replay BUG-161.
+        foreground_vedastro_future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
+            _run_foreground_vedastro_gateway, handler, dict(body),
+        )
 
     for step in runtime_planner.get('sync_steps', []):
         if step == 'run_prashna':
@@ -1790,12 +1919,10 @@ def execute_consultation_workflow(
 
     vedastro_gateway = rectification.get('vedastro_gateway') if isinstance(rectification, dict) else None
     if defer_optional_external_evidence:
-        vedastro_gateway = {
-            'scope': 'vedastro_gateway_run',
-            'status': 'local_fallback',
-            'official_closure_state': 'official_blocked',
-            'official_closure_reason': 'foreground_optional_evidence_deferred',
-        }
+        vedastro_gateway = _join_foreground_vedastro(
+            foreground_vedastro_future,
+            timeout=_foreground_vedastro_join_seconds(),
+        )
     elif not isinstance(vedastro_gateway, dict):
         try:
             vedastro_gateway = handler._compute_vedastro_gateway_run(body)
@@ -1884,6 +2011,14 @@ def execute_consultation_workflow(
         quality_gate=runtime_evidence_log.get('quality_gate') if isinstance(runtime_evidence_log.get('quality_gate'), dict) else {},
         entry_mode=str(entry_mode or 'direct_chart'),
     )
+    consumer_context['vedastro_cross_check'] = _compact_vedastro_cross_check(
+        vedastro_official,
+        vedastro_gateway if isinstance(vedastro_gateway, dict) else {},
+    )
+    consumer_context['vedastro_cross_check'] = _compact_vedastro_cross_check(
+        vedastro_official,
+        vedastro_gateway if isinstance(vedastro_gateway, dict) else {},
+    )
 
     result = {
         'success': True,