"""Foreground VedAstro wait, cancel, and snapshot-cache coordination. Join, budget, and worker count bound the same wall clock, so they are declared together and must be changed together. join: how long the consultation thread waits for the official gateway (default 1.5s, cap 3s). budget: how long a worker may keep running after it has started. workers: process-wide pool. Extra work queues; join still caps the wait. A timed-out join must cancel its future so queued work never starts. budget <= k * join (k=2) keeps leftover occupancy of an already-running worker from stretching later requests past the join cap. The previous default budget of 8s with a 1.5s join left cancelled-too-late workers holding the two-thread pool for the rest of the 8s. """ from __future__ import annotations import os import threading from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError from typing import Any try: from scripts.vedastro_runtime_context import temporary_timeout_seconds from scripts.vedastro_snapshot_cache import ( annotate_stale_gateway, is_cacheable_gateway, lookup_snapshot, official_snapshot_reference_date, requires_today_snapshot, store_snapshot, ) except ModuleNotFoundError: # pragma: no cover - script execution path from vedastro_runtime_context import temporary_timeout_seconds from vedastro_snapshot_cache import ( annotate_stale_gateway, is_cacheable_gateway, lookup_snapshot, official_snapshot_reference_date, requires_today_snapshot, store_snapshot, ) # --- join / budget / workers (one declaration block; change together) --- FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT = 1.5 FOREGROUND_VEDASTRO_JOIN_SECONDS_MAX = 3.0 FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO = 2.0 FOREGROUND_VEDASTRO_WORKERS_DEFAULT = 2 # ---------------------------------------------------------------------- _FOREGROUND_VEDASTRO_WORKERS = max( int(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_WORKERS", str(FOREGROUND_VEDASTRO_WORKERS_DEFAULT))), 1, ) _FOREGROUND_VEDASTRO_EXECUTOR = ThreadPoolExecutor( max_workers=_FOREGROUND_VEDASTRO_WORKERS, thread_name_prefix="jyotish-vedastro-fg", ) def _foreground_vedastro_join_seconds() -> float: raw = str(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", str(FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT))).strip() try: return min(max(float(raw), 0.0), FOREGROUND_VEDASTRO_JOIN_SECONDS_MAX) except ValueError: return FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT def _foreground_vedastro_budget_seconds() -> float: join = _foreground_vedastro_join_seconds() cap = FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO * join raw = str(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "")).strip() if not raw: return cap try: budget = float(raw) except ValueError: return cap if budget < 0: return cap return min(budget, cap) 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, cancel_event: threading.Event | None = None) -> dict: if cancel_event is not None and cancel_event.is_set(): return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout") 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 _run_foreground_vedastro_gateway_and_store( handler, body: dict, cancel_event: threading.Event | None = None, ) -> dict: result = _run_foreground_vedastro_gateway(handler, body, cancel_event) if is_cacheable_gateway(result): try: store_snapshot(body, result) except OSError: pass return result def _join_foreground_vedastro(future, *, timeout: float, cancel_event: threading.Event | None = None) -> dict: if future is None: return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout") try: result = future.result(timeout=timeout) except FuturesTimeoutError: if cancel_event is not None: cancel_event.set() future.cancel() 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 should_include_western_evidence_packet(body: dict | None, *, surface: str = "") -> bool: payload = body if isinstance(body, dict) else {} if payload.get("include_western_evidence_packet") or payload.get("return_western_evidence_packet"): return True if payload.get("return_high_rigor_shape"): return True if isinstance(payload.get("western_oracle_payload"), dict) or isinstance(payload.get("western_astrology_oracle"), dict): return True if isinstance(payload.get("western_evidence_packet"), dict): return True return str(surface or "") in {"skill_mcp", "professional_reading_web", "mcp"} class ForegroundVedastroSession: def __init__(self, handler, body: dict, *, defer: bool) -> None: self.handler = handler self.body = dict(body) self.defer = bool(defer) self.cached_gateway: dict | None = None self.future = None self.cancel_event = threading.Event() self.meta: dict[str, Any] = {} def start(self) -> "ForegroundVedastroSession": today = official_snapshot_reference_date(self.body) hit = lookup_snapshot(self.body, today=today) if hit is not None: record = hit["record"] freshness = hit["freshness"] gateway = record["gateway"] if freshness == "stale": gateway = annotate_stale_gateway( gateway, reference_date=str(record.get("reference_date") or ""), served_on_utc_date=today, ) refresh_body = dict(self.body) refresh_body["reference_date"] = today refresh_body["today"] = today refresh_body["current_date"] = today self.future = _FOREGROUND_VEDASTRO_EXECUTOR.submit( _run_foreground_vedastro_gateway_and_store, self.handler, refresh_body, None, ) self.cached_gateway = gateway self.meta = { "freshness": freshness, "reference_date": str(record.get("reference_date") or today), "served_on_utc_date": today, "refresh_submitted": freshness == "stale", } return self self.meta = { "freshness": "miss", "reference_date": today, "served_on_utc_date": today, "refresh_submitted": False, "require_today": requires_today_snapshot(self.body), } if self.defer: self.future = _FOREGROUND_VEDASTRO_EXECUTOR.submit( _run_foreground_vedastro_gateway_and_store, self.handler, dict(self.body), self.cancel_event, ) return self def finish(self) -> dict: if self.cached_gateway is not None: return self.cached_gateway if self.defer: return _join_foreground_vedastro( self.future, timeout=_foreground_vedastro_join_seconds(), cancel_event=self.cancel_event, ) try: result = self.handler._compute_vedastro_gateway_run(self.body) except Exception as exc: return _blocked_foreground_vedastro( reason="gateway_invocation_error", error_type=type(exc).__name__, ) if not isinstance(result, dict): return _blocked_foreground_vedastro(reason="gateway_invocation_error") if is_cacheable_gateway(result): try: store_snapshot(self.body, result) except OSError: pass return result def start_foreground_vedastro(handler, body: dict, *, defer_optional_external_evidence: bool) -> ForegroundVedastroSession: return ForegroundVedastroSession( handler, body, defer=defer_optional_external_evidence, ).start() def finish_foreground_vedastro(session: ForegroundVedastroSession | None) -> dict: if session is None: return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout") return session.finish()