Jyotisha/scripts/vedastro_service_adapter.py

#!/usr/bin/env python3
"""Minimal VedAstro service-boundary adapter skeleton.

This module does not replace the local SwissEph path. It only defines the
request/response schema and a controlled "not configured" status so the
workspace can evolve from research notes to an executable adapter contract.
"""

from __future__ import annotations

import argparse
import http.client
import hashlib
import json
import os
import socket
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib import request, error
from urllib.parse import urlparse

try:
    from scripts.local_env import load_local_env
    from scripts.vedastro_runtime_context import timeout_override_seconds
except ModuleNotFoundError:  # pragma: no cover - script execution path
    from local_env import load_local_env
    from vedastro_runtime_context import timeout_override_seconds


ROOT = Path(__file__).resolve().parents[1]
load_local_env(ROOT)

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,
        "month": 2,
        "day": 24,
        "hour": 19,
        "minute": 15,
        "lat": 37.7749,
        "lon": -122.4194,
        "tz": -8.0,
        "ayanamsa_policy": "lahiri",
        "node_policy": "mean",
    },
    "beijing_first_use_demo": {
        "year": 1990,
        "month": 1,
        "day": 1,
        "hour": 12,
        "minute": 0,
        "lat": 39.9042,
        "lon": 116.4074,
        "tz": 8.0,
        "ayanamsa_policy": "lahiri",
        "node_policy": "mean",
    },
    "delhi_lagna_boundary": {
        "year": 1984,
        "month": 10,
        "day": 31,
        "hour": 6,
        "minute": 30,
        "lat": 28.6139,
        "lon": 77.2090,
        "tz": 5.5,
        "ayanamsa_policy": "lahiri",
        "node_policy": "mean",
    },
    "new_york_moon_boundary": {
        "year": 2001,
        "month": 9,
        "day": 11,
        "hour": 8,
        "minute": 46,
        "lat": 40.7128,
        "lon": -74.0060,
        "tz": -4.0,
        "ayanamsa_policy": "lahiri",
        "node_policy": "mean",
    },
}

SUPPORTED_RANGE_SCAN_DOMAINS = {"marriage", "wealth", "career"}
SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS = {"marriage", "wealth", "career", "general"}
OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH = "/Calculate/SearchEvents"
OFFICIAL_SEARCH_EVENTS_METHOD = "POST"
OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION = "official_builder_search_events_v1"
OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION = "official_full_snapshot_v1"
OFFICIAL_METHOD_CATALOG_URL = "https://vedastro.org/Complete-List-VedAstro-API-Methods-Calculators.html"
OFFICIAL_FULL_SNAPSHOT_METHODS = [
    {
        "section": "events_overview",
        "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH,
        "calculator_name": "SearchEvents",
        "role": "life_event_raw_evidence",
        "description": "Official event radar using SearchEvents for career, marriage and wealth tags.",
    },
    {
        "section": "dasha_all",
        "endpoint_path": "/Calculate/DasaAtRange",
        "calculator_name": "DasaAtRange",
        "role": "all_dasha_raw_evidence",
        "description": "Official dasha timeline snapshot where available.",
    },
    {
        "section": "chart_core",
        "endpoint_path": "/Calculate/AllPlanetData",
        "calculator_name": "AllPlanetData",
        "role": "core_chart_raw_evidence",
        "description": "Core planet, ascendant, house, nakshatra, ayanamsa and node-mode evidence when supported by the official service.",
        "fanout": "planetName",
    },
    {
        "section": "house_core",
        "endpoint_path": "/Calculate/AllHouseData",
        "calculator_name": "AllHouseData",
        "role": "core_house_raw_evidence",
        "description": "Official house data snapshot when supported by the official service.",
        "fanout": "houseName",
    },
]
OFFICIAL_FULL_SNAPSHOT_BACKLOG_SECTIONS = [
    {
        "section": "varga_all",
        "role": "all_varga_raw_evidence",
        "status": "catalog_pending",
        "description": "Awaiting official method mapping for all divisional charts; local varga remains fallback until mapped.",
    },
]
OFFICIAL_SNAPSHOT_PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu", "Ascendant"]
OFFICIAL_SNAPSHOT_HOUSES = [f"House{i}" for i in range(1, 13)]
OFFICIAL_RANGE_SCAN_EVENT_TAGS = {
    "marriage": ["Marriage", "Personal", "General"],
    "wealth": ["LendingMoney", "BorrowingMoney", "BuyingSelling", "General"],
    "career": ["Personal", "General", "Building", "Travel"],
}
VEDASTRO_CALCULATION_COVERAGE = {
    "official_python_library_calculations": "596+",
    "official_api_builder_calculators": "600+",
    "official_events_builder_events": "400+",
    "official_events_builder_methods": ["SearchEvents", "GetEventTiming", "ListEventTypes"],
    "range_scan_role": "high_frequency_life_event_radar",
    "intended_use": "external_timing_evidence_for_strict_workflow",
}
EXTERNAL_TECHNIQUE_ROLE = "external_technique_evidence"
EXTERNAL_TECHNIQUE_OPERATION = "calculation_method"
EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY = {
    "role": EXTERNAL_TECHNIQUE_ROLE,
    "can_change_score": False,
    "can_set_dominant_label": False,
    "can_set_payout_label": False,
    "allowed_destinations": ["secondary_context", "technique_audit"],
}
RANGE_SCAN_EVENT_ALLOWLIST = {
    "marriage": {
        "event_ids": {
            "GocharJupiterIn7th",
            "GocharJupiterAspect7th",
            "GocharSaturnAspect7th",
            "JupiterSupportsMarriageAxis",
        },
        "tags": {"marriage", "relationship", "spouse", "transit"},
    },
    "wealth": {
        "event_ids": {
            "GocharJupiterIn2nd",
            "GocharJupiterIn11th",
            "GocharJupiterAspect2nd",
            "GocharJupiterAspect11th",
            "WealthExpansionWindow",
        },
        "tags": {"wealth", "finance", "income", "gains", "transit"},
    },
    "career": {
        "event_ids": {
            "GocharJupiterIn10th",
            "GocharSaturnIn10th",
            "GocharJupiterAspect10th",
            "CareerExpansionWindow",
        },
        "tags": {"career", "profession", "work", "transit"},
    },
}
RANGE_SCAN_SIGNAL_METADATA = {
    "marriage": {
        "GocharJupiterIn7th": {
            "signal_key": "gochar_jupiter_7th_marriage",
            "signal_label": "Jupiter in 7th marriage window",
            "signal_family": "marriage_trigger",
        },
        "GocharJupiterAspect7th": {
            "signal_key": "gochar_jupiter_aspect_7th_marriage",
            "signal_label": "Jupiter aspecting 7th marriage window",
            "signal_family": "marriage_trigger",
        },
        "GocharSaturnAspect7th": {
            "signal_key": "gochar_saturn_aspect_7th_relationship_pressure",
            "signal_label": "Saturn aspecting 7th relationship window",
            "signal_family": "relationship_pressure",
        },
        "JupiterSupportsMarriageAxis": {
            "signal_key": "jupiter_supports_marriage_axis",
            "signal_label": "Jupiter supports marriage axis",
            "signal_family": "marriage_trigger",
        },
    },
    "wealth": {
        "GocharJupiterIn2nd": {
            "signal_key": "gochar_jupiter_2nd_wealth",
            "signal_label": "Jupiter in 2nd wealth window",
            "signal_family": "wealth_trigger",
        },
        "GocharJupiterIn11th": {
            "signal_key": "gochar_jupiter_11th_gains",
            "signal_label": "Jupiter in 11th gains window",
            "signal_family": "gains_trigger",
        },
        "GocharJupiterAspect2nd": {
            "signal_key": "gochar_jupiter_aspect_2nd_wealth",
            "signal_label": "Jupiter aspecting 2nd wealth window",
            "signal_family": "wealth_trigger",
        },
        "GocharJupiterAspect11th": {
            "signal_key": "gochar_jupiter_aspect_11th_gains",
            "signal_label": "Jupiter aspecting 11th gains window",
            "signal_family": "gains_trigger",
        },
        "WealthExpansionWindow": {
            "signal_key": "wealth_expansion_window",
            "signal_label": "Wealth expansion window",
            "signal_family": "wealth_trigger",
        },
    },
    "career": {
        "GocharJupiterIn10th": {
            "signal_key": "gochar_jupiter_10th_career",
            "signal_label": "Jupiter in 10th career window",
            "signal_family": "career_trigger",
        },
        "GocharSaturnIn10th": {
            "signal_key": "gochar_saturn_10th_career",
            "signal_label": "Saturn in 10th career window",
            "signal_family": "career_pressure",
        },
        "GocharJupiterAspect10th": {
            "signal_key": "gochar_jupiter_aspect_10th_career",
            "signal_label": "Jupiter aspecting 10th career window",
            "signal_family": "career_trigger",
        },
        "CareerExpansionWindow": {
            "signal_key": "career_expansion_window",
            "signal_label": "Career expansion window",
            "signal_family": "career_trigger",
        },
    },
}
RANGE_SCAN_OFFICIAL_TAG_MATCHES = {
    "marriage": {"Marriage"},
    "wealth": {"LendingMoney", "BorrowingMoney", "BuyingSelling"},
    "career": {"Building", "Travel"},
}
RANGE_SCAN_ALIAS_TERMS = {
    "marriage": {
        "marriage",
        "spouse",
        "wedding",
        "relationship",
        "partner",
        "partnership",
    },
    "wealth": {
        "wealth",
        "money",
        "finance",
        "financial",
        "income",
        "gain",
        "gains",
        "lending",
        "borrowing",
        "business",
        "cash",
    },
    "career": {
        "career",
        "profession",
        "work",
        "job",
        "business",
        "travel",
        "building",
        "public",
        "status",
    },
}
MATCH_METADATA_BY_TYPE = {
    "exact_id": {"signal_lift": 3, "confidence": "high"},
    "official_tag": {"signal_lift": 2, "confidence": "medium_high"},
    "alias": {"signal_lift": 1, "confidence": "low"},
    "rejected": {"signal_lift": 0, "confidence": "rejected"},
}

OFFICIAL_PYTHON_BUNDLE_SECTIONS = [
    "chart_core",
    "house_core",
    "dasha_all",
    "vimshottari_now",
    "chara_dasha_now",
    "shadbala",
    "ashtakavarga",
]
OFFICIAL_FAST_PRIMARY_SECTIONS = {
    "chart_core",
    "house_core",
    "dasha_all",
    "vimshottari_now",
    "chara_dasha_now",
    "shadbala",
    "ashtakavarga",
}
ALIAS_NEGATIVE_GUARD_TERMS = {"noise", "without", "generic", "irrelevant", "insignificance", "not"}
DEFAULT_TIMEOUT_SECONDS = 4
TIMEOUT_ENV = "VEDASTRO_TIMEOUT_SECONDS"
BACKOFF_ENV = "VEDASTRO_RETRY_BACKOFF_SECONDS"
RETRY_POLICY = {
    "max_attempts": 2,
    "backoff_seconds": 1,
    "retry_on": ["timeout", "429", "502", "503", "504"],
}
ALLOW_NETWORK_ENV = "VEDASTRO_ENABLE_NETWORK"
CACHE_TTL_ENV = "VEDASTRO_CACHE_TTL_SECONDS"
FREE_TIER_MAX_REQUESTS_ENV = "VEDASTRO_FREE_TIER_MAX_REQUESTS"
FREE_TIER_WINDOW_SECONDS_ENV = "VEDASTRO_FREE_TIER_WINDOW_SECONDS"
DEFAULT_CACHE_TTL_SECONDS = 86400.0
DEFAULT_FREE_TIER_MAX_REQUESTS = 5
DEFAULT_FREE_TIER_WINDOW_SECONDS = 60.0
ARTIFACT_DIR = ROOT / "scratch" / "local" / "vedastro_adapter"
_FREE_TIER_REQUEST_TIMESTAMPS: list[float] = []
_FREE_TIER_REQUEST_LOCK = threading.Lock()


def _timeout_seconds() -> float:
    override = timeout_override_seconds()
    if override is not None:
        return override
    raw = os.environ.get(TIMEOUT_ENV, "").strip()
    if not raw:
        return DEFAULT_TIMEOUT_SECONDS
    try:
        return float(raw)
    except ValueError:
        return DEFAULT_TIMEOUT_SECONDS



def _backoff_seconds() -> float:
    raw = os.environ.get(BACKOFF_ENV, "").strip()
    if not raw:
        return float(RETRY_POLICY["backoff_seconds"])
    try:
        return max(0.0, float(raw))
    except ValueError:
        return float(RETRY_POLICY["backoff_seconds"])


def _json_bytes(payload: dict[str, Any]) -> bytes:
    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")


def _hash_payload(payload: dict[str, Any]) -> str:
    return hashlib.sha256(_json_bytes(payload)).hexdigest()


def _utc_timestamp() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def _endpoint_host(endpoint: str) -> str:
    parsed = urlparse(endpoint)
    return parsed.netloc or endpoint


def _artifact_path(operation: str, request_hash: str, response_hash: str) -> Path:
    ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
    filename = f"{operation}-{request_hash[:12]}-{response_hash[:12]}.json"
    return ARTIFACT_DIR / filename


def _cache_dir() -> Path:
    path = ARTIFACT_DIR / "response_cache"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _official_full_snapshot_cache_dir() -> Path:
    path = ARTIFACT_DIR / "official_full_snapshot_cache"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _repo_relative(path: Path) -> str:
    try:
        return str(path.relative_to(ROOT))
    except ValueError:
        return str(path)


