diff --git a/docs/research/vedastro_external_technique_evidence_contract_2026_06_28.md b/docs/research/vedastro_external_technique_evidence_contract_2026_06_28.md new file mode 100644 index 00000000..1f2d883e --- /dev/null +++ b/docs/research/vedastro_external_technique_evidence_contract_2026_06_28.md @@ -0,0 +1,77 @@ +# VedAstro External Technique Evidence Contract - 2026-06-28 + +## Scope + +This pass adds a thin `external_technique_evidence` bridge for VedAstro +calculation methods / API endpoints. + +VedAstro's `596+` surface is treated here as a broad calculation-method and +astrological-event compute library. It is not modeled as `596` user-facing life +event categories. + +Examples of supported external method intent: + +- low-level astronomy and longitude methods +- dignity and strength methods +- Vargas and per-chart placement methods +- Dasha calculation methods +- Yoga / Gochar / FindLifeEvents-triggered calculation records + +## Boundary + +`external_technique_evidence` may enter: + +- `secondary_context` +- `technique_audit` + +It must not: + +- change strict workflow `score` +- set or lift `dominant_label` +- set or lift `payout_label` +- replace local Dasha, Varga, Shadbala, Ashtakavarga, Jaimini, or functional benefic/malefic judgement + +## Adapter Contract + +`scripts/vedastro_service_adapter.py` now exposes: + +```bash +python3 scripts/vedastro_service_adapter.py \ + --external-technique \ + --domain wealth \ + --method CalculateShadbala \ + --api-endpoint Calculate/Shadbala +``` + +When network execution is disabled, the adapter emits a request preview and +provenance metadata. When network execution is enabled, the normalized response +is wrapped into an `evidence_ledger` item with: + +- `source = vedastro_service_adapter_candidate` +- `operation = calculation_method` +- `role = external_technique_evidence` +- `domain` +- `method` +- `api_endpoint` + +## Strict Workflow Policy + +The local adjudicator remains final. VedAstro external method records are +filtered by source, operation, role, and domain before being accepted. + +Accepted evidence produces: + +- `present_evidence.external_technique_evidence.level = context_only` +- `event_judgement.secondary_context += ["external_technique_evidence"]` +- a top-level `technique_audit` row marking the evidence as `external_evidence_only` + +No scoring branch reads this evidence. + +## Verification + +Fresh checks: + +- `python3 -m pytest tests/test_vedastro_external_technique_evidence.py -q` +- `python3 -m pytest tests/test_vedastro_service_adapter_executor.py tests/test_vedastro_parity_matrix.py tests/test_vedastro_adapter_candidate_guard.py tests/test_vedastro_external_technique_evidence.py -q` +- `python3 -m pytest tests/test_mcp_strict_workflow_finance.py tests/test_mcp_strict_workflow_career.py tests/test_mcp_strict_workflow_relationship.py tests/test_life_event_graph_v1.py -q` + diff --git a/docs/research/vedastro_required_high_frequency_radar_contract_2026_06_28.md b/docs/research/vedastro_required_high_frequency_radar_contract_2026_06_28.md new file mode 100644 index 00000000..e48d0004 --- /dev/null +++ b/docs/research/vedastro_required_high_frequency_radar_contract_2026_06_28.md @@ -0,0 +1,119 @@ +# VedAstro Required High-Frequency Radar Contract - 2026-06-28 + +## Scope + +This pass turns VedAstro from an optional comparison note into a required +external timing-radar boundary for strict event workflows. + +The local engine still owns final adjudication. VedAstro is used as an external +high-frequency calculation/event source, especially for range timing. + +## Official Surface Interpreted + +Public VedAstro surfaces reviewed on 2026-06-28: + +- `https://vedastro.org/APIBuilder.html` + - General API Builder surface. + - Treated as the broad calculator layer, advertised by VedAstro as 600+ + calculators. +- `https://vedastro.org/EventsChartAPIBuilder.html` + - Events API Builder surface. + - Page metadata describes three event timing endpoints: + `SearchEvents`, `GetEventTiming`, `ListEventTypes`. + - Page metadata describes 400+ pre-defined events. + - The UI exposes `Scan precision (hours)`, confirming that the events surface + is a range-scan / high-frequency timing tool, not a fixed life-category + dictionary. +- `https://pypi.org/pypi/vedastro/json` + - Current observed Python package version: `1.23.25`. + - The local contract keeps the project-level claim as `596+` calculations + because that is the Python-library surface previously documented in this + project. + +## Adapter Contract + +`scripts/vedastro_service_adapter.py --print-schema` now exposes: + +```json +{ + "vedastro_calculation_coverage": { + "official_python_library_calculations": "596+", + "official_api_builder_calculators": "600+", + "official_events_builder_events": "400+", + "official_events_builder_methods": [ + "SearchEvents", + "GetEventTiming", + "ListEventTypes" + ], + "range_scan_role": "high_frequency_life_event_radar", + "intended_use": "external_timing_evidence_for_strict_workflow" + } +} +``` + +Range-scan request previews now include: + +```json +{ + "operation": "range_scan", + "vedastro_event_method": "SearchEvents" +} +``` + +The adapter remains configurable through `VEDASTRO_API_ENDPOINT` and gated by +`VEDASTRO_ENABLE_NETWORK`. If the endpoint or network flag is absent, it returns +a controlled preview / blocked status instead of fabricating external evidence. + +## Strict Workflow Contract + +For `career`, `relationship`, and `finance` strict workflows: + +- `present_evidence.external_activation` is required as the VedAstro timing + radar slot. +- If no VedAstro range-scan ledger is supplied, the slot becomes: + +```json +{ + "level": "missing_required_external_radar", + "source": "vedastro_service_adapter_candidate", + "required": true, + "operation": "range_scan" +} +``` + +- `event_judgement.secondary_context` receives + `vedastro_range_scan_missing`. +- `technique_audit` receives a blocked row: + +```json +{ + "technique": "VedAstro EventsAtRange / 596+ Calculator Radar", + "status": "blocked", + "role": "required_external_timing_radar", + "effect": "confidence_boundary_only_no_score_or_label_lift" +} +``` + +When valid VedAstro range-scan events are supplied, the audit row changes to +`status = used`, with `event_count`. + +## Boundary + +VedAstro range-scan evidence may support timing context, but it must not bypass +the local core gates: + +- It does not replace Vimshottari + Narayana dual dasha. +- It does not replace D9/UL, D2/D10, Shadbala, Ashtakavarga, Jaimini, or + Functional Benefic/Malefic layers. +- It does not directly set `dominant_label` or `payout_label`. +- It cannot bypass missing local promise layers; tests verify that a missing + `wealth_promise_strength` still caps the finance score even when VedAstro + range evidence is present. + +## Verification + +Fresh verification: + +- `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` +- `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` + diff --git a/mcp_server.py b/mcp_server.py index 6ce1122a..e5ca731e 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -417,10 +417,17 @@ def _derive_external_activation_support(modules: Dict[str, Any], domain: str) -> ledger = _safe_get(modules, "external_activation", "evidence_ledger") if not isinstance(ledger, list): return { - "level": "none", - "source": None, + "level": "missing_required_external_radar", + "source": "vedastro_service_adapter_candidate", "signals": [], "events": [], + "required": True, + "operation": "range_scan", + "external_calculation_coverage": "VedAstro 596+/600+ calculation nodes", + "reason": ( + "VedAstro EventsAtRange / FindLifeEvents-style high-frequency radar " + "was not provided; keep timing confidence bounded." + ), } events: List[Dict[str, Any]] = [] @@ -447,9 +454,102 @@ def _derive_external_activation_support(modules: Dict[str, Any], domain: str) -> "source": "vedastro_service_adapter_candidate" if events else None, "signals": ["vedastro_range_scan"] if events else [], "events": events, + "required": True, + "operation": "range_scan", + "external_calculation_coverage": "VedAstro 596+/600+ calculation nodes", } +def _external_activation_audit(external_activation: Any) -> List[Dict[str, Any]]: + if not isinstance(external_activation, dict): + return [] + if external_activation.get("level") == "missing_required_external_radar": + return [ + { + "technique": "VedAstro EventsAtRange / 596+ Calculator Radar", + "status": "blocked", + "role": "required_external_timing_radar", + "effect": "confidence_boundary_only_no_score_or_label_lift", + } + ] + if external_activation.get("source") == "vedastro_service_adapter_candidate": + events = external_activation.get("events") or [] + if isinstance(events, list) and events: + return [ + { + "technique": "VedAstro EventsAtRange / 596+ Calculator Radar", + "status": "used", + "role": "external_timing_evidence", + "event_count": len(events), + "effect": "activation_context_only_guarded_score_bump", + } + ] + return [] + + +def _derive_external_technique_evidence(modules: Dict[str, Any], domain: str) -> Dict[str, Any]: + ledger = _safe_get(modules, "external_technique_evidence", "evidence_ledger") + if not isinstance(ledger, list): + return { + "level": "none", + "source": None, + "signals": [], + "methods": [], + "evidence": [], + "policy": { + "can_change_score": False, + "can_set_dominant_label": False, + "can_set_payout_label": False, + }, + } + + evidence: List[Dict[str, Any]] = [] + methods: List[str] = [] + for item in ledger: + if not isinstance(item, dict): + continue + if item.get("source") != "vedastro_service_adapter_candidate": + continue + if item.get("operation") != "calculation_method": + continue + if item.get("role") != "external_technique_evidence": + continue + item_domain = item.get("domain") or "general" + if item_domain not in {domain, "general"}: + continue + evidence.append(item) + method = item.get("method") + if isinstance(method, str) and method and method not in methods: + methods.append(method) + + return { + "level": "context_only" if evidence else "none", + "source": "vedastro_service_adapter_candidate" if evidence else None, + "signals": ["vedastro_external_calculation_method"] if evidence else [], + "methods": methods, + "evidence": evidence, + "policy": { + "can_change_score": False, + "can_set_dominant_label": False, + "can_set_payout_label": False, + }, + } + + +def _external_technique_audit(external_technique: Any) -> List[Dict[str, Any]]: + if not isinstance(external_technique, dict) or external_technique.get("level") != "context_only": + return [] + return [ + { + "technique": "VedAstro External Technique Evidence", + "status": "used", + "role": "external_evidence_only", + "methods": external_technique.get("methods") or [], + "effect": "secondary_context_only_no_score_or_label_lift", + } + ] + + def _derive_functional_benefic_malefic(modules: Dict[str, Any]) -> Dict[str, Any]: chart = _safe_get(modules, "chart") ascendant = _safe_get(chart, "ascendant") if isinstance(chart, dict) else None @@ -507,10 +607,20 @@ def _derive_synastry_relationship_support(modules: Dict[str, Any]) -> Dict[str, return str(result).lower() if result is not None else None return str(value).lower() if value is not None else None + if _kuta_result(additional_kutas.get("Mahendra")) == "good": + signals.append("mahendra_support") + if _kuta_result(additional_kutas.get("StreeDeergha")) == "good": + signals.append("stree_deergha_support") vedha_good = _kuta_result(additional_kutas.get("Vedha")) == "good" bad_constellations_good = _kuta_result(additional_kutas.get("BadConstellations")) == "good" rajju = additional_kutas.get("Rajju") rajju_good = isinstance(rajju, dict) and rajju.get("result") == "good" + if vedha_good: + signals.append("vedha_clean") + if rajju_good: + signals.append("rajju_clean") + if bad_constellations_good: + signals.append("bad_constellations_clean") if vedha_good and bad_constellations_good and rajju_good: signals.append("kuta_exception_clean") @@ -945,6 +1055,11 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s secondary_context.append("virodhargala_obstruction") if external_activation.get("level") == "moderate": secondary_context.append("external_activation_support") + elif external_activation.get("level") == "missing_required_external_radar": + secondary_context.append("vedastro_range_scan_missing") + external_technique = present.get("external_technique_evidence") or {} + if external_technique.get("level") == "context_only": + secondary_context.append("external_technique_evidence") hard_gate_missing = any( key in missing for key in ( @@ -1018,8 +1133,32 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s secondary_context.append("jaimini_support") if present.get("upapada_lagna"): secondary_context.append("ul_support") - if synastry_support.get("level") in {"supportive", "moderate"}: + synastry_level = synastry_support.get("level") + if synastry_level in {"supportive", "moderate"}: secondary_context.append("synastry_support") + synastry_signals = synastry_support.get("signals") or [] + if synastry_level in {"supportive", "moderate"} and any( + signal in synastry_signals for signal in ("mahendra_support", "stree_deergha_support") + ): + secondary_context.append("synastry_compatibility_support") + if synastry_level in {"supportive", "moderate"} and any( + signal in synastry_signals for signal in ("vedha_clean", "rajju_clean", "bad_constellations_clean") + ): + secondary_context.append("synastry_protective_kuta_support") + if synastry_level in {"supportive", "moderate"} and "exception_mitigated_match" in synastry_signals: + secondary_context.append("synastry_exception_mitigated") + if synastry_level in {"supportive", "moderate"} and "kuta_exception_clean" in synastry_signals: + secondary_context.append("synastry_kuta_exception_clean") + if synastry_level in {"supportive", "moderate"} and "mahendra_support" in synastry_signals: + secondary_context.append("mahendra_support") + if synastry_level in {"supportive", "moderate"} and "stree_deergha_support" in synastry_signals: + secondary_context.append("stree_deergha_support") + if synastry_level in {"supportive", "moderate"} and "vedha_clean" in synastry_signals: + secondary_context.append("vedha_clean") + if synastry_level in {"supportive", "moderate"} and "rajju_clean" in synastry_signals: + secondary_context.append("rajju_clean") + if synastry_level in {"supportive", "moderate"} and "bad_constellations_clean" in synastry_signals: + secondary_context.append("bad_constellations_clean") shadbala_component_audit = present.get("shadbala_component_audit") or {} if shadbala_component_audit.get("status") in {"blocked", "incomplete"}: secondary_context.append("shadbala_component_gap") @@ -1033,6 +1172,11 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s external_activation = present.get("external_activation") or {} if external_activation.get("level") == "moderate": secondary_context.append("external_activation_support") + elif external_activation.get("level") == "missing_required_external_radar": + secondary_context.append("vedastro_range_scan_missing") + external_technique = present.get("external_technique_evidence") or {} + if external_technique.get("level") == "context_only": + secondary_context.append("external_technique_evidence") if dignity_guardrail.get("status") == "conflict": secondary_context.append("dignity_conflict") elif dignity_guardrail.get("score_delta") == 5: @@ -1056,6 +1200,13 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s and jaimini_support.get("level") == "moderate" ): dominant_label = "legal_marriage" + elif ( + not hard_gate_missing + and not label_support_present + and jaimini_support.get("level") == "moderate" + and external_activation.get("level") == "moderate" + ): + secondary_context.append("public_formalization_candidate") return { "event_family": "relationship", "score": score, @@ -1179,6 +1330,11 @@ def _derive_event_judgement(route: str, present: Dict[str, Any], missing: List[s external_activation = present.get("external_activation") or {} if external_activation.get("level") == "moderate": secondary_context.append("external_activation_support") + elif external_activation.get("level") == "missing_required_external_radar": + secondary_context.append("vedastro_range_scan_missing") + external_technique = present.get("external_technique_evidence") or {} + if external_technique.get("level") == "context_only": + secondary_context.append("external_technique_evidence") if dignity_guardrail.get("status") == "conflict": secondary_context.append("dignity_conflict") elif dignity_guardrail.get("score_delta") == 5: @@ -1235,6 +1391,10 @@ def _build_life_event_graph(route: str, strict: Dict[str, Any]) -> Dict[str, Any if isinstance(vim, dict): md = vim.get("mahadasha") ad = vim.get("antardasha") + if isinstance(md, dict): + md = md.get("lord") or md.get("mahadasha") + if isinstance(ad, dict): + ad = ad.get("lord") or ad.get("antardasha") label = "/".join([part for part in (md, ad) if part]) if label: nodes.append( @@ -1299,6 +1459,18 @@ def _build_life_event_graph(route: str, strict: Dict[str, Any]) -> Dict[str, Any } ) + for label in event_judgement.get("secondary_context") or []: + if not isinstance(label, str): + continue + if label.startswith("synastry_"): + nodes.append( + { + "kind": "context", + "label": label, + "source": "event_judgement.secondary_context", + } + ) + return { "version": "life_event_graph_v1", "route": route, @@ -1335,7 +1507,10 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present = { "d10_dasamsa": _safe_get(modules, "varga_full", "D10_Dasamsa"), "a10_karma_pada": _safe_get(modules, "special_lagnas", "A10_Karma_Pada"), - "amatyakaraka": _safe_get(modules, "jaimini", "karakas", "Amatyakaraka"), + "amatyakaraka": ( + _safe_get(modules, "jaimini", "chara_karaka_7", "karaka_table", "Amatyakaraka") + or _safe_get(modules, "jaimini", "karakas", "Amatyakaraka") + ), "karakamsha": _safe_get(modules, "jaimini", "karakamsha"), "vimshottari_current": _safe_get(modules, "dasha", "current_dasha"), "narayana_current": _safe_get(modules, "narayana_dasha", "current_dasha"), @@ -1346,9 +1521,10 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present["kakshya_career_support"] = _derive_kakshya_career_support(_safe_get(modules, "kakshya")) present["argala_support"] = _derive_argala_support(modules, 10) present["external_activation"] = _derive_external_activation_support(modules, "career") + present["external_technique_evidence"] = _derive_external_technique_evidence(modules, "career") present["functional_benefic_malefic"] = _derive_functional_benefic_malefic(modules) missing = [key for key, value in present.items() if key not in { - "external_activation", "argala_support", "shadbala", "shadbala_component_audit", "kakshya_career_support", "functional_benefic_malefic" + "external_activation", "external_technique_evidence", "argala_support", "shadbala", "shadbala_component_audit", "kakshya_career_support", "functional_benefic_malefic" } and value in (None, {}, [], "")] convergence = present["career_convergence"] or {} confidence_cap = "medium" @@ -1363,7 +1539,7 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An else: confidence_cap = "medium-low" event_judgement = _derive_event_judgement(route, present, missing) - return _with_life_event_graph(route, { + strict = { "question_type": route, "required_evidence": required, "present_evidence": present, @@ -1375,7 +1551,14 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An "Career timing requires D10 + A10/Karma Pada + AmK/Karakamsha " "plus dual dasha and career convergence support." ), - }) + } + audit = ( + _external_activation_audit(present.get("external_activation")) + + _external_technique_audit(present.get("external_technique_evidence")) + ) + if audit: + strict["technique_audit"] = audit + return _with_life_event_graph(route, strict) if route == "relationship": required = [ @@ -1404,11 +1587,12 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An present["synastry_relationship_support"] = _derive_synastry_relationship_support(modules) present["argala_support"] = _derive_argala_support(modules, 7) present["external_activation"] = _derive_external_activation_support(modules, "marriage") + present["external_technique_evidence"] = _derive_external_technique_evidence(modules, "marriage") present["dignity_guardrail"] = _derive_dignity_guardrail(route, present) present["functional_benefic_malefic"] = _derive_functional_benefic_malefic(modules) missing = [ key for key, value in present.items() - if key not in {"chart", "external_activation", "dignity_guardrail", "jaimini_marriage_support", "jaimini_timing_support", "synastry_relationship_support", "argala_support", "shadbala", "shadbala_component_audit", "functional_benefic_malefic"} + if key not in {"chart", "external_activation", "external_technique_evidence", "dignity_guardrail", "jaimini_marriage_support", "jaimini_timing_support", "synastry_relationship_support", "argala_support", "shadbala", "shadbala_component_audit", "functional_benefic_malefic"} and value in (None, {}, [], "") ] convergence = present["marriage_convergence"] or {} @@ -1426,7 +1610,7 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An else: confidence_cap = "medium-low" event_judgement = _derive_event_judgement(route, present, missing) - return _with_life_event_graph(route, { + strict = { "question_type": route, "required_evidence": required, "present_evidence": present, @@ -1438,7 +1622,14 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An "Marriage timing requires D9 + UL + DK + dual dasha + Vivah Saham " "and convergence support; missing links cap confidence." ), - }) + } + audit = ( + _external_activation_audit(present.get("external_activation")) + + _external_technique_audit(present.get("external_technique_evidence")) + ) + if audit: + strict["technique_audit"] = audit + return _with_life_event_graph(route, strict) if route == "finance": avayogi_risk = _check_external_avayogi_risk(result) @@ -1477,10 +1668,11 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An ) present["kakshya_finance_support"] = _derive_kakshya_finance_support(_safe_get(modules, "kakshya")) present["external_activation"] = _derive_external_activation_support(modules, "wealth") + present["external_technique_evidence"] = _derive_external_technique_evidence(modules, "wealth") present["dignity_guardrail"] = _derive_dignity_guardrail(route, present) present["functional_benefic_malefic"] = _derive_functional_benefic_malefic(modules) missing = [key for key, value in present.items() if key not in { - "chart", "external_activation", "dignity_guardrail", "gains_convergence", "career_convergence", "avayogi_risk", "ashtakavarga_finance_support", "shadbala_component_audit", "asc_sign", "pav_finance_support", "sodhita_finance_support", "kakshya_finance_support", "functional_benefic_malefic" + "chart", "external_activation", "external_technique_evidence", "dignity_guardrail", "gains_convergence", "career_convergence", "avayogi_risk", "ashtakavarga_finance_support", "shadbala_component_audit", "asc_sign", "pav_finance_support", "sodhita_finance_support", "kakshya_finance_support", "functional_benefic_malefic" } and value in (None, {}, [], "")] convergence_hits: List[Dict[str, Any]] = [ item for item in [ @@ -1507,7 +1699,7 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An promise = present.get("wealth_promise_strength") or {} if "yogi" in promise.get("supporting_sources", []) and event_judgement.get("dominant_label") and "yogi_active" not in event_judgement.get("secondary_context", []): event_judgement["secondary_context"] = event_judgement.get("secondary_context", []) + ["yogi_active"] - return _with_life_event_graph(route, { + strict = { "question_type": route, "required_evidence": required, "present_evidence": present, @@ -1519,7 +1711,14 @@ def _collect_strict_evidence(route: str, result: Dict[str, Any]) -> Dict[str, An "Finance timing requires D2/D10 + strength + SAV + dual dasha " "plus at least one wealth-related convergence domain." ), - }) + } + audit = ( + _external_activation_audit(present.get("external_activation")) + + _external_technique_audit(present.get("external_technique_evidence")) + ) + if audit: + strict["technique_audit"] = audit + return _with_life_event_graph(route, strict) return _with_life_event_graph(route, { "question_type": route, diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py index 291142df..2bebb8df 100644 --- a/scripts/vedastro_service_adapter.py +++ b/scripts/vedastro_service_adapter.py @@ -21,6 +21,18 @@ ROOT = Path(__file__).resolve().parents[1] PARITY_CASES = { + "user_REDACTED_YEAR_test": { + "year": REDACTED_YEAR, + "month": 4, + "day": 17, + "hour": 14, + "minute": 49, + "lat": 36.42, + "lon": 114.2, + "tz": 8.0, + "ayanamsa_policy": "lahiri", + "node_policy": "mean", + }, "beijing_first_use_demo": { "year": 1990, "month": 1, @@ -60,6 +72,24 @@ PARITY_CASES = { } SUPPORTED_RANGE_SCAN_DOMAINS = {"marriage", "wealth", "career"} +SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS = {"marriage", "wealth", "career", "general"} +VEDASTRO_CALCULATION_COVERAGE = { + "official_python_library_calculations": "596+", + "official_api_builder_calculators": "600+", + "official_events_builder_events": "400+", + "official_events_builder_methods": ["SearchEvents", "GetEventTiming", "ListEventTypes"], + "range_scan_role": "high_frequency_life_event_radar", + "intended_use": "external_timing_evidence_for_strict_workflow", +} +EXTERNAL_TECHNIQUE_ROLE = "external_technique_evidence" +EXTERNAL_TECHNIQUE_OPERATION = "calculation_method" +EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY = { + "role": EXTERNAL_TECHNIQUE_ROLE, + "can_change_score": False, + "can_set_dominant_label": False, + "can_set_payout_label": False, + "allowed_destinations": ["secondary_context", "technique_audit"], +} RANGE_SCAN_EVENT_ALLOWLIST = { "marriage": { "event_ids": { @@ -163,7 +193,7 @@ RANGE_SCAN_SIGNAL_METADATA = { }, }, } -DEFAULT_TIMEOUT_SECONDS = 8 +DEFAULT_TIMEOUT_SECONDS = 120 TIMEOUT_ENV = "VEDASTRO_TIMEOUT_SECONDS" RETRY_POLICY = { "max_attempts": 2, @@ -230,6 +260,7 @@ def schema() -> dict[str, Any]: ], "range_scan_request_contract": [ "operation", + "vedastro_event_method", "domain", "start_date", "end_date", @@ -254,6 +285,36 @@ def schema() -> dict[str, Any]: "evidence_ledger", "source_metadata", ], + "vedastro_calculation_coverage": VEDASTRO_CALCULATION_COVERAGE, + "external_technique_request_contract": [ + "operation", + "role", + "domain", + "method", + "api_endpoint", + "year", + "month", + "day", + "hour", + "minute", + "lat", + "lon", + "tz", + "ayanamsa_policy", + "node_policy", + ], + "external_technique_response_contract": [ + "backend", + "available", + "status", + "operation", + "role", + "domain", + "evidence_ledger", + "adjudicator_policy", + "source_metadata", + ], + "external_technique_adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, "range_scan_event_allowlist": range_scan_allowlist, "request_example": request_example, "provenance_contract": { @@ -296,6 +357,7 @@ 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", + "vedastro_event_method": "SearchEvents", "domain": domain, "start_date": start_date, "end_date": end_date, @@ -304,6 +366,22 @@ def _range_scan_preview(case: dict[str, Any], domain: str, start_date: str, end_ } +def _external_technique_preview( + case: dict[str, Any], + domain: str, + method: str, + api_endpoint: str, +) -> dict[str, Any]: + return { + "operation": EXTERNAL_TECHNIQUE_OPERATION, + "role": EXTERNAL_TECHNIQUE_ROLE, + "domain": domain, + "method": method, + "api_endpoint": api_endpoint, + **case, + } + + def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any]: return { "backend": "vedastro_service_adapter_candidate", @@ -324,16 +402,61 @@ def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any] } +def _normalize_external_technique_success( + payload: dict[str, Any], + endpoint: str, + request_preview: dict[str, Any], +) -> dict[str, Any]: + evidence = { + "source": "vedastro_service_adapter_candidate", + "operation": EXTERNAL_TECHNIQUE_OPERATION, + "role": EXTERNAL_TECHNIQUE_ROLE, + "domain": request_preview["domain"], + "method": request_preview["method"], + "api_endpoint": request_preview["api_endpoint"], + "status": payload.get("status") or "ok", + "summary": payload.get("summary"), + "nature": payload.get("nature"), + "tags": payload.get("tags") if isinstance(payload.get("tags"), list) else [], + "raw": payload, + } + return { + "backend": "vedastro_service_adapter_candidate", + "available": True, + "status": "ok", + "operation": EXTERNAL_TECHNIQUE_OPERATION, + "role": EXTERNAL_TECHNIQUE_ROLE, + "domain": request_preview["domain"], + "request_preview": request_preview, + "evidence_ledger": [evidence], + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + "source_metadata": { + "transport": "http_json_service_boundary", + "endpoint": endpoint, + "provenance_mode": "external_service_candidate", + "timeout_seconds": _timeout_seconds(), + "retry_policy": RETRY_POLICY, + **(payload.get("source_metadata") or {}), + }, + } + + def _normalize_range_scan_success( payload: dict[str, Any], endpoint: str, request_preview: dict[str, Any], ) -> dict[str, Any]: - events = payload.get("events") + # Handle actual VedAstro response format: {"Status": "Pass", "Payload": [...]} + if payload.get("Status") == "Pass": + events = payload.get("Payload", []) + else: + # Fallback to local stub format if not VedAstro format + events = payload.get("events", []) + if not isinstance(events, list): events = [] - domain = request_preview["domain"] + domain = request_preview.get("domain", "") allowlist = RANGE_SCAN_EVENT_ALLOWLIST.get(domain, {}) allowed_ids = allowlist.get("event_ids", set()) allowed_tags = allowlist.get("tags", set()) @@ -342,8 +465,9 @@ def _normalize_range_scan_success( for index, event in enumerate(events, start=1): if not isinstance(event, dict): continue - event_id = event.get("id") or event.get("name") or f"event_{index}" - tags = event.get("tags") or [] + # VedAstro uses "Name" for event id, and "EventTags" for tags + event_id = event.get("Name") or event.get("id") or event.get("name") or f"event_{index}" + tags = event.get("EventTags") or event.get("tags") or [] if not isinstance(tags, list): tags = [] tag_set = {str(tag) for tag in tags} @@ -359,8 +483,8 @@ def _normalize_range_scan_success( "signal_key": signal_metadata.get("signal_key"), "signal_label": signal_metadata.get("signal_label") or event.get("name") or event_id, "signal_family": signal_metadata.get("signal_family"), - "start": event.get("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"), + "start": event.get("StartTime") or event.get("start") or event.get("start_time") or event.get("start_date"), + "end": event.get("EndTime") or event.get("end") or event.get("end_time") or event.get("end_date"), "score": event.get("score") if event.get("score") is not None else event.get("strength"), "tags": tags, "raw": event, @@ -417,9 +541,65 @@ def _source_metadata(endpoint: str) -> dict[str, Any]: def _post_json(endpoint: str, request_preview: dict[str, Any]) -> dict[str, Any] | str: + # If this is a range scan, format it for VedAstro's EventsAtRange API + if request_preview.get("operation") == "range_scan": + # VedAstro EventsAtRange schema requires: birthTime, startTime, endTime, eventTagList + from datetime import datetime + + # Parse dates + start_dt = datetime.strptime(request_preview["start_date"], "%Y-%m-%d") + end_dt = datetime.strptime(request_preview["end_date"], "%Y-%m-%d") + + # Extract location and timezone + tz = f"+0{int(request_preview['tz'])}:00" if request_preview['tz'] > 0 else f"-0{abs(int(request_preview['tz']))}:00" + location = { + "name": "AutoLocation", + "lat": request_preview["lat"], + "lng": request_preview["lon"] + } + + vedastro_payload = { + "birthTime": { + "year": request_preview["year"], + "month": request_preview["month"], + "date": request_preview["day"], + "hour": request_preview["hour"], + "minute": request_preview["minute"], + "location": location, + "timeOffset": tz + }, + "startTime": { + "year": start_dt.year, + "month": start_dt.month, + "date": start_dt.day, + "hour": 0, + "minute": 0, + "location": location, + "timeOffset": tz + }, + "endTime": { + "year": end_dt.year, + "month": end_dt.month, + "date": end_dt.day, + "hour": 0, + "minute": 0, + "location": location, + "timeOffset": tz + }, + "eventTagList": ["Gochara"] # Can map based on domain in future + } + # Append /EventsAtRange if the endpoint is the base Calculate API + if endpoint.endswith("Calculate"): + endpoint = f"{endpoint}/EventsAtRange" + elif not endpoint.endswith("EventsAtRange"): + endpoint = f"{endpoint}/api/Calculate/EventsAtRange" + + else: + vedastro_payload = request_preview + req = request.Request( endpoint, - data=json.dumps(request_preview).encode("utf-8"), + data=json.dumps(vedastro_payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) @@ -510,11 +690,15 @@ def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> "reason": f"Unsupported range scan domain: {domain}", } + request_preview = _range_scan_preview(PARITY_CASES[case_id], domain, start_date, end_date) 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.") + result = _unconfigured("VEDASTRO_API_ENDPOINT is not configured; range scan stops before network access.") + result["operation"] = "range_scan" + result["domain"] = domain + result["request_preview"] = request_preview + return result - 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", @@ -567,18 +751,120 @@ def run_range_scan(case_id: str, domain: str, start_date: str, end_date: str) -> return _normalize_range_scan_success(payload, endpoint, request_preview) +def run_external_technique(case_id: str, domain: str, method: str, api_endpoint: str) -> dict[str, Any]: + if case_id not in PARITY_CASES: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "unknown_case_id", + "reason": f"Unknown parity case: {case_id}", + } + if domain not in SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "unsupported_external_technique_domain", + "reason": f"Unsupported external technique domain: {domain}", + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + } + if not method.strip() or not api_endpoint.strip(): + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "missing_external_technique_method", + "reason": "Both method and api_endpoint are required for external technique evidence.", + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + } + + endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip() + if not endpoint: + result = _unconfigured( + "VEDASTRO_API_ENDPOINT is not configured; external technique evidence stops before network access." + ) + result["operation"] = EXTERNAL_TECHNIQUE_OPERATION + result["role"] = EXTERNAL_TECHNIQUE_ROLE + result["domain"] = domain + result["adjudicator_policy"] = EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY + return result + + request_preview = _external_technique_preview(PARITY_CASES[case_id], domain, method, api_endpoint) + if os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() not in {"1", "true", "yes"}: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "network_execution_disabled", + "reason": ( + f"{ALLOW_NETWORK_ENV} is not enabled; external technique evidence stops " + "after building request/provenance metadata." + ), + "request_preview": request_preview, + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + "source_metadata": _source_metadata(endpoint), + } + + try: + payload = _post_json(endpoint, request_preview) + except error.HTTPError as exc: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "http_error", + "reason": f"VedAstro external technique HTTP error: {exc.code}", + "request_preview": request_preview, + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + "source_metadata": _source_metadata(endpoint), + } + except error.URLError as exc: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "network_error", + "reason": f"VedAstro external technique network error: {exc.reason}", + "request_preview": request_preview, + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + "source_metadata": _source_metadata(endpoint), + } + except (TimeoutError, socket.timeout): + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "timeout", + "reason": "VedAstro external technique timed out", + "request_preview": request_preview, + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + "source_metadata": _source_metadata(endpoint), + } + except json.JSONDecodeError: + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "invalid_json", + "reason": "VedAstro external technique received non-JSON response", + "request_preview": request_preview, + "adjudicator_policy": EXTERNAL_TECHNIQUE_ADJUDICATOR_POLICY, + "source_metadata": _source_metadata(endpoint), + } + + return _normalize_external_technique_success(payload, endpoint, request_preview) + + def main() -> int: parser = argparse.ArgumentParser(description="VedAstro service adapter skeleton") parser.add_argument("--print-schema", action="store_true") parser.add_argument("--case", default="beijing_first_use_demo") parser.add_argument("--range-scan", action="store_true") - parser.add_argument("--domain", choices=sorted(SUPPORTED_RANGE_SCAN_DOMAINS), default="marriage") + parser.add_argument("--domain", choices=sorted(SUPPORTED_EXTERNAL_TECHNIQUE_DOMAINS), default="marriage") parser.add_argument("--start-date", default="2026-01-01") parser.add_argument("--end-date", default="2031-01-01") + parser.add_argument("--external-technique", action="store_true") + parser.add_argument("--method", default="") + parser.add_argument("--api-endpoint", default="") args = parser.parse_args() if args.print_schema: result = schema() + elif args.external_technique: + result = run_external_technique(args.case, args.domain, args.method, args.api_endpoint) elif args.range_scan: result = run_range_scan(args.case, args.domain, args.start_date, args.end_date) else: diff --git a/tests/test_mcp_strict_workflow_career.py b/tests/test_mcp_strict_workflow_career.py index ee9cbbdf..f59b4542 100644 --- a/tests/test_mcp_strict_workflow_career.py +++ b/tests/test_mcp_strict_workflow_career.py @@ -56,6 +56,7 @@ def test_career_collects_a10_amk_karakamsha_as_strict_evidence() -> None: "amk_active", "karakamsha_context", "argala_support", + "vedastro_range_scan_missing", ] @@ -73,6 +74,7 @@ def test_career_blocks_label_when_d10_is_missing_but_preserves_jaimini_context() "amk_active", "karakamsha_context", "argala_support", + "vedastro_range_scan_missing", ] diff --git a/tests/test_vedastro_external_technique_evidence.py b/tests/test_vedastro_external_technique_evidence.py new file mode 100644 index 00000000..59b836fb --- /dev/null +++ b/tests/test_vedastro_external_technique_evidence.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""VedAstro external calculation-method evidence boundary tests.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from copy import deepcopy +from pathlib import Path + +from mcp_server import _collect_strict_evidence + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_vedastro_adapter_declares_external_technique_evidence_policy() -> None: + completed = subprocess.run( + [sys.executable, "scripts/vedastro_service_adapter.py", "--print-schema"], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + schema = json.loads(completed.stdout) + + assert "external_technique_request_contract" in schema + assert "method" in schema["external_technique_request_contract"] + assert "api_endpoint" in schema["external_technique_request_contract"] + assert schema["external_technique_adjudicator_policy"] == { + "role": "external_technique_evidence", + "can_change_score": False, + "can_set_dominant_label": False, + "can_set_payout_label": False, + "allowed_destinations": ["secondary_context", "technique_audit"], + } + + +def test_vedastro_adapter_builds_external_technique_preview_without_network() -> None: + env = os.environ.copy() + env["VEDASTRO_API_ENDPOINT"] = "https://example.invalid/vedastro" + env.pop("VEDASTRO_ENABLE_NETWORK", None) + + completed = subprocess.run( + [ + sys.executable, + "scripts/vedastro_service_adapter.py", + "--external-technique", + "--domain", + "wealth", + "--method", + "CalculateShadbala", + "--api-endpoint", + "Calculate/Shadbala", + "--case", + "beijing_first_use_demo", + ], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + env=env, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + + assert report["backend"] == "vedastro_service_adapter_candidate" + assert report["status"] == "network_execution_disabled" + assert report["request_preview"]["operation"] == "calculation_method" + assert report["request_preview"]["role"] == "external_technique_evidence" + assert report["request_preview"]["domain"] == "wealth" + assert report["request_preview"]["method"] == "CalculateShadbala" + assert report["request_preview"]["api_endpoint"] == "Calculate/Shadbala" + assert report["adjudicator_policy"]["can_change_score"] is False + assert report["adjudicator_policy"]["can_set_dominant_label"] is False + assert report["adjudicator_policy"]["can_set_payout_label"] is False + + +def _finance_modules() -> dict: + return { + "varga_full": { + "D2_Hora": {"available": True}, + "D10_Dasamsa": {"available": True}, + }, + "shadbala": { + "planets": { + "Jupiter": { + "sthana_bala": 1, + "dig_bala": 1, + "kala_bala": 1, + "chesta_bala": 1, + "naisargika_bala": 1, + "drik_bala": 1, + } + } + }, + "ashtakavarga": { + "house_scores": { + "2": 28, + "11": 29, + } + }, + "dasha": { + "current_dasha": { + "mahadasha": "Saturn", + "antardasha": "Mercury", + } + }, + "narayana_dasha": { + "current_dasha": { + "sign": "Gemini", + "lord": "Mercury", + } + }, + "dasa_convergence": { + "domain_activations": { + "wealth_family": { + "convergence_level": "L1", + "probability": "+15-20%", + } + } + }, + "chart": { + "ascendant": { + "sign": "Leo", + } + }, + } + + +def test_strict_workflow_uses_external_technique_as_context_only() -> None: + base_result = {"modules": _finance_modules()} + with_external = {"modules": deepcopy(base_result["modules"])} + with_external["modules"]["external_technique_evidence"] = { + "evidence_ledger": [ + { + "source": "vedastro_service_adapter_candidate", + "operation": "calculation_method", + "role": "external_technique_evidence", + "domain": "wealth", + "method": "CalculateShadbala", + "api_endpoint": "Calculate/Shadbala", + "status": "ok", + "summary": "VedAstro Shadbala node returned as external evidence.", + } + ] + } + + base = _collect_strict_evidence("finance", base_result) + strict = _collect_strict_evidence("finance", with_external) + + base_judgement = base["event_judgement"] + judgement = strict["event_judgement"] + + assert strict["present_evidence"]["external_technique_evidence"]["level"] == "context_only" + assert strict["present_evidence"]["external_technique_evidence"]["methods"] == ["CalculateShadbala"] + assert "external_technique_evidence" in judgement["secondary_context"] + assert { + "technique": "VedAstro EventsAtRange / 596+ Calculator Radar", + "status": "blocked", + "role": "required_external_timing_radar", + "effect": "confidence_boundary_only_no_score_or_label_lift", + } in strict["technique_audit"] + assert { + "technique": "VedAstro External Technique Evidence", + "status": "used", + "role": "external_evidence_only", + "methods": ["CalculateShadbala"], + "effect": "secondary_context_only_no_score_or_label_lift", + } in strict["technique_audit"] + assert judgement["score"] == base_judgement["score"] + assert judgement["dominant_label"] == base_judgement["dominant_label"] + assert judgement["payout_label"] == base_judgement["payout_label"] + + +def test_strict_workflow_requires_vedastro_range_scan_boundary_for_timing_routes() -> None: + strict = _collect_strict_evidence("finance", {"modules": _finance_modules()}) + + assert strict["present_evidence"]["external_activation"]["level"] == "missing_required_external_radar" + assert strict["present_evidence"]["external_activation"]["source"] == "vedastro_service_adapter_candidate" + assert strict["present_evidence"]["external_activation"]["required"] is True + assert "vedastro_range_scan_missing" in strict["event_judgement"]["secondary_context"] + assert { + "technique": "VedAstro EventsAtRange / 596+ Calculator Radar", + "status": "blocked", + "role": "required_external_timing_radar", + "effect": "confidence_boundary_only_no_score_or_label_lift", + } in strict["technique_audit"] + + +def test_strict_workflow_marks_vedastro_range_scan_used_without_score_or_label_override() -> None: + base_result = {"modules": _finance_modules()} + with_range_scan = {"modules": deepcopy(base_result["modules"])} + with_range_scan["modules"]["external_activation"] = { + "evidence_ledger": [ + { + "source": "vedastro_service_adapter_candidate", + "operation": "range_scan", + "domain": "wealth", + "event_id": "GocharJupiterIn11th", + "signal_label": "Jupiter in 11th gains window", + "score": 76, + "tags": ["wealth", "transit"], + } + ] + } + + base = _collect_strict_evidence("finance", base_result) + strict = _collect_strict_evidence("finance", with_range_scan) + + assert strict["present_evidence"]["external_activation"]["level"] == "moderate" + assert "external_activation_support" in strict["event_judgement"]["secondary_context"] + assert "vedastro_range_scan_missing" not in strict["event_judgement"]["secondary_context"] + assert any( + row["technique"] == "VedAstro EventsAtRange / 596+ Calculator Radar" + and row["status"] == "used" + and row["event_count"] == 1 + for row in strict["technique_audit"] + ) + assert strict["missing_evidence"] == ["wealth_promise_strength"] + assert strict["event_judgement"]["score"] == base["event_judgement"]["score"] + assert strict["event_judgement"]["dominant_label"] == base["event_judgement"]["dominant_label"] + assert strict["event_judgement"]["payout_label"] == base["event_judgement"]["payout_label"] diff --git a/tests/test_vedastro_service_adapter_executor.py b/tests/test_vedastro_service_adapter_executor.py index 16f5c0d5..71de675d 100644 --- a/tests/test_vedastro_service_adapter_executor.py +++ b/tests/test_vedastro_service_adapter_executor.py @@ -38,7 +38,16 @@ def test_vedastro_service_adapter_executor_schema_is_declared() -> None: assert "endpoint" in report["provenance_contract"]["required_fields"] assert "range_scan_request_contract" in report assert "domain" in report["range_scan_request_contract"] + assert "vedastro_event_method" in report["range_scan_request_contract"] assert "start_date" in report["range_scan_request_contract"] + assert report["vedastro_calculation_coverage"] == { + "official_python_library_calculations": "596+", + "official_api_builder_calculators": "600+", + "official_events_builder_events": "400+", + "official_events_builder_methods": ["SearchEvents", "GetEventTiming", "ListEventTypes"], + "range_scan_role": "high_frequency_life_event_radar", + "intended_use": "external_timing_evidence_for_strict_workflow", + } assert "range_scan_response_contract" in report assert "evidence_ledger" in report["range_scan_response_contract"] assert report["range_scan_event_allowlist"]["marriage"]["event_ids"] @@ -139,6 +148,7 @@ def test_vedastro_service_adapter_builds_range_scan_preview_before_real_network_ assert report["backend"] == "vedastro_service_adapter_candidate" assert report["status"] == "network_execution_disabled" assert report["request_preview"]["operation"] == "range_scan" + assert report["request_preview"]["vedastro_event_method"] == "SearchEvents" assert report["request_preview"]["domain"] == "marriage" assert report["request_preview"]["start_date"] == "2026-01-01" assert report["request_preview"]["end_date"] == "2031-01-01" @@ -146,6 +156,41 @@ def test_vedastro_service_adapter_builds_range_scan_preview_before_real_network_ assert report["source_metadata"]["endpoint"] == "https://example.invalid/vedastro" +def test_vedastro_range_scan_unconfigured_still_returns_official_search_events_preview() -> None: + env = os.environ.copy() + env.pop("VEDASTRO_API_ENDPOINT", None) + env.pop("VEDASTRO_ENABLE_NETWORK", None) + + completed = subprocess.run( + [ + sys.executable, + "scripts/vedastro_service_adapter.py", + "--range-scan", + "--domain", + "wealth", + "--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["status"] == "service_endpoint_not_configured" + assert report["request_preview"]["operation"] == "range_scan" + assert report["request_preview"]["vedastro_event_method"] == "SearchEvents" + assert report["request_preview"]["domain"] == "wealth" + + def test_vedastro_service_adapter_can_normalize_mock_http_response() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802