From cb5e978a41564341a1e835821b9987e8a88942ab Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Sun, 28 Jun 2026 13:25:10 +0800 Subject: [PATCH] Add VedAstro range scan adapter MVP --- ...astro_adapter_range_scan_mvp_2026_06_28.md | 39 +++ scripts/vedastro_service_adapter.py | 246 ++++++++++++++---- .../test_vedastro_service_adapter_executor.py | 123 +++++++++ 3 files changed, 358 insertions(+), 50 deletions(-) create mode 100644 docs/research/vedastro_adapter_range_scan_mvp_2026_06_28.md diff --git a/docs/research/vedastro_adapter_range_scan_mvp_2026_06_28.md b/docs/research/vedastro_adapter_range_scan_mvp_2026_06_28.md new file mode 100644 index 00000000..8c86fd42 --- /dev/null +++ b/docs/research/vedastro_adapter_range_scan_mvp_2026_06_28.md @@ -0,0 +1,39 @@ +# VedAstro Adapter Range Scan MVP - 2026-06-28 + +## Scope + +This pass adds the first executable VedAstro range-scan boundary. It is not a +full VedAstro API integration and does not replace local Jyotish adjudication. + +## Implemented + +- `scripts/vedastro_service_adapter.py` now exposes a `--range-scan` mode. +- Supported first-pass domains are `marriage`, `wealth`, and `career`. +- The adapter builds a provenance-preserving request preview when + `VEDASTRO_API_ENDPOINT` is configured but `VEDASTRO_ENABLE_NETWORK` is not + enabled. +- If network execution is explicitly enabled, the adapter can normalize a + VedAstro-like event payload into an `evidence_ledger`. + +## Boundary + +- VedAstro output is treated as external candidate evidence only. +- The adapter does not emit final marriage, wealth, or career verdicts. +- The local adjudicators remain responsible for promise, activation, + manifestation, confidence, and conflict handling. +- Network execution remains opt-in through `VEDASTRO_ENABLE_NETWORK=1`. + +## Verification + +- `python3 -m pytest tests/test_vedastro_service_adapter_executor.py -q` +- Result: `10 passed` + +## Next Work + +1. Map real VedAstro endpoint response fields into the current `evidence_ledger` + contract. +2. Add domain-specific event allowlists for marriage and wealth windows. +3. Feed the normalized range-scan ledger into the local relationship and finance + strict workflow as external activation evidence. +4. Only after the above is stable, expand from MVP boundary to full VedAstro API + radar integration. diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py index 56ee1b7b..0a3f81dc 100644 --- a/scripts/vedastro_service_adapter.py +++ b/scripts/vedastro_service_adapter.py @@ -59,6 +59,7 @@ PARITY_CASES = { }, } +SUPPORTED_RANGE_SCAN_DOMAINS = {"marriage", "wealth", "career"} DEFAULT_TIMEOUT_SECONDS = 8 TIMEOUT_ENV = "VEDASTRO_TIMEOUT_SECONDS" RETRY_POLICY = { @@ -117,6 +118,32 @@ def schema() -> dict[str, Any]: "bodies", "source_metadata", ], + "range_scan_request_contract": [ + "operation", + "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", + ], "request_example": request_example, "provenance_contract": { "external_service": True, @@ -155,6 +182,17 @@ def _request_preview(case: dict[str, Any]) -> dict[str, Any]: } +def _range_scan_preview(case: dict[str, Any], domain: str, start_date: str, end_date: str) -> dict[str, Any]: + return { + "operation": "range_scan", + "domain": domain, + "start_date": start_date, + "end_date": end_date, + "event_model": "vedastro_events_at_range_candidate", + **case, + } + + def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any]: return { "backend": "vedastro_service_adapter_candidate", @@ -175,6 +213,75 @@ def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any] } +def _normalize_range_scan_success( + payload: dict[str, Any], + endpoint: str, + request_preview: dict[str, Any], +) -> dict[str, Any]: + events = payload.get("events") + if not isinstance(events, list): + events = [] + + evidence_ledger = [] + for index, event in enumerate(events, start=1): + if not isinstance(event, dict): + continue + evidence_ledger.append( + { + "source": "vedastro_service_adapter_candidate", + "operation": "range_scan", + "domain": request_preview["domain"], + "event_id": event.get("id") or event.get("name") or f"event_{index}", + "start": event.get("start") or event.get("start_time") or event.get("start_date"), + "end": 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": event.get("tags") or [], + "raw": event, + } + ) + + return { + "backend": "vedastro_service_adapter_candidate", + "available": True, + "status": "ok", + "operation": "range_scan", + "domain": request_preview["domain"], + "request_preview": request_preview, + "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 {}), + }, + } + + +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 _post_json(endpoint: str, request_preview: dict[str, Any]) -> dict[str, Any] | str: + req = request.Request( + endpoint, + data=json.dumps(request_preview).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with request.urlopen(req, timeout=_timeout_seconds()) as resp: + raw = resp.read().decode("utf-8") + return json.loads(raw) + + def run_case(case_id: str) -> dict[str, Any]: if case_id not in PARITY_CASES: return { @@ -197,25 +304,10 @@ def run_case(case_id: str) -> dict[str, Any]: "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": _timeout_seconds(), - "retry_policy": RETRY_POLICY, - "network_execution_env": ALLOW_NETWORK_ENV, - }, + "source_metadata": _source_metadata(endpoint), } try: - req = request.Request( - endpoint, - data=json.dumps(request_preview).encode("utf-8"), - headers={"Content-Type": "application/json"}, - method="POST", - ) - with request.urlopen(req, timeout=_timeout_seconds()) as resp: - raw = resp.read().decode("utf-8") - payload = json.loads(raw) + payload = _post_json(endpoint, request_preview) except error.HTTPError as exc: return { "backend": "vedastro_service_adapter_candidate", @@ -223,14 +315,7 @@ def run_case(case_id: str) -> dict[str, Any]: "status": "http_error", "reason": f"VedAstro adapter HTTP error: {exc.code}", "request_preview": request_preview, - "source_metadata": { - "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, - }, + "source_metadata": _source_metadata(endpoint), } except error.URLError as exc: return { @@ -239,14 +324,7 @@ def run_case(case_id: str) -> dict[str, Any]: "status": "network_error", "reason": f"VedAstro adapter network error: {exc.reason}", "request_preview": request_preview, - "source_metadata": { - "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, - }, + "source_metadata": _source_metadata(endpoint), } except (TimeoutError, socket.timeout): return { @@ -255,14 +333,7 @@ def run_case(case_id: str) -> dict[str, Any]: "status": "timeout", "reason": "VedAstro adapter timed out", "request_preview": request_preview, - "source_metadata": { - "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, - }, + "source_metadata": _source_metadata(endpoint), } except json.JSONDecodeError: return { @@ -271,26 +342,101 @@ def run_case(case_id: str) -> dict[str, Any]: "status": "invalid_json", "reason": "VedAstro adapter received non-JSON response", "request_preview": request_preview, - "source_metadata": { - "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, - }, + "source_metadata": _source_metadata(endpoint), } return _normalize_success(payload, endpoint) +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}", + } + 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}", + } + + endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip() + if not endpoint: + return _unconfigured("VEDASTRO_API_ENDPOINT is not configured; range scan stops before network access.") + + request_preview = _range_scan_preview(PARITY_CASES[case_id], domain, start_date, end_date) + 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), + } + + 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 range scan 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 range scan 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 range scan 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 range scan received non-JSON response", + "request_preview": request_preview, + "source_metadata": _source_metadata(endpoint), + } + + return _normalize_range_scan_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("--domain", choices=sorted(SUPPORTED_RANGE_SCAN_DOMAINS), default="marriage") + parser.add_argument("--start-date", default="2026-01-01") + parser.add_argument("--end-date", default="2031-01-01") args = parser.parse_args() - result = schema() if args.print_schema else run_case(args.case) + if args.print_schema: + result = schema() + 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 diff --git a/tests/test_vedastro_service_adapter_executor.py b/tests/test_vedastro_service_adapter_executor.py index bc4f6873..0d3f2f23 100644 --- a/tests/test_vedastro_service_adapter_executor.py +++ b/tests/test_vedastro_service_adapter_executor.py @@ -36,6 +36,11 @@ def test_vedastro_service_adapter_executor_schema_is_declared() -> None: assert "request_example" in report assert report["provenance_contract"]["external_service"] is True assert "endpoint" in report["provenance_contract"]["required_fields"] + assert "range_scan_request_contract" in report + assert "domain" in report["range_scan_request_contract"] + assert "start_date" in report["range_scan_request_contract"] + assert "range_scan_response_contract" in report + assert "evidence_ledger" in report["range_scan_response_contract"] def test_vedastro_service_adapter_returns_controlled_unconfigured_status() -> None: @@ -101,6 +106,44 @@ def test_vedastro_service_adapter_builds_request_preview_before_real_network_use assert report["source_metadata"]["provenance_mode"] == "external_service_candidate" +def test_vedastro_service_adapter_builds_range_scan_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", + "--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, + ) + + 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"]["operation"] == "range_scan" + assert report["request_preview"]["domain"] == "marriage" + assert report["request_preview"]["start_date"] == "2026-01-01" + assert report["request_preview"]["end_date"] == "2031-01-01" + assert report["request_preview"]["event_model"] == "vedastro_events_at_range_candidate" + assert report["source_metadata"]["endpoint"] == "https://example.invalid/vedastro" + + def test_vedastro_service_adapter_can_normalize_mock_http_response() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 @@ -164,6 +207,86 @@ def test_vedastro_service_adapter_can_normalize_mock_http_response() -> None: assert report["source_metadata"]["endpoint"].startswith("http://127.0.0.1:") +def test_vedastro_service_adapter_can_normalize_mock_range_scan_response() -> None: + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(length).decode("utf-8")) + assert payload["operation"] == "range_scan" + assert payload["domain"] == "marriage" + assert payload["start_date"] == "2026-01-01" + response = { + "events": [ + { + "id": "jupiter_7h_window", + "name": "Jupiter supports marriage axis", + "start": "2026-05-01", + "end": "2026-06-01", + "score": 72, + "tags": ["marriage", "transit"], + } + ], + "source_metadata": { + "service": "mock-vedastro", + "version": "range-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) + assert report["backend"] == "vedastro_service_adapter_candidate" + assert report["available"] is True + assert report["status"] == "ok" + assert report["operation"] == "range_scan" + assert report["domain"] == "marriage" + assert report["evidence_ledger"][0]["event_id"] == "jupiter_7h_window" + assert report["evidence_ledger"][0]["domain"] == "marriage" + assert report["evidence_ledger"][0]["score"] == 72 + assert report["evidence_ledger"][0]["raw"]["name"] == "Jupiter supports marriage axis" + assert report["source_metadata"]["endpoint"].startswith("http://127.0.0.1:") + + def test_vedastro_service_adapter_classifies_http_error() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802