diff --git a/frontend/src/mastra/index.ts b/frontend/src/mastra/index.ts index 207285be..420cc387 100644 --- a/frontend/src/mastra/index.ts +++ b/frontend/src/mastra/index.ts @@ -88,6 +88,7 @@ export function toAgentConsultationContext(data: JsonRecord) { lagna_boundary: rectification.lagna_boundary, }, thematic_evidence: selectedTheme, + reference_transparency: record(data.reference_transparency), }; } @@ -107,6 +108,11 @@ Treat consumer_context as the authoritative answer policy: - When core_status is ready and can_answer_direction is true, answer the user's actual question directly. Do not begin with infrastructure or confidence disclaimers. - An unavailable optional provider or external cross-check is not a calculation failure. Never call it an internal error. - Do not mention VedAstro, snapshot, fallback, gateway, archive, provider, MEVG, or calibration unless the user explicitly asks about methodology, or the missing layer materially blocks the exact claim they requested. + +When reference_transparency is present: +- Present candidate_windows and exact_triggers when relevant, but describe exact_triggers as technical trigger points, never guaranteed events. +- Share a public case only when similar_public_cases.status is high_similarity_public_references_available. State the listed matching factors, dissimilar factors, event source URL, and that the case is reference-only. +- When method_variants applies, present parallel methods and their source paths rather than silently picking one result as the only truth. - If should_lead_with_limitations is false, do not lead with limitations. If a limitation is relevant, put it in one short sentence at the end. - Only say the chart calculation failed when hard_blockers is non-empty. - Never claim D9, D10, A10, UL, or Narayana Dasha is missing when it appears in available_layers or local_layers. diff --git a/frontend/tests/consultation-context.test.ts b/frontend/tests/consultation-context.test.ts new file mode 100644 index 00000000..c5bd6fc8 --- /dev/null +++ b/frontend/tests/consultation-context.test.ts @@ -0,0 +1,11 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; + +test("passes transparent public-case references into the agent context", () => { + const source = readFileSync(new URL("../src/mastra/index.ts", import.meta.url), "utf8"); + + assert.match(source, /reference_transparency:\s*record\(data\.reference_transparency\)/); + assert.match(source, /high_similarity_public_references_available/); + assert.match(source, /exact_triggers as technical trigger points/); +}); diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 8a5b6393..6a589912 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -452,6 +452,7 @@ def execute_consultation_workflow( chart_override: dict | None = None, ) -> dict: from scripts.timing_precision_contract import build_timing_precision_contract + from scripts.reference_transparency_contract import build_reference_transparency_contract birth_payload = handler._high_rigor_birth_payload(body) themes = handler._high_rigor_requested_themes(body) @@ -715,6 +716,11 @@ def execute_consultation_workflow( if body.get('return_high_rigor_shape'): result['endpoint'] = 'high_rigor_workflow' result['timing_precision_contract'] = build_timing_precision_contract(body.get('timing')) + result['reference_transparency'] = build_reference_transparency_contract( + chart, + themes, + timing=body.get('timing'), + ) return result diff --git a/scripts/reference_transparency_contract.py b/scripts/reference_transparency_contract.py new file mode 100644 index 00000000..dbebfc37 --- /dev/null +++ b/scripts/reference_transparency_contract.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""User-visible disclosure for timing, engine observations, and public case references.""" + +from __future__ import annotations + +import json +import sys +from functools import lru_cache +from pathlib import Path +from typing import Any + +SCRIPT_DIR = Path(__file__).resolve().parent +if str(SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPT_DIR)) + +try: + from scripts.domain_calculation_service import compute_chart + from scripts.timing_precision_contract import build_timing_precision_contract +except ModuleNotFoundError: # pragma: no cover - direct script execution + from domain_calculation_service import compute_chart + from timing_precision_contract import build_timing_precision_contract + + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_MANIFEST = ROOT / "references" / "real_case_calibration" / "replay_manifest.json" +DOMAIN_HOUSES = { + "career": "house_10", + "marriage": "house_7", + "wealth": "house_2", + "health": "house_6", +} +FEATURE_WEIGHTS = { + "ascendant": 0.35, + "moon_sign": 0.30, + "domain_lord_sign": 0.25, + "node_axis": 0.10, +} +HIGH_SIMILARITY_THRESHOLD = 0.75 + + +def _sign(chart: dict[str, Any], planet: str) -> str | None: + planets = chart.get("planets") if isinstance(chart, dict) else None + value = planets.get(planet) if isinstance(planets, dict) else None + return value.get("sign") if isinstance(value, dict) else None + + +def _domain_lord_sign(chart: dict[str, Any], domain: str) -> str | None: + house = DOMAIN_HOUSES.get(domain) + houses = chart.get("houses") if isinstance(chart, dict) else None + house_value = houses.get(house) if house and isinstance(houses, dict) else None + lord = house_value.get("lord") if isinstance(house_value, dict) else None + return _sign(chart, lord) if isinstance(lord, str) else None + + +def _features(chart: dict[str, Any], domain: str) -> dict[str, Any]: + ascendant = chart.get("ascendant") if isinstance(chart, dict) else None + return { + "ascendant": ascendant.get("sign") if isinstance(ascendant, dict) else None, + "moon_sign": _sign(chart, "Moon"), + "domain_lord_sign": _domain_lord_sign(chart, domain), + "node_axis": (_sign(chart, "Rahu"), _sign(chart, "Ketu")), + } + + +@lru_cache(maxsize=64) +def _case_chart(case_id: str, year: int, month: int, day: int, hour: int, minute: int, + lat: float, lon: float, tz: float, node_mode: str) -> dict[str, Any] | None: + try: + return compute_chart({ + "year": year, "month": month, "day": day, "hour": hour, "minute": minute, + "lat": lat, "lon": lon, "tz": tz, "ayanamsa": "lahiri", "node_mode": node_mode, + }) + except Exception: + return None + + +def _chart_for_case(case: dict[str, Any]) -> dict[str, Any] | None: + provided = case.get("chart") + if isinstance(provided, dict): + return provided + subject = case.get("subject") + if not isinstance(subject, dict): + return None + required = ("year", "month", "day", "hour", "minute", "lat", "lon", "tz") + if any(subject.get(field) is None for field in required): + return None + return _case_chart( + str(case.get("case_id", "")), int(subject["year"]), int(subject["month"]), int(subject["day"]), + int(subject["hour"]), int(subject["minute"]), float(subject["lat"]), float(subject["lon"]), + float(subject["tz"]), str(subject.get("node_mode", "mean")), + ) + + +def _similarity(user_chart: dict[str, Any], case_chart: dict[str, Any], domain: str) -> dict[str, Any]: + user = _features(user_chart, domain) + candidate = _features(case_chart, domain) + matching, dissimilar, total = [], [], 0.0 + for name, weight in FEATURE_WEIGHTS.items(): + if user[name] is None or candidate[name] is None: + continue + total += weight + if user[name] == candidate[name]: + matching.append(name) + else: + dissimilar.append(name) + score = round(sum(FEATURE_WEIGHTS[name] for name in matching) / total, 3) if total else 0.0 + return { + "score": score, + "matching_factors": matching, + "dissimilar_factors": dissimilar, + "feature_scope": "D1 ascendant, Moon, theme-house lord, and Rahu/Ketu axis only", + "uncompared_layers": ["D9", "D10", "dasha_event_state", "transit_event_state"], + } + + +def _load_cases(manifest_path: Path) -> list[dict[str, Any]]: + try: + payload = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return [] + cases = payload.get("cases") if isinstance(payload, dict) else None + return [case for case in cases if isinstance(case, dict)] if isinstance(cases, list) else [] + + +def select_similar_public_cases( + user_chart: dict[str, Any], + themes: list[str], + *, + cases: list[dict[str, Any]] | None = None, + threshold: float = HIGH_SIMILARITY_THRESHOLD, + max_cases: int = 3, +) -> dict[str, Any]: + candidates = cases if cases is not None else _load_cases(DEFAULT_MANIFEST) + selected: list[dict[str, Any]] = [] + for case in candidates: + replay = case.get("replay") + if not isinstance(replay, dict) or replay.get("outcome_replay_status") != "replayed": + continue + if replay.get("do_not_use_for_prediction") is True: + continue + case_chart = _chart_for_case(case) + if not isinstance(case_chart, dict): + continue + for event in case.get("event_outcomes", []): + if not isinstance(event, dict) or event.get("domain") not in themes: + continue + similarity = _similarity(user_chart, case_chart, event["domain"]) + if similarity["score"] < threshold: + continue + source = case.get("source") if isinstance(case.get("source"), dict) else {} + event_source = event.get("source") if isinstance(event.get("source"), dict) else {} + subject = case.get("subject") if isinstance(case.get("subject"), dict) else {} + selected.append({ + "case_id": case.get("case_id"), + "subject": subject.get("name"), + "domain": event.get("domain"), + "event_type": event.get("event_type"), + "event_date": event.get("event_date"), + "outcome": event.get("outcome"), + "case_source": {"url": source.get("url"), "source_grade": source.get("source_grade")}, + "event_source": {"url": event_source.get("url"), "source_grade": event_source.get("source_grade")}, + "similarity": similarity, + "reference_only": True, + "difference_notice": "相似仅限列出的 D1 特征；未比较层不得推断为相同。", + }) + selected.sort(key=lambda item: (-item["similarity"]["score"], item["case_id"] or "")) + selected = selected[:max_cases] + return { + "status": "high_similarity_public_references_available" if selected else "no_high_similarity_public_reference", + "cases": selected, + "threshold": threshold, + "manifest": "references/real_case_calibration/replay_manifest.json", + "public_figures_only": True, + "does_not_predict_user_outcome": True, + "boundary": "公开案例用于比较与理解，不表示用户会复现该事件。", + } + + +def build_reference_transparency_contract( + chart: dict[str, Any], themes: list[str], *, timing: dict[str, Any] | None = None, + cases: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + timing_contract = build_timing_precision_contract(timing) + return { + "version": "transparent_reference_v1", + "timing_display": { + "claim_status": timing_contract["claim_status"], + "verified_window": "display_with_evidence_scope", + "candidate_windows": "display_with_signals_and_confidence_cap", + "exact_triggers": "display_as_technical_trigger_not_guarantee", + "boundary": timing_contract["boundary"], + }, + "external_engine_observations": { + "Local native": {"role": "primary_calculation", "source": "current request calculation contract"}, + "VedAstro hosted": { + "role": "external_observation", "deployment_identity": "not_publicly_proven", + "source": "references/oracle/vedastro_contract_arbitration_2026_07_17.json", + }, + "Xalen": {"role": "formula_isolation_observation", "source": "references/oracle/xalen_fourth_oracle_comparison_2026_07_17.json"}, + "jyotishyamitra": {"role": "independent_observation", "source": "references/oracle/jyotishyamitra_steve_jobs_probe_2026_07_18.json"}, + }, + "method_variants": { + "display": "show_parallel_methods_with_sources", + "source": "references/oracle/xalen_formula_unit_attribution_2026_07_17.json", + "boundary": "流派/公式差异并列展示；不以单一引擎多数投票决定真值。", + }, + "similar_public_cases": select_similar_public_cases(chart, themes, cases=cases), + } diff --git a/tests/test_api_server_security.py b/tests/test_api_server_security.py index 52ca2ab8..a0176d8c 100644 --- a/tests/test_api_server_security.py +++ b/tests/test_api_server_security.py @@ -2334,6 +2334,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: diff --git a/tests/test_reference_transparency_contract.py b/tests/test_reference_transparency_contract.py new file mode 100644 index 00000000..2be6f1b3 --- /dev/null +++ b/tests/test_reference_transparency_contract.py @@ -0,0 +1,90 @@ +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_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 selected["cases"][0]["event_source"]["url"] == "https://example.com/event" + assert selected["does_not_predict_user_outcome"] is True + + +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" + + +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_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