Gate VedAstro network execution stub

scripts/vedastro_service_adapter.py

@@ -63,6 +63,7 @@ RETRY_POLICY = {
     "backoff_seconds": 1,
     "retry_on": ["timeout", "429", "502", "503", "504"],
 }
+ALLOW_NETWORK_ENV = "VEDASTRO_ENABLE_NETWORK"
 
 
 def schema() -> dict[str, Any]:
@@ -134,6 +135,13 @@ def _unconfigured(reason: str) -> dict[str, Any]:
     }
 
 
+def _request_preview(case: dict[str, Any]) -> dict[str, Any]:
+    return {
+        **case,
+        "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"],
+    }
+
+
 def run_case(case_id: str) -> dict[str, Any]:
     if case_id not in PARITY_CASES:
         return {
@@ -148,21 +156,36 @@ def run_case(case_id: str) -> dict[str, Any]:
         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": {
+                "transport": "http_json_service_boundary",
+                "endpoint": endpoint,
+                "provenance_mode": "external_service_candidate",
+                "timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
+                "retry_policy": RETRY_POLICY,
+                "network_execution_env": ALLOW_NETWORK_ENV,
+            },
+        }
     return {
         "backend": "vedastro_service_adapter_candidate",
         "available": False,
         "status": "service_execution_not_implemented",
         "reason": "Endpoint is configured, but the real HTTP execution/normalization layer is intentionally not implemented yet.",
-        "request_preview": {
-            **case,
-            "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"],
-        },
+        "request_preview": request_preview,
         "source_metadata": {
             "transport": "http_json_service_boundary",
             "endpoint": endpoint,
             "provenance_mode": "external_service_candidate",
             "timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
             "retry_policy": RETRY_POLICY,
+            "network_execution_env": ALLOW_NETWORK_ENV,
         },
     }
 

tests/test_vedastro_service_adapter_executor.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import os
 import subprocess
 import sys
 from pathlib import Path
@@ -72,3 +73,27 @@ def test_ephemeris_adapter_contract_can_call_vedastro_candidate_stub() -> None:
     first = report["rows"][0]["candidate_backend"]
     assert first["backend"] == "vedastro_service_adapter_candidate"
     assert first["status"] == "service_endpoint_not_configured"
+
+
+def test_vedastro_service_adapter_builds_request_preview_before_real_network_use() -> None:
+    env = os.environ.copy()
+    env["VEDASTRO_API_ENDPOINT"] = "https://example.invalid/vedastro"
+
+    completed = subprocess.run(
+        [sys.executable, "scripts/vedastro_service_adapter.py", "--case", "beijing_first_use_demo"],
+        cwd=ROOT,
+        text=True,
+        capture_output=True,
+        timeout=120,
+        check=False,
+        env=env,
+    )
+
+    assert completed.returncode == 0, completed.stderr or completed.stdout
+    report = json.loads(completed.stdout)
+    assert report["backend"] == "vedastro_service_adapter_candidate"
+    assert report["status"] == "network_execution_disabled"
+    assert report["request_preview"]["year"] == 1990
+    assert report["request_preview"]["ayanamsa_policy"] == "lahiri"
+    assert report["source_metadata"]["endpoint"] == "https://example.invalid/vedastro"
+    assert report["source_metadata"]["provenance_mode"] == "external_service_candidate"