Wire VedAstro status and live quality gate

README.md

@@ -83,6 +83,7 @@ Static demo / PWA 发布要求：必须保留 `static_demo_boundary_visible` 说
 - browser：完整浏览器守门，覆盖 runtime smoke 与真实浏览器用户路径：`python3 scripts/run_quality_gate.py --profile browser`
 - release：发布前守门，包含关键产品文件未跟踪检查、慢速 golden cases、真实案例复验与 Yoga 逻辑报告：`python3 scripts/run_quality_gate.py --profile release`
 - accuracy：本地准确率守门，跳过浏览器点击重活，但强制运行真实案例复验、Dasha/Oracle 审计、Yoga 逻辑对照和本地准确率总报告：`python3 scripts/run_quality_gate.py --profile accuracy`
+- vedastro-live：外部 VedAstro 雷达守门，只跑可选 live smoke，默认不依赖网络；只有配置 `VEDASTRO_API_ENDPOINT` 与 `VEDASTRO_ENABLE_NETWORK=1` 时才真正出网：`python3 scripts/run_quality_gate.py --profile vedastro-live`
 
 ### 真实案例复验与准确率边界
 

docs/superpowers/plans/2026-06-29-vedastro-adapter-mvp.md

@@ -0,0 +1,180 @@
+# VedAstro Adapter MVP Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Complete the VedAstro Adapter MVP as a gated external timing radar with provenance, strict workflow injection, Trust Center status, and optional live smoke.
+
+**Architecture:** Extend the existing `scripts/vedastro_service_adapter.py` service boundary instead of replacing local computation. Add deterministic mock-tested behavior by default and a separate optional live profile when `VEDASTRO_API_ENDPOINT` and `VEDASTRO_ENABLE_NETWORK` are configured.
+
+**Tech Stack:** Python standard library HTTP/JSON, pytest, existing `scripts/run_quality_gate.py`, existing `jyotish-app` JavaScript Trust Center.
+
+## Global Constraints
+
+- Local Jyotish computation remains authoritative.
+- VedAstro evidence may only enter `secondary_context`, `technique_audit`, `external_activation`, and Life Event Graph external nodes.
+- Default CI must not depend on live network.
+- No secrets may be rendered in the frontend.
+- New production behavior must be introduced with failing tests first.
+
+---
+
+### Task 1: Adapter Provenance, Retry, and Artifact Persistence
+
+**Files:**
+- Modify: `scripts/vedastro_service_adapter.py`
+- Test: `tests/test_vedastro_service_adapter_executor.py`
+
+**Interfaces:**
+- Produces: `run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> dict`
+- Produces metadata keys: `request_hash`, `response_hash`, `called_at`, `artifact_path`, `allowlist_event_count`, `filtered_event_count`
+
+- [ ] **Step 1: Write failing tests**
+
+Add tests asserting live mock range-scan writes an artifact and records request/response hashes, plus retry succeeds after one `503`.
+
+- [ ] **Step 2: Run red tests**
+
+Run: `python3 -m pytest tests/test_vedastro_service_adapter_executor.py -k "artifact or retry" -q`
+
+Expected: fails because provenance/artifact/retry fields are missing.
+
+- [ ] **Step 3: Implement minimal adapter changes**
+
+Add deterministic SHA-256 helpers, UTC `called_at`, scratch artifact writer, and retry loop around `_post_json`.
+
+- [ ] **Step 4: Run green tests**
+
+Run: `python3 -m pytest tests/test_vedastro_service_adapter_executor.py -q`
+
+Expected: all adapter executor tests pass.
+
+### Task 2: Backend Status Route and Live Quality Gate
+
+**Files:**
+- Modify: `scripts/jyotish_api_server.py`
+- Modify: `scripts/run_quality_gate.py`
+- Test: `tests/test_api_server_security.py`
+- Test: `tests/test_vedastro_service_adapter_executor.py`
+
+**Interfaces:**
+- Produces: `GET /api/vedastro/status`
+- Produces quality profile: `vedastro-live`
+
+- [ ] **Step 1: Write failing tests**
+
+Add API test requiring `/api/vedastro/status` to report `configured`, `network_enabled`, `status`, and safe provenance fields. Add quality-gate static test requiring `vedastro-live`.
+
+- [ ] **Step 2: Run red tests**
+
+Run: `python3 -m pytest tests/test_api_server_security.py -k vedastro_status -q`
+
+Expected: fails because the route is absent.
+
+- [ ] **Step 3: Implement route and quality profile**
+
+Expose safe status from adapter schema/env and add a `vedastro-live` profile that runs the adapter live smoke only when configured, otherwise reports a controlled skip/blocked message.
+
+- [ ] **Step 4: Run green tests**
+
+Run: `python3 -m pytest tests/test_api_server_security.py tests/test_vedastro_service_adapter_executor.py -k "vedastro" -q`
+
+Expected: focused VedAstro tests pass.
+
+### Task 3: Strict Workflow Auto-Injection
+
+**Files:**
+- Modify: `mcp_server.py`
+- Test: `tests/test_mcp_strict_workflow_relationship.py`
+- Test: `tests/test_mcp_strict_workflow_career.py`
+- Test: `tests/test_mcp_strict_workflow_finance.py`
+- Test: `tests/test_life_event_graph_v1.py`
+
+**Interfaces:**
+- Consumes: `modules.external_activation.evidence_ledger`
+- Produces: strict workflow `external_activation` when adapter evidence is supplied or generated by a helper.
+
+- [ ] **Step 1: Write failing tests**
+
+Add tests requiring strict workflow to accept an adapter result and render external nodes in Life Event Graph without score/label override.
+
+- [ ] **Step 2: Run red tests**
+
+Run: `python3 -m pytest tests/test_mcp_strict_workflow_relationship.py tests/test_life_event_graph_v1.py -k "vedastro or external" -q`
+
+Expected: fails for missing auto-injection or graph fields.
+
+- [ ] **Step 3: Implement minimal strict workflow bridge**
+
+Normalize adapter result into `modules.external_activation` and preserve the existing blocked row when no evidence is available.
+
+- [ ] **Step 4: Run green tests**
+
+Run: `python3 -m pytest tests/test_mcp_strict_workflow_relationship.py tests/test_mcp_strict_workflow_career.py tests/test_mcp_strict_workflow_finance.py tests/test_life_event_graph_v1.py -q`
+
+Expected: strict workflow tests pass.
+
+### Task 4: Trust Center Status Surface
+
+**Files:**
+- Modify: `jyotish-app/main.js`
+- Modify: `jyotish-app/export.js` if status is exported
+- Test: `tests/test_frontend_productization.py`
+
+**Interfaces:**
+- Consumes: `/api/vedastro/status`
+- Produces visible Trust Center status labels without rendering endpoint secrets.
+
+- [ ] **Step 1: Write failing frontend static test**
+
+Require `renderVedAstroStatus`, `/api/vedastro/status`, `VEDASTRO_API_ENDPOINT`, and a non-secret status label in Trust Center.
+
+- [ ] **Step 2: Run red test**
+
+Run: `python3 -m pytest tests/test_frontend_productization.py -k vedastro -q`
+
+Expected: fails because Trust Center status is not rendered.
+
+- [ ] **Step 3: Implement frontend status card**
+
+Add a compact Trust Center row showing unconfigured/network disabled/live-ready/last artifact states.
+
+- [ ] **Step 4: Run green tests**
+
+Run: `python3 -m pytest tests/test_frontend_productization.py -k vedastro -q && npm run build --prefix jyotish-app`
+
+Expected: frontend tests and build pass.
+
+### Task 5: Verification and Progress Update
+
+**Files:**
+- Modify: `progress.md`
+- Modify: `findings.md`
+
+**Interfaces:**
+- Produces: permanent project log with exact verification commands.
+
+- [ ] **Step 1: Run focused verification**
+
+Run: `python3 -m pytest tests/test_vedastro_service_adapter_executor.py tests/test_vedastro_external_technique_evidence.py tests/test_vedastro_parity_matrix.py tests/test_vedastro_adapter_candidate_guard.py -q`
+
+- [ ] **Step 2: Run workflow verification**
+
+Run: `python3 -m pytest tests/test_mcp_strict_workflow_finance.py tests/test_mcp_strict_workflow_relationship.py tests/test_mcp_strict_workflow_career.py tests/test_life_event_graph_v1.py -q`
+
+- [ ] **Step 3: Run product verification**
+
+Run: `python3 -m pytest tests/test_api_server_security.py tests/test_frontend_productization.py -k "vedastro or trust or api_runtime" -q && npm run build --prefix jyotish-app`
+
+- [ ] **Step 4: Run quick gate**
+
+Run: `python3 scripts/run_quality_gate.py --profile quick --skip-frontend-runtime`
+
+- [ ] **Step 5: Update progress files**
+
+Record implemented files, blocked live endpoint boundary, and verification results in `progress.md` and `findings.md`.
+
+## Self-Review
+
+- Spec coverage: adapter provenance, live gate, strict workflow injection, Trust Center, and verification are each assigned to a task.
+- Placeholder scan: no TBD/TODO placeholders remain.
+- Type consistency: adapter result fields use existing `evidence_ledger` and `source_metadata` contracts.