def _write_artifact(result: dict[str, Any]) -> str:
    metadata = result.get("source_metadata") or {}
    artifact = _artifact_path(
        str(metadata.get("operation") or result.get("operation") or "calculation"),
        str(metadata.get("request_hash") or "no-request-hash"),
        str(metadata.get("response_hash") or "no-response-hash"),
    )
    artifact.write_text(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    return _repo_relative(artifact)


def list_official_full_snapshot_artifacts() -> dict[str, Any]:
    artifacts: list[dict[str, Any]] = []
    if ARTIFACT_DIR.exists():
        for path in sorted(ARTIFACT_DIR.glob("official_full_snapshot-*.json")):
            try:
                payload = json.loads(path.read_text(encoding="utf-8"))
            except json.JSONDecodeError:
                continue
            raw = payload.get("official_raw_response") or payload.get("raw_response")
            raw_source = str(raw.get("source") or "") if isinstance(raw, dict) else ""
            official_raw_available = raw_source.startswith("vedastro_official")
            artifacts.append(
                {
                    "path": _repo_relative(path),
                    "status": payload.get("status"),
                    "operation": payload.get("operation"),
                    "official_raw_response_available": official_raw_available,
                    "section_count": len(payload.get("snapshot_sections") or {}),
                    "request_manifest_available": bool(payload.get("request_manifest")),
                }
            )
    return {
        "scope": "vedastro_official_full_snapshot_artifact_manifest",
        "artifact_count": len(artifacts),
        "artifacts": artifacts,
    }


def _cache_ttl_seconds() -> float:
    raw = os.environ.get(CACHE_TTL_ENV, "").strip()
    if not raw:
        return DEFAULT_CACHE_TTL_SECONDS
    try:
        return max(0.0, float(raw))
    except ValueError:
        return DEFAULT_CACHE_TTL_SECONDS


def _official_full_snapshot_cache_ttl_seconds() -> float:
    raw = os.environ.get("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "").strip()
    if not raw:
        raw = os.environ.get(CACHE_TTL_ENV, "").strip()
    if not raw:
        return 0.0
    try:
        return max(0.0, float(raw))
    except ValueError:
        return 0.0


def _free_tier_max_requests() -> int:
    raw = os.environ.get(FREE_TIER_MAX_REQUESTS_ENV, "").strip()
    if not raw:
        return DEFAULT_FREE_TIER_MAX_REQUESTS
    try:
        return max(0, int(raw))
    except ValueError:
        return DEFAULT_FREE_TIER_MAX_REQUESTS


def _free_tier_window_seconds() -> float:
    raw = os.environ.get(FREE_TIER_WINDOW_SECONDS_ENV, "").strip()
    if not raw:
        return DEFAULT_FREE_TIER_WINDOW_SECONDS
    try:
        return max(0.0, float(raw))
    except ValueError:
        return DEFAULT_FREE_TIER_WINDOW_SECONDS


def schema() -> dict[str, Any]:
    request_example = {
        **PARITY_CASES["beijing_first_use_demo"],
        "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"],
    }
    range_scan_allowlist = {
        domain: {
            "event_ids": sorted(values["event_ids"]),
            "tags": sorted(values["tags"]),
        }
        for domain, values in sorted(RANGE_SCAN_EVENT_ALLOWLIST.items())
    }
    return {
        "adapter": "vedastro_service_adapter",
        "backend": "vedastro_service_adapter_candidate",
        "transport": "http_json_service_boundary",
        "default_timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
        "retry_policy": RETRY_POLICY,
        "required_env": {
            "endpoint": "VEDASTRO_API_ENDPOINT",
            "api_key_optional": "VEDASTRO_API_KEY",
        },
        "request_contract": [
            "year",
            "month",
            "day",
            "hour",
            "minute",
            "lat",
            "lon",
            "tz",
            "ayanamsa_policy",
            "node_policy",
            "body_list",
        ],
        "response_contract": [
            "backend",
            "available",
            "status",
            "ayanamsa_value",
            "node_policy",
            "body_list",
            "bodies",
            "source_metadata",
        ],
        "range_scan_request_contract": [
            "operation",
            "vedastro_event_method",
            "domain",
            "start_date",
            "end_date",
            "year",
            "month",
            "day",
            "hour",
            "minute",
            "lat",
            "lon",
            "tz",
            "ayanamsa_policy",
            "node_policy",
            "event_model",
        ],
        "range_scan_response_contract": [
            "backend",
            "available",
            "status",
            "operation",
            "domain",
            "evidence_ledger",
            "source_metadata",
        ],
        "official_search_events_profile_contract": {
            "profile_version": OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION,
            "base_url_requirement": "VEDASTRO_API_ENDPOINT must end with /api",
            "route_template": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH,
            "method": OFFICIAL_SEARCH_EVENTS_METHOD,
            "content_type": "application/json",
            "optional_auth_header": "x-api-key",
            "body_fields": [
                "BirthTime",
                "Ayanamsa",
                "EventTagList",
            ],
            "range_mode_fields": [
                "AtTime | StartTime + EndTime + PrecisionHours",
            ],
        },
        "official_full_snapshot_request_contract": {
            "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION,
            "primary_source": "vedastro_official",
            "method_catalog_url": OFFICIAL_METHOD_CATALOG_URL,
            "strategy": "fetch_official_raw_sections_first_then_local_crosscheck",
            "common_body_fields": [
                "BirthTime",
                "Ayanamsa",
                "NodeMode",
                "CalculationPreferences",
            ],
            "request_sections": [
                {
                    "section": item["section"],
                    "endpoint_path": item["endpoint_path"],
                    "calculator_name": item.get("calculator_name"),
                    "role": item["role"],
                }
                for item in OFFICIAL_FULL_SNAPSHOT_METHODS
            ],
            "backlog_sections": OFFICIAL_FULL_SNAPSHOT_BACKLOG_SECTIONS,
            "user_visibility": "backend_raw_evidence_not_direct_user_report",
        },
        "vedastro_calculation_coverage": VEDASTRO_CALCULATION_COVERAGE,
        "official_full_snapshot_response_contract": [
            "backend",
            "available",
            "status",
            "operation",
            "primary_source",
            "snapshot_sections",
            "request_manifest",
            "source_metadata",
        ],
        "external_technique_request_contract": [
            "operation",
            "role",
            "domain",
            "method",
            "api_endpoint",
            "year",
            "month",
            "day",
            "hour",
            "minute",
            "lat",
            "lon",
            "tz",
            "ayanamsa_policy",
            "node_policy",
        ],
        "external_technique_response_contract": [
            "backend",
            "available",
            "status",
            "operation",
            "role",
            "domain",
            "evidence_ledger",
            "adjudicator_policy",
            "source_metadata",
        ],
        "external_technique_adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
        "range_scan_event_allowlist": range_scan_allowlist,
        "request_example": request_example,
        "provenance_contract": {
            "external_service": True,
            "required_fields": [
                "endpoint",
                "endpoint_host",
                "transport",
                "provenance_mode",
                "retry_policy",
                "timeout_seconds",
                "request_hash",
                "response_hash",
                "called_at",
                "artifact_path",
            ],
        },
    }


def _unconfigured(reason: str) -> dict[str, Any]:
    return {
        "backend": "vedastro_service_adapter_candidate",
        "available": False,
        "status": "service_endpoint_not_configured",
        "reason": reason,
        "source_metadata": {
            "transport": "http_json_service_boundary",
            "endpoint_env": "VEDASTRO_API_ENDPOINT",
            "api_key_env": "VEDASTRO_API_KEY",
            "provenance_mode": "external_service_candidate",
            "timeout_seconds": _timeout_seconds(),
            "retry_policy": RETRY_POLICY,
        },
    }


def _request_preview(case: dict[str, Any]) -> dict[str, Any]:
    return {
        **case,
        "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"],
    }


def _range_scan_preview(case: dict[str, Any], domain: str, start_date: str, end_date: str) -> dict[str, Any]:
    preview = {
        "operation": "range_scan",
        "vedastro_event_method": "SearchEvents",
        "domain": domain,
        "start_date": start_date,
        "end_date": end_date,
        "event_model": "vedastro_events_at_range_candidate",
        "search_mode": "single_point",
        **case,
    }
    preview["official_request_profile"] = _build_official_search_events_profile(preview)
    preview["live_sampling_request_profile"] = _build_live_sampling_search_events_profile(preview)
    return preview


def _format_std_time(date_text: str, hour: Any, minute: Any, tz: Any) -> str:
    year, month, day = str(date_text).split("-")
    hour_int = int(float(hour))
    minute_int = int(float(minute))
    return f"{hour_int:02d}:{minute_int:02d} {day}/{month}/{year} {tz}"


def _time_json_from_case(
    case: dict[str, Any],
    date_text: str,
    *,
    hour: Any | None = None,
    minute: Any | None = None,
) -> dict[str, Any]:
    location_name = (
        case.get("location_name")
        or case.get("place_name")
        or case.get("city")
        or "UserLocation"
    )
    return {
        "StdTime": _format_std_time(
            date_text,
            case.get("hour", 0) if hour is None else hour,
            case.get("minute", 0) if minute is None else minute,
            case.get("tz", "+00:00"),
        ),
        "Location": {
            "Name": str(location_name),
            "Latitude": case.get("lat"),
            "Longitude": case.get("lon"),
        },
    }


def _bridge_geo_from_case(case: dict[str, Any]) -> dict[str, Any]:
    location_name = (
        case.get("location_name")
        or case.get("place_name")
        or case.get("city")
        or "UserLocation"
    )
    return {
        "__vedastro_type__": "GeoLocation",
        "location_name": str(location_name),
        "longitude": case.get("lon"),
        "latitude": case.get("lat"),
    }


def _bridge_time_from_case(
    case: dict[str, Any],
    date_text: str,
    *,
    hour: Any | None = None,
    minute: Any | None = None,
) -> dict[str, Any]:
    year, month, day = str(date_text).split("-")
    return {
        "__vedastro_type__": "Time",
        "year": int(year),
        "month": int(month),
        "day": int(day),
        "hour": int(float(case.get("hour", 0) if hour is None else hour)),
        "minute": int(float(case.get("minute", 0) if minute is None else minute)),
        "offset": case.get("tz", "+00:00"),
        "geolocation": _bridge_geo_from_case(case),
    }


def _normalize_tz(case: dict[str, Any]) -> str:
    tz = case.get("tz")
    if isinstance(tz, str):
        return tz
    if tz is None:
        return "+00:00"
    sign = "+" if float(tz) >= 0 else "-"
    value = abs(float(tz))
    hours = int(value)
    minutes = int(round((value - hours) * 60))
    return f"{sign}{hours:02d}:{minutes:02d}"


def _build_official_search_events_profile(request_preview: dict[str, Any]) -> dict[str, Any]:
    case = dict(request_preview)
    case["tz"] = _normalize_tz(case)
    body = {
        "BirthTime": _time_json_from_case(case, f"{case['year']:04d}-{case['month']:02d}-{case['day']:02d}"),
        "Ayanamsa": str(case.get("ayanamsa_policy") or "lahiri"),
        "EventTagList": OFFICIAL_RANGE_SCAN_EVENT_TAGS.get(str(request_preview.get("domain") or ""), ["General"]),
    }
    start_time = _time_json_from_case(case, str(request_preview["start_date"]))
    end_time = _time_json_from_case(case, str(request_preview["end_date"]))
    if str(request_preview["start_date"]) == str(request_preview["end_date"]):
        body["AtTime"] = start_time
    else:
        body["StartTime"] = start_time
        body["EndTime"] = end_time
        body["PrecisionHours"] = 100
    headers: dict[str, str] = {"Content-Type": "application/json"}
    return {
        "profile_version": OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION,
        "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH,
        "method": OFFICIAL_SEARCH_EVENTS_METHOD,
        "headers": headers,
        "body": body,
    }


def _build_live_sampling_search_events_profile(request_preview: dict[str, Any]) -> dict[str, Any]:
    case = dict(request_preview)
    case["tz"] = _normalize_tz(case)
    body = {
        "BirthTime": _time_json_from_case(case, f"{case['year']:04d}-{case['month']:02d}-{case['day']:02d}"),
        "Ayanamsa": str(case.get("ayanamsa_policy") or "lahiri"),
        "EventTagList": OFFICIAL_RANGE_SCAN_EVENT_TAGS.get(str(request_preview.get("domain") or ""), ["General"]),
        "AtTime": _time_json_from_case(case, str(request_preview["start_date"])),
    }
    headers: dict[str, str] = {"Content-Type": "application/json"}
    return {
        "profile_version": f"{OFFICIAL_SEARCH_EVENTS_PROFILE_VERSION}_live_sampling",
        "endpoint_path": OFFICIAL_SEARCH_EVENTS_ENDPOINT_PATH,
        "method": OFFICIAL_SEARCH_EVENTS_METHOD,
        "headers": headers,
        "body": body,
    }


def _official_common_body(case: dict[str, Any]) -> dict[str, Any]:
    normalized = dict(case)
    normalized["tz"] = _normalize_tz(normalized)
    return {
        "time": _time_json_from_case(
            normalized,
            f"{int(normalized['year']):04d}-{int(normalized['month']):02d}-{int(normalized['day']):02d}",
        ),
        "Ayanamsa": str(normalized.get("ayanamsa_policy") or "lahiri"),
        "NodeMode": str(normalized.get("node_policy") or "mean"),
        "CalculationPreferences": {
            "scope": "all_supported_official_calculations",
            "user_visibility": "backend_raw_evidence_not_direct_user_report",
        },
    }


def _official_snapshot_reference_date(case: dict[str, Any]) -> str:
    for key in ("reference_date", "today", "transit_date", "current_date"):
        value = case.get(key)
        if not value:
            continue
        raw = str(value)[:10]
        try:
            datetime.strptime(raw, "%Y-%m-%d")
            return raw
        except ValueError:
            continue
    return datetime.utcnow().strftime("%Y-%m-%d")


