Harden VedAstro external evidence boundaries

2026-06-29 03:17:02 +08:00
parent cf5ed67645
commit 1127cca790
7 changed files with 982 additions and 24 deletions
@@ -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",
    ]


@@ -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"]
@@ -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