jyotish-app/api-bridge.js

@@ -228,6 +228,10 @@ async function getCapabilityAudit() {
   throw lastError || new Error(buildAPIRecoveryMessage('/api/capability_audit', '能力审计接口不可用'));
 }
 
+async function getVedAstroStatus() {
+  return fetchJson('/api/vedastro/status');
+}
+
 async function getTechniqueCatalog() {
   let lastError = null;
   for (const base of getApiBases(true)) {
@@ -450,6 +454,7 @@ window.JyotishAPI = {
   computeThematicReport,
   getAPIHealth,
   getCapabilityAudit,
+  getVedAstroStatus,
   getTechniqueCatalog,
   runTechniqueExample,
   computeAnnual,

jyotish-app/public/api-bridge.js

@@ -228,6 +228,10 @@ async function getCapabilityAudit() {
   throw lastError || new Error(buildAPIRecoveryMessage('/api/capability_audit', '能力审计接口不可用'));
 }
 
+async function getVedAstroStatus() {
+  return fetchJson('/api/vedastro/status');
+}
+
 async function getTechniqueCatalog() {
   let lastError = null;
   for (const base of getApiBases(true)) {
@@ -450,6 +454,7 @@ window.JyotishAPI = {
   computeThematicReport,
   getAPIHealth,
   getCapabilityAudit,
+  getVedAstroStatus,
   getTechniqueCatalog,
   runTechniqueExample,
   computeAnnual,

mcp_server.py

@@ -413,8 +413,22 @@ def _derive_jaimini_marriage_support(present: Dict[str, Any]) -> Dict[str, Any]:
     }
 
 
-def _derive_external_activation_support(modules: Dict[str, Any], domain: str) -> Dict[str, Any]:
+def _external_activation_ledger(modules: Dict[str, Any]) -> tuple[Any, Dict[str, Any]]:
     ledger = _safe_get(modules, "external_activation", "evidence_ledger")
+    if isinstance(ledger, list):
+        activation = modules.get("external_activation") if isinstance(modules, dict) else {}
+        metadata = activation.get("source_metadata") if isinstance(activation, dict) else {}
+        return ledger, metadata if isinstance(metadata, dict) else {}
+    adapter_result = modules.get("vedastro_range_scan_result") if isinstance(modules, dict) else {}
+    if isinstance(adapter_result, dict) and adapter_result.get("backend") == "vedastro_service_adapter_candidate":
+        ledger = adapter_result.get("evidence_ledger")
+        metadata = adapter_result.get("source_metadata")
+        return ledger, metadata if isinstance(metadata, dict) else {}
+    return None, {}
+
+
+def _derive_external_activation_support(modules: Dict[str, Any], domain: str) -> Dict[str, Any]:
+    ledger, provenance = _external_activation_ledger(modules)
     if not isinstance(ledger, list):
         return {
             "level": "missing_required_external_radar",
@@ -457,6 +471,7 @@ def _derive_external_activation_support(modules: Dict[str, Any], domain: str) ->
         "required": True,
         "operation": "range_scan",
         "external_calculation_coverage": "VedAstro 596+/600+ calculation nodes",
+        "provenance": provenance,
     }
 
 

scripts/run_quality_gate.py

@@ -98,6 +98,7 @@ QUALITY_GATE_PROFILES = {
         "skip_dasha_audit": True,
         "skip_oracle_audit": True,
         "skip_local_accuracy_report": True,
+        "skip_vedastro_live": True,
     },
     "browser": {
         "skip_slow": True,
@@ -110,6 +111,7 @@ QUALITY_GATE_PROFILES = {
         "skip_dasha_audit": True,
         "skip_oracle_audit": True,
         "skip_local_accuracy_report": True,
+        "skip_vedastro_live": True,
     },
     "release": {
         "skip_slow": False,
@@ -122,6 +124,7 @@ QUALITY_GATE_PROFILES = {
         "skip_dasha_audit": False,
         "skip_oracle_audit": False,
         "skip_local_accuracy_report": False,
+        "skip_vedastro_live": True,
     },
     "accuracy": {
         "skip_slow": True,
@@ -134,6 +137,20 @@ QUALITY_GATE_PROFILES = {
         "skip_dasha_audit": False,
         "skip_oracle_audit": False,
         "skip_local_accuracy_report": False,
+        "skip_vedastro_live": True,
+    },
+    "vedastro-live": {
+        "skip_slow": True,
+        "skip_yoga_logic": True,
+        "skip_frontend_runtime": True,
+        "skip_frontend_click": True,
+        "frontend_click_mode": "core",
+        "check_release_hygiene": False,
+        "skip_real_cases": True,
+        "skip_dasha_audit": True,
+        "skip_oracle_audit": True,
+        "skip_local_accuracy_report": True,
+        "skip_vedastro_live": False,
     },
 }
 
@@ -341,6 +358,36 @@ def release_hygiene_check() -> None:
     print("release_hygiene_check ok: no release-critical product files are untracked")
 
 
+def run_vedastro_live_smoke() -> None:
+    print("\n== VedAstro live adapter smoke ==")
+    endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
+    network_enabled = os.environ.get("VEDASTRO_ENABLE_NETWORK", "").strip().lower() in {"1", "true", "yes"}
+    if not endpoint or not network_enabled:
+        print(json.dumps({
+            "status": "blocked",
+            "reason": "vedastro_live_endpoint_or_network_flag_missing",
+            "required_env": {
+                "endpoint": "VEDASTRO_API_ENDPOINT",
+                "network": "VEDASTRO_ENABLE_NETWORK",
+            },
+            "boundary": "Default CI stays deterministic; configure both env vars to run a real VedAstro live smoke.",
+        }, ensure_ascii=False, indent=2))
+        return
+    run([
+        PYTHON,
+        "scripts/vedastro_service_adapter.py",
+        "--range-scan",
+        "--domain",
+        "career",
+        "--case",
+        "beijing_first_use_demo",
+        "--start-date",
+        "2026-01-01",
+        "--end-date",
+        "2026-12-31",
+    ])
+
+
 def compile_targets() -> None:
     print("\n== Compile core Python files ==")
     targets: list[Path] = []
@@ -386,6 +433,7 @@ def run_profile(args: argparse.Namespace) -> dict:
         "skip_dasha_audit",
         "skip_oracle_audit",
         "skip_local_accuracy_report",
+        "skip_vedastro_live",
     ]:
         if getattr(args, key):
             profile[key] = True
@@ -396,7 +444,7 @@ def run_profile(args: argparse.Namespace) -> dict:
 
 def main() -> int:
     parser = argparse.ArgumentParser(description="Run Jyotish skill quality gate")
-    parser.add_argument("--profile", choices=["quick", "browser", "release", "accuracy"], default="browser", help="Quality gate profile: quick, browser, release, or accuracy")
+    parser.add_argument("--profile", choices=["quick", "browser", "release", "accuracy", "vedastro-live"], default="browser", help="Quality gate profile: quick, browser, release, accuracy, or vedastro-live")
     parser.add_argument("--skip-slow", action="store_true", help="Skip slow golden-case regressions")
     parser.add_argument("--skip-yoga-logic", action="store_true", help="Skip Yoga logic comparison report refresh")
     parser.add_argument("--skip-frontend-runtime", action="store_true", help="Skip frontend build and runtime smoke")
@@ -405,6 +453,7 @@ def main() -> int:
     parser.add_argument("--skip-dasha-audit", action="store_true", help="Skip Dasha reference-drift audit")
     parser.add_argument("--skip-oracle-audit", action="store_true", help="Skip combined Dasha/Shadbala external oracle boundary audit")
     parser.add_argument("--skip-local-accuracy-report", action="store_true", help="Skip consolidated local accuracy report")
+    parser.add_argument("--skip-vedastro-live", action="store_true", help="Skip optional VedAstro live endpoint smoke")
     parser.add_argument("--frontend-click-mode", choices=["core", "mobile", "offline", "pdf", "workspace", "mobile-trust", "import-files", "all"], default=None, help="Browser click smoke mode for browser/release profiles")
     parser.add_argument("--frontend-click-timeout", type=int, default=240, help="Timeout seconds for browser click smoke")
     parser.add_argument("--all-tests", action="store_true", help="Run every pytest file, including optional-dependency suites")
@@ -449,6 +498,8 @@ def main() -> int:
         run([PYTHON, "scripts/validate_logic_v2.py"], optional=True)
     if not profile["skip_local_accuracy_report"]:
         run([PYTHON, "scripts/local_accuracy_report.py", "--format", "json"])
+    if not profile["skip_vedastro_live"]:
+        run_vedastro_live_smoke()
     print("\nQuality gate passed.")
     return 0
 

scripts/vedastro_service_adapter.py

@@ -9,12 +9,15 @@ workspace can evolve from research notes to an executable adapter contract.
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import socket
+import time
 from pathlib import Path
 from typing import Any
 from urllib import request, error
+from urllib.parse import urlparse
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -195,12 +198,14 @@ RANGE_SCAN_SIGNAL_METADATA = {
 }
 DEFAULT_TIMEOUT_SECONDS = 120
 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"
+ARTIFACT_DIR = ROOT / "scratch" / "local" / "vedastro_adapter"
 
 
 def _timeout_seconds() -> float:
@@ -213,6 +218,57 @@ def _timeout_seconds() -> float:
         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 _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 schema() -> dict[str, Any]:
     request_example = {
         **PARITY_CASES["beijing_first_use_demo"],
@@ -321,10 +377,15 @@ def schema() -> dict[str, Any]:
             "external_service": True,
             "required_fields": [
                 "endpoint",
+                "endpoint_host",
                 "transport",
                 "provenance_mode",
                 "retry_policy",
                 "timeout_seconds",
+                "request_hash",
+                "response_hash",
+                "called_at",
+                "artifact_path",
             ],
         },
     }
@@ -382,8 +443,44 @@ def _external_technique_preview(
     }
 
 
-def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any]:
+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]:
     return {
+        "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 [],
+    }
+
+
+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",
@@ -391,15 +488,10 @@ def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any]
         "node_policy": payload.get("node_policy"),
         "body_list": payload.get("body_list"),
         "bodies": payload.get("bodies"),
-        "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 {}),
-        },
+        "source_metadata": metadata,
     }
