feat: add transparent case reference contract

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.

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/);
+});

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
 
 

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),
+    }

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:

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