def _official_dasha_range_body(case: dict[str, Any], common_body: dict[str, Any]) -> dict[str, Any]:
    normalized = dict(case)
    normalized["tz"] = _normalize_tz(normalized)
    reference = datetime.strptime(_official_snapshot_reference_date(normalized), "%Y-%m-%d").date()
    start_date = reference.replace(month=1, day=1)
    end_date = reference.replace(month=12, day=31)
    return {
        "birthTime": common_body["time"],
        "startTime": _time_json_from_case(normalized, start_date.isoformat(), hour=0, minute=0),
        "endTime": _time_json_from_case(normalized, end_date.isoformat(), hour=23, minute=59),
        "levels": int(normalized.get("dasha_levels") or 3),
        "precisionHours": int(normalized.get("dasha_precision_hours") or 100),
        "Ayanamsa": common_body["Ayanamsa"],
    }


def _official_dasha_range_dates(case: dict[str, Any]) -> tuple[str, str]:
    normalized = dict(case)
    reference = datetime.strptime(_official_snapshot_reference_date(normalized), "%Y-%m-%d").date()
    start_date = reference.replace(month=1, day=1)
    end_date = reference.replace(month=12, day=31)
    return start_date.isoformat(), end_date.isoformat()


def _official_full_snapshot_manifest(case: dict[str, Any], case_id: str = "user_chart") -> dict[str, Any]:
    common_body = _official_common_body(case)
    reference_date = _official_snapshot_reference_date(case)
    headers: dict[str, str] = {"Content-Type": "application/json"}
    requests = []
    for item in OFFICIAL_FULL_SNAPSHOT_METHODS:
        body = dict(common_body)
        if item["section"] == "events_overview":
            body = {
                "BirthTime": common_body["time"],
                "Ayanamsa": common_body["Ayanamsa"],
                "EventTagList": sorted({tag for tags in OFFICIAL_RANGE_SCAN_EVENT_TAGS.values() for tag in tags}),
                "AtTime": common_body["time"],
            }
        if item["section"] == "dasha_all":
            body = _official_dasha_range_body(case, common_body)
        fanout_values = []
        if item.get("fanout") == "planetName":
            fanout_values = OFFICIAL_SNAPSHOT_PLANETS
        elif item.get("fanout") == "houseName":
            fanout_values = OFFICIAL_SNAPSHOT_HOUSES
        requests.append(
            {
                "section": item["section"],
                "role": item["role"],
                "calculator_name": item.get("calculator_name"),
                "endpoint_path": item["endpoint_path"],
                "method": "POST",
                "headers": headers,
                "body": body,
                "fanout_parameter": item.get("fanout"),
                "fanout_values": fanout_values,
                "description": item["description"],
            }
        )
    return {
        "operation": "official_full_snapshot",
        "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION,
        "source_role": "primary_official_raw_evidence",
        "primary_source": "vedastro_official",
        "case_id": case_id,
        "reference_date": reference_date,
        "method_catalog": {
            "url": OFFICIAL_METHOD_CATALOG_URL,
            "declared_coverage": VEDASTRO_CALCULATION_COVERAGE,
            "catalog_role": "all_supported_method_reference_not_user_visible_output",
            "backlog_sections": OFFICIAL_FULL_SNAPSHOT_BACKLOG_SECTIONS,
        },
        "requests": requests,
    }


def _call_vedastro_python_bridge_high_value(method_key: str, payload: dict[str, Any]) -> dict[str, Any]:
    if not VEDASTRO_PYTHON_BRIDGE.exists():
        return {
            "available": False,
            "status": "python_bridge_missing",
            "method": method_key,
            "source": "vedastro_service_adapter_candidate",
        }
    try:
        completed = subprocess.run(
            [
                _vedastro_python_bin(),
                str(VEDASTRO_PYTHON_BRIDGE),
                "--high-value",
                method_key,
                "--params-json",
                json.dumps(payload, ensure_ascii=False),
            ],
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=_timeout_seconds(),
            check=False,
            env=os.environ.copy(),
        )
    except subprocess.TimeoutExpired:
        return {
            "available": False,
            "status": "python_bridge_timeout",
            "method": method_key,
            "reason": "VedAstro official Python bridge timed out",
            "timeout_seconds": _timeout_seconds(),
            "source": "vedastro_service_adapter_candidate",
        }
    if completed.returncode != 0:
        return {
            "available": False,
            "status": "python_bridge_runtime_error",
            "method": method_key,
            "stderr": (completed.stderr or "").strip(),
            "stdout_excerpt": (completed.stdout or "").strip()[:500],
            "source": "vedastro_service_adapter_candidate",
        }
    try:
        return json.loads(completed.stdout)
    except json.JSONDecodeError:
        return {
            "available": False,
            "status": "python_bridge_invalid_json",
            "method": method_key,
            "stdout_excerpt": (completed.stdout or "").strip()[:500],
            "source": "vedastro_service_adapter_candidate",
        }


def _try_official_python_bridge_snapshot_bundle(case: dict[str, Any]) -> dict[str, Any]:
    normalized = dict(case)
    normalized["tz"] = _normalize_tz(normalized)
    birth_date = f"{int(normalized['year']):04d}-{int(normalized['month']):02d}-{int(normalized['day']):02d}"
    reference_date = _official_snapshot_reference_date(normalized)
    start_date, end_date = _official_dasha_range_dates(normalized)
    payload = {
        "birth_time": _bridge_time_from_case(normalized, birth_date),
        "check_time": _bridge_time_from_case(normalized, reference_date),
        "start_time": _bridge_time_from_case(normalized, start_date, hour=0, minute=0),
        "end_time": _bridge_time_from_case(normalized, end_date, hour=23, minute=59),
        "levels": int(normalized.get("dasha_levels") or 3),
        "precision_hours": int(normalized.get("dasha_precision_hours") or 100),
        "planets": OFFICIAL_SNAPSHOT_PLANETS,
        "houses": OFFICIAL_SNAPSHOT_HOUSES,
    }
    report = _call_vedastro_python_bridge_high_value("official_full_snapshot_bundle", payload)
    if report.get("status") != "ok":
        return {
            "available": bool(report.get("available")),
            "status": report.get("status") or "blocked",
            "source": report.get("source") or "vedastro_python_bridge",
            "reason": report.get("reason"),
            "snapshot_sections": {},
            "section_statuses": {},
            "coverage": {"source_mode": "official_python_bridge_bundle", "filled_sections": []},
            "raw": report,
        }
    result = report.get("result") if isinstance(report.get("result"), dict) else {}
    return {
        "available": bool(report.get("available")),
        "status": report.get("status") or "ok",
        "source": report.get("source") or "vedastro_python_bridge",
        "snapshot_sections": result.get("snapshot_sections") if isinstance(result.get("snapshot_sections"), dict) else {},
        "section_statuses": result.get("section_statuses") if isinstance(result.get("section_statuses"), dict) else {},
        "coverage": result.get("coverage") if isinstance(result.get("coverage"), dict) else {"source_mode": "official_python_bridge_bundle", "filled_sections": []},
        "python_bin": report.get("python_bin"),
        "module_name": report.get("module_name"),
    }