+    result["source_metadata"]["artifact_path"] = _write_artifact(result)
+    return result
 
 
 def _normalize_external_technique_success(
@@ -445,6 +537,8 @@ 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 format: {"Status": "Pass", "Payload": [...]}
     if payload.get("Status") == "Pass":
@@ -461,6 +555,7 @@ def _normalize_range_scan_success(
     allowed_ids = allowlist.get("event_ids", set())
     allowed_tags = allowlist.get("tags", set())
 
+    original_event_count = len(events)
     evidence_ledger = []
     for index, event in enumerate(events, start=1):
         if not isinstance(event, dict):
@@ -508,7 +603,16 @@ def _normalize_range_scan_success(
             "tags": top.get("tags") or [],
         }
 
-    return {
+    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,
+        **(payload.get("source_metadata") or {}),
+    }
+    result = {
         "backend": "vedastro_service_adapter_candidate",
         "available": True,
         "status": "ok",
@@ -518,15 +622,10 @@ def _normalize_range_scan_success(
         "event_count": len(evidence_ledger),
         "top_event": top_event,
         "evidence_ledger": evidence_ledger,
-        "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 {}),
-        },
+        "source_metadata": metadata,
     }
+    result["source_metadata"]["artifact_path"] = _write_artifact(result)
+    return result
 
 
 def _source_metadata(endpoint: str) -> dict[str, Any]:
@@ -553,6 +652,35 @@ def _post_json(endpoint: str, request_preview: dict[str, Any]) -> dict[str, Any]
         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]]:
