diff --git a/README.md b/README.md index 1bb14682..f4b2f1d4 100644 --- a/README.md +++ b/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` ### 真实案例复验与准确率边界 diff --git a/docs/superpowers/plans/2026-06-29-vedastro-adapter-mvp.md b/docs/superpowers/plans/2026-06-29-vedastro-adapter-mvp.md new file mode 100644 index 00000000..f4f6bca5 --- /dev/null +++ b/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. diff --git a/jyotish-app/api-bridge.js b/jyotish-app/api-bridge.js index c58d1320..2147ba36 100644 --- a/jyotish-app/api-bridge.js +++ b/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, diff --git a/jyotish-app/public/api-bridge.js b/jyotish-app/public/api-bridge.js index c58d1320..2147ba36 100644 --- a/jyotish-app/public/api-bridge.js +++ b/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, diff --git a/mcp_server.py b/mcp_server.py index e5ca731e..e6e8cbd0 100644 --- a/mcp_server.py +++ b/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, } diff --git a/scripts/run_quality_gate.py b/scripts/run_quality_gate.py index d39053de..26f40de6 100644 --- a/scripts/run_quality_gate.py +++ b/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 diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py index 61fe7986..aa0ab296 100644 --- a/scripts/vedastro_service_adapter.py +++ b/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]: diff --git a/tests/test_api_server_security.py b/tests/test_api_server_security.py index eea8e771..4cff2382 100644 --- a/tests/test_api_server_security.py +++ b/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'), [ diff --git a/tests/test_frontend_productization.py b/tests/test_frontend_productization.py index 72f5617d..567639d1 100644 --- a/tests/test_frontend_productization.py +++ b/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 [ diff --git a/tests/test_life_event_graph_v1.py b/tests/test_life_event_graph_v1.py index 38be9ddb..90a3ae50 100644 --- a/tests/test_life_event_graph_v1.py +++ b/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"]) diff --git a/tests/test_vedastro_service_adapter_executor.py b/tests/test_vedastro_service_adapter_executor.py index 71de675d..2e90c55a 100644 --- a/tests/test_vedastro_service_adapter_executor.py +++ b/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