def _try_official_capability_runner_snapshot_bundle(case: dict[str, Any]) -> dict[str, Any]:
    if not VEDASTRO_OFFICIAL_CAPABILITY_RUNNER.exists():
        return {
            "available": False,
            "status": "official_capability_runner_missing",
            "source": "vedastro_official_capability_runner",
            "snapshot_sections": {},
            "section_statuses": {},
            "coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
        }
    try:
        completed = subprocess.run(
            [
                _vedastro_python_bin(),
                str(VEDASTRO_OFFICIAL_CAPABILITY_RUNNER),
                "--bundle",
                "official_full_snapshot",
                "--birth-json",
                json.dumps(case, ensure_ascii=False),
            ],
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=_timeout_seconds(),
            check=False,
            env=os.environ.copy(),
        )
    except subprocess.TimeoutExpired:
        return {
            "available": False,
            "status": "official_capability_runner_timeout",
            "source": "vedastro_official_capability_runner",
            "snapshot_sections": {},
            "section_statuses": {},
            "coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
            "reason": "VedAstro official capability runner timed out",
            "timeout_seconds": _timeout_seconds(),
        }
    if completed.returncode != 0:
        return {
            "available": False,
            "status": "official_capability_runner_runtime_error",
            "source": "vedastro_official_capability_runner",
            "snapshot_sections": {},
            "section_statuses": {},
            "coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
            "stderr": (completed.stderr or "").strip(),
        }
    try:
        report = json.loads(completed.stdout)
    except json.JSONDecodeError:
        return {
            "available": False,
            "status": "official_capability_runner_invalid_json",
            "source": "vedastro_official_capability_runner",
            "snapshot_sections": {},
            "section_statuses": {},
            "coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
            "stdout_excerpt": (completed.stdout or "").strip()[:500],
        }
    result = report.get("result") if isinstance(report.get("result"), dict) else {}
    return {
        "available": bool(report.get("available")),
        "status": report.get("status") or "blocked",
        "source": "vedastro_official_capability_runner",
        "bundle": report.get("bundle") or "official_full_snapshot",
        "summary": report.get("summary") if isinstance(report.get("summary"), dict) else {},
        "snapshot_sections": result.get("snapshot_sections") if isinstance(result.get("snapshot_sections"), dict) else {},
        "section_statuses": result.get("section_statuses") if isinstance(result.get("section_statuses"), dict) else {},
        "coverage": result.get("coverage") if isinstance(result.get("coverage"), dict) else {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
    }


def _try_official_full_capability_catalog_bundle(case: dict[str, Any]) -> dict[str, Any]:
    if not VEDASTRO_OFFICIAL_CAPABILITY_RUNNER.exists():
        return {
            "available": False,
            "status": "official_capability_runner_missing",
            "source": "vedastro_official_capability_runner",
            "bundle": "official_full_capability_catalog",
            "summary": {},
            "coverage": {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
            "domain_routing": {},
            "dynamic_selection": {},
            "bucket_statuses": {},
            "method_statuses": {},
        }
    try:
        completed = subprocess.run(
            [
                _vedastro_python_bin(),
                str(VEDASTRO_OFFICIAL_CAPABILITY_RUNNER),
                "--bundle",
                "official_full_capability_catalog",
                "--birth-json",
                json.dumps(case, ensure_ascii=False),
            ],
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=_timeout_seconds(),
            check=False,
            env=os.environ.copy(),
        )
    except subprocess.TimeoutExpired:
        return {
            "available": False,
            "status": "official_full_capability_catalog_timeout",
            "source": "vedastro_official_capability_runner",
            "bundle": "official_full_capability_catalog",
            "summary": {},
            "coverage": {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
            "domain_routing": {},
            "dynamic_selection": {},
            "bucket_statuses": {},
            "method_statuses": {},
            "reason": "VedAstro official capability catalog runner timed out",
            "timeout_seconds": _timeout_seconds(),
        }
    if completed.returncode != 0:
        return {
            "available": False,
            "status": "official_full_capability_catalog_runtime_error",
            "source": "vedastro_official_capability_runner",
            "bundle": "official_full_capability_catalog",
            "summary": {},
            "coverage": {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
            "domain_routing": {},
            "dynamic_selection": {},
            "bucket_statuses": {},
            "method_statuses": {},
            "stderr": (completed.stderr or "").strip(),
        }
    try:
        report = json.loads(completed.stdout)
    except json.JSONDecodeError:
        return {
            "available": False,
            "status": "official_full_capability_catalog_invalid_json",
            "source": "vedastro_official_capability_runner",
            "bundle": "official_full_capability_catalog",
            "summary": {},
            "coverage": {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
            "domain_routing": {},
            "dynamic_selection": {},
            "bucket_statuses": {},
            "method_statuses": {},
            "stdout_excerpt": (completed.stdout or "").strip()[:500],
        }
    return {
        "available": bool(report.get("available")),
        "status": report.get("status") or "blocked",
        "source": "vedastro_official_capability_runner",
        "bundle": report.get("bundle") or "official_full_capability_catalog",
        "summary": report.get("summary") if isinstance(report.get("summary"), dict) else {},
        "coverage": report.get("coverage") if isinstance(report.get("coverage"), dict) else {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
        "domain_routing": report.get("domain_routing") if isinstance(report.get("domain_routing"), dict) else {},
        "dynamic_selection": report.get("dynamic_selection") if isinstance(report.get("dynamic_selection"), dict) else {},
        "bucket_statuses": report.get("bucket_statuses") if isinstance(report.get("bucket_statuses"), dict) else {},
        "method_statuses": report.get("method_statuses") if isinstance(report.get("method_statuses"), dict) else {},
    }


def _external_technique_preview(
    case: dict[str, Any],
    domain: str,
    method: str,
    api_endpoint: str,
) -> dict[str, Any]:
    return {
        "operation": EXTERNAL_TECHNIQUE_OPERATION,
        "role": EXTERNAL_TECHNIQUE_ROLE,
        "domain": domain,
        "method": method,
        "api_endpoint": api_endpoint,
        **case,
    }


def _base_live_metadata(
    endpoint: str,
    request_preview: dict[str, Any],
    payload: dict[str, Any],
    operation: str,
    attempt_count: int = 1,
    retry_error_codes: list[int] | None = None,
) -> dict[str, Any]:
    official_request_profile = request_preview.get("official_request_profile") if isinstance(request_preview, dict) else None
    transport_metadata = payload.get("source_metadata") if isinstance(payload, dict) else None
    metadata = {
        "transport": "http_json_service_boundary",
        "endpoint": endpoint,
        "endpoint_host": _endpoint_host(endpoint),
        "method": "POST",
        "operation": operation,
        "provenance_mode": "external_service_candidate",
        "timeout_seconds": _timeout_seconds(),
        "retry_policy": {**RETRY_POLICY, "backoff_seconds": _backoff_seconds()},
        "network_execution_env": ALLOW_NETWORK_ENV,
        "called_at": _utc_timestamp(),
        "request_hash": _hash_payload(request_preview),
        "response_hash": _hash_payload(payload),
        "attempt_count": attempt_count,
        "retry_error_codes": retry_error_codes or [],
    }
    if isinstance(official_request_profile, dict):
        redacted_headers = dict(official_request_profile.get("headers") or {})
        if "x-api-key" in redacted_headers:
            redacted_headers["x-api-key"] = "[redacted]"
        redacted_profile = {
            **official_request_profile,
            "headers": redacted_headers,
        }
        metadata["official_endpoint_path"] = official_request_profile.get("endpoint_path")
        metadata["official_request_profile"] = redacted_profile
        metadata["official_request_profile_hash"] = _hash_payload(redacted_profile)
    if isinstance(transport_metadata, dict):
        for key in (
            "cache_hit",
            "cache_key",
            "cache_created_at",
            "cache_expires_at",
            "cache_ttl_seconds",
            "free_tier_rate_limit",
        ):
            if key in transport_metadata:
                metadata[key] = transport_metadata[key]
    return metadata


def _normalize_success(
    payload: dict[str, Any],
    endpoint: str,
    request_preview: dict[str, Any],
    attempt_count: int = 1,
    retry_error_codes: list[int] | None = None,
) -> dict[str, Any]:
    metadata = {
        **_base_live_metadata(endpoint, request_preview, payload, "calculation", attempt_count, retry_error_codes),
        **(payload.get("source_metadata") or {}),
    }
    result = {
        "backend": "vedastro_service_adapter_candidate",
        "available": True,
        "status": "ok",
        "ayanamsa_value": payload.get("ayanamsa_value"),
        "node_policy": payload.get("node_policy"),
        "body_list": payload.get("body_list"),
        "bodies": payload.get("bodies"),
        "source_metadata": metadata,
    }
    result["source_metadata"]["artifact_path"] = _write_artifact(result)
    return result


def _normalize_external_technique_success(
    payload: dict[str, Any],
    endpoint: str,
    request_preview: dict[str, Any],
) -> dict[str, Any]:
    evidence = {
        "source": "vedastro_service_adapter_candidate",
        "operation": EXTERNAL_TECHNIQUE_OPERATION,
        "role": EXTERNAL_TECHNIQUE_ROLE,
        "domain": request_preview["domain"],
        "method": request_preview["method"],
        "api_endpoint": request_preview["api_endpoint"],
        "status": payload.get("status") or "ok",
        "summary": payload.get("summary"),
        "nature": payload.get("nature"),
        "tags": payload.get("tags") if isinstance(payload.get("tags"), list) else [],
        "raw": payload,
    }
    return {
        "backend": "vedastro_service_adapter_candidate",
        "available": True,
        "status": "ok",
        "operation": EXTERNAL_TECHNIQUE_OPERATION,
        "role": EXTERNAL_TECHNIQUE_ROLE,
        "domain": request_preview["domain"],
        "request_preview": request_preview,
        "evidence_ledger": [evidence],
        "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
        "source_metadata": {
            "transport": "http_json_service_boundary",
            "endpoint": endpoint,
            "provenance_mode": "external_service_candidate",
            "timeout_seconds": _timeout_seconds(),
            "retry_policy": RETRY_POLICY,
            **(payload.get("source_metadata") or {}),
        },
    }


def _normalize_range_scan_success(
    payload: dict[str, Any],
    endpoint: str,
    request_preview: dict[str, Any],
    attempt_count: int = 1,
    retry_error_codes: list[int] | None = None,
) -> dict[str, Any]:
    # Handle actual VedAstro response formats:
    # {"Status": "Pass", "Payload": {"SearchEvents": [...]}}
    # {"Status": "Pass", "Payload": [...]}
    if payload.get("Status") == "Pass":
        payload_body = payload.get("Payload", [])
        if isinstance(payload_body, dict):
            events = payload_body.get("SearchEvents", [])
        else:
            events = payload_body
    else:
        # Fallback to local stub format if not VedAstro format
        events = payload.get("events", [])
        
    if not isinstance(events, list):
        events = []

    domain = request_preview.get("domain", "")
    allowlist = RANGE_SCAN_EVENT_ALLOWLIST.get(domain, {})
    allowed_ids = allowlist.get("event_ids", set())
    allowed_tags = allowlist.get("tags", set())
    official_tags = RANGE_SCAN_OFFICIAL_TAG_MATCHES.get(domain, set())
    alias_terms = RANGE_SCAN_ALIAS_TERMS.get(domain, set())

    original_event_count = len(events)
    evidence_ledger = []
    mapping_details = []
    matched_tags: set[str] = set()
    recommended_allowlist_candidates: set[str] = set()
    match_counts = {
        "exact_id": 0,
        "official_tag": 0,
        "alias": 0,
        "rejected": 0,
    }
    for index, event in enumerate(events, start=1):
        if not isinstance(event, dict):
            continue
        # VedAstro uses "Name" for event id, and may expose tags as EventTags, tags, or Tag.
        event_id = event.get("Name") or event.get("id") or event.get("name") or f"event_{index}"
        tags = event.get("EventTags") or event.get("tags") or event.get("Tag") or []
        if isinstance(tags, str):
            tags = [part.strip() for part in tags.split(",") if part.strip()]
        elif not isinstance(tags, list):
            tags = []
        tag_set = {str(tag) for tag in tags}
        matched_by = "rejected"
        matched_terms: list[str] = []
        drop_reason = "no_supported_match"
        signal_metadata = RANGE_SCAN_SIGNAL_METADATA.get(domain, {}).get(event_id, {})
        if event_id in allowed_ids:
            matched_by = "exact_id"
            matched_terms = [event_id]
            drop_reason = ""
        else:
            official_tag_hits = sorted(tag_set.intersection(official_tags))
            if official_tag_hits:
                matched_by = "official_tag"
                matched_terms = official_tag_hits
                drop_reason = ""
            else:
                haystack_parts = [
                    str(event_id),
                    str(event.get("Description") or ""),
                    str(event.get("description") or ""),
                    str(event.get("Name") or ""),
                    " ".join(str(tag) for tag in tags),
                ]
                haystack = " ".join(part.lower() for part in haystack_parts if part)
                alias_hits = sorted(term for term in alias_terms if term in haystack)
                guard_hits = sorted(term for term in ALIAS_NEGATIVE_GUARD_TERMS if term in haystack)
                if alias_hits and not guard_hits:
                    matched_by = "alias"
                    matched_terms = alias_hits
                    drop_reason = ""
        match_counts[matched_by] += 1
        if matched_by == "rejected":
            mapping_details.append(
                {
                    "event_id": event_id,
                    "matched_by": matched_by,
                    "matched_terms": matched_terms,
                    "drop_reason": drop_reason,
                    "tags": tags,
                }
            )
            continue
        if matched_by == "official_tag":
            matched_tags.update(matched_terms)
            if event_id not in allowed_ids and not tag_set.intersection(allowed_tags):
                recommended_allowlist_candidates.add(event_id)
        elif matched_by == "alias":
            if event_id not in allowed_ids:
                recommended_allowlist_candidates.add(event_id)
        match_meta = MATCH_METADATA_BY_TYPE[matched_by]
        mapping_details.append(
            {
                "event_id": event_id,
                "matched_by": matched_by,
                "matched_terms": matched_terms,
                "drop_reason": drop_reason,
                "tags": tags,
            }
        )
        evidence_ledger.append(
            {
                "source": "vedastro_service_adapter_candidate",
                "operation": "range_scan",
                "domain": domain,
                "event_id": event_id,
                "matched_by": matched_by,
                "matched_terms": matched_terms,
                "signal_lift": match_meta["signal_lift"],
                "confidence": match_meta["confidence"],
                "drop_reason": None,
                "signal_key": signal_metadata.get("signal_key"),
                "signal_label": signal_metadata.get("signal_label") or event.get("name") or event_id,
                "signal_family": signal_metadata.get("signal_family"),
                "start": event.get("StartTime") or event.get("start") or event.get("start_time") or event.get("start_date"),
                "end": event.get("EndTime") or event.get("end") or event.get("end_time") or event.get("end_date"),
                "score": event.get("score") if event.get("score") is not None else event.get("strength"),
                "tags": tags,
                "raw": event,
            }
        )

    top_event = None
    if evidence_ledger:
        top = max(
            evidence_ledger,
            key=lambda item: item.get("score") if isinstance(item.get("score"), (int, float)) else float("-inf"),
        )
        top_event = {
            "event_id": top.get("event_id"),
            "signal_key": top.get("signal_key"),
            "signal_label": top.get("signal_label"),
            "signal_family": top.get("signal_family"),
            "score": top.get("score"),
            "start": top.get("start"),
            "end": top.get("end"),
            "tags": top.get("tags") or [],
        }
    daily_windows, top_daily_window = _build_daily_windows(domain, evidence_ledger)

    metadata = {
        **_base_live_metadata(endpoint, request_preview, payload, "range_scan", attempt_count, retry_error_codes),
        "vedastro_event_method": request_preview.get("vedastro_event_method"),
        "allowlist_domain": domain,
        "allowlist_event_count": len(evidence_ledger),
        "filtered_event_count": len(evidence_ledger),
        "raw_event_count": original_event_count,
        "mapping_replay": {
            "raw_event_count": original_event_count,
            "filtered_event_count": len(evidence_ledger),
            "zero_event_domains": [domain] if original_event_count > 0 and not evidence_ledger else [],
            "match_counts": match_counts,
            "matched_tags": sorted(matched_tags),
            "recommended_allowlist_candidates": sorted(recommended_allowlist_candidates),
            "events": mapping_details,
        },
        **(payload.get("source_metadata") or {}),
    }
    result = {
        "backend": "vedastro_service_adapter_candidate",
        "available": True,
        "status": "ok",
        "operation": "range_scan",
        "domain": domain,
        "request_preview": request_preview,
        "event_count": len(evidence_ledger),
        "top_event": top_event,
        "daily_windows": daily_windows,
        "top_daily_window": top_daily_window,
        "evidence_ledger": evidence_ledger,
        "source_metadata": metadata,
    }
    result["source_metadata"]["artifact_path"] = _write_artifact(result)
    return result


def _event_date_text(value: Any) -> str | None:
    if isinstance(value, dict):
        std_time = value.get("StdTime")
        if isinstance(std_time, str):
            parts = std_time.split()
            if len(parts) >= 2 and "/" in parts[1]:
                day, month, year = parts[1].split("/")
                return f"{year}-{month}-{day}"
        return None
    if isinstance(value, str):
        raw = value.strip()
        if len(raw) >= 10 and raw[4] == "-" and raw[7] == "-":
            return raw[:10]
    return None


def _confidence_rank(value: str) -> int:
    order = {
        "high": 4,
        "medium_high": 3,
        "medium": 2,
        "low": 1,
        "rejected": 0,
    }
    return order.get(str(value or ""), 0)


def _build_daily_windows(domain: str, evidence_ledger: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, Any] | None]:
    grouped: dict[str, dict[str, Any]] = {}
    for event in evidence_ledger:
        if not isinstance(event, dict):
            continue
        event_date = _event_date_text(event.get("start")) or _event_date_text(event.get("end"))
        if not event_date:
            continue
        bucket = grouped.setdefault(
            event_date,
            {
                "date": event_date,
                "domain": domain,
                "score": 0,
                "confidence": "low",
                "event_count": 0,
                "signal_families": set(),
                "event_ids": [],
                "top_signal_label": None,
                "_top_conf_rank": -1,
            },
        )
        bucket["event_count"] += 1
        bucket["score"] += int(event.get("signal_lift") or 0)
        family = event.get("signal_family")
        if family:
            bucket["signal_families"].add(str(family))
        event_id = event.get("event_id")
        if isinstance(event_id, str) and event_id and event_id not in bucket["event_ids"]:
            bucket["event_ids"].append(event_id)
        conf = str(event.get("confidence") or "low")
        conf_rank = _confidence_rank(conf)
        if conf_rank > bucket["_top_conf_rank"]:
            bucket["_top_conf_rank"] = conf_rank
            bucket["confidence"] = conf
            bucket["top_signal_label"] = event.get("signal_label") or event.get("event_id")

    windows: list[dict[str, Any]] = []
    for date_text, bucket in grouped.items():
        windows.append(
            {
                "date": date_text,
                "domain": bucket["domain"],
                "score": bucket["score"],
                "confidence": bucket["confidence"],
                "event_count": bucket["event_count"],
                "signal_families": sorted(bucket["signal_families"]),
                "event_ids": list(bucket["event_ids"]),
                "top_signal_label": bucket["top_signal_label"],
            }
        )
    windows.sort(key=lambda item: (-int(item.get("score") or 0), -int(item.get("event_count") or 0), str(item.get("date") or "")))
    top = windows[0] if windows else None
    return windows, top


def _source_metadata(endpoint: str) -> dict[str, Any]:
    return {
        "transport": "http_json_service_boundary",
        "endpoint": endpoint,
        "provenance_mode": "external_service_candidate",
        "timeout_seconds": _timeout_seconds(),
        "retry_policy": RETRY_POLICY,
        "network_execution_env": ALLOW_NETWORK_ENV,
    }


def _build_live_request(
    endpoint: str,
    request_preview: dict[str, Any],
) -> tuple[str, dict[str, Any], dict[str, Any]]:
    official_request_profile = None
    if isinstance(request_preview, dict):
        official_request_profile = (
            request_preview.get("live_sampling_request_profile")
            or request_preview.get("official_request_profile")
        )
    request_url = endpoint
    headers = {"Content-Type": "application/json"}
    vedastro_payload = request_preview
    if isinstance(official_request_profile, dict):
        request_url = f"{endpoint.rstrip('/')}{official_request_profile.get('endpoint_path', '')}"
        headers = dict(official_request_profile.get("headers") or headers)
        vedastro_payload = dict(official_request_profile.get("body") or {})
    api_key = os.environ.get("VEDASTRO_API_KEY", "").strip()
    if api_key:
        headers["x-api-key"] = api_key
    return request_url, headers, vedastro_payload


def _cache_key_for_request(
    request_url: str,
    headers: dict[str, Any],
    vedastro_payload: dict[str, Any],
) -> str:
    normalized_headers = dict(headers)
    if "x-api-key" in normalized_headers:
        normalized_headers["x-api-key"] = "[redacted]"
    return _hash_payload(
        {
            "request_url": request_url,
            "headers": normalized_headers,
            "body": vedastro_payload,
        }
    )


def _cache_path(cache_key: str) -> Path:
    return _cache_dir() / f"{cache_key}.json"


def _official_full_snapshot_cache_key(case: dict[str, Any], case_id: str) -> str:
    return _hash_payload(
        {
            "scope": "official_full_snapshot",
            "birth": {
                "year": case.get("year"),
                "month": case.get("month"),
                "day": case.get("day"),
                "hour": case.get("hour"),
                "minute": case.get("minute"),
                "second": case.get("second", 0),
                "lat": case.get("lat"),
                "lon": case.get("lon"),
                "tz": case.get("tz"),
            },
            "policies": {
                "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri",
                "node_policy": case.get("node_policy") or case.get("node_mode") or "mean",
                "reference_date": case.get("reference_date") or case.get("today") or case.get("transit_date") or case.get("current_date"),
                "dasha_levels": case.get("dasha_levels"),
                "dasha_precision_hours": case.get("dasha_precision_hours"),
            },
            "runtime": {
                "endpoint": os.environ.get("VEDASTRO_API_ENDPOINT", "").strip(),
                "network_enabled": os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() in {"1", "true", "yes"},
                "api_key_present": bool(os.environ.get("VEDASTRO_API_KEY", "").strip()),
                "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION,
            },
        }
    )


def _official_full_snapshot_cache_path(cache_key: str) -> Path:
    return _official_full_snapshot_cache_dir() / f"{cache_key}.json"


def _attach_official_full_snapshot_semantic_cache_metadata(
    payload: dict[str, Any],
    *,
    cache_key: str,
    created_at_unix: float,
    cache_hit: bool,
) -> dict[str, Any]:
    payload_copy = json.loads(json.dumps(payload))
    ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
    created_at = datetime.utcfromtimestamp(created_at_unix).strftime("%Y-%m-%dT%H:%M:%SZ")
    expires_at = datetime.utcfromtimestamp(created_at_unix + ttl_seconds).strftime("%Y-%m-%dT%H:%M:%SZ")
    metadata = dict(payload_copy.get("source_metadata") or {})
    metadata["semantic_cache"] = {
        "scope": "official_full_snapshot",
        "cache_hit": cache_hit,
        "cache_key": cache_key,
        "cache_created_at": created_at,
        "cache_expires_at": expires_at,
        "cache_ttl_seconds": ttl_seconds,
    }
    payload_copy["source_metadata"] = metadata
    return payload_copy


def _rebind_official_full_snapshot_case_identity(payload: dict[str, Any], case_id: str) -> dict[str, Any]:
    payload_copy = json.loads(json.dumps(payload))
    manifest = payload_copy.get("request_manifest")
    if isinstance(manifest, dict):
        manifest["case_id"] = case_id
    return payload_copy


def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str) -> dict[str, Any] | None:
    ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
    if ttl_seconds <= 0:
        return None
    cache_key = _official_full_snapshot_cache_key(case, case_id)
    cache_path = _official_full_snapshot_cache_path(cache_key)
    if not cache_path.exists():
        return None
    try:
        record = json.loads(cache_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None
    if not isinstance(record, dict):
        return None
    created_at_raw = record.get("created_at")
    payload = record.get("payload")
    if not isinstance(created_at_raw, (int, float)) or not isinstance(payload, dict):
        return None
    if time.time() - float(created_at_raw) > ttl_seconds:
        return None
    rebound = _rebind_official_full_snapshot_case_identity(payload, case_id)
    return _attach_official_full_snapshot_semantic_cache_metadata(
        rebound,
        cache_key=cache_key,
        created_at_unix=float(created_at_raw),
        cache_hit=True,
    )


def _store_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    cache_key = _official_full_snapshot_cache_key(case, case_id)
    created_at_unix = time.time()
    payload_copy = _attach_official_full_snapshot_semantic_cache_metadata(
        payload,
        cache_key=cache_key,
        created_at_unix=created_at_unix,
        cache_hit=False,
    )
    ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
    if ttl_seconds > 0:
        record = {
            "cache_key": cache_key,
            "created_at": created_at_unix,
            "payload": payload_copy,
        }
        _official_full_snapshot_cache_path(cache_key).write_text(
            json.dumps(record, ensure_ascii=False, sort_keys=True),
            encoding="utf-8",
        )
    return payload_copy


def _is_official_public_endpoint(request_url: str) -> bool:
    host = (urlparse(request_url).netloc or "").lower()
    if not host:
        return False
    if host.startswith("127.0.0.1") or host.startswith("localhost"):
        return False
    if host.startswith("192.168.") or host.startswith("10.") or host.startswith("172.16."):
        return False
    return "vedastro" in host


def _load_cached_payload(cache_key: str) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
    ttl_seconds = _cache_ttl_seconds()
    if ttl_seconds <= 0:
        return None, None
    cache_path = _cache_path(cache_key)
    if not cache_path.exists():
        return None, None
    try:
        record = json.loads(cache_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None, None
    if not isinstance(record, dict):
        return None, None
    created_at_raw = record.get("created_at")
    payload = record.get("payload")
    if not isinstance(created_at_raw, (int, float)) or not isinstance(payload, dict):
        return None, None
    age_seconds = time.time() - float(created_at_raw)
    if age_seconds > ttl_seconds:
        return None, None
    created_at = datetime.utcfromtimestamp(float(created_at_raw)).strftime("%Y-%m-%dT%H:%M:%SZ")
    expires_at = datetime.utcfromtimestamp(float(created_at_raw) + ttl_seconds).strftime("%Y-%m-%dT%H:%M:%SZ")
    payload_copy = json.loads(json.dumps(payload))
    metadata = dict(payload_copy.get("source_metadata") or {})
    metadata.update(
        {
            "cache_hit": True,
            "cache_key": cache_key,
            "cache_created_at": created_at,
            "cache_expires_at": expires_at,
            "cache_ttl_seconds": ttl_seconds,
        }
    )
    payload_copy["source_metadata"] = metadata
    return payload_copy, metadata


def _is_cacheable_payload(payload: dict[str, Any]) -> bool:
    status = str(payload.get("Status") or "")
    if status == "Pass":
        return True
    if isinstance(payload.get("events"), list):
        return True
    if status.lower() == "fail":
        payload_text = json.dumps(payload.get("Payload"), ensure_ascii=False).lower()
        if "rate limit" in payload_text or "calls/minute" in payload_text or "too many requests" in payload_text:
            return False
    return False


def _store_cached_payload(cache_key: str, payload: dict[str, Any]) -> dict[str, Any]:
    ttl_seconds = _cache_ttl_seconds()
    payload_copy = json.loads(json.dumps(payload))
    created_at_unix = time.time()
    created_at = datetime.utcfromtimestamp(created_at_unix).strftime("%Y-%m-%dT%H:%M:%SZ")
    expires_at = datetime.utcfromtimestamp(created_at_unix + ttl_seconds).strftime("%Y-%m-%dT%H:%M:%SZ")
    metadata = dict(payload_copy.get("source_metadata") or {})
    metadata.update(
        {
            "cache_hit": False,
            "cache_key": cache_key,
            "cache_created_at": created_at,
            "cache_expires_at": expires_at,
            "cache_ttl_seconds": ttl_seconds,
        }
    )
    payload_copy["source_metadata"] = metadata
    if ttl_seconds > 0 and _is_cacheable_payload(payload_copy):
        cache_path = _cache_path(cache_key)
        record = {
            "cache_key": cache_key,
            "created_at": created_at_unix,
            "payload": payload_copy,
        }
        cache_path.write_text(json.dumps(record, ensure_ascii=False, sort_keys=True), encoding="utf-8")
    return payload_copy


def _acquire_free_tier_slot(request_url: str) -> dict[str, Any]:
    if not _is_official_public_endpoint(request_url):
        return {
            "mode": "non_public_or_self_host_endpoint",
            "queue_active": False,
            "waited_seconds": 0.0,
            "window_seconds": _free_tier_window_seconds(),
            "max_requests": _free_tier_max_requests(),
        }
    if os.environ.get("VEDASTRO_API_KEY", "").strip():
        return {
            "mode": "api_key_present",
            "queue_active": False,
            "waited_seconds": 0.0,
            "window_seconds": _free_tier_window_seconds(),
            "max_requests": _free_tier_max_requests(),
        }
    max_requests = _free_tier_max_requests()
    window_seconds = _free_tier_window_seconds()
    if max_requests <= 0 or window_seconds <= 0:
        return {
            "mode": "free_tier_queue_disabled",
            "queue_active": False,
            "waited_seconds": 0.0,
            "window_seconds": window_seconds,
            "max_requests": max_requests,
        }

    waited_seconds = 0.0
    with _FREE_TIER_REQUEST_LOCK:
        while True:
            now = time.monotonic()
            _FREE_TIER_REQUEST_TIMESTAMPS[:] = [
                ts for ts in _FREE_TIER_REQUEST_TIMESTAMPS if now - ts < window_seconds
            ]
            if len(_FREE_TIER_REQUEST_TIMESTAMPS) < max_requests:
                _FREE_TIER_REQUEST_TIMESTAMPS.append(now)
                break
            sleep_seconds = max(window_seconds - (now - _FREE_TIER_REQUEST_TIMESTAMPS[0]), 0.0)
            waited_seconds += sleep_seconds
            if sleep_seconds > 0:
                time.sleep(sleep_seconds)
            else:
                _FREE_TIER_REQUEST_TIMESTAMPS.pop(0)
    return {
        "mode": "free_tier_queue_active",
        "queue_active": True,
        "waited_seconds": round(waited_seconds, 6),
        "window_seconds": window_seconds,
        "max_requests": max_requests,
    }


def _post_json(endpoint: str, request_preview: dict[str, Any]) -> dict[str, Any] | str:
    request_url, headers, vedastro_payload = _build_live_request(endpoint, request_preview)
    req = request.Request(
        request_url,
        data=json.dumps(vedastro_payload).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    with request.urlopen(req, timeout=_timeout_seconds()) as resp:
        raw = resp.read().decode("utf-8")
        return json.loads(raw)


def _retry_status_codes() -> set[int]:
    codes = set()
    for value in RETRY_POLICY.get("retry_on", []):
        try:
            codes.add(int(str(value)))
        except ValueError:
            continue
    return codes


def _post_json_with_retry(endpoint: str, request_preview: dict[str, Any]) -> tuple[dict[str, Any], int, list[int]]:
    request_url, headers, vedastro_payload = _build_live_request(endpoint, request_preview)
    cache_key = _cache_key_for_request(request_url, headers, vedastro_payload)
    cached_payload, _ = _load_cached_payload(cache_key)
    if isinstance(cached_payload, dict):
        return cached_payload, 1, []

    retry_codes = _retry_status_codes()
    retry_error_codes: list[int] = []
    max_attempts = int(RETRY_POLICY["max_attempts"])
    for attempt in range(1, max_attempts + 1):
        try:
            rate_limit_metadata = _acquire_free_tier_slot(request_url)
            payload = _post_json(endpoint, request_preview)
            if not isinstance(payload, dict):
                return {}, attempt, retry_error_codes
            payload_copy = _store_cached_payload(cache_key, payload)
            transport_metadata = dict(payload_copy.get("source_metadata") or {})
            transport_metadata["free_tier_rate_limit"] = rate_limit_metadata
            payload_copy["source_metadata"] = transport_metadata
            return payload_copy, attempt, retry_error_codes
        except error.HTTPError as exc:
            if attempt >= max_attempts or exc.code not in retry_codes:
                raise
            retry_error_codes.append(exc.code)
            if _backoff_seconds():
                time.sleep(_backoff_seconds())
    return {}, max_attempts, retry_error_codes


def _iter_sample_dates(start_date: str, end_date: str) -> list[str]:
    from datetime import datetime, timedelta

    start = datetime.strptime(start_date, "%Y-%m-%d").date()
    end = datetime.strptime(end_date, "%Y-%m-%d").date()
    if end <= start:
        return [start_date]
    span_days = max((end - start).days, 1)
    step_days = max(1, span_days // 11)
    dates: list[str] = []
    current = start
    while current <= end:
        dates.append(current.isoformat())
        current += timedelta(days=step_days)
    if dates[-1] != end.isoformat():
        dates.append(end.isoformat())
    return dates


def _merge_range_scan_reports(
    reports: list[dict[str, Any]],
    endpoint: str,
    base_preview: dict[str, Any],
) -> dict[str, Any]:
    if not reports:
        return _normalize_range_scan_success({"Status": "Pass", "Payload": []}, endpoint, base_preview)

    deduped: dict[tuple[str, str, str], dict[str, Any]] = {}
    for report in reports:
        for item in report.get("evidence_ledger", []):
            key = (
                str(item.get("event_id") or ""),
                str(item.get("start") or ""),
                str(item.get("end") or ""),
            )
            existing = deduped.get(key)
            existing_score = existing.get("signal_lift", 0) if existing else -1
            score = item.get("signal_lift", 0)
            if existing is None or score > existing_score:
                deduped[key] = item

    evidence_ledger = list(deduped.values())
    top_event = None
    if evidence_ledger:
        top = max(
            evidence_ledger,
            key=lambda item: (
                item.get("signal_lift") if isinstance(item.get("signal_lift"), (int, float)) else 0,
                item.get("confidence") == "high",
            ),
        )
        top_event = {
            "event_id": top.get("event_id"),
            "signal_key": top.get("signal_key"),
            "signal_label": top.get("signal_label"),
            "signal_family": top.get("signal_family"),
            "score": top.get("score"),
            "start": top.get("start"),
            "end": top.get("end"),
            "tags": top.get("tags") or [],
        }
    daily_windows, top_daily_window = _build_daily_windows(str(base_preview.get("domain") or ""), evidence_ledger)

    metadata = dict(reports[-1].get("source_metadata") or {})
    metadata["sampling_mode"] = "at_time_sweep"
    metadata["sample_dates"] = [report.get("request_preview", {}).get("start_date") for report in reports]
    metadata["sample_count"] = len(reports)
    metadata["raw_event_count"] = sum(int((report.get("source_metadata") or {}).get("raw_event_count", 0)) for report in reports)
    metadata["filtered_event_count"] = len(evidence_ledger)
    metadata["allowlist_event_count"] = len(evidence_ledger)
    metadata["attempt_count"] = sum(int((report.get("source_metadata") or {}).get("attempt_count", 1)) for report in reports)
    retry_codes: list[int] = []
    for report in reports:
        retry_codes.extend(list((report.get("source_metadata") or {}).get("retry_error_codes") or []))
    metadata["retry_error_codes"] = retry_codes

    result = {
        "backend": "vedastro_service_adapter_candidate",
        "available": True,
        "status": "ok",
        "operation": "range_scan",
        "domain": base_preview.get("domain"),
        "request_preview": base_preview,
        "event_count": len(evidence_ledger),
        "top_event": top_event,
        "daily_windows": daily_windows,
        "top_daily_window": top_daily_window,
        "evidence_ledger": evidence_ledger,
        "source_metadata": metadata,
    }
    result["source_metadata"]["artifact_path"] = _write_artifact(result)
    return result


def run_case(case_id: str) -> dict[str, Any]:
    if case_id not in PARITY_CASES:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "unknown_case_id",
            "reason": f"Unknown parity case: {case_id}",
        }

    endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
    if not endpoint:
        return _unconfigured("VEDASTRO_API_ENDPOINT is not configured; adapter skeleton stops before network access.")

    case = PARITY_CASES[case_id]
    request_preview = _request_preview(case)
    if os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() not in {"1", "true", "yes"}:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "network_execution_disabled",
            "reason": f"{ALLOW_NETWORK_ENV} is not enabled; adapter stops after building request/provenance metadata.",
            "request_preview": request_preview,
            "source_metadata": _source_metadata(endpoint),
        }
    try:
        payload, attempt_count, retry_error_codes = _post_json_with_retry(endpoint, request_preview)
    except error.HTTPError as exc:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "http_error",
            "reason": f"VedAstro adapter HTTP error: {exc.code}",
            "request_preview": request_preview,
            "source_metadata": _source_metadata(endpoint),
        }
    except error.URLError as exc:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "network_error",
            "reason": f"VedAstro adapter network error: {exc.reason}",
            "request_preview": request_preview,
            "source_metadata": _source_metadata(endpoint),
        }
    except (TimeoutError, socket.timeout):
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "timeout",
            "reason": "VedAstro adapter timed out",
            "request_preview": request_preview,
            "source_metadata": _source_metadata(endpoint),
        }
    except json.JSONDecodeError:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "invalid_json",
            "reason": "VedAstro adapter received non-JSON response",
            "request_preview": request_preview,
            "source_metadata": _source_metadata(endpoint),
        }

    return _normalize_success(payload, endpoint, request_preview)


def _official_full_snapshot_metadata(endpoint: str | None, manifest: dict[str, Any]) -> dict[str, Any]:
    metadata = {
        "transport": "http_json_service_boundary",
        "operation": "official_full_snapshot",
        "primary_source": "vedastro_official",
        "provenance_mode": "vedastro_official_primary_candidate",
        "timeout_seconds": _timeout_seconds(),
        "retry_policy": {**RETRY_POLICY, "backoff_seconds": _backoff_seconds()},
        "network_execution_env": ALLOW_NETWORK_ENV,
        "method_catalog_url": OFFICIAL_METHOD_CATALOG_URL,
        "reference_date": manifest.get("reference_date"),
        "request_hash": _hash_payload(manifest),
    }
    if endpoint:
        metadata["endpoint"] = endpoint
        metadata["endpoint_host"] = _endpoint_host(endpoint)
    return metadata


def _official_snapshot_budget_exhausted_bundle(reason: str) -> dict[str, Any]:
    return {
        "available": False,
        "status": "official_snapshot_budget_exhausted",
        "source": "vedastro_official_capability_runner",
        "bundle": "official_full_snapshot",
        "snapshot_sections": {},
        "section_statuses": {},
        "coverage": {"source_mode": "official_foreground_budget", "filled_sections": []},
        "reason": reason,
        "timeout_seconds": _timeout_seconds(),
    }


def _payload_status(payload: dict[str, Any]) -> str:
    if not isinstance(payload, dict):
        return "invalid"
    if str(payload.get("Status") or "").lower() == "fail":
        failure_text = json.dumps(payload.get("Payload"), ensure_ascii=False).lower()
        if "rate limit" in failure_text or "calls/minute" in failure_text or "too many requests" in failure_text:
            return "rate_limited"
    return "ok" if payload.get("Status") == "Pass" else "fail"


def _aggregate_section_status(statuses: list[str]) -> str:
    if statuses and all(status == "ok" for status in statuses):
        return "ok"
    if any(status == "rate_limited" for status in statuses):
        return "rate_limited"
    return "partial"


def _degrees_from_sign_payload(value: Any) -> float | None:
    if not isinstance(value, dict):
        return None
    degrees = value.get("DegreesIn") if isinstance(value.get("DegreesIn"), dict) else {}
    raw = degrees.get("TotalDegrees")
    try:
        return float(raw)
    except (TypeError, ValueError):
        return None


def _sign_position(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, dict):
        return None
    sign = value.get("Name")
    degree = _degrees_from_sign_payload(value)
    if not sign:
        return None
    return {
        "sign": sign,
        "degree_in_sign": degree,
    }


def _extract_all_planet_data(payload: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(payload, dict):
        return {}
    body = payload.get("Payload") if isinstance(payload.get("Payload"), dict) else {}
    data = body.get("AllPlanetData") if isinstance(body.get("AllPlanetData"), dict) else {}
    return data if isinstance(data, dict) else {}


def _extract_all_house_data(payload: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(payload, dict):
        return {}
    body = payload.get("Payload") if isinstance(payload.get("Payload"), dict) else {}
    data = body.get("AllHouseData") if isinstance(body.get("AllHouseData"), dict) else {}
    return data if isinstance(data, dict) else {}


def _official_planet_snapshot(planet_name: str, data: dict[str, Any]) -> dict[str, Any]:
    d1 = _sign_position(data.get("PlanetRasiD1Sign")) or {}
    raw_lon = None
    nirayana = data.get("PlanetNirayanaLongitude")
    if isinstance(nirayana, dict):
        try:
            raw_lon = float(nirayana.get("TotalDegrees"))
        except (TypeError, ValueError):
            raw_lon = None
    house_text = data.get("HousePlanetOccupiesBasedOnSign") or data.get("HousePlanetOccupiesBasedOnLongitudes")
    house = None
    if isinstance(house_text, str) and house_text.lower().startswith("house"):
        try:
            house = int("".join(ch for ch in house_text if ch.isdigit()))
        except ValueError:
            house = None
    return {
        "source": "vedastro_official",
        "name": planet_name,
        "sign": d1.get("sign"),
        "degree_in_sign": d1.get("degree_in_sign"),
        "degree": raw_lon,
        "lon": raw_lon,
        "house": house,
        "vargas": {
            "D1": d1,
            "D2": _sign_position(data.get("PlanetHoraD2Signs")),
            "D3": _sign_position(data.get("PlanetDrekkanaD3Sign")),
            "D4": _sign_position(data.get("PlanetChaturthamshaD4Sign")),
            "D7": _sign_position(data.get("PlanetSaptamshaD7Sign")),
            "D9": _sign_position(data.get("PlanetNavamshaD9Sign")),
            "D10": _sign_position(data.get("PlanetDashamamshaD10Sign")),
            "D12": _sign_position(data.get("PlanetDwadashamshaD12Sign")),
            "D16": _sign_position(data.get("PlanetShodashamshaD16Sign")),
            "D20": _sign_position(data.get("PlanetVimshamshaD20Sign")),
            "D24": _sign_position(data.get("PlanetChaturvimshamshaD24Sign")),
            "D27": _sign_position(data.get("PlanetBhamshaD27Sign")),
            "D30": _sign_position(data.get("PlanetTrimshamshaD30Sign")),
            "D40": _sign_position(data.get("PlanetKhavedamshaD40Sign")),
            "D45": _sign_position(data.get("PlanetAkshavedamshaD45Sign")),
            "D60": _sign_position(data.get("PlanetShashtyamshaD60Sign")),
        },
        "nakshatra": data.get("PlanetConstellation"),
        "raw_source_keys": sorted(data.keys()),
    }


def _official_house_snapshot(house_name: str, data: dict[str, Any]) -> dict[str, Any]:
    d1 = _sign_position(data.get("HouseRasiD1Sign") or data.get("HouseBhavaChalitSign")) or {}
    return {
        "source": "vedastro_official",
        "name": house_name,
        "sign": d1.get("sign"),
        "degree_in_sign": d1.get("degree_in_sign"),
        "vargas": {
            "D1": d1,
            "D2": _sign_position(data.get("HouseHoraD2Sign") or data.get("HouseHoraD2Signs")),
            "D3": _sign_position(data.get("HouseDrekkanaD3Sign")),
            "D4": _sign_position(data.get("HouseChaturthamshaD4Sign")),
            "D7": _sign_position(data.get("HouseSaptamshaD7Sign")),
            "D9": _sign_position(data.get("HouseNavamshaD9Sign") or data.get("HouseNavamsaD9Sign")),
            "D10": _sign_position(data.get("HouseDashamamshaD10Sign")),
            "D12": _sign_position(data.get("HouseDwadashamshaD12Sign")),
            "D16": _sign_position(data.get("HouseShodashamshaD16Sign")),
            "D20": _sign_position(data.get("HouseVimshamshaD20Sign")),
            "D24": _sign_position(data.get("HouseChaturvimshamshaD24Sign")),
            "D27": _sign_position(data.get("HouseBhamshaD27Sign")),
            "D30": _sign_position(data.get("HouseTrimshamshaD30Sign")),
            "D40": _sign_position(data.get("HouseKhavedamshaD40Sign")),
            "D45": _sign_position(data.get("HouseAkshavedamshaD45Sign")),
            "D60": _sign_position(data.get("HouseShashtyamshaD60Sign")),
        },
        "nakshatra": data.get("HouseConstellation"),
        "raw_source_keys": sorted(data.keys()),
    }


def _build_official_chart_from_snapshot(sections: dict[str, Any]) -> dict[str, Any]:
    chart_core = sections.get("chart_core") if isinstance(sections.get("chart_core"), dict) else {}
    house_core = sections.get("house_core") if isinstance(sections.get("house_core"), dict) else {}
    planets: dict[str, Any] = {}
    for planet_name, payload in chart_core.items():
        data = _extract_all_planet_data(payload)
        if data:
            planets[planet_name] = _official_planet_snapshot(planet_name, data)
    houses: dict[str, Any] = {}
    for house_name, payload in house_core.items():
        data = _extract_all_house_data(payload)
        if data:
            houses[house_name] = _official_house_snapshot(house_name, data)
    ascendant = houses.get("House1") or {}
    return {
        "source": "vedastro_official",
        "primary_source": "vedastro_official",
        "planets": planets,
        "houses": houses,
        "ascendant": ascendant,
        "coverage": {
            "planet_count": len(planets),
            "house_count": len(houses),
            "varga_keys": ["D1", "D2", "D3", "D4", "D7", "D9", "D10", "D12", "D16", "D20", "D24", "D27", "D30", "D40", "D45", "D60"],
        },
    }


def _post_official_snapshot_section(endpoint: str, request_item: dict[str, Any]) -> tuple[dict[str, Any], int, list[int]]:
    body = dict(request_item["body"])
    fanout_parameter = request_item.get("fanout_parameter")
    fanout_value = request_item.get("fanout_value")
    if fanout_parameter and fanout_value is not None:
        if fanout_parameter == "planetName":
            body[fanout_parameter] = {"Name": str(fanout_value)}
        else:
            body[fanout_parameter] = str(fanout_value)
    section_preview = {
        "operation": "official_full_snapshot",
        "section": request_item["section"],
        "official_request_profile": {
            "profile_version": OFFICIAL_FULL_SNAPSHOT_PROFILE_VERSION,
            "endpoint_path": request_item["endpoint_path"],
            "method": request_item["method"],
            "headers": request_item["headers"],
            "body": body,
        },
    }
    return _post_json_with_retry(endpoint, section_preview)


def _normalize_official_full_snapshot_success(
    endpoint: str,
    manifest: dict[str, Any],
    sections: dict[str, Any],
    section_statuses: dict[str, str],
    attempt_count: int,
    retry_error_codes: list[int],
    official_python_bundle: dict[str, Any] | None = None,
    official_full_capability_catalog: dict[str, Any] | None = None,
) -> dict[str, Any]:
    primary_sections = [item["section"] for item in manifest["requests"]]
    ok_count = sum(1 for section in primary_sections if section_statuses.get(section) == "ok")
    fast_primary_present = OFFICIAL_FAST_PRIMARY_SECTIONS.intersection(set(section_statuses))
    fast_primary_ok = bool(fast_primary_present) and all(
        section_statuses.get(section) == "ok" for section in fast_primary_present
    )
    status = "ok" if ok_count == len(primary_sections) else "partial"
    if fast_primary_ok and len(fast_primary_present) >= 4:
        status = "ok"
    rate_limited_sections = [
        section
        for section in primary_sections
        if section_statuses.get(section) == "rate_limited"
    ]
    metadata = {
        **_official_full_snapshot_metadata(endpoint, manifest),
        "called_at": _utc_timestamp(),
        "section_statuses": section_statuses,
        "section_count": len(primary_sections),
        "section_ok_count": ok_count,
        "fast_primary_sections": sorted(OFFICIAL_FAST_PRIMARY_SECTIONS),
        "fast_primary_ok": fast_primary_ok,
        "rate_limited_sections": rate_limited_sections,
        "attempt_count": attempt_count,
        "retry_error_codes": retry_error_codes,
        "response_hash": _hash_payload({"sections": sections, "section_statuses": section_statuses}),
    }
    if isinstance(official_python_bundle, dict):
        metadata["official_python_bundle"] = {
            "source": official_python_bundle.get("source"),
            "status": official_python_bundle.get("status"),
            "coverage": official_python_bundle.get("coverage"),
            "bundle": official_python_bundle.get("bundle"),
            "summary": official_python_bundle.get("summary"),
            "python_bin": official_python_bundle.get("python_bin"),
            "module_name": official_python_bundle.get("module_name"),
        }
        metadata["official_python_path"] = str(official_python_bundle.get("source") or "vedastro_python_bridge")
        metadata["python_bridge"] = metadata["official_python_bundle"]
    if isinstance(official_full_capability_catalog, dict):
        metadata["official_full_capability_catalog"] = {
            "source": official_full_capability_catalog.get("source"),
            "status": official_full_capability_catalog.get("status"),
            "bundle": official_full_capability_catalog.get("bundle"),
            "summary": official_full_capability_catalog.get("summary"),
            "coverage": official_full_capability_catalog.get("coverage"),
            "domain_routing": official_full_capability_catalog.get("domain_routing") or {},
            "dynamic_selection": official_full_capability_catalog.get("dynamic_selection") or {},
        }
    if rate_limited_sections:
        metadata["production_hint"] = "configure_vedastro_api_key_or_self_host_official_api"
    result = {
        "backend": "vedastro_service_adapter_candidate",
        "available": ok_count > 0,
        "status": status,
        "operation": "official_full_snapshot",
        "primary_source": "vedastro_official",
        "snapshot_sections": sections,
        "raw_response": {
            "source": "vedastro_official_full_snapshot",
            "sections": sections,
            "section_statuses": section_statuses,
            "request_manifest": manifest,
            "response_hash": metadata["response_hash"],
        },
        "official_chart": _build_official_chart_from_snapshot(sections),
        "official_full_capability_catalog": official_full_capability_catalog or {},
        "section_statuses": section_statuses,
        "request_manifest": manifest,
        "user_visibility": "backend_raw_evidence_not_direct_user_report",
        "source_metadata": metadata,
    }
    result["source_metadata"]["artifact_path"] = _write_artifact(result)
    return result


def _run_official_full_snapshot_case(case: dict[str, Any], case_id: str = "user_chart") -> dict[str, Any]:
    user_case = {
        "case_id": case_id,
        "year": case.get("year"),
        "month": case.get("month"),
        "day": case.get("day"),
        "hour": case.get("hour"),
        "minute": case.get("minute"),
        "second": case.get("second", 0),
        "lat": case.get("lat"),
        "lon": case.get("lon"),
        "tz": case.get("tz"),
        "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri",
        "node_policy": case.get("node_policy") or case.get("node_mode") or "mean",
        "reference_date": case.get("reference_date") or case.get("today") or case.get("transit_date") or case.get("current_date"),
        "dasha_levels": case.get("dasha_levels"),
        "dasha_precision_hours": case.get("dasha_precision_hours"),
        "themes": case.get("themes") or case.get("theme"),
    }
    manifest = _official_full_snapshot_manifest(user_case, case_id)
    budget_started_at = time.monotonic()
    official_full_capability_catalog: dict[str, Any] = {}
    elapsed_seconds = time.monotonic() - budget_started_at
    if elapsed_seconds >= _timeout_seconds():
        official_python_bundle = _official_snapshot_budget_exhausted_bundle(
            "VedAstro official full-snapshot budget was exhausted before the snapshot runner started."
        )
    else:
        official_python_bundle = _try_official_capability_runner_snapshot_bundle(user_case)
    if (
        not official_python_bundle.get("snapshot_sections")
        and official_python_bundle.get("status") != "official_snapshot_budget_exhausted"
        and time.monotonic() - budget_started_at >= _timeout_seconds()
    ):
        official_python_bundle = _official_snapshot_budget_exhausted_bundle(
            "VedAstro official snapshot runner consumed the foreground full-snapshot budget."
        )
    if (
        not official_python_bundle.get("snapshot_sections")
        and official_python_bundle.get("status") != "official_snapshot_budget_exhausted"
    ):
        official_python_bundle = _try_official_python_bridge_snapshot_bundle(user_case)
    budget_exhausted = official_python_bundle.get("status") == "official_snapshot_budget_exhausted"
    if not budget_exhausted and time.monotonic() - budget_started_at < _timeout_seconds():
        official_full_capability_catalog = _try_official_full_capability_catalog_bundle(user_case)
    else:
        official_full_capability_catalog = {
            "available": False,
            "status": "official_full_capability_catalog_skipped_budget_exhausted",
            "source": "vedastro_official_capability_runner",
            "bundle": "official_full_capability_catalog",
            "summary": {},
            "coverage": {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
            "domain_routing": {},
            "dynamic_selection": {},
            "bucket_statuses": {},
            "method_statuses": {},
            "reason": "Skipped capability catalog after preserving foreground budget for the official full snapshot bundle.",
            "timeout_seconds": _timeout_seconds(),
        }
    bridge_sections = (
        official_python_bundle.get("snapshot_sections")
        if isinstance(official_python_bundle.get("snapshot_sections"), dict)
        else {}
    )
    bridge_section_statuses = (
        official_python_bundle.get("section_statuses")
        if isinstance(official_python_bundle.get("section_statuses"), dict)
        else {}
    )
    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",
            "available": False,
            "status": "official_snapshot_budget_exhausted",
            "operation": "official_full_snapshot",
            "primary_source": "vedastro_official",
            "reason": official_python_bundle.get("reason"),
            "snapshot_sections": {},
            "official_full_capability_catalog": official_full_capability_catalog,
            "request_manifest": manifest,
            "user_visibility": "backend_raw_evidence_not_direct_user_report",
            "source_metadata": _official_full_snapshot_metadata(endpoint or None, manifest),
        }
        result["source_metadata"]["official_full_capability_catalog"] = {
            "source": official_full_capability_catalog.get("source"),
            "status": official_full_capability_catalog.get("status"),
            "bundle": official_full_capability_catalog.get("bundle"),
            "summary": official_full_capability_catalog.get("summary"),
            "coverage": official_full_capability_catalog.get("coverage"),
            "domain_routing": official_full_capability_catalog.get("domain_routing") or {},
            "dynamic_selection": official_full_capability_catalog.get("dynamic_selection") or {},
        }
        result["source_metadata"]["official_python_bundle"] = {
            "source": official_python_bundle.get("source"),
            "status": official_python_bundle.get("status"),
            "coverage": official_python_bundle.get("coverage"),
            "bundle": official_python_bundle.get("bundle"),
            "summary": official_python_bundle.get("summary"),
            "python_bin": official_python_bundle.get("python_bin"),
            "module_name": official_python_bundle.get("module_name"),
        }
        result["source_metadata"]["official_python_path"] = str(official_python_bundle.get("source") or "vedastro_python_bridge")
        result["source_metadata"]["python_bridge"] = result["source_metadata"]["official_python_bundle"]
        return result
    if not endpoint:
        if bridge_sections:
            return _normalize_official_full_snapshot_success(
                "",
                manifest,
                bridge_sections,
                bridge_section_statuses,
                1,
                [],
                official_python_bundle=official_python_bundle,
                official_full_capability_catalog=official_full_capability_catalog,
            )
        result = {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "service_endpoint_not_configured",
            "operation": "official_full_snapshot",
            "primary_source": "vedastro_official",
            "reason": "VEDASTRO_API_ENDPOINT is not configured; official full snapshot stops before network access.",
            "snapshot_sections": {},
            "official_full_capability_catalog": official_full_capability_catalog,
            "request_manifest": manifest,
            "user_visibility": "backend_raw_evidence_not_direct_user_report",
            "source_metadata": _official_full_snapshot_metadata(None, manifest),
        }
        result["source_metadata"]["official_full_capability_catalog"] = {
            "source": official_full_capability_catalog.get("source"),
            "status": official_full_capability_catalog.get("status"),
            "bundle": official_full_capability_catalog.get("bundle"),
            "summary": official_full_capability_catalog.get("summary"),
            "coverage": official_full_capability_catalog.get("coverage"),
            "domain_routing": official_full_capability_catalog.get("domain_routing") or {},
            "dynamic_selection": official_full_capability_catalog.get("dynamic_selection") or {},
        }
        result["source_metadata"]["official_python_bundle"] = {
            "source": official_python_bundle.get("source"),
            "status": official_python_bundle.get("status"),
            "coverage": official_python_bundle.get("coverage"),
            "bundle": official_python_bundle.get("bundle"),
            "summary": official_python_bundle.get("summary"),
            "python_bin": official_python_bundle.get("python_bin"),
            "module_name": official_python_bundle.get("module_name"),
        }
        result["source_metadata"]["official_python_path"] = str(official_python_bundle.get("source") or "vedastro_python_bridge")
        result["source_metadata"]["python_bridge"] = result["source_metadata"]["official_python_bundle"]
        return result

    if not network_enabled:
        if bridge_sections:
            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,
            )
        result = {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "network_execution_disabled",
            "operation": "official_full_snapshot",
            "primary_source": "vedastro_official",
            "reason": f"{ALLOW_NETWORK_ENV} is not enabled; official full snapshot stops after building request manifest.",
            "snapshot_sections": {},
            "official_full_capability_catalog": official_full_capability_catalog,
            "request_manifest": manifest,
            "user_visibility": "backend_raw_evidence_not_direct_user_report",
            "source_metadata": _official_full_snapshot_metadata(endpoint, manifest),
        }
        result["source_metadata"]["official_full_capability_catalog"] = {
            "source": official_full_capability_catalog.get("source"),
            "status": official_full_capability_catalog.get("status"),
            "bundle": official_full_capability_catalog.get("bundle"),
            "summary": official_full_capability_catalog.get("summary"),
            "coverage": official_full_capability_catalog.get("coverage"),
            "domain_routing": official_full_capability_catalog.get("domain_routing") or {},
            "dynamic_selection": official_full_capability_catalog.get("dynamic_selection") or {},
        }
        result["source_metadata"]["official_python_bundle"] = {
            "source": official_python_bundle.get("source"),
            "status": official_python_bundle.get("status"),
            "coverage": official_python_bundle.get("coverage"),
            "bundle": official_python_bundle.get("bundle"),
            "summary": official_python_bundle.get("summary"),
            "python_bin": official_python_bundle.get("python_bin"),
            "module_name": official_python_bundle.get("module_name"),
        }
        result["source_metadata"]["official_python_path"] = str(official_python_bundle.get("source") or "vedastro_python_bridge")
        result["source_metadata"]["python_bridge"] = result["source_metadata"]["official_python_bundle"]
        return result

    sections: dict[str, Any] = dict(bridge_sections)
    section_statuses: dict[str, str] = dict(bridge_section_statuses)
    attempt_count = 0
    retry_error_codes: list[int] = []
    for request_item in manifest["requests"]:
        section = request_item["section"]
        if section in sections and section_statuses.get(section) == "ok":
            continue
        fanout_values = request_item.get("fanout_values") if isinstance(request_item.get("fanout_values"), list) else []
        if fanout_values:
            section_payloads: dict[str, Any] = {}
            fanout_statuses: dict[str, str] = {}
            for value in fanout_values:
                fanout_request = {**request_item, "fanout_value": value}
                try:
                    payload, attempts, retries = _post_official_snapshot_section(endpoint, fanout_request)
                    section_payloads[str(value)] = payload
                    fanout_statuses[str(value)] = _payload_status(payload)
                    attempt_count += attempts
                    retry_error_codes.extend(retries)
                except error.HTTPError as exc:
                    fanout_statuses[str(value)] = f"http_error:{exc.code}"
                except (error.URLError, http.client.RemoteDisconnected) as exc:
                    fanout_statuses[str(value)] = f"network_error:{getattr(exc, 'reason', str(exc))}"
                except (TimeoutError, socket.timeout):
                    fanout_statuses[str(value)] = "timeout"
                except json.JSONDecodeError:
                    fanout_statuses[str(value)] = "invalid_json"
            sections[section] = section_payloads
            section_statuses[section] = _aggregate_section_status(list(fanout_statuses.values()))
            section_statuses[f"{section}_fanout"] = fanout_statuses
            continue

        try:
            payload, attempts, retries = _post_official_snapshot_section(endpoint, request_item)
            sections[section] = payload
            section_statuses[section] = _payload_status(payload)
            attempt_count += attempts
            retry_error_codes.extend(retries)
        except error.HTTPError as exc:
            section_statuses[section] = f"http_error:{exc.code}"
        except (error.URLError, http.client.RemoteDisconnected) as exc:
            section_statuses[section] = f"network_error:{getattr(exc, 'reason', str(exc))}"
        except (TimeoutError, socket.timeout):
            section_statuses[section] = "timeout"
        except json.JSONDecodeError:
            section_statuses[section] = "invalid_json"

    return _normalize_official_full_snapshot_success(
        endpoint,
        manifest,
        sections,
        section_statuses,
        attempt_count or 1,
        retry_error_codes,
        official_python_bundle=official_python_bundle,
        official_full_capability_catalog=official_full_capability_catalog,
    )


def run_official_full_snapshot(case_id: str, reference_date: str | None = None) -> dict[str, Any]:
    if case_id not in PARITY_CASES:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "unknown_case_id",
            "operation": "official_full_snapshot",
            "primary_source": "vedastro_official",
            "reason": f"Unknown parity case: {case_id}",
        }
    case = dict(PARITY_CASES[case_id])
    if reference_date:
        case["reference_date"] = reference_date
    return _run_official_full_snapshot_case(case, case_id=case_id)


def run_official_full_snapshot_for_case(
    case: dict[str, Any],
    *,
    case_id: str = "user_chart",
) -> dict[str, Any]:
    cached = _load_official_full_snapshot_semantic_cache(case, case_id)
    if isinstance(cached, dict):
        return cached
    result = _run_official_full_snapshot_case(case, case_id=case_id)
    if not isinstance(result, dict):
        return result
    return _store_official_full_snapshot_semantic_cache(case, case_id, result)


def _rectification_position(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, dict) or not value.get("sign"):
        return None
    position = {"sign": str(value["sign"])}
    degree = value.get("degree_in_sign")
    if isinstance(degree, (int, float)):
        position["degree_in_sign"] = round(float(degree), 6)
    return position


def _rectification_varga_positions(
    chart: dict[str, Any],
    varga: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
    houses = chart.get("houses") if isinstance(chart.get("houses"), dict) else {}
    planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
    house_positions = {
        name: position
        for name, item in sorted(houses.items())
        if isinstance(item, dict)
        and (position := _rectification_position((item.get("vargas") or {}).get(varga)))
    }
    planet_positions = {
        name: position
        for name, item in sorted(planets.items())
        if isinstance(item, dict)
        and (position := _rectification_position((item.get("vargas") or {}).get(varga)))
    }
    return house_positions, planet_positions


def _rectification_timeline_item_count(value: Any) -> int:
    if isinstance(value, list):
        return len(value)
    if isinstance(value, dict):
        return max(
            (_rectification_timeline_item_count(item) for item in value.values()),
            default=0,
        )
    return 0


def run_rectification_minute_snapshot_for_case(
    case: dict[str, Any],
    *,
    case_id: str = "user_chart",
) -> dict[str, Any]:
    """Return a safe official fingerprint for minute-candidate comparison.

    The summary intentionally excludes raw official responses. SearchEvents is
    not part of this fingerprint; it remains a separate background check.
    """
    snapshot = run_official_full_snapshot_for_case(case, case_id=case_id)
    chart = snapshot.get("official_chart") if isinstance(snapshot.get("official_chart"), dict) else {}
    houses = chart.get("houses") if isinstance(chart.get("houses"), dict) else {}
    ascendant = _rectification_position(chart.get("ascendant"))
    house_positions = {
        name: position
        for name, item in sorted(houses.items())
        if (position := _rectification_position(item))
    }
    d9_houses, d9_planets = _rectification_varga_positions(chart, "D9")
    d10_houses, d10_planets = _rectification_varga_positions(chart, "D10")
    sections = snapshot.get("snapshot_sections") if isinstance(snapshot.get("snapshot_sections"), dict) else {}
    section_statuses = snapshot.get("section_statuses") if isinstance(snapshot.get("section_statuses"), dict) else {}
    dasha_payload = sections.get("dasha_all")

    ascendant_house_payload = {
        "ascendant": ascendant,
        "houses": house_positions,
    }
    d9_payload = {"houses": d9_houses, "planets": d9_planets}
    d10_payload = {"houses": d10_houses, "planets": d10_planets}
    dasha_available = section_statuses.get("dasha_all") == "ok" and isinstance(dasha_payload, dict)
    layers = {
        "ascendant_house_boundaries": {
            "status": "ok" if ascendant and house_positions else "missing",
            **ascendant_house_payload,
            "fingerprint": _hash_payload(ascendant_house_payload) if ascendant and house_positions else None,
        },
        "D9": {
            "status": "ok" if d9_houses or d9_planets else "missing",
            **d9_payload,
            "fingerprint": _hash_payload(d9_payload) if d9_houses or d9_planets else None,
        },
        "D10": {
            "status": "ok" if d10_houses or d10_planets else "missing",
            **d10_payload,
            "fingerprint": _hash_payload(d10_payload) if d10_houses or d10_planets else None,
        },
        "dasha_boundaries": {
            "status": "ok" if dasha_available else "missing",
            "boundary_count": _rectification_timeline_item_count(dasha_payload) if dasha_available else 0,
            "fingerprint": _hash_payload(dasha_payload) if dasha_available else None,
        },
        "kp_cusp_sub_lord": {
            "status": "unsupported_by_verified_official_interface",
            "reason": "The verified VedAstro interface exposes KP house-membership helpers, not an auditable cusp/sub-lord result.",
        },
    }
    required_statuses = [layers[name]["status"] for name in (
        "ascendant_house_boundaries", "D9", "D10", "dasha_boundaries"
    )]
    available = bool(snapshot.get("available")) and all(status == "ok" for status in required_statuses)
    source_metadata = snapshot.get("source_metadata") if isinstance(snapshot.get("source_metadata"), dict) else {}
    return {
        "available": available,
        "status": "ok" if available else "partial" if any(status == "ok" for status in required_statuses) else "blocked",
        "source": "vedastro_official",
        "operation": "rectification_minute_snapshot",
        "candidate_time": f'{int(case.get("hour") or 0):02d}:{int(case.get("minute") or 0):02d}',
        "layers": layers,
        "kp_cusp_sub_lord": layers["kp_cusp_sub_lord"],
        "source_metadata": {
            "response_hash": source_metadata.get("response_hash"),
            "official_snapshot_status": snapshot.get("status"),
        },
    }


def _run_range_scan_case(case: dict[str, Any], domain: str, start_date: str, end_date: str) -> dict[str, Any]:
    if domain not in SUPPORTED_RANGE_SCAN_DOMAINS:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "unsupported_range_scan_domain",
            "reason": f"Unsupported range scan domain: {domain}",
        }

    request_preview = _range_scan_preview(case, domain, start_date, end_date)
    endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
    if not endpoint:
        result = _unconfigured("VEDASTRO_API_ENDPOINT is not configured; range scan stops before network access.")
        result["operation"] = "range_scan"
        result["domain"] = domain
        result["request_preview"] = request_preview
        return result

    if os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() not in {"1", "true", "yes"}:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "network_execution_disabled",
            "reason": f"{ALLOW_NETWORK_ENV} is not enabled; range scan stops after building request/provenance metadata.",
            "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:
        sample_preview = dict(request_preview)
        sample_preview["start_date"] = sample_date
        sample_preview["end_date"] = sample_date
        sample_preview["official_request_profile"] = _build_official_search_events_profile(sample_preview)
        sample_preview["live_sampling_request_profile"] = _build_live_sampling_search_events_profile(sample_preview)
        try:
            payload, attempt_count, retry_error_codes = _post_json_with_retry(endpoint, sample_preview)
        except error.HTTPError as exc:
            return {
                "backend": "vedastro_service_adapter_candidate",
                "available": False,
                "status": "http_error",
                "reason": f"VedAstro range scan HTTP error: {exc.code}",
                "request_preview": sample_preview,
                "source_metadata": _source_metadata(endpoint),
            }
        except (error.URLError, http.client.RemoteDisconnected) as exc:
            return {
                "backend": "vedastro_service_adapter_candidate",
                "available": False,
                "status": "network_error",
                "reason": f"VedAstro range scan network error: {getattr(exc, 'reason', str(exc))}",
                "request_preview": sample_preview,
                "source_metadata": _source_metadata(endpoint),
            }
        except (TimeoutError, socket.timeout):
            return {
                "backend": "vedastro_service_adapter_candidate",
                "available": False,
                "status": "timeout",
                "reason": "VedAstro range scan timed out",
                "request_preview": sample_preview,
                "source_metadata": _source_metadata(endpoint),
            }
        except json.JSONDecodeError:
            return {
                "backend": "vedastro_service_adapter_candidate",
                "available": False,
                "status": "invalid_json",
                "reason": "VedAstro range scan received non-JSON response",
                "request_preview": sample_preview,
                "source_metadata": _source_metadata(endpoint),
            }
        reports.append(_normalize_range_scan_success(payload, endpoint, sample_preview, attempt_count, retry_error_codes))

    return _merge_range_scan_reports(reports, endpoint, request_preview)


def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> dict[str, Any]:
    if case_id not in PARITY_CASES:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "unknown_case_id",
            "reason": f"Unknown parity case: {case_id}",
        }
    return _run_range_scan_case(PARITY_CASES[case_id], domain, start_date, end_date)


def run_range_scan_for_case(
    case: dict[str, Any],
    domain: str,
    start_date: str,
    end_date: str,
    case_id: str = "user_chart",
) -> dict[str, Any]:
    user_case = {
        "case_id": case_id,
        "year": case.get("year"),
        "month": case.get("month"),
        "day": case.get("day"),
        "hour": case.get("hour"),
        "minute": case.get("minute"),
        "second": case.get("second", 0),
        "lat": case.get("lat"),
        "lon": case.get("lon"),
        "tz": case.get("tz"),
        "location_name": case.get("location_name") or case.get("place_name") or case.get("city") or "UserLocation",
        "ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri",
        "node_policy": case.get("node_policy") or case.get("node_mode") or "mean",
    }
    return _run_range_scan_case(user_case, domain, start_date, end_date)


def run_external_technique(case_id: str, domain: str, method: str, api_endpoint: str) -> dict[str, Any]:
    if case_id not in PARITY_CASES:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "unknown_case_id",
            "reason": f"Unknown parity case: {case_id}",
        }
    if domain not in SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "unsupported_external_technique_domain",
            "reason": f"Unsupported external technique domain: {domain}",
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
        }
    if not method.strip() or not api_endpoint.strip():
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "missing_external_technique_method",
            "reason": "Both method and api_endpoint are required for external technique evidence.",
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
        }

    endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
    if not endpoint:
        result = _unconfigured(
            "VEDASTRO_API_ENDPOINT is not configured; external technique evidence stops before network access."
        )
        result["operation"] = EXTERNAL_TECHNIQUE_OPERATION
        result["role"] = EXTERNAL_TECHNIQUE_ROLE
        result["domain"] = domain
        result["adjudicator_policy"] = EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY
        return result

    request_preview = _external_technique_preview(PARITY_CASES[case_id], domain, method, api_endpoint)
    if os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() not in {"1", "true", "yes"}:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "network_execution_disabled",
            "reason": (
                f"{ALLOW_NETWORK_ENV} is not enabled; external technique evidence stops "
                "after building request/provenance metadata."
            ),
            "request_preview": request_preview,
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
            "source_metadata": _source_metadata(endpoint),
        }

    try:
        payload = _post_json(endpoint, request_preview)
    except error.HTTPError as exc:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "http_error",
            "reason": f"VedAstro external technique HTTP error: {exc.code}",
            "request_preview": request_preview,
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
            "source_metadata": _source_metadata(endpoint),
        }
    except error.URLError as exc:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "network_error",
            "reason": f"VedAstro external technique network error: {exc.reason}",
            "request_preview": request_preview,
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
            "source_metadata": _source_metadata(endpoint),
        }
    except (TimeoutError, socket.timeout):
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "timeout",
            "reason": "VedAstro external technique timed out",
            "request_preview": request_preview,
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
            "source_metadata": _source_metadata(endpoint),
        }
    except json.JSONDecodeError:
        return {
            "backend": "vedastro_service_adapter_candidate",
            "available": False,
            "status": "invalid_json",
            "reason": "VedAstro external technique received non-JSON response",
            "request_preview": request_preview,
            "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY,
            "source_metadata": _source_metadata(endpoint),
        }

    return _normalize_external_technique_success(payload, endpoint, request_preview)


def main() -> int:
    parser = argparse.ArgumentParser(description="VedAstro service adapter skeleton")
    parser.add_argument("--print-schema", action="store_true")
    parser.add_argument("--case", default="beijing_first_use_demo")
    parser.add_argument("--range-scan", action="store_true")
    parser.add_argument("--official-full-snapshot", action="store_true")
    parser.add_argument("--domain", choices=sorted(SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS), default="marriage")
    parser.add_argument("--start-date", default="2026-01-01")
    parser.add_argument("--end-date", default="2031-01-01")
    parser.add_argument("--reference-date", default=None)
    parser.add_argument("--external-technique", action="store_true")
    parser.add_argument("--method", default="")
    parser.add_argument("--api-endpoint", default="")
    args = parser.parse_args()

    if args.print_schema:
        result = schema()
    elif args.official_full_snapshot:
        result = run_official_full_snapshot(args.case, reference_date=args.reference_date)
    elif args.external_technique:
        result = run_external_technique(args.case, args.domain, args.method, args.api_endpoint)
    elif args.range_scan:
        result = run_range_scan(args.case, args.domain, args.start_date, args.end_date)
    else:
        result = run_case(args.case)
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())