+    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:
+            payload = _post_json(endpoint, request_preview)
+            if not isinstance(payload, dict):
+                return {}, attempt, retry_error_codes
+            return payload, 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 run_case(case_id: str) -> dict[str, Any]:
     if case_id not in PARITY_CASES:
         return {
@@ -578,7 +706,7 @@ def run_case(case_id: str) -> dict[str, Any]:
             "source_metadata": _source_metadata(endpoint),
         }
     try:
-        payload = _post_json(endpoint, request_preview)
+        payload, attempt_count, retry_error_codes = _post_json_with_retry(endpoint, request_preview)
     except error.HTTPError as exc:
         return {
             "backend": "vedastro_service_adapter_candidate",
@@ -616,7 +744,7 @@ def run_case(case_id: str) -> dict[str, Any]:
             "source_metadata": _source_metadata(endpoint),
         }
 
-    return _normalize_success(payload, endpoint)
+    return _normalize_success(payload, endpoint, request_preview)
 
 
 def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> dict[str, Any]:
@@ -655,7 +783,7 @@ def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) ->
         }
 
     try:
-        payload = _post_json(endpoint, request_preview)
+        payload, attempt_count, retry_error_codes = _post_json_with_retry(endpoint, request_preview)
     except error.HTTPError as exc:
         return {
             "backend": "vedastro_service_adapter_candidate",
@@ -693,7 +821,7 @@ def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) ->
             "source_metadata": _source_metadata(endpoint),
         }
 
