Merge origin/main into agent-guided birth time rectification

This commit is contained in:
Jesse_Chen
2026-07-18 18:54:21 +08:00
170 changed files with 77128 additions and 355 deletions
+14
View File
@@ -2033,6 +2033,10 @@ def test_panchanga_range_rejects_large_ranges() -> None:
def test_rectification_gate_returns_varga_risk_summary() -> None:
handler = _handler()
handler._compute_vedastro_gateway_run = lambda body: { # type: ignore[method-assign]
'status': 'official_verified',
'official_raw_response': {'request_id': 'rectification-live-call'},
}
result = handler._compute_rectification_gate({
'planets': _sample_planets(),
'ascendant': {'lon': 92.0},
@@ -2042,6 +2046,7 @@ def test_rectification_gate_returns_varga_risk_summary() -> None:
assert result['success'] is True
assert result['endpoint'] == 'rectification_gate'
assert result['vedastro_gateway']['official_raw_response']['request_id'] == 'rectification-live-call'
assert result['effective_accuracy'] == '15min'
assert 'headline' in result['summary']
assert result['summary']['recommended_events']
@@ -2261,6 +2266,11 @@ def test_consultation_workflow_uses_unified_orchestrator_contract(monkeypatch) -
'endpoint': 'rectification_gate',
'summary': {'recommended_events': ['career_change']},
})
monkeypatch.setattr(handler, '_compute_vedastro_gateway_run', lambda body: {
'scope': 'vedastro_gateway_run',
'status': 'official_verified',
'official_raw_response': {'request_id': 'test-live-call'},
})
monkeypatch.setattr(handler, '_compute_thematic_report', lambda body: {
'success': True,
'endpoint': 'thematic_report',
@@ -2324,6 +2334,8 @@ def test_consultation_workflow_uses_unified_orchestrator_contract(monkeypatch) -
'official_blocked',
'local_fallback',
}
assert result['vedastro_gateway']['status'] == 'official_verified'
assert result['vedastro_gateway']['official_raw_response']['request_id'] == 'test-live-call'
assert result['runtime_evidence_log']['quality_gate']['technique_audit_table_required'] is True
assert result['runtime_evidence_log']['quality_gate']['technique_audit_table'][0]['technique'] == 'VedAstro Cloud State'
assert result['machine_evidence_packet']['status'] == 'partial'
@@ -2334,6 +2346,8 @@ def test_consultation_workflow_uses_unified_orchestrator_contract(monkeypatch) -
assert result['runtime_evidence_log']['blind_technical_mode']['enabled'] is True
assert 'conversation_feedback' in result['runtime_evidence_log']['blind_technical_mode']['disallowed_sources']
assert result['chart']['special_lagnas']['precision'] == 'sunrise_correct'
assert result['reference_transparency']['timing_display']['exact_triggers'] == 'display_as_technical_trigger_not_guarantee'
assert result['reference_transparency']['similar_public_cases']['does_not_predict_user_outcome'] is True
def test_consultation_workflow_accepts_western_oracle_payload(monkeypatch) -> None:
@@ -0,0 +1,12 @@
import json
from pathlib import Path
from scripts.archive_vedastro_nuget_probe import build_archive
def test_archive_contains_all_identity_layers(tmp_path: Path) -> None:
candidate=tmp_path/"candidate.json";runtime=tmp_path/"runtime.json"
candidate.write_text(json.dumps({"status":"reproducible_library_executable_verified","package_sha256":"a","library_dll_sha256":"b"}))
runtime.write_text(json.dumps({"version":"1.2.0.0","informational_version":"1.2.0","methods":["M"],"method_contracts":[{"name":"M"}]}))
report=build_archive(candidate,runtime)
assert report["assembly_version"]=="1.2.0.0"
assert report["public_methods"]==["M"]
assert report["runtime_image_id"].startswith("sha256:")
+13
View File
@@ -78,6 +78,19 @@ def test_sade_sati_uses_real_saturn_transit_for_reference_date() -> None:
assert result["provenance"]["reference_date"] == "2026-07-11"
def test_jupiter_transit_is_available_from_the_native_oracle() -> None:
result = calculation_service.compute_transit_longitude(
planet="Jupiter",
reference_date="2026-07-11",
tz=5.5,
ayanamsa="lahiri",
)
assert result["planet"] == "Jupiter"
assert 0 <= result["longitude"] < 360
assert result["data_layer"] == "true_transit_positions"
def test_timezone_inference_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
calculation_service,
@@ -0,0 +1,59 @@
from __future__ import annotations
import json
from scripts import collect_vedastro_official_parity_raw as collector
def test_collect_chart_core_writes_secret_free_hashed_packet(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api")
monkeypatch.setenv("VEDASTRO_API_KEY", "must-not-leak")
def fake_post(endpoint, request_item):
if request_item["section"] == "shadbala_chesta":
return {"Status": "Pass", "Payload": {"PlanetChestaBala": "12.5"}}, 1, []
if request_item["section"] == "shadbala":
return {"Status": "Pass", "Payload": {"AllPlanetStrength": "(400, Sun)"}}, 1, []
if request_item["section"] == "ashtakavarga_sav":
return {"Status": "Pass", "Payload": {"AshtakvargaLifeMap": {"TotalBindus": 337}}}, 1, []
if request_item["section"] == "ashtakavarga_bav":
return {"Status": "Pass", "Payload": {"BhinnashtakavargaChart": {"Sun": {"Rows": [1] * 12}}}}, 1, []
if request_item["section"] == "ashtakavarga_sav_chart":
return {"Status": "Pass", "Payload": {"SarvashtakavargaChart": {"Sarvashtakavarga": {"Rows": [28] * 12}}}}, 1, []
planet = request_item["fanout_value"]
return {
"Status": "Pass",
"Payload": {
"AllPlanetData": {
"PlanetHoraD2Signs": {"Name": f"{planet}-D2"},
"PlanetChaturthamshaD4Sign": {"Name": f"{planet}-D4"},
"PlanetNavamshaD9Sign": {"Name": f"{planet}-D9"},
"PlanetDashamamshaD10Sign": {"Name": f"{planet}-D10"},
}
},
}, 1, []
monkeypatch.setattr(collector.adapter, "_post_official_snapshot_section", fake_post)
output = tmp_path / "packet.json"
report = collector.collect_chart_core(
collector.adapter.PARITY_CASES["beijing_first_use_demo"],
case_id="beijing_first_use_demo",
planets=["Sun", "Moon"],
output_path=output,
)
stored = json.loads(output.read_text(encoding="utf-8"))
assert report["status"] == "ok"
assert stored["coverage"]["sections"] == [
"D2", "D4", "D9", "D10", "shadbala_total", "shadbala_components", "ashtakavarga_bav", "ashtakavarga_sav"
]
assert stored["fanout_statuses"] == {"Moon": "ok", "Sun": "ok"}
assert stored["scalar_statuses"] == {
"ashtakavarga_bav": "ok",
"ashtakavarga_sav": "ok",
"ashtakavarga_sav_chart": "ok",
"shadbala": "ok",
}
assert stored["component_statuses"] == {"Moon.chesta": "ok", "Sun.chesta": "ok"}
assert len(stored["response_hash"]) == 64
assert "must-not-leak" not in output.read_text(encoding="utf-8")
@@ -0,0 +1,49 @@
from pathlib import Path
PAGE = Path("frontend/src/app/page.tsx")
DAILY_ROUTE = Path("frontend/src/app/api/daily-starlanguage/route.ts")
RECTIFICATION_ROUTE = Path("frontend/src/app/api/birth-rectification/route.ts")
def test_daily_starlanguage_entrypoint_is_productized() -> None:
source = PAGE.read_text(encoding="utf-8")
assert "今日星语" in source
assert "fetchDailyStarlanguage" in source
assert "buildDailyStarlanguageCard" in source
assert "daily-starlanguage-card" in source
assert "今日趋势" in source
assert "行动建议" in source
assert "今日提醒" in source
assert "draftDailyStarlanguageQuestion" in source
assert "探索性日提示" in source
assert "不是确定预测" in source
def test_daily_starlanguage_api_declares_honest_source_boundary() -> None:
source = DAILY_ROUTE.read_text(encoding="utf-8")
assert "status: \"ok\"" in source
assert "/api/chart" in source
assert "/api/transit" in source
assert "jyotish_api_transit_lite" in source
assert "calculation_lite" in source
assert "exploratory_unvalidated" in source
assert "not_deterministic_prediction" in source
def test_birth_time_rectification_entrypoint_is_productized() -> None:
source = PAGE.read_text(encoding="utf-8")
assert "生时校正" in source
assert "fetchBirthRectificationPreview" in source
assert "birth-rectification-card" in source
assert "draftBirthTimeRectificationQuestion" in source
assert "候选出生时间段" in source
assert "不能直接改写默认星盘" in source
def test_birth_rectification_api_proxies_active_questionnaire_with_boundary() -> None:
source = RECTIFICATION_ROUTE.read_text(encoding="utf-8")
assert "/api/active_rectification_questions" in source
assert "candidate_scan" in source
assert "question_count" in source
assert "not_auto_rectified" in source
+33
View File
@@ -0,0 +1,33 @@
from __future__ import annotations
from scripts.daily_guidance_service import build_daily_guidance
def test_daily_guidance_returns_short_positive_evidence_packet() -> None:
packet = build_daily_guidance({
"year": 1990,
"month": 1,
"day": 1,
"hour": 12,
"minute": 0,
"lat": 39.9,
"lon": 116.4,
"tz": 8,
"date": "2026-07-17",
"ayanamsa": "lahiri",
"node_mode": "mean",
})
assert packet["success"] is True
assert packet["endpoint"] == "daily_guidance"
assert packet["word_count"] <= 100
assert packet["daily_star_words"].startswith("今日星语:")
assert packet["audit"]["not_a_prediction"] is True
assert {"D1", "Daily Transit"} <= {row["layer"] for row in packet["evidence"]}
assert packet["suggested_actions"]
def test_daily_guidance_endpoint_is_registered() -> None:
source = __import__("pathlib").Path("scripts/jyotish_api_server.py").read_text(encoding="utf-8")
assert "/api/daily_guidance" in source
assert "daily_guidance_service" in source
+4 -4
View File
@@ -36,8 +36,8 @@ def test_dasha_oracle_closure_status_reports_current_dasha_closure() -> None:
report = json.loads(completed.stdout)
assert report["scope"] == "dasha_external_oracle_closure_status"
assert report["schema_version"] == 1
assert report["summary"]["dasha_task_count"] == 3
assert report["summary"]["external_verified_dasha_tasks"] == 3
assert report["summary"]["dasha_task_count"] == 2
assert report["summary"]["external_verified_dasha_tasks"] == 2
assert report["summary"]["can_claim_dasha_oracle_closure"] is True
assert report["first_priority"] is None
@@ -91,7 +91,7 @@ def test_dasha_oracle_closure_status_advances_after_first_packet_is_filled(tmp_p
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["summary"]["valid_dasha_packets"] == 3
assert report["summary"]["valid_dasha_packets"] == 2
assert report["summary"]["all_dasha_packets_external_verified"] is True
@@ -100,7 +100,7 @@ def test_dasha_oracle_closure_status_has_no_first_priority_after_all_dasha_packe
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["summary"]["external_verified_dasha_tasks"] == 3
assert report["summary"]["external_verified_dasha_tasks"] == 2
assert report["summary"]["can_claim_dasha_oracle_closure"] is True
assert report["first_priority"] is None
assert report["next_actions"] == [
+26
View File
@@ -0,0 +1,26 @@
from pathlib import Path
import json
from scripts.day_level_holdout_validator import validate
ROOT=Path(__file__).resolve().parents[1]
def test_empty_preregistered_holdout_remains_honestly_blocked() -> None:
report=validate(ROOT/"references/real_case_calibration/day_level_holdout_v3_preregistration.json")
assert report["status"] == "awaiting_independent_labels"
assert report["annotation_count"] == 0
assert report["production_tuning_allowed"] is False
def test_raman_source_candidates_are_not_holdout_labels() -> None:
payload = json.loads((ROOT / "references/real_case_calibration/raman_vol1_holdout_source_candidates_2026_07_18.json").read_text())
assert payload["candidate_count"] >= 20
assert all(row["status"] == "needs_independent_normalization" for row in payload["candidates"])
assert all("day_level_label" in row["prohibited_uses"] for row in payload["candidates"])
def test_observational_holdout_is_ready_but_not_independent() -> None:
report = validate(ROOT / "references/real_case_calibration/day_level_observational_holdout_v1.json")
assert report["status"] == "observational_ready_not_independent"
assert report["positive_count"] >= 20
assert report["negative_count"] >= 80
assert report["production_tuning_allowed"] is False
@@ -31,15 +31,30 @@ def test_external_engine_adapter_diagnostics_aggregates_three_engines() -> None:
assert report["engines"]["PyJHora/JHora"]["license_boundary"].startswith("AGPL external benchmark")
assert report["engines"]["jyotishganit"]["license"] == "MIT"
contract = report["same_chart_parity_contract"]
assert contract["status"] in {"ready", "blocked"}
expected_contract_status = (
"mismatch"
if all(state["available"] for state in contract["engine_states"].values())
else "blocked"
)
assert contract["status"] == expected_contract_status
assert contract["required_outputs"] == ["D1", "D9", "D10", "D2", "D4", "Vimshottari", "Shadbala", "Ashtakavarga"]
assert "official_raw_response" in contract["expected_oracle_fields"]["VedAstro"]
assert "raw_output_path" in contract["expected_oracle_fields"]["PyJHora/JHora"]
assert contract["engine_states"]["jyotishganit"]["available"] is True
assert contract["engine_states"]["jyotishganit"]["tested"] is True
assert contract["engine_states"]["VedAstro"]["tested"] is True
assert contract["engine_states"]["PyJHora/JHora"]["tested"] is True
pyjhora = contract["partial_verifications"]["PyJHora/JHora"]
assert pyjhora["status"] == "partial_verified"
assert pyjhora["missing_required_outputs"] == ["D2", "D4", "Shadbala", "Ashtakavarga"]
assert contract["replay_manifest"]["tested"] is False
assert contract["replay_manifest"]["blocked_reason"] == "no_same_chart_oracle_rows_imported"
assert pyjhora["missing_required_outputs"] == ["Shadbala"]
assert {"D2", "D4", "Ashtakavarga"} <= set(pyjhora["covered_outputs"])
assert pyjhora["output_sample_counts"]["D2"] == 1
assert pyjhora["output_sample_counts"]["D4"] == 1
assert pyjhora["output_sample_counts"]["Ashtakavarga"] == 1
assert pyjhora["output_sample_counts"]["D1"] == 10
assert pyjhora["supplemental_verifications"][0]["partial_outputs"] == ["Shadbala"]
assert contract["replay_manifest"]["tested"] is True
assert contract["replay_manifest"]["status"] == "mismatch"
assert contract["replay_manifest"]["blocked_reason"] is None
assert contract["replay_manifest"]["missing_high_rigor_sections"] == []
assert report["status"] in {"complete", "partial"}
+1 -1
View File
@@ -41,7 +41,7 @@ def test_external_official_sanity_closure_reports_all_three_oracles() -> None:
pyjhora = report["oracle_ledger"]["pyjhora"]
assert pyjhora["role"] == "black_box_external_oracle"
assert pyjhora["artifact_count"] >= 8
assert pyjhora["packet_count"] >= 8
assert pyjhora["packet_count"] >= 6
assert pyjhora["license_boundary"] == "black_box_artifacts_only_no_agpl_code_import"
jyotishganit = report["oracle_ledger"]["jyotishganit"]
@@ -0,0 +1,53 @@
"""Regression coverage for the public external-validation release boundary."""
from __future__ import annotations
import json
from pathlib import Path
import scripts.external_validation_release_gate as gate
ROOT = Path(__file__).resolve().parents[1]
MANIFEST = ROOT / "references" / "evidence_manifests" / "commercial_external_validation_release.v1.json"
def _manifest() -> dict:
return json.loads(MANIFEST.read_text(encoding="utf-8"))
def _copy_manifest_with_one_bad_digest(tmp_path: Path) -> Path:
manifest = _manifest()
manifest["assets"][0]["sha256"] = "0" * 64
path = tmp_path / "release.json"
path.write_text(json.dumps(manifest), encoding="utf-8")
return path
def test_release_manifest_declares_public_assets_and_external_boundaries() -> None:
manifest = _manifest()
assert manifest["schema_version"] == 1
assert manifest["release_scope"] == "public_research_evidence_snapshot"
assert manifest["engines"]["PyJHora"]["status"] == "available"
assert manifest["engines"]["jyotishganit"]["status"] == "available"
assert manifest["engines"]["VedAstro"]["status"] == "blocked"
assert manifest["engines"]["JHora"]["official_raw_status"] != "verified"
def test_evaluate_manifest_accepts_current_public_release() -> None:
report = gate.evaluate_manifest(MANIFEST)
assert report["status"] == "pass"
assert report["summary"]["assets_verified"] == report["summary"]["assets_total"]
assert report["summary"]["production_tuning_allowed"] is False
assert report["engines"]["VedAstro"]["status"] == "blocked"
def test_evaluate_manifest_reports_digest_drift(tmp_path: Path) -> None:
report = gate.evaluate_manifest(_copy_manifest_with_one_bad_digest(tmp_path))
assert report["status"] == "blocked"
assert report["assets"][0]["integrity"] == "mismatch"
def test_runtime_truth_profile_runs_external_validation_release_gate() -> None:
text = (ROOT / "scripts" / "run_quality_gate.py").read_text(encoding="utf-8")
assert "external_validation_release_gate.py" in text
+5 -5
View File
@@ -946,7 +946,7 @@ 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\", \"vedastro-live\"]",
"choices=[\"quick\", \"browser\", \"release\", \"accuracy\", \"vedastro-live\", \"runtime-truth\"]",
"QUALITY_GATE_PROFILES",
"quick",
"browser",
@@ -2893,12 +2893,12 @@ def test_api_birth_seconds_are_preserved_in_user_facing_flows() -> None:
payload = sample_second_precision_payload()
birth_dt = handler._parse_birth_datetime(payload)
assert birth_dt.isoformat() == "1955-02-24T19:15:00"
assert birth_dt.isoformat() == "1955-02-24T19:45:20"
without_seconds = handler._compute_chart({**payload, "second": 0})
with_seconds = handler._compute_chart(payload)
assert with_seconds["success"] is True
assert with_seconds["birth"]["time"] == "19:15:00"
assert with_seconds["birth"]["time"] == "19:45:20"
assert with_seconds["birth"]["second"] == 20
assert with_seconds["birth"]["julian_day"] > without_seconds["birth"]["julian_day"]
@@ -2912,9 +2912,9 @@ def test_api_birth_seconds_are_preserved_in_user_facing_flows() -> None:
assert with_seconds["shadbala"]["Sun"]["rupas"] == round(expected_shadbala["planets"]["Sun"]["total_rupas"], 2)
full_reading = handler._compute_full_reading_for_thematic(payload)
assert full_reading["birth_info"]["time"] == "19:15:00"
assert full_reading["birth_info"]["time"] == "19:45:20"
assert full_reading["birth_info"]["second"] == 20
assert full_reading["modules"]["chart"]["birth_info"]["time"] == "19:15:00"
assert full_reading["modules"]["chart"]["birth_info"]["time"] == "19:45:20"
def test_local_frontend_and_api_runtime_smoke() -> None:
@@ -0,0 +1,19 @@
"""The PyJHora artifact inventory must be safe to read during verification."""
from __future__ import annotations
import scripts.generate_pyjhora_oracle_artifact_manifest as manifest
def test_build_manifest_does_not_write_tracked_output(monkeypatch) -> None:
output = manifest.ROOT / "references" / "oracle" / "artifacts" / ".pytest-manifest.json"
output.unlink(missing_ok=True)
monkeypatch.setattr(manifest, "OUTPUT_PATH", output)
try:
report = manifest.build_manifest()
assert not output.exists()
finally:
output.unlink(missing_ok=True)
assert report["scope"] == "pyjhora_oracle_artifact_manifest"
+19 -2
View File
@@ -1,11 +1,15 @@
from datetime import datetime
from scripts.gulika import GHATIKA_END, calculate_gulika
import json
from pathlib import Path
from scripts.gulika import GHATIKA_END, SATURN_PART_START, calculate_gulika
def test_gulika_uses_prasna_marga_weekday_table() -> None:
assert GHATIKA_END[6] == {"day": 26, "night": 10}
assert GHATIKA_END[0] == {"day": 22, "night": 6}
assert SATURN_PART_START[0] == {"day": 5, "night": 1}
def test_gulika_returns_sidereal_segment_ascendant_with_audit_trace() -> None:
@@ -13,5 +17,18 @@ def test_gulika_returns_sidereal_segment_ascendant_with_audit_trace() -> None:
assert result["status"] == "partial"
assert 0 <= result["longitude"] < 360
assert result["ghatika_end"] in range(0, 31)
assert result["part_index"] in range(0, 8)
assert result["ghatika_end"] is None
assert result["rule_source"].endswith("#3.5")
def test_gulika_matches_public_pyjhora_smoke_oracle() -> None:
packet = json.loads(
Path("references/oracle/prashna_sphuta_pyjhora_public_smoke.json").read_text(encoding="utf-8")
)
expected = packet["outputs"]["gulika"]
result = calculate_gulika(datetime(1990, 1, 1, 12, 0), lat=39.9042, lon=116.4074, tz=8)
assert result["method"] == "saturn_part_start"
assert result["sign_idx"] == expected["sign_index"]
assert abs(result["degree_in_sign"] - expected["degree_in_sign"]) <= 0.5
+152
View File
@@ -0,0 +1,152 @@
from pathlib import Path
from scripts.jyotishyamitra_adapter_probe import (
build_report,
build_varga_comparison,
canonical_request,
compare_with_existing_oracles,
extract_local_varga_signs,
extract_varga_signs,
extract_xalen_varga_signs,
normalize_raw,
schema_fingerprint,
)
WHEEL = Path("/tmp/jyotishyamitra_probe/jyotishyamitra-1.4.0-py3-none-any.whl")
WHEEL_SHA256 = "4f1a16facfa86ef01cb44de505c6de2e66b2f8d2c8b5e125c090e3f8485c6a9d"
def test_jyotishyamitra_probe_contract_uses_pinned_identity_fixture(tmp_path: Path) -> None:
wheel = WHEEL
assert wheel.exists(), "fixture_missing: run pinned wheel download outside CI; tests must not fetch network"
report = build_report(
wheel_path=wheel,
commit="86f7eb610a66b06b3f0817d2c53355bec8b3bf8d",
raw={"D1": {"Sun": "Aquarius"}, "metadata": {"ayanamsa": "Lahiri"}},
)
assert report["oracle"] == "jyotishyamitra"
assert report["license"] == "MIT"
assert report["version"] == "1.4.0"
assert report["source_commit"] == "86f7eb610a66b06b3f0817d2c53355bec8b3bf8d"
assert report["wheel_sha256"]
assert report["wheel_sha256"] == WHEEL_SHA256
assert report["raw_sha256"]
assert report["normalized_raw_sha256"]
assert report["package_metadata"]["version"] == "1.4.0"
assert report["package_metadata"]["requires_python"] == ">=3.7"
assert report["package_metadata"]["license"] is None
assert report["package_metadata"]["license_file_spdx_inferred"] == "MIT"
assert report["python_runtime"]["version"]
assert report["schema_fingerprint"]["sha256"]
assert report["isolated_subprocess"]["used"] is False
assert report["truth_policy"] == "independent_observation_not_truth"
def test_jyotishyamitra_missing_wheel_is_explicitly_blocked(tmp_path: Path) -> None:
report = build_report(tmp_path / "missing.whl")
assert report["status"] == "blocked"
assert report["blocked_reason"] == "fixture_missing"
assert report["promotion_allowed"] is False
def test_jyotishyamitra_canonical_request_records_defaults() -> None:
request = canonical_request(
{
"name": "Steve Jobs",
"gender": "male",
"place": "San Francisco",
"longitude": -122.4194,
"latitude": 37.7749,
"timezone": -8.0,
"birth": {"year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15},
}
)
assert request["ayanamsa"] == "package_default"
assert request["node_mode"] == "package_default"
assert request["returnval"] == "ASTRODATA_DICTIONARY"
assert request["birth"]["second"] == 0
def test_jyotishyamitra_schema_fingerprint_is_stable() -> None:
fp = schema_fingerprint({"D1": {"Sun": "Aquarius"}, "scores": [1, 2]})
assert fp["path_count"] == 3
assert fp["sha256"]
assert "$.D1.Sun:str" in fp["sample_paths"]
def test_jyotishyamitra_normalizes_volatile_current_dasha_timestamp() -> None:
a = {"Dashas": {"Vimshottari": {"current": {"date": "run-a", "lord": "Sun"}}}}
b = {"Dashas": {"Vimshottari": {"current": {"date": "run-b", "lord": "Sun"}}}}
assert normalize_raw(a) == normalize_raw(b)
def test_jyotishyamitra_field_comparison_is_observation_not_promotion() -> None:
comparison = compare_with_existing_oracles(
{"D1": {"Sun": "Aquarius"}, "D9": {"Moon": "Cancer"}},
{"D1": {"Sun": "Aquarius"}, "D9": {"Moon": "Leo"}},
{"D1": {"Sun": "Aquarius"}, "D9": {"Moon": "Cancer"}},
)
assert comparison["row_count"] == 2
assert comparison["match_counts"]["local"] == 1
assert comparison["match_counts"]["xalen"] == 2
assert comparison["promotion_allowed"] is False
def test_extract_varga_signs_keeps_only_supported_planet_signs() -> None:
raw = {
"D1": {"planets": {"Sun": {"sign": "Aquarius"}, "Rahu": {"sign": "Saggitarius"}}},
"D9": {"planets": {"Moon": {"sign": "Cancer"}, "Mars": {"sign": None}}},
"Balas": {"Shadbala": {}},
}
assert extract_varga_signs(raw) == {
"D1": {"Sun": "Aquarius", "Rahu": "Sagittarius"},
"D9": {"Moon": "Cancer"},
}
def test_existing_oracle_sign_adapters_share_the_same_surface() -> None:
local = {
"planets": {"Sun": {"sign": "Aquarius"}},
"varga": {"D9": {"Moon": {"sign": "Cancer"}}},
}
xalen = {"varga": {"D1": {"Sun": "Kumbha"}, "D9": {"Moon": "Karka"}}}
assert extract_local_varga_signs(local) == {
"D1": {"Sun": "Aquarius"},
"D9": {"Moon": "Cancer"},
}
assert extract_xalen_varga_signs(xalen) == {
"D1": {"Sun": "Aquarius"},
"D9": {"Moon": "Cancer"},
}
def test_varga_comparison_never_promotes_the_new_oracle() -> None:
jyotishyamitra = {"D1": {"planets": {"Sun": {"sign": "Aquarius"}}}}
local = {"planets": {"Sun": {"sign": "Aquarius"}}}
xalen = {"varga": {"D1": {"Sun": "Kumbha"}}}
comparison = build_varga_comparison(jyotishyamitra, local, xalen)
assert comparison["row_count"] == 1
assert comparison["match_counts"] == {"local": 1, "xalen": 1}
assert comparison["promotion_allowed"] is False
def test_varga_comparison_marks_missing_engine_fields_not_comparable() -> None:
comparison = compare_with_existing_oracles(
{"D1": {"Rahu": "Leo"}},
{"D1": {"Rahu": "Leo"}},
{"D1": {}},
)
assert comparison["rows"][0]["local_status"] == "match"
assert comparison["rows"][0]["xalen_status"] == "not_comparable"
+27 -9
View File
@@ -4,12 +4,16 @@
from __future__ import annotations
import json
import re
import subprocess
import sys
from pathlib import Path
import scripts.oracle_closure_master_dashboard as dashboard
ROOT = Path(__file__).resolve().parents[1]
DASHBOARD = ROOT / "docs" / "research" / "oracle_closure_master_dashboard_latest.md"
def run_dashboard(*args: str) -> subprocess.CompletedProcess[str]:
@@ -31,6 +35,20 @@ def run_dashboard(*args: str) -> subprocess.CompletedProcess[str]:
)
def _without_generated_at(markdown: str) -> str:
return re.sub(r"Generated: `[^`]+`", "Generated: `<timestamp>`", markdown)
def test_checked_in_dashboard_matches_current_render_except_timestamp() -> None:
current = dashboard.render_markdown(
dashboard.build_dashboard(
"references/oracle/dasha_shadbala_oracle_cases.json",
"references/oracle/tajika_annual_oracle_cases.json",
)
)
assert _without_generated_at(DASHBOARD.read_text(encoding="utf-8")) == _without_generated_at(current)
def test_oracle_closure_master_dashboard_aggregates_all_hard_fronts() -> None:
completed = run_dashboard("--format", "json")
@@ -38,17 +56,17 @@ def test_oracle_closure_master_dashboard_aggregates_all_hard_fronts() -> None:
report = json.loads(completed.stdout)
assert report["scope"] == "jyotish_external_oracle_closure_master_dashboard"
assert report["schema_version"] == 1
assert report["summary"]["total_tasks"] == 12
assert report["summary"]["external_verified_tasks"] == 12
assert report["summary"]["total_tasks"] == 9
assert report["summary"]["external_verified_tasks"] == 9
assert report["summary"]["open_tasks"] == 0
assert report["summary"]["can_claim_current_target_set_closure"] is True
assert report["summary"]["can_claim_global_oracle_closure"] is False
assert report["fronts"]["dasha"]["task_count"] == 3
assert report["fronts"]["shadbala"]["task_count"] == 4
assert report["fronts"]["dasha"]["task_count"] == 2
assert report["fronts"]["shadbala"]["task_count"] == 2
assert report["fronts"]["tajika_sahams"]["task_count"] == 5
assert report["fronts"]["dasha"]["external_verified_tasks"] == 3
assert report["fronts"]["dasha"]["external_verified_tasks"] == 2
assert report["fronts"]["dasha"]["first_priority"] is None
assert report["fronts"]["shadbala"]["external_verified_tasks"] == 4
assert report["fronts"]["shadbala"]["external_verified_tasks"] == 2
assert report["fronts"]["shadbala"]["open_tasks"] == 0
assert report["fronts"]["shadbala"]["first_priority"] is None
assert report["fronts"]["tajika_sahams"]["external_verified_tasks"] == 5
@@ -66,11 +84,11 @@ def test_oracle_closure_master_dashboard_markdown_can_be_written(tmp_path: Path)
assert output.exists()
markdown = output.read_text(encoding="utf-8")
assert "# Jyotish External Oracle Closure Master Dashboard" in markdown
assert "total_tasks: `12`" in markdown
assert "total_tasks: `9`" in markdown
assert "can_claim_current_target_set_closure: `true`" in markdown
assert "can_claim_global_oracle_closure: `false`" in markdown
assert "`dasha` | 3 | 3 | `complete`" in markdown
assert "`shadbala` | 4 | 4 | `complete`" in markdown
assert "`dasha` | 2 | 2 | `complete`" in markdown
assert "`shadbala` | 2 | 2 | `complete`" in markdown
assert "template_einstein_varshaphala_1905_lahiri" in markdown
assert "manual entries" in markdown
assert "metadata missing" in markdown
@@ -0,0 +1,24 @@
import json
from pathlib import Path
def test_public_pyjhora_prashna_sphuta_packet_is_numeric_and_guarded() -> None:
packet = json.loads(
Path("references/oracle/prashna_sphuta_pyjhora_public_smoke.json").read_text(encoding="utf-8")
)
assert packet["status"] == "local_parity_verified_single_case"
assert packet["birth_data_policy"] == "public_synthetic_smoke_only"
assert packet["capture"]["raw_response_available"] is True
assert packet["engine"]["license_boundary"].startswith("AGPL external benchmark")
assert set(packet["outputs"]) == {
"gulika", "kunda_lagna", "tri_sphuta", "chatur_sphuta", "pancha_sphuta"
}
for row in packet["outputs"].values():
assert 0 <= row["sign_index"] < 12
assert 0 <= row["degree_in_sign"] < 30
assert packet["local_comparison"]["status"] == "pass"
assert max(
value for key, value in packet["local_comparison"].items() if key.endswith("delta_degrees")
) <= packet["local_comparison"]["tolerance_degrees"]
assert "remain required" in packet["boundary"]
+28
View File
@@ -0,0 +1,28 @@
from scripts.production_smoke import check
def test_health_route_declares_required_checks() -> None:
source = open("frontend/src/app/api/health/route.ts", encoding="utf-8").read()
for expected in (
"supabasePublicConfig",
"supabaseServiceRole",
"modelProvider",
"jyotishApi",
"JYOTISH_API_BASE",
):
assert expected in source
def test_production_smoke_accepts_health_degraded_status(monkeypatch) -> None:
def fake_fetch(url: str, timeout: float) -> tuple[int, str, float]:
if url.endswith("/api/health"):
return 503, '{"status":"blocked","checks":{"web":{},"jyotishApi":{}}}', 0.01
return 200, "Jyotisha", 0.01
monkeypatch.setattr("scripts.production_smoke.fetch", fake_fetch)
report = check("https://example.invalid", 1.0)
assert report["ok"] is True
assert report["checks"][1]["health_status"] == "blocked"
+10 -10
View File
@@ -38,19 +38,19 @@ def test_public_benchmark_dashboard_outputs_stable_json_summary() -> None:
assert report["schema_version"] == 1
assert report["summary"]["technique_count"] >= 60
assert report["summary"]["capability_valid"] is True
assert report["oracle_readiness"]["total_packets"] == 6
assert report["oracle_readiness"]["valid_packets"] == 5
assert report["oracle_readiness"]["ready_for_calibration"] == 5
assert report["oracle_readiness"]["total_packets"] == 4
assert report["oracle_readiness"]["valid_packets"] == 3
assert report["oracle_readiness"]["ready_for_calibration"] == 3
assert report["oracle_readiness"]["production_tuning_allowed"] is False
assert report["dasha_oracle_readiness"]["valid_dasha_packets"] == 3
assert report["dasha_oracle_readiness"]["total_dasha_packets"] == 3
assert report["dasha_oracle_readiness"]["valid_dasha_packets"] == 2
assert report["dasha_oracle_readiness"]["total_dasha_packets"] == 2
assert report["boundary_audit"]["production_tuning_recommended"] is False
assert report["pyjhora_blackbox_assets"]["artifact_count"] >= 8
assert report["pyjhora_blackbox_assets"]["packet_count"] >= 8
assert report["pyjhora_blackbox_assets"]["artifact_count"] == 9
assert report["pyjhora_blackbox_assets"]["packet_count"] == 6
assert report["pyjhora_blackbox_assets"]["fronts"]["tajika_sahams"]["artifact_count"] >= 1
assert "Dasha-only external oracle readiness is 3/3" in report["global_first_gap"]
assert "Shadbala external absolute-value readiness is 4/4" in report["global_first_gap"]
assert "PyJHora black-box assets are 12 artifacts / 8 packets" in report["global_first_gap"]
assert "Dasha-only external oracle readiness is 2/2" in report["global_first_gap"]
assert "Shadbala external absolute-value readiness is 2/2" in report["global_first_gap"]
assert "PyJHora black-box assets are 9 artifacts / 6 packets" in report["global_first_gap"]
assert report["public_claim"]["can_claim_global_first"] is False
assert report["public_claim"]["reason"]
@@ -2,11 +2,16 @@ from __future__ import annotations
from scripts.public_real_case_negative_controls import (
generate_control_dates,
parse_offsets,
rank_positive_against_controls,
summarize_negative_control_rows,
)
def test_parse_offsets_rejects_zero_and_deduplicates() -> None:
assert parse_offsets("-60,-30,0,30,60,-30") == (-60, -30, 30, 60)
def test_generate_control_dates_uses_fixed_offsets_without_positive_date() -> None:
dates = generate_control_dates("2018-05-19", offsets=(-60, -30, 30, 60))
@@ -47,7 +52,10 @@ def test_negative_control_summary_reports_false_activations() -> None:
assert summary["control_date_count"] == 4
assert summary["control_activation_rate"] == 0.5
assert summary["control_strong_activation_rate"] == 0.25
assert summary["specificity_proxy"] == 0.5
assert summary["strong_specificity_proxy"] == 0.75
assert summary["positive_top_1_rate"] == 0.5
assert summary["positive_top_3_rate"] == 1.0
assert summary["mean_reciprocal_rank"] == 0.75
assert summary["mean_score_margin"] == 0.5
assert summary["timing_precision_gate"]["status"] == "blocked"
+8
View File
@@ -19,6 +19,14 @@ def test_real_case_replay_manifest_contains_ten_research_grade_cases() -> None:
assert result["birth_time_ratings"] == {"A": 2, "AA": 8}
def test_public_context_manifest_is_structured_but_not_replay_calibration() -> None:
result = validate_manifest(ROOT / "references/real_case_calibration/public_context_manifest.json")
assert result["status"] == "partial"
assert result["errors"] == []
assert result["replay_ready_count"] == 0
assert result["domain_counts"] == {"health": 1, "wealth": 1}
def test_holdout_manifest_contains_ten_new_balanced_cases() -> None:
batch1_path = ROOT / "references/real_case_calibration/replay_manifest.json"
holdout_path = ROOT / "references/real_case_calibration/replay_manifest_holdout_v2.json"
@@ -0,0 +1,245 @@
from scripts.reference_transparency_contract import (
build_reference_transparency_contract,
select_similar_public_cases,
)
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def _chart(ascendant: str, moon: str, domain_lord_sign: str) -> dict:
return {
"ascendant": {"sign": ascendant},
"planets": {
"Moon": {"sign": moon},
"Saturn": {"sign": domain_lord_sign},
},
"houses": {"house_7": {"lord": "Venus"}, "house_10": {"lord": "Saturn"}},
}
def _varga_chart() -> dict:
return {
"ascendant": {"sign": "Leo", "lon": 149.0634},
"planets": {
"Moon": {"sign": "Pisces", "lon": 344.5165},
"Saturn": {"sign": "Libra", "lon": 207.9320},
"Sun": {"sign": "Aquarius", "lon": 312.5175},
"Venus": {"sign": "Sagittarius", "lon": 267.9412},
"Rahu": {"sign": "Sagittarius", "lon": 249.2743},
"Ketu": {"sign": "Gemini", "lon": 69.2743},
},
"houses": {"house_10": {"lord": "Saturn"}},
}
def test_select_similar_cases_shares_only_high_similarity_same_domain() -> None:
user_chart = _chart("Leo", "Pisces", "Libra")
cases = [
{
"case_id": "matching_career_case",
"subject": {"name": "Public Example"},
"chart": _chart("Leo", "Pisces", "Libra"),
"source": {"url": "https://example.com/birth", "source_grade": "primary"},
"event_outcomes": [{
"domain": "career", "event_type": "career_breakthrough", "event_date": "2007-01-09",
"outcome": "Public career event", "source": {"url": "https://example.com/event", "source_grade": "primary"},
}],
"replay": {"outcome_replay_status": "replayed", "do_not_use_for_prediction": False},
},
{
"case_id": "wrong_domain_case",
"subject": {"name": "Different Example"},
"chart": _chart("Aries", "Aries", "Aries"),
"source": {"url": "https://example.com/birth-2", "source_grade": "primary"},
"event_outcomes": [{
"domain": "marriage", "event_type": "legal_marriage", "event_date": "2011-04-29",
"outcome": "Public marriage event", "source": {"url": "https://example.com/event-2", "source_grade": "primary"},
}],
"replay": {"outcome_replay_status": "replayed", "do_not_use_for_prediction": False},
},
]
selected = select_similar_public_cases(user_chart, ["career"], cases=cases)
assert selected["status"] == "high_similarity_public_references_available"
assert [case["case_id"] for case in selected["cases"]] == ["matching_career_case"]
assert selected["cases"][0]["similarity"]["score"] == 1.0
assert "node_axis" not in selected["cases"][0]["similarity"]["matching_factors"]
assert selected["cases"][0]["event_source"]["url"] == "https://example.com/event"
assert selected["does_not_predict_user_outcome"] is True
assert selected["coverage"] == {
"available_event_domains": ["career", "marriage"],
"requested_uncovered_domains": [],
}
def test_reference_contract_preserves_dates_and_discloses_parallel_methods() -> None:
contract = build_reference_transparency_contract(
_chart("Leo", "Pisces", "Libra"),
["career"],
timing={"candidate_windows": [{"start": "2026-08-12", "end": "2026-08-16"}]},
cases=[],
)
assert contract["timing_display"]["exact_triggers"] == "display_as_technical_trigger_not_guarantee"
assert contract["external_engine_observations"]["VedAstro hosted"]["deployment_identity"] == "not_publicly_proven"
assert contract["method_variants"]["display"] == "show_parallel_methods_with_sources"
assert contract["similar_public_cases"]["status"] == "no_high_similarity_public_reference"
assert contract["similar_public_cases"]["coverage"]["requested_uncovered_domains"] == ["career"]
def test_default_public_manifest_can_surface_a_matching_replayed_case() -> None:
from scripts.domain_calculation_service import compute_chart
jobs_chart = compute_chart({
"year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15,
"lat": 37.7833, "lon": -122.4167, "tz": -8.0,
"ayanamsa": "lahiri", "node_mode": "mean",
})
selected = select_similar_public_cases(jobs_chart, ["career"])
assert selected["status"] == "high_similarity_public_references_available"
assert selected["cases"][0]["case_id"] == "jobs_iphone_2007"
assert selected["cases"][0]["reference_only"] is True
def test_same_event_date_compares_vimshottari_mahadasha() -> None:
from scripts.domain_calculation_service import compute_chart
jobs_chart = compute_chart({
"year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15,
"lat": 37.7833, "lon": -122.4167, "tz": -8.0,
"ayanamsa": "lahiri", "node_mode": "mean",
})
selected = select_similar_public_cases(
jobs_chart,
["career"],
reference_date="2007-01-09",
)
similarity = selected["cases"][0]["similarity"]
assert {
"vimshottari_mahadasha",
"vimshottari_antardasha",
"narayana_mahadasha_sign",
"narayana_antardasha_sign",
"jupiter_transit_house",
"saturn_transit_house",
}.issubset(similarity["matching_factors"])
assert similarity["timing_state"]["status"] == "matched"
assert similarity["timing_state"]["narayana_status"] == "matched"
assert similarity["timing_state"]["transit_status"] == "matched"
assert "vimshottari_antardasha" not in similarity["uncompared_layers"]
assert "narayana_dasha" not in similarity["uncompared_layers"]
assert "transit_event_state" not in similarity["uncompared_layers"]
def test_pending_health_case_is_context_only_not_calibration() -> None:
user_chart = _chart("Leo", "Pisces", "Libra")
cases = [{
"case_id": "public_health_context",
"subject": {"name": "Public Example"},
"chart": _chart("Leo", "Pisces", "Libra"),
"source": {"url": "https://example.com/birth", "source_grade": "primary"},
"event_outcomes": [{
"domain": "health", "event_type": "serious_injury", "event_date": "1925-09",
"outcome": "Public health event", "source": {"url": "https://example.com/event", "source_grade": "verified_secondary"},
}],
"replay": {"outcome_replay_status": "pending", "do_not_use_for_prediction": True},
}]
selected = select_similar_public_cases(user_chart, ["health"], cases=cases)
assert selected["status"] == "high_similarity_public_references_available"
assert selected["cases"][0]["reference_status"] == "public_context_only"
assert selected["cases"][0]["reference_only"] is True
assert selected["coverage"]["available_event_domains"] == ["health"]
def test_default_public_context_catalog_covers_wealth_as_reference_only() -> None:
selected = select_similar_public_cases(_chart("Leo", "Pisces", "Libra"), ["wealth"])
assert "wealth" in selected["coverage"]["available_event_domains"]
assert "wealth" not in selected["coverage"]["requested_uncovered_domains"]
def test_career_similarity_adds_d10_only_when_both_charts_have_longitudes() -> None:
chart = _varga_chart()
cases = [{
"case_id": "d10_match", "subject": {"name": "Public Example"}, "chart": chart,
"source": {"url": "https://example.com/birth", "source_grade": "primary"},
"event_outcomes": [{
"domain": "career", "event_type": "career_breakthrough", "event_date": "2007-01-09",
"outcome": "Public career event", "source": {"url": "https://example.com/event", "source_grade": "primary"},
}],
"replay": {"outcome_replay_status": "replayed", "do_not_use_for_prediction": False},
}]
selected = select_similar_public_cases(chart, ["career"], cases=cases)
similarity = selected["cases"][0]["similarity"]
assert {"d10_ascendant", "d10_sun"}.issubset(similarity["matching_factors"])
assert "D10" not in similarity["uncompared_layers"]
def test_marriage_similarity_adds_d9_only_when_both_charts_have_longitudes() -> None:
chart = _varga_chart()
cases = [{
"case_id": "d9_match", "subject": {"name": "Public Example"}, "chart": chart,
"source": {"url": "https://example.com/birth", "source_grade": "primary"},
"event_outcomes": [{
"domain": "marriage", "event_type": "legal_marriage", "event_date": "2011-04-29",
"outcome": "Public marriage event", "source": {"url": "https://example.com/event", "source_grade": "primary"},
}],
"replay": {"outcome_replay_status": "replayed", "do_not_use_for_prediction": False},
}]
selected = select_similar_public_cases(chart, ["marriage"], cases=cases)
similarity = selected["cases"][0]["similarity"]
assert {"d9_ascendant", "d9_venus"}.issubset(similarity["matching_factors"])
assert "D9" not in similarity["uncompared_layers"]
def test_api_house_shape_keeps_theme_lord_in_similarity() -> None:
chart = _varga_chart()
chart["houses"] = {10: {"sign": "Taurus", "sign_idx": 1}}
cases = [{
"case_id": "api_house_shape", "subject": {"name": "Public Example"}, "chart": chart,
"source": {"url": "https://example.com/birth", "source_grade": "primary"},
"event_outcomes": [{
"domain": "career", "event_type": "career_breakthrough", "event_date": "2007-01-09",
"outcome": "Public career event", "source": {"url": "https://example.com/event", "source_grade": "primary"},
}],
"replay": {"outcome_replay_status": "replayed", "do_not_use_for_prediction": False},
}]
selected = select_similar_public_cases(chart, ["career"], cases=cases)
assert "domain_lord_sign" in selected["cases"][0]["similarity"]["matching_factors"]
def test_default_manifest_exposes_kahlo_health_as_context_only() -> None:
from scripts.domain_calculation_service import compute_chart
kahlo_chart = compute_chart({
"year": 1907, "month": 7, "day": 6, "hour": 8, "minute": 30,
"lat": 19.3333, "lon": -99.1667, "tz": -6.6111,
"ayanamsa": "lahiri", "node_mode": "mean",
})
selected = select_similar_public_cases(kahlo_chart, ["health"])
assert [case["case_id"] for case in selected["cases"]] == ["kahlo_bus_injury_1925"]
assert selected["cases"][0]["reference_status"] == "public_context_only"
assert selected["coverage"]["available_event_domains"] == ["career", "health", "marriage", "wealth"]
def test_consultation_api_exposes_reference_transparency_contract() -> None:
source = (ROOT / "scripts" / "jyotish_api_server.py").read_text(encoding="utf-8")
assert "result['reference_transparency'] = build_reference_transparency_contract(" in source
assert "reference_date=_consultation_reference_date(body).date().isoformat()" in source
@@ -0,0 +1,32 @@
from pathlib import Path
PAGE = Path("frontend/src/app/page.tsx")
STYLES = Path("frontend/src/app/globals.css")
def test_chat_history_management_actions_are_exposed() -> None:
source = PAGE.read_text(encoding="utf-8") + STYLES.read_text(encoding="utf-8")
for expected in (
"renameSession",
"deleteSession",
"togglePinnedSession",
"toggleArchivedSession",
"showArchivedSessions",
"shareSession",
"share_payload_version",
"messages.map",
"onContextMenu",
"session-menu-trigger",
'role="menu"',
"closeSessionMenu",
"Escape",
"focus-within",
"置顶",
"重命名",
"归档",
"恢复",
"删除",
"转发",
):
assert expected in source
+4 -1
View File
@@ -136,7 +136,10 @@ class TestDigBala:
assert 0 < calc_dig_bala('Sun', 7) < 60
def test_dig_bala_jupiter_synthetic_north_china_case_needs_better_than_house_only_linear_model(self):
comparison = compare_case("references/oracle/dasha_shadbala_oracle_cases.json", "template_synthetic_north_china_shadbala_raman")
try:
comparison = compare_case("references/oracle/dasha_shadbala_oracle_cases.json", "template_synthetic_north_china_shadbala_raman")
except KeyError:
pytest.skip("synthetic North China oracle case is not distributed in the public release")
jupiter_dig_gap = comparison["comparison"]["Jupiter"]["components"]["dig"]["abs_diff_rupa"]
assert jupiter_dig_gap < 3.6348
+83
View File
@@ -0,0 +1,83 @@
from scripts.shadbala import (
build_shadbala_context,
calc_dig_bala_precise,
calc_drik_bala_precise,
calc_kala_bala_precise,
calc_chesta_bala_precise,
calc_sthana_bala_precise,
classify_drik_planets,
)
PLANET_LONGITUDES = {
"Sun": 312.51819523181405,
"Moon": 344.51168181221095,
"Mars": 5.862334338243072,
"Mercury": 291.12589362817465,
"Jupiter": 87.27055150216494,
"Venus": 267.94211928370015,
"Saturn": 207.92738390514674,
}
PYJHORA_DIG = {"Sun": 24.81, "Moon": 95.48, "Mars": 77.41, "Mercury": 12.64, "Jupiter": 80.60, "Venus": 69.95, "Saturn": 19.62}
PYJHORA_DRIK = {"Sun": 1.74, "Moon": 20.06, "Mars": 13.49, "Mercury": -10.63, "Jupiter": 6.89, "Venus": -2.75, "Saturn": 9.49}
PYJHORA_STHANA = {"Sun": 217.09, "Moon": 246.34, "Mars": 198.63, "Mercury": 167.96, "Jupiter": 201.80, "Venus": 154.06, "Saturn": 218.61}
PYJHORA_KALA = {"Sun": 140.99, "Moon": 149.58, "Mars": 127.93, "Mercury": 121.78, "Jupiter": 199.31, "Venus": 113.44, "Saturn": 136.77}
def _planets() -> dict:
return {
planet: {
"degree": longitude,
"sign": ["Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces"][int(longitude // 30)],
}
for planet, longitude in PLANET_LONGITUDES.items()
}
def test_precise_dig_bala_matches_pyjhora_same_chart() -> None:
context = build_shadbala_context(2435163.6354166665, 37.7749, -122.4194, "lahiri", -8.0)
for planet, longitude in PLANET_LONGITUDES.items():
assert calc_dig_bala_precise(planet, longitude, context["house_midpoints"]) == PYJHORA_DIG[planet]
def test_drik_classification_matches_pyjhora_same_chart() -> None:
benefics, malefics = classify_drik_planets(_planets())
assert benefics == {"Moon", "Mercury", "Jupiter", "Venus"}
assert malefics == {"Sun", "Mars", "Saturn"}
def test_precise_drik_bala_matches_pyjhora_same_chart() -> None:
planets = _planets()
for planet, expected in PYJHORA_DRIK.items():
assert abs(calc_drik_bala_precise(planet, planets) - expected) <= 0.02
def test_precise_sthana_bala_matches_pyjhora_same_chart() -> None:
planets = _planets()
houses = {"Sun": 7, "Moon": 8, "Mars": 9, "Mercury": 6, "Jupiter": 11, "Venus": 5, "Saturn": 3}
for planet, expected in PYJHORA_STHANA.items():
assert calc_sthana_bala_precise(planet, planets, houses[planet])["total"] == expected
def test_precise_kala_bala_matches_pyjhora_same_chart() -> None:
planets = _planets()
context = build_shadbala_context(2435163.6354166665, 37.7749, -122.4194, "lahiri", -8.0)
assert abs(context["sunrise_hour"] - 6.886306) < 0.00001
assert abs(context["sunset_hour"] - 17.890427) < 0.00001
for planet, expected in PYJHORA_KALA.items():
assert calc_kala_bala_precise(planet, planets, context)["total"] == expected
def test_precise_chesta_uses_classical_bounded_variant() -> None:
planets = _planets()
context = build_shadbala_context(2435163.6354166665, 37.7749, -122.4194, "lahiri", -8.0)
kala = {planet: calc_kala_bala_precise(planet, planets, context) for planet in planets}
result = {planet: calc_chesta_bala_precise(planet, planets, kala[planet], context) for planet in planets}
assert result["Sun"] == kala["Sun"]["ayana"]
assert result["Moon"] == kala["Moon"]["paksha"]
for planet in ("Mars", "Mercury", "Jupiter", "Venus", "Saturn"):
assert 0.0 <= result[planet] <= 60.0
vedastro = {"Mars": 21.7377, "Mercury": 47.5107, "Jupiter": 46.0549, "Venus": 30.8469, "Saturn": 39.4390}
tolerances = {"Mars": 1.0, "Mercury": 2.0, "Jupiter": 1.0, "Venus": 2.0, "Saturn": 3.5}
for planet, expected in vedastro.items():
assert abs(result[planet] - expected) <= tolerances[planet]
+10 -1
View File
@@ -8,11 +8,20 @@ import subprocess
import sys
from pathlib import Path
import scripts.skill_gap_truth_audit as audit
ROOT = Path(__file__).resolve().parents[1]
REGISTRY = ROOT / "references" / "skill_gap_truth_registry.json"
def test_quarantined_local_draft_sources_do_not_break_clean_release_audit() -> None:
registry = json.loads(REGISTRY.read_text(encoding="utf-8"))
problems = audit._validate_registry(registry)
assert not any(problem.startswith("correction:missing_source_ref:docs/research/local_drafts/") for problem in problems)
def test_skill_gap_truth_registry_lists_hard_fronts_and_past_corrections() -> None:
data = json.loads(REGISTRY.read_text(encoding="utf-8"))
@@ -68,7 +77,7 @@ def test_skill_gap_truth_audit_outputs_current_truth_boundary() -> None:
assert report["summary"]["capability_valid"] is True
assert report["summary"]["hard_front_count"] >= 5
assert report["summary"]["pyjhora_artifact_count"] >= 8
assert report["summary"]["pyjhora_packet_count"] >= 8
assert report["summary"]["pyjhora_packet_count"] >= 6
assert report["public_claim"]["can_claim_global_first"] is False
assert report["public_claim"]["can_claim_all_skills_complete"] is False
assert report["public_claim"]["can_claim_perfect_accuracy"] is False
+137 -3
View File
@@ -23,7 +23,25 @@ CONSULTATION_MIGRATION = (
/ "migrations"
/ "20260717000000_consultation_request_lifecycle.sql"
)
CHART_PROFILE_MIGRATION = (
Path(__file__).resolve().parents[1]
/ "frontend"
/ "supabase"
/ "migrations"
/ "20260717010000_chart_profiles.sql"
)
SYNASTRY_REPORT_MIGRATION = (
Path(__file__).resolve().parents[1]
/ "frontend"
/ "supabase"
/ "migrations"
/ "20260717020000_synastry_reports.sql"
)
PAGE = Path(__file__).resolve().parents[1] / "frontend" / "src" / "app" / "page.tsx"
CHART_PROFILE_ROUTE = Path(__file__).resolve().parents[1] / "frontend" / "src" / "app" / "api" / "chart-profiles" / "route.ts"
CHART_PROFILE_DELETE_ROUTE = Path(__file__).resolve().parents[1] / "frontend" / "src" / "app" / "api" / "chart-profiles" / "[id]" / "route.ts"
SYNASTRY_ROUTE = Path(__file__).resolve().parents[1] / "frontend" / "src" / "app" / "api" / "synastry" / "route.ts"
SYNASTRY_REPORT_ROUTE = Path(__file__).resolve().parents[1] / "frontend" / "src" / "app" / "api" / "synastry-reports" / "route.ts"
def _sql() -> str:
@@ -92,9 +110,125 @@ def test_chat_page_uses_authenticated_cloud_persistence() -> None:
assert 'await persistSession(interruptedSession)' in source
assert 'await persistence' in source
assert 'disabled={Boolean(pendingSessionId) || cancellationPending}' in source
assert "localStorage" not in source
assert "ayanam-profile" not in source
assert "ayanam-sessions" not in source
assert 'localStorage.setItem(chartLibraryStorageKey(accountId)' in source
assert 'localStorage.setItem("chat_sessions"' not in source
def test_chart_profile_library_has_cloud_table_api_and_local_fallback() -> None:
sql = re.sub(r"\s+", " ", CHART_PROFILE_MIGRATION.read_text(encoding="utf-8").lower()).strip()
route = CHART_PROFILE_ROUTE.read_text(encoding="utf-8")
delete_route = CHART_PROFILE_DELETE_ROUTE.read_text(encoding="utf-8")
page = PAGE.read_text(encoding="utf-8")
for token in (
"create table if not exists public.chart_profiles",
"user_id uuid not null references auth.users(id) on delete cascade",
"role text not null default 'other' check (role in ('self', 'other'))",
"profile jsonb not null check (jsonb_typeof(profile) = 'object')",
"create unique index if not exists chart_profiles_one_self_per_user_idx",
"alter table public.chart_profiles enable row level security",
"create policy chart_profiles_select_own",
"create policy chart_profiles_insert_own",
"create policy chart_profiles_update_own",
"create policy chart_profiles_delete_own",
"grant delete on table public.chart_profiles to authenticated",
):
assert token in sql
for token in (
'from("chart_profiles")',
'eq("user_id", user.id)',
'eq("role", "self")',
'insert({ user_id: user.id, role, profile: body.profile',
'upsert(record, { onConflict: "id" })',
):
assert token in route
assert 'eq("role", "other")' in delete_route
for token in (
"fetchCloudChartLibrary",
"saveCloudChartProfile",
"deleteCloudChartProfile",
"buildSynastryQuestion",
"draftSynastryQuestionFromChart",
"synastryReportCard",
"synastryHistory",
"synastryHistoryStorageKey",
"synastry-report-card",
"synastry-history-list",
'fetch("/api/synastry"',
"Ashtakoot",
"chartLibraryStorageKey",
"Cloud chart library is best-effort",
"星盘库",
"添加其他星盘",
"用于合盘",
"设为默认",
):
assert token in page
assert "ayanam-profile" not in page
assert "ayanam-sessions" not in page
def test_synastry_route_orchestrates_python_chart_and_ashtakoot() -> None:
route = SYNASTRY_ROUTE.read_text(encoding="utf-8")
for token in (
'const apiBase = process.env.JYOTISH_API_BASE ?? "http://127.0.0.1:5200"',
'postPython("/api/chart", birthPayload(body.selfProfile))',
'postPython("/api/chart", birthPayload(body.partnerProfile))',
'postPython("/api/varga_full"',
'postPython("/api/synastry"',
"moonLongitude(selfChart)",
"moonSummary(selfChart)",
"d9Summary(selfD9)",
"relationshipReport",
"scoreBand",
"nextEvidence",
"ashtakoot_plus_moon_nakshatra_d9",
'evidenceLayers: ["ashtakoot", "moon_nakshatra", "d9_navamsa"]',
'status: "blocked"',
):
assert token in route
def test_synastry_reports_are_cloud_persisted_per_user() -> None:
sql = re.sub(r"\s+", " ", SYNASTRY_REPORT_MIGRATION.read_text(encoding="utf-8").lower()).strip()
route = SYNASTRY_REPORT_ROUTE.read_text(encoding="utf-8")
page = PAGE.read_text(encoding="utf-8")
for token in (
"create table if not exists public.synastry_reports",
"user_id uuid not null references auth.users(id) on delete cascade",
"partner_name text not null default '对方'",
"report jsonb not null check (jsonb_typeof(report) = 'object')",
"alter table public.synastry_reports enable row level security",
"create policy synastry_reports_select_own on public.synastry_reports for select to authenticated using ((select auth.uid()) = user_id)",
"create policy synastry_reports_insert_own on public.synastry_reports for insert to authenticated with check ((select auth.uid()) = user_id)",
"create policy synastry_reports_delete_own on public.synastry_reports for delete to authenticated using ((select auth.uid()) = user_id)",
"grant select on table public.synastry_reports to authenticated",
"grant insert (id, user_id, partner_name, report, created_at) on table public.synastry_reports to authenticated",
):
assert token in sql
for token in (
'.from("synastry_reports")',
'.select("id, partner_name, report, created_at")',
".eq(\"user_id\", user.id)",
".limit(10)",
"partner_name: partnerName",
"report: body.report",
):
assert token in route
for token in (
"fetchCloudSynastryHistory",
"saveCloudSynastryReport",
'fetch("/api/synastry-reports"',
"writeSynastryHistory(accountId, next)",
"cloud_synastry_history_unavailable",
"cloud_synastry_report_save_failed",
):
assert token in page
def test_consultation_credit_lifecycle_is_idempotent_and_server_only() -> None:
@@ -0,0 +1,56 @@
import os
from pathlib import Path
import subprocess
import sys
from benchmarks.jyotish.scripts.run_skill_baseline import run_sample
from benchmarks.jyotish.scripts.run_pyjhora_compare import build_pyjhora_sample, compare_one
from scripts import three_engine_high_rigor_parity as parity
ROOT = Path(__file__).resolve().parents[1]
def test_runner_imports_from_commercial_root() -> None:
env = os.environ.copy()
env["PYTHONPATH"] = str(ROOT)
result = subprocess.run(
[sys.executable, "-c", "import scripts.three_engine_high_rigor_parity"],
cwd=ROOT,
env=env,
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0, result.stderr
def test_public_sample_exposes_shadbala_components() -> None:
result = run_sample(parity.SAMPLE)
assert result["ok"] is True
components = result["canonical"]["shadbala_components"]
assert set(components["Sun"]) == set(parity.VED_COMPONENT_FIELDS)
def test_pyjhora_sample_exposes_shadbala_components() -> None:
sample = build_pyjhora_sample(parity.SAMPLE)
assert set(sample["shadbala_components"]["Sun"]) == set(parity.VED_COMPONENT_FIELDS)
def test_pyjhora_matrix_includes_shadbala_components() -> None:
baseline = {
"ascendant": {}, "planets": {}, "varga": {}, "ashtakavarga": {}, "shadbala": {}, "dasha": {},
"shadbala_components": {"Sun": {name: 1.0 for name in parity.VED_COMPONENT_FIELDS}},
}
rows = compare_one("sample", baseline, baseline)
assert {row["field"] for row in rows if row["section"] == "Shadbala_Component" and row["body"] == "Sun"} == set(parity.VED_COMPONENT_FIELDS)
def test_jyotishganit_planet_map_reads_divisional_occupants() -> None:
chart = {"houses": [{"occupants": [{"celestialBody": "Sun", "sign": "Leo"}]}]}
assert parity._jyotish_planet_signs(chart) == {"Sun": "Leo"}
def test_vedastro_components_read_six_raw_fields() -> None:
raw = {"Payload": {"AllPlanetData": {field: index for index, field in enumerate(parity.VED_COMPONENT_FIELDS.values(), 1)}}}
assert parity._ved_components(raw) == {name: index for index, name in enumerate(parity.VED_COMPONENT_FIELDS, 1)}
@@ -0,0 +1,35 @@
import json
from pathlib import Path
from scripts.three_engine_mismatch_arbitrator import arbitrate_manifest
ROOT = Path(__file__).resolve().parents[1]
def test_arbitrator_classifies_every_mismatch_without_majority_truth() -> None:
report = arbitrate_manifest(ROOT / "references/oracle/three_engine_parity_replay_manifest.json")
assert report["mismatch_count"] == 60
assert report["classified_count"] == 60
assert report["unclassified_count"] == 0
assert report["truth_policy"] == "no_majority_vote"
assert sum(report["category_counts"].values()) == 60
assert all(row["differing_engines"] for row in report["rows"])
def test_vedastro_only_d2_difference_is_endpoint_semantics(tmp_path: Path) -> None:
manifest = {
"engines": {name: {} for name in ("VedAstro", "PyJHora_JHora", "jyotishganit")},
"comparison_rows": [{
"section": "D2", "field": "Sun.sign", "local_value": "Leo",
"oracle_values": {"VedAstro": "Gemini", "PyJHora_JHora": "Leo", "jyotishganit": "Leo"},
"status": "mismatch",
}],
}
path = tmp_path / "manifest.json"
path.write_text(json.dumps(manifest), encoding="utf-8")
row = arbitrate_manifest(path)["rows"][0]
assert row["category"] == "endpoint_or_varga_semantics"
assert row["differing_engines"] == ["VedAstro"]
@@ -12,14 +12,16 @@ from scripts.three_engine_parity_replay_validator import validate_manifest
ROOT = Path(__file__).resolve().parents[1]
def test_three_engine_parity_manifest_passes_d1_with_vedastro_and_pyjhora_raw() -> None:
def test_three_engine_parity_manifest_exposes_raw_coverage_and_formula_mismatches() -> None:
result = validate_manifest(ROOT / "references/oracle/three_engine_parity_replay_manifest.json")
assert result["status"] == "pass"
assert result["comparison_row_count"] == 15
assert result["match_count"] == 14
assert result["status"] == "mismatch"
assert result["comparison_row_count"] == 92
assert result["match_count"] == 32
assert result["mismatch_count"] == 60
assert result["tested"] is True
assert result["blocked_reason"] is None
assert result["missing_high_rigor_sections"] == []
def test_three_engine_parity_validator_accepts_one_same_chart_row(tmp_path: Path) -> None:
@@ -54,7 +56,8 @@ def test_three_engine_parity_validator_accepts_one_same_chart_row(tmp_path: Path
result = validate_manifest(path)
assert result["status"] == "pass"
assert result["status"] == "partial"
assert result["blocked_reason"] == "missing_high_rigor_sections"
assert result["tested"] is True
assert result["comparison_row_count"] == 1
assert result["match_count"] == 1
@@ -0,0 +1,38 @@
from scripts.timing_negative_holdout_audit import build_report
def test_timing_negative_holdout_blocks_positive_only_sources() -> None:
report = build_report(
[
{
"name": "positive-only public timeline",
"positive_events": True,
"explicit_non_event_intervals": False,
"independent_human_reviewed": False,
"observed_before_preregistration": False,
}
]
)
assert report["claim_status"] == "exploratory_unvalidated"
assert report["production_tuning_allowed"] is False
assert report["sources"][0]["usable_for_promotion"] is False
assert "missing_explicit_non_event_intervals" in report["sources"][0]["blockers"]
def test_timing_negative_holdout_can_only_promote_locked_independent_labels() -> None:
report = build_report(
[
{
"name": "locked blind label packet",
"positive_events": True,
"explicit_non_event_intervals": True,
"independent_human_reviewed": True,
"observed_before_preregistration": False,
}
]
)
assert report["claim_status"] == "ready_for_blind_holdout"
assert report["production_tuning_allowed"] is True
assert report["sources"][0]["blockers"] == []
+50
View File
@@ -0,0 +1,50 @@
from scripts.timing_precision_contract import build_timing_precision_contract
import json
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_timing_contract_preserves_dates_but_caps_claim_status() -> None:
contract = build_timing_precision_contract({
"verified_window": "2026-08 to 2026-10",
"candidate_windows": [{
"start": "2026-08-12", "end": "2026-08-16", "rank": 1,
"signals": ["Vimshottari", "Narayana", "D10", "transit"],
"confidence_cap": "low",
}],
"exact_triggers": [{"at": "2026-08-14T09:30:00Z", "technique": "exact_transit"}],
})
assert contract["timing_precision"] == "candidate_day_window"
assert contract["claim_status"] == "exploratory_unvalidated"
assert contract["verified_window"] == "2026-08 to 2026-10"
assert contract["candidate_windows"][0]["start"] == "2026-08-12"
assert contract["exact_triggers"][0]["at"] == "2026-08-14T09:30:00Z"
assert contract["promotion_gate"]["status"] == "blocked"
def test_empty_timing_contract_still_exposes_boundary() -> None:
contract = build_timing_precision_contract()
assert contract["timing_precision"] == "broad_window_only"
assert contract["candidate_windows"] == []
assert "不能作为确定事件承诺" in contract["boundary"]
def test_predict_cli_exposes_exploratory_timing_contract() -> None:
completed = subprocess.run(
[sys.executable, "scripts/jyotish_engine.py", "predict", "--chart", '{"planets":{}}',
"--lat", "0", "--lon", "0", "--tz", "0"],
cwd=ROOT, text=True, capture_output=True, timeout=30, check=False,
)
assert completed.returncode == 0, completed.stderr
contract = json.loads(completed.stdout)["timing_precision_contract"]
assert contract["claim_status"] == "exploratory_unvalidated"
assert contract["promotion_gate"]["current_negative_controls_reusable_for_tuning"] is False
def test_consultation_api_attaches_same_contract() -> None:
source = (ROOT / "scripts/jyotish_api_server.py").read_text(encoding="utf-8")
assert source.count("result['timing_precision_contract'] = build_timing_precision_contract") == 2
@@ -0,0 +1,12 @@
from scripts.vedastro_contract_arbitrator import arbitrate
def test_arbitrator_blocks_when_method_contract_changes_across_runs() -> None:
runs = [
{"contract_status": "blocked", "method_contract": {"status": "blocked"}, "time_contract": {"equivalent_local_utc": True}, "api_version_contract": {"status": "blocked"}, "normalized_vectors": {"x": {"Sun": 1}}},
{"contract_status": "blocked", "method_contract": {"status": "resolved"}, "time_contract": {"equivalent_local_utc": True}, "api_version_contract": {"status": "blocked"}, "normalized_vectors": {"x": {"Sun": 2}}},
]
report = arbitrate(runs)
assert report["status"] == "blocked"
assert report["cross_run_normalized_stable"] is False
assert report["method_statuses"] == ["blocked", "resolved"]
+30
View File
@@ -0,0 +1,30 @@
from scripts import vedastro_contract_probe as probe
def test_equivalent_time_gate_uses_normalized_values_not_raw_hashes() -> None:
runs = {
"local_minus_08": [{"normalized": {"Sun": 312.5}, "raw_hash": "a"}, {"normalized": {"Sun": 312.5}, "raw_hash": "b"}],
"utc_equivalent": [{"normalized": {"Sun": 312.5}, "raw_hash": "c"}],
"positive_08_control": [{"normalized": {"Sun": 100.0}, "raw_hash": "d"}],
}
result = probe.evaluate_time_contract(runs)
assert result["equivalent_local_utc"] is True
assert result["positive_offset_control_distinct"] is True
assert result["repeat_normalized_stable"] is True
assert result["raw_hash_stable"] is False
def test_method_gate_blocks_ambiguous_all_planet_longitude() -> None:
result = probe.evaluate_method_contract(
all_planet={"Sun": 311.84},
all_planet_data={"Sun": 312.51},
nirayana={"Sun": 312.51},
tolerance=0.01,
)
assert result["all_planet_data_matches_nirayana"] is True
assert result["all_planet_longitude_matches_nirayana"] is False
assert result["status"] == "blocked"
def test_redact_headers_never_persists_credentials() -> None:
assert probe.redact_headers({"x-api-key": "secret", "Server": "Kestrel"}) == {"server": "Kestrel"}
+17 -5
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import json
def test_gateway_status_defaults_to_local_first_cn_safe(monkeypatch):
def test_gateway_status_defaults_to_official_first_cn_safe(monkeypatch):
from scripts import vedastro_gateway
monkeypatch.delenv("VEDASTRO_GATEWAY_MODE", raising=False)
@@ -17,13 +17,12 @@ def test_gateway_status_defaults_to_local_first_cn_safe(monkeypatch):
status = vedastro_gateway.gateway_status()
assert status["scope"] == "vedastro_gateway"
assert status["mode"] == "local_first"
assert status["mode"] == "official_first"
assert status["direct_browser_access_allowed"] is False
assert status["frontend_secret_safe"] is True
assert status["backend_priority"] == ["self_host", "official", "cache", "queue", "local_fallback"]
assert status["active_backend"] == "local_fallback"
assert status["official_readiness"]["official_ready"] is False
assert "missing_endpoint" in status["official_readiness"]["readiness_blockers"]
assert status["active_backend"] == "official"
assert status["official_readiness"]["official_ready"] is True
assert status["boundary"] == "Users never call VedAstro directly; backend gateway owns cache, queue, and fallback."
@@ -261,6 +260,19 @@ def test_gateway_completion_archives_official_raw_response(monkeypatch, tmp_path
assert '"vedastro_official"' in path.read_text(encoding="utf-8")
def test_gateway_archives_official_full_snapshot_raw_response(monkeypatch, tmp_path):
from scripts import vedastro_gateway
monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_DIR", str(tmp_path))
job = vedastro_gateway.enqueue_gateway_job({"year": 1990}, question="public validation")
completed = vedastro_gateway.complete_gateway_job(
job["job_id"],
{"raw_response": {"source": "vedastro_official_full_snapshot", "response_hash": "public-demo"}},
)
assert completed["raw_response_archive"]["official_raw_response_available"] is True
def test_gateway_lists_official_raw_response_archives(monkeypatch, tmp_path):
from scripts import vedastro_gateway
+38
View File
@@ -0,0 +1,38 @@
from scripts.vedastro_identity_archive import build_archive
def test_vedastro_identity_archive_records_self_host_identity(monkeypatch) -> None:
fixtures = {
"https://api.nuget.org/v3/registration5-semver1/vedastro.library/1.2.0.json": {
"catalogEntry": "https://example.invalid/catalog.json",
"packageContent": "https://example.invalid/pkg.nupkg",
"published": "2023-03-22T22:22:01.81+00:00",
},
"https://example.invalid/catalog.json": {
"packageHashAlgorithm": "SHA512",
"packageHash": "hash",
"packageSize": 230836,
"published": "2023-03-22T22:22:01.81Z",
"catalog:commitId": "commit",
"catalog:commitTimeStamp": "2023-03-22T22:24:11.811524Z",
"licenseExpression": "MIT",
"projectUrl": "https://vedastro.org/",
"dependencyGroups": [
{
"targetFramework": "net7.0",
"dependencies": [{"id": "SwissEphNet", "range": "[2.8.0.2, )"}],
}
],
},
}
monkeypatch.setattr("scripts.vedastro_identity_archive._json_url", lambda url, timeout: fixtures[url])
report = build_archive("1.2.0")
assert report["package_hash_algorithm"] == "SHA512"
assert report["license"] == "MIT"
assert report["self_host_candidate_status"] == "reproducible_package_identity_archived"
assert report["hosted_api_status"] == "blocked"
assert report["dependencies"] == [
{"target_framework": "net7.0", "id": "SwissEphNet", "range": "[2.8.0.2, )"}
]
@@ -0,0 +1,21 @@
from pathlib import Path
from scripts.vedastro_official_source_contract import audit_source
def test_official_source_contract_resolves_zero_offset_semantics(tmp_path: Path) -> None:
(tmp_path / "Library/Data").mkdir(parents=True)
(tmp_path / "Library/Logic/Calculate").mkdir(parents=True)
(tmp_path / "Library/Logic").mkdir(parents=True, exist_ok=True)
(tmp_path / "Library/Library.csproj").write_text("<Version>1.2.0</Version><PackageLicenseExpression>MIT</PackageLicenseExpression>")
(tmp_path / "Library/Data/Time.cs").write_text('DateTimeOffset.ParseExact(stdDateTimeText, Time.DateTimeFormat, null); public const string DateTimeFormat = "HH:mm dd/MM/yyyy zzz";')
(tmp_path / "Library/Logic/Tools.cs").write_text('var parsedTimezone = Tools.StringToTimezone(offsetStr); if (parsedTimezone == null || parsedTimezone == TimeSpan.Zero) offsetStr = await Calculate.GeoLocationToTimezone(geoLocation, parsedInputTime); DateTimeOffset.ParseExact(timezoneRaw, "zzz", CultureInfo.InvariantCulture).Offset;')
(tmp_path / "Library/Logic/Calculate/Core.cs").write_text("public static List<PlanetLongitude> AllPlanetLongitude(Time time) { return PlanetNirayanaLongitude(Sun, time); }")
report = audit_source(tmp_path)
assert report["source_contract_status"] == "verified"
assert report["timezone_contract"]["zero_offset"] == "auto_lookup_sentinel"
assert report["timezone_contract"]["negative_offset"] == "literal_offset"
assert report["longitude_contract"]["AllPlanetLongitude"] == "nirayana"
assert report["deployment_identity_status"] == "blocked"
@@ -37,7 +37,7 @@ def test_vedastro_diagnostics_reports_fast_fallback_mode_without_endpoint() -> N
assert "VEDASTRO_API_ENDPOINT" in report["missing"]
assert "missing_endpoint" in report["readiness_blockers"]
assert "network_disabled" in report["readiness_blockers"]
assert "timeout_too_low" in report["readiness_blockers"]
assert "timeout_too_low" not in report["readiness_blockers"]
assert report["expected_fallback_status"] == "official_snapshot_budget_exhausted_or_endpoint_blocked"
assert "VEDASTRO_API_ENDPOINT" in report["official_closure_plan"]["required_env"]
assert report["official_closure_plan"]["raw_response_acceptance"] == "vedastro_official.raw_response must be present before claiming official cloud closure."
+37
View File
@@ -0,0 +1,37 @@
from scripts.vp_jain_shadbala_benchmark import build_report
def test_vp_jain_published_component_benchmark_is_fully_classified() -> None:
report = build_report()
assert report["case"]["name"] == "VP Jain-attributed Shadbala fixture"
assert report["source"] == {
"upstream": "PyJHora V4.8.7 pvr_tests.py::shadbala_VPJainBook_tests",
"upstream_commit": "ca22995709bd60e371e7820a1a5efc80ce4cf821",
"upstream_url": "https://github.com/naturalstupid/PyJHora/blob/ca22995709bd60e371e7820a1a5efc80ce4cf821/src/jhora/tests/pvr_tests.py#L6853",
"upstream_issue": None,
"license_boundary": "AGPL-3.0 process-isolated numeric expectations only; no implementation copied.",
"truth_status": "candidate_attributed_fixture_replay_not_independent_arbitration",
}
assert report["summary"]["row_count"] == 42
assert report["summary"]["classified_count"] == 42
assert report["summary"]["within_tolerance_count"] == 34
assert report["summary"]["method_variant_count"] == 8
def test_vp_jain_benchmark_keeps_known_method_variants_explicit() -> None:
rows = build_report()["rows"]
variants = {(row["component"], row["planet"]): row for row in rows if row["status"] != "within_tolerance"}
assert set(variants) == {
("sthana", "Sun"),
("sthana", "Mercury"),
("chesta", "Sun"),
("chesta", "Moon"),
("chesta", "Mars"),
("chesta", "Mercury"),
("chesta", "Jupiter"),
("chesta", "Saturn"),
}
assert variants[("sthana", "Sun")]["variant"] == "moolatrikona_degree_range_vs_whole_sign"
assert variants[("chesta", "Moon")]["variant"] == "luminary_chesta_policy"
@@ -0,0 +1,17 @@
from scripts.western_real_case_calibration import build_case
def test_jobs_public_case_replays_progressed_angles_and_parans() -> None:
report = build_case("jobs_iphone_2007")
assert report["status"] == "calculated_not_predictive_validation"
assert report["event"]["source"]["source_grade"] == "primary"
assert report["birth_source"]["time_accuracy_rating"] == "AA"
assert report["layers"]["secondary_progressions"]["progressed_angles"]["status"] == "used"
assert report["layers"]["parans"]["status"] == "used"
assert report["summary"]["paran_event_count"] > 0
assert "cannot establish" in report["boundary"]
def test_unknown_case_is_blocked() -> None:
assert build_case("missing")["reason"] == "case_id_not_found"
+11 -2
View File
@@ -97,10 +97,12 @@ def test_transit_duration_scan_groups_daily_windows() -> None:
assert isinstance(scan["windows"], list)
def test_parans_are_explicitly_blocked_until_solver_exists() -> None:
def test_parans_emit_latitude_aware_angular_events() -> None:
parans = calculate_parans_status(**_BIRTH, target_date="2026-07-09")
assert parans["technique"] == "parans"
assert parans["status"] == "blocked"
assert parans["status"] == "used"
assert parans["method"].startswith("Swiss Ephemeris rise_trans")
assert parans["event_count"] > 0
def test_timing_builder_can_emit_advanced_layers() -> None:
@@ -122,3 +124,10 @@ def test_timing_builder_can_emit_advanced_layers() -> None:
"transit_duration_scan",
"parans",
} <= set(timing)
progressed = build_timing_techniques(**_BIRTH, secondary_progression_date="2026-07-09")["secondary_progressions"]
assert progressed["progressed_angles"]["status"] == "used"
assert set(progressed["progressed_angles"]["angles"]) == {"ascendant", "mc", "descendant", "ic"}
parans = timing["parans"]
assert parans["status"] == "used"
assert parans["event_count"] > 0
assert all(row["separation_minutes"] <= 4 for row in parans["paran_pairs_within_4_minutes"])
+51
View File
@@ -0,0 +1,51 @@
import json
from scripts.xalen_arbitration_gate import build_report
def test_xalen_arbitration_gate_blocks_unresolved_method_variants(tmp_path) -> None:
attribution = tmp_path / "attr.json"
public_batch = tmp_path / "batch.json"
ephemeris = tmp_path / "ephem.json"
attribution.write_text(
json.dumps(
{
"rows": [
{
"field": "Sun.chesta",
"truth_status": "method_variant_unresolved",
}
]
}
),
encoding="utf-8",
)
public_batch.write_text(json.dumps({"case_count": 5}), encoding="utf-8")
ephemeris.write_text(
json.dumps({"maximum_absolute_longitude_delta_deg": 0.001, "varga_difference_count": 0}),
encoding="utf-8",
)
report = build_report(attribution, public_batch, ephemeris)
assert report["multi_case_replay_status"] == "ready"
assert report["independent_ephemeris_status"] == "ready"
assert report["truth_status"] == "blocked"
assert report["promotion_allowed"] is False
def test_xalen_arbitration_gate_allows_only_when_variants_are_arbitrated(tmp_path) -> None:
attribution = tmp_path / "attr.json"
public_batch = tmp_path / "batch.json"
ephemeris = tmp_path / "ephem.json"
attribution.write_text(json.dumps({"rows": [{"truth_status": "external_worked_example_match"}]}), encoding="utf-8")
public_batch.write_text(json.dumps({"case_count": 5}), encoding="utf-8")
ephemeris.write_text(
json.dumps({"maximum_absolute_longitude_delta_deg": 0.001, "varga_difference_count": 0}),
encoding="utf-8",
)
report = build_report(attribution, public_batch, ephemeris)
assert report["truth_status"] == "ready"
assert report["promotion_allowed"] is True
@@ -0,0 +1,13 @@
from pathlib import Path
from scripts.xalen_difference_attribution import build_report
ROOT=Path(__file__).resolve().parents[1]
def test_all_xalen_shadbala_and_av_differences_are_formula_unit_attributed() -> None:
report=build_report(ROOT/"references/oracle/xalen_fourth_oracle_comparison_2026_07_17.json")
assert report["row_count"]==46
assert report["classified_count"]==46
assert sum(report["category_counts"].values())==46
assert all(row["unit"] in {"Virupa","bindu_count"} for row in report["rows"])
av=[r for r in report["rows"] if r["unit"]=="bindu_count"]
assert len(av)==4
assert all(r["row_total_local"]==r["row_total_xalen"] for r in av)
@@ -0,0 +1,10 @@
from pathlib import Path
from scripts.xalen_ephemeris_mode_comparison import compare
ROOT=Path(__file__).resolve().parents[1]
def test_independent_xalen_ephemeris_is_separately_disclosed() -> None:
r=compare(ROOT/'references/oracle/artifacts/xalen_steve_jobs_high_rigor_raw.json',ROOT/'references/oracle/artifacts/xalen_steve_jobs_independent_ephemeris_raw.json')
assert r['shared_mode']=='shared_input'
assert r['independent_mode']=='independent_ephemeris'
assert r['maximum_absolute_longitude_delta_deg']<0.02
assert r['varga_difference_count']==0
assert 'house numbers remain shared' in r['boundary']
+9
View File
@@ -0,0 +1,9 @@
from scripts.xalen_oracle_adapter import COMMIT, MANIFEST
def test_xalen_adapter_is_pinned_and_license_bounded() -> None:
text = MANIFEST.read_text(encoding="utf-8")
assert COMMIT in text
assert "xalen-vedic" in text
assert len(COMMIT) == 40
assert "independent_ephemeris" in (MANIFEST.parent.parent.parent / "scripts/xalen_oracle_adapter.py").read_text(encoding="utf-8")
+11
View File
@@ -0,0 +1,11 @@
from pathlib import Path
from scripts.xalen_oracle_comparison import compare
ROOT=Path(__file__).resolve().parents[1]
def test_xalen_compares_all_high_rigor_rows_as_fourth_observation() -> None:
report=compare(ROOT/"references/oracle/three_engine_parity_replay_manifest.json",ROOT/"references/oracle/artifacts/xalen_steve_jobs_high_rigor_raw.json")
assert report["row_count"] == 92
assert report["match_count"] + report["mismatch_count"] == 92
assert report["truth_policy"] == "fourth_observation_not_truth"
assert report["license"] == "Apache-2.0"
+5
View File
@@ -0,0 +1,5 @@
from scripts.xalen_public_case_batch import PLANETS,SIGNS
def test_batch_contract_uses_seven_planets_and_twelve_signs() -> None:
assert PLANETS==['Sun','Moon','Mars','Mercury','Jupiter','Venus','Saturn']
assert len(SIGNS)==12