-    return _normalize_range_scan_success(payload, endpoint, request_preview)
+    return _normalize_range_scan_success(payload, endpoint, request_preview, attempt_count, retry_error_codes)
 
 
 def run_external_technique(case_id: str, domain: str, method: str, api_endpoint: str) -> dict[str, Any]:

tests/test_api_server_security.py

@@ -86,6 +86,28 @@ class _HealthCaptureHandler(JyotishAPIHandler):
         return json.loads(self.wfile.getvalue().decode('utf-8'))
 
 
+class _VedAstroStatusCaptureHandler(JyotishAPIHandler):
+    def __init__(self) -> None:
+        self.headers = _FakeHeaders()
+        self.server = _FakeServer()
+        self.path = '/api/vedastro/status'
+        self.wfile = BytesIO()
+        self.status_code = None
+        self.response_headers = []
+
+    def send_response(self, code, message=None):  # noqa: ANN001
+        self.status_code = code
+
+    def send_header(self, key, value):  # noqa: ANN001
+        self.response_headers.append((key, value))
+
+    def end_headers(self):
+        return None
+
+    def payload(self) -> dict:
+        return json.loads(self.wfile.getvalue().decode('utf-8'))
+
+
 def test_default_cors_origins_are_local_only() -> None:
     assert 'http://localhost:3456' in DEFAULT_ALLOWED_ORIGINS
     assert '*' not in DEFAULT_ALLOWED_ORIGINS
@@ -125,6 +147,25 @@ def test_health_endpoint_exposes_runtime_accuracy_metadata() -> None:
     assert 'swisseph_version' in payload
 
 
+def test_vedastro_status_endpoint_exposes_safe_adapter_state(monkeypatch) -> None:
+    monkeypatch.setenv('VEDASTRO_API_ENDPOINT', 'https://vedastro.example.test/secret/path')
+    monkeypatch.delenv('VEDASTRO_ENABLE_NETWORK', raising=False)
+    handler = _VedAstroStatusCaptureHandler()
+
+    handler.do_GET()
+
+    assert handler.status_code == 200
+    payload = handler.payload()
+    assert payload['adapter'] == 'vedastro_service_adapter'
+    assert payload['configured'] is True
+    assert payload['network_enabled'] is False
+    assert payload['status'] == 'network_execution_disabled'
+    assert payload['endpoint_host'] == 'vedastro.example.test'
+    assert 'secret/path' not in json.dumps(payload)
+    assert payload['required_env']['endpoint'] == 'VEDASTRO_API_ENDPOINT'
+    assert payload['live_profile'] == 'vedastro-live'
+
+
 @pytest.mark.parametrize(
     ('key', 'value', 'minimum', 'maximum'),
     [

tests/test_frontend_productization.py

@@ -891,13 +891,15 @@ def test_quality_gate_declares_fast_browser_release_profiles() -> None:
     readme = (ROOT / "README.md").read_text(encoding="utf-8")
     for token in [
         "--profile",
-        "choices=[\"quick\", \"browser\", \"release\", \"accuracy\"]",
+        "choices=[\"quick\", \"browser\", \"release\", \"accuracy\", \"vedastro-live\"]",
         "QUALITY_GATE_PROFILES",
         "quick",
         "browser",
         "release",
         "accuracy",
+        "vedastro-live",
         "skip_local_accuracy_report",
+        "skip_vedastro_live",
         "scripts/local_accuracy_report.py",
         "skip_slow",
         "skip_yoga_logic",
@@ -917,10 +919,12 @@ def test_quality_gate_declares_fast_browser_release_profiles() -> None:
         "browser：完整浏览器守门",
         "release：发布前守门",
         "accuracy：本地准确率守门",
+        "vedastro-live：外部 VedAstro 雷达守门",
         "python3 scripts/run_quality_gate.py --profile quick",
         "python3 scripts/run_quality_gate.py --profile browser",
         "python3 scripts/run_quality_gate.py --profile release",
         "python3 scripts/run_quality_gate.py --profile accuracy",
+        "python3 scripts/run_quality_gate.py --profile vedastro-live",
     ]:
         assert token in readme
 
@@ -956,6 +960,33 @@ def test_accuracy_quality_gate_runs_local_accuracy_report_without_frontend_click
     assert profile["skip_local_accuracy_report"] is False
 
 
+def test_vedastro_live_quality_gate_is_optional_and_network_gated() -> None:
+    quality_gate = load_quality_gate_module()
+    quality_gate_text = (ROOT / "scripts" / "run_quality_gate.py").read_text(encoding="utf-8")
+
+    profile = quality_gate.QUALITY_GATE_PROFILES["vedastro-live"]
+    assert profile["skip_frontend_click"] is True
+    assert profile["skip_frontend_runtime"] is True
+    assert profile["skip_vedastro_live"] is False
+    assert "VEDASTRO_API_ENDPOINT" in quality_gate_text
+    assert "VEDASTRO_ENABLE_NETWORK" in quality_gate_text
+    assert "scripts/vedastro_service_adapter.py" in quality_gate_text
+    assert '"vedastro-live"' in quality_gate_text
+
+
+def test_trust_center_surfaces_vedastro_adapter_status_without_endpoint_secret() -> None:
+    main = (ROOT / "jyotish-app" / "main.js").read_text(encoding="utf-8")
+    api_bridge = (ROOT / "jyotish-app" / "api-bridge.js").read_text(encoding="utf-8")
+
+    assert "renderVedAstroStatus" in main
+    assert "getVedAstroStatus" in main
+    assert "/api/vedastro/status" in api_bridge
+    assert "VedAstro 外部雷达" in main
+    assert "VEDASTRO_API_ENDPOINT" in main
+    assert "endpoint_host" in main
+    assert "secret/path" not in main
+
+
 def test_github_release_quality_gate_runs_browser_release_profile() -> None:
     workflow = (ROOT / ".github" / "workflows" / "release-quality-gate.yml").read_text(encoding="utf-8")
     for token in [

tests/test_life_event_graph_v1.py

@@ -144,3 +144,59 @@ def test_life_event_graph_is_returned_from_strict_relationship_evidence() -> Non
     assert strict["life_event_graph"]["route"] == "relationship"
     assert strict["life_event_graph"]["dominant_label"] == "legal_marriage"
     assert any(node["kind"] == "external_window" for node in strict["life_event_graph"]["event_nodes"])
+
+
+def test_strict_workflow_accepts_adapter_range_scan_result_without_manual_repackaging() -> None:
+    result = {
+        "modules": {
+            "varga_full": {"D9_Navamsa": {"summary": "ok"}},
+            "special_lagnas": {"Upapada_Lagna": {"sign": "Libra", "lord": "Venus"}},
+            "jaimini": {
+                "darakaraka": {"planet": "Venus", "house": 7},
+                "marriage_support": {"dk_7h_link": True},
+            },
+            "vivah_saham": {"sign": "Taurus", "house": 7},
+            "dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Moon"}},
+            "narayana_dasha": {"current_dasha": {"sign": "Libra", "lord": "Venus"}},
+            "dasa_convergence": {
+                "domain_activations": {
+                    "marriage_partnership": {"convergence_level": "L4", "probability": "70-85%"}
+                }
+            },
+            "vedastro_range_scan_result": {
+                "backend": "vedastro_service_adapter_candidate",
+                "available": True,
+                "status": "ok",
+                "operation": "range_scan",
+                "domain": "marriage",
+                "evidence_ledger": [
+                    {
+                        "source": "vedastro_service_adapter_candidate",
+                        "operation": "range_scan",
+                        "domain": "marriage",
+                        "event_id": "GocharJupiterIn7th",
+                        "signal_key": "gochar_jupiter_7th_marriage",
+                        "signal_label": "Jupiter in 7th marriage window",
+                        "signal_family": "marriage_trigger",
+                        "score": 72,
+                        "start": "2026-05-01",
+                        "end": "2026-06-01",
+                        "tags": ["marriage", "transit"],
+                    }
+                ],
+                "source_metadata": {
+                    "request_hash": "a" * 64,
+                    "response_hash": "b" * 64,
+                    "artifact_path": "scratch/local/vedastro_adapter/range.json",
+                },
+            },
+        }
+    }
+
+    strict = _collect_strict_evidence("relationship", result)
+
+    external = strict["present_evidence"]["external_activation"]
+    assert external["level"] == "moderate"
+    assert external["source"] == "vedastro_service_adapter_candidate"
+    assert external["provenance"]["request_hash"] == "a" * 64
+    assert any(node["kind"] == "external_window" for node in strict["life_event_graph"]["event_nodes"])

tests/test_vedastro_service_adapter_executor.py

@@ -348,6 +348,161 @@ def test_vedastro_service_adapter_can_normalize_mock_range_scan_response() -> No
     assert report["source_metadata"]["endpoint"].startswith("http://127.0.0.1:")
 
 
+def test_vedastro_range_scan_records_hashes_and_artifact_path() -> None:
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self) -> None:  # noqa: N802
+            response = {
+                "events": [
+                    {
+                        "id": "GocharJupiterIn7th",
+                        "name": "Jupiter enters 7th house",
+                        "start": "2026-05-01",
+                        "end": "2026-06-01",
+                        "score": 72,
+                        "tags": ["marriage", "transit"],
+                    }
+                ],
+                "source_metadata": {
+                    "service": "mock-vedastro",
+                    "version": "artifact-test",
+                },
+            }
+            body = json.dumps(response).encode("utf-8")
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, format: str, *args) -> None:  # noqa: A003
+            return
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        env = os.environ.copy()
+        env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro"
+        env["VEDASTRO_ENABLE_NETWORK"] = "1"
+        completed = subprocess.run(
+            [
+                sys.executable,
+                "scripts/vedastro_service_adapter.py",
+                "--range-scan",
+                "--domain",
+                "marriage",
+                "--case",
+                "beijing_first_use_demo",
+                "--start-date",
+                "2026-01-01",
+                "--end-date",
+                "2031-01-01",
+            ],
+            cwd=ROOT,
+            text=True,
+            capture_output=True,
+            timeout=120,
+            check=False,
+            env=env,
+        )
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+    assert completed.returncode == 0, completed.stderr or completed.stdout
+    report = json.loads(completed.stdout)
+    metadata = report["source_metadata"]
+    assert len(metadata["request_hash"]) == 64
+    assert len(metadata["response_hash"]) == 64
+    assert metadata["method"] == "POST"
+    assert metadata["operation"] == "range_scan"
+    assert metadata["vedastro_event_method"] == "SearchEvents"
+    assert metadata["allowlist_domain"] == "marriage"
+    assert metadata["allowlist_event_count"] == 1
+    assert metadata["filtered_event_count"] == 1
+    assert metadata["attempt_count"] == 1
+    artifact_path = ROOT / metadata["artifact_path"]
+    assert artifact_path.exists()
+    artifact = json.loads(artifact_path.read_text(encoding="utf-8"))
+    assert artifact["source_metadata"]["request_hash"] == metadata["request_hash"]
+    assert artifact["source_metadata"]["response_hash"] == metadata["response_hash"]
+    assert artifact["evidence_ledger"][0]["event_id"] == "GocharJupiterIn7th"
+
+
+def test_vedastro_range_scan_retries_transient_http_error() -> None:
+    class Handler(BaseHTTPRequestHandler):
+        attempts = 0
+
+        def do_POST(self) -> None:  # noqa: N802
+            Handler.attempts += 1
+            if Handler.attempts == 1:
+                self.send_response(503)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                return
+            response = {
+                "events": [
+                    {
+                        "id": "GocharJupiterIn7th",
+                        "name": "Jupiter enters 7th house",
+                        "start": "2026-05-01",
+                        "end": "2026-06-01",
+                        "score": 72,
+                        "tags": ["marriage", "transit"],
+                    }
+                ]
+            }
+            body = json.dumps(response).encode("utf-8")
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, format: str, *args) -> None:  # noqa: A003
+            return
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        env = os.environ.copy()
+        env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro"
+        env["VEDASTRO_ENABLE_NETWORK"] = "1"
+        env["VEDASTRO_RETRY_BACKOFF_SECONDS"] = "0"
+        completed = subprocess.run(
+            [
+                sys.executable,
+                "scripts/vedastro_service_adapter.py",
+                "--range-scan",
+                "--domain",
+                "marriage",
+                "--case",
+                "beijing_first_use_demo",
+                "--start-date",
+                "2026-01-01",
+                "--end-date",
+                "2031-01-01",
+            ],
+            cwd=ROOT,
+            text=True,
+            capture_output=True,
+            timeout=120,
+            check=False,
+            env=env,
+        )
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+    assert completed.returncode == 0, completed.stderr or completed.stdout
+    report = json.loads(completed.stdout)
+    assert report["status"] == "ok"
+    assert report["event_count"] == 1
+    assert report["source_metadata"]["attempt_count"] == 2
+    assert report["source_metadata"]["retry_error_codes"] == [503]
+
+
 def test_vedastro_service_adapter_applies_domain_allowlist_to_range_scan_noise() -> None:
     class Handler(BaseHTTPRequestHandler):
         def do_POST(self) -> None:  # noqa: N802