Jyotisha/tests/test_consultation_consumer_context.py

#!/usr/bin/env python3
"""Chat-facing consultation contract regression tests."""

from __future__ import annotations

import os
import re
import sys

SCRIPTS = os.path.join(os.path.dirname(__file__), '..', 'scripts')
if SCRIPTS not in sys.path:
    sys.path.insert(0, SCRIPTS)

from jyotish_api_server import (  # noqa: E402
    _PRECISE_BIRTH_TIME_ACCURACY,
    _ROUTE_DOMAIN_CONTEXT,
    _ROUTE_REQUIRED_LAYERS,
    JyotishAPIHandler,
    _attach_local_consultation_layers,
    _build_consumer_context,
)
from unified_consultation_orchestrator import UnifiedConsultationOrchestrator  # noqa: E402


def _handler() -> JyotishAPIHandler:
    return JyotishAPIHandler.__new__(JyotishAPIHandler)


def _base_chart() -> dict:
    longitudes = {
        'Sun': 120.9,
        'Moon': 31.4,
        'Mars': 173.2,
        'Mercury': 140.1,
        'Jupiter': 222.1,
        'Venus': 120.2,
        'Saturn': 329.5,
        'Rahu': 215.8,
        'Ketu': 35.8,
    }
    asc_lon = 205.19
    asc_idx = int(asc_lon / 30)
    planets = {
        name: {
            'lon': lon,
            'degree': lon % 30,
            'sign_idx': int(lon / 30),
            'house': ((int(lon / 30) - asc_idx) % 12) + 1,
        }
        for name, lon in longitudes.items()
    }
    return {
        'success': True,
        'ascendant': {'lon': asc_lon, 'sign_idx': asc_idx, 'sign': 'Libra'},
        'planets': planets,
        'houses': {house: {'sign_idx': (asc_idx + house - 1) % 12} for house in range(1, 13)},
        'dasha': {'current_md': 'Sun'},
        'modules': {'dasha': {'current_md': 'Sun'}},
    }


def test_local_consultation_layers_supply_d10_a10_and_narayana_without_vedastro() -> None:
    chart = _attach_local_consultation_layers(
        _handler(),
        _base_chart(),
        {'year': 1995, 'month': 8, 'day': 18, 'hour': 12, 'minute': 0},
        {'current_date': '2026-07-14'},
    )

    modules = chart['modules']
    assert 'D9_Navamsa' in modules['varga_full']
    assert 'D10_Dasamsa' in modules['varga_full']
    assert modules['arudha_padas']['padas']['A10']['name'] == 'Karma Pada (A10)'
    assert modules['narayana_dasha']['current_dasha']['md']
    assert modules['ashtakavarga']['sav']['total'] > 0
    assert len(modules['kp_cusps']['houses']) == 12
    assert chart['local_consultation_layers']['status'] == 'ready'

    packet = UnifiedConsultationOrchestrator().machine_evidence_packet(
        chart=chart,
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        vedastro_official={'status': 'blocked', 'runtime_truth': {'status': 'blocked'}},
    )
    assert packet['sections']['D10']['status'] == 'used'
    assert packet['sections']['A10']['status'] == 'used'
    assert packet['sections']['narayana_dasha']['status'] == 'used'
    assert packet['sections']['ashtakavarga']['status'] == 'used'
    assert packet['sections']['KP_cusp']['status'] == 'used'


def test_consumer_context_treats_unconfigured_vedastro_as_optional_cross_check() -> None:
    chart = _attach_local_consultation_layers(
        _handler(),
        _base_chart(),
        {'year': 1995, 'month': 8, 'day': 18, 'hour': 12, 'minute': 0},
        {'current_date': '2026-07-14'},
    )
    orchestrator = UnifiedConsultationOrchestrator()
    route = orchestrator.resolve_route('请分析我的事业方向', ['career'])
    official = {
        'status': 'service_endpoint_not_configured',
        'runtime_truth': {
            'status': 'partial',
            'official_execution_layers': {'chart_core': 'blocked'},
            'fallback_active': False,
        },
    }
    packet = orchestrator.machine_evidence_packet(
        chart=chart,
        route_packet=route,
        vedastro_official=official,
    )
    context = _build_consumer_context(
        question='请分析我的事业方向',
        route_packet=route,
        chart=chart,
        rectification={
            'summary': {
                'headline': '可读主盘，但高敏分盘需要降级',
                'warned': ['D9', 'D10'],
                'disabled': [],
            },
        },
        machine_evidence_packet=packet,
        vedastro_official=official,
    )

    assert context['core_status'] == 'ready'
    assert context['hard_blockers'] == []
    assert context['missing_route_layers'] == []
    assert context['answer_policy']['can_answer_direction'] is True
    assert context['answer_policy']['should_lead_with_limitations'] is False
    assert context['answer_policy']['provider_unavailable_is_fatal'] is False
    assert context['optional_unavailable_layers'][0]['layer'] == 'vedastro_official_cross_check'


def test_consumer_context_only_leads_with_limits_for_unavailable_precise_timing() -> None:
    orchestrator = UnifiedConsultationOrchestrator()
    route = orchestrator.resolve_route('具体哪一个月份适合跳槽？', ['career'])
    packet = {
        'sections': {
            'D1': {'status': 'used'},
            'D10': {'status': 'used'},
            'A10': {'status': 'used'},
            'dasha_boundaries': {'status': 'used'},
            'narayana_dasha': {'status': 'missing'},
            'external_oracle_status': {'status': 'official_blocked'},
        },
    }
    context = _build_consumer_context(
        question='具体哪一个月份适合跳槽？',
        route_packet=route,
        chart={'success': True},
        rectification={'summary': {'warned': ['D10'], 'disabled': []}},
        machine_evidence_packet=packet,
        vedastro_official={'status': 'blocked'},
    )

    assert context['core_status'] == 'degraded'
    assert context['answer_policy']['can_answer_direction'] is True
    assert context['answer_policy']['can_answer_precise_timing'] is False
    assert context['answer_policy']['should_lead_with_limitations'] is True


def test_thematic_career_evidence_uses_local_d10_a10_and_narayana() -> None:
    handler = _handler()
    chart = _attach_local_consultation_layers(
        handler,
        _base_chart(),
        {'year': 1995, 'month': 8, 'day': 18, 'hour': 12, 'minute': 0},
        {'current_date': '2026-07-14'},
    )

    items = handler._derived_career_evidence(
        chart,
        {
            'career': {'summary': 'career ready'},
            'shadbala': {},
            'full_modules': chart['modules'],
        },
    )

    by_technique = {item['technique']: item for item in items}
    assert by_technique['D10-Dashamsha-local']['chart'] == 'D10'
    assert by_technique['A10-Karma-Pada-local']['chart'] == 'A10'
    assert by_technique['Narayana-Dasha-local']['chart'] == 'Narayana'
    assert '已完成' in by_technique['D10-Dashamsha-local']['conclusion']


def test_thematic_report_reuses_attached_chart_modules_without_claiming_d10_missing(monkeypatch) -> None:
    handler = _handler()
    chart = _attach_local_consultation_layers(
        handler,
        _base_chart(),
        {'year': 1995, 'month': 8, 'day': 18, 'hour': 12, 'minute': 0},
        {'current_date': '2026-07-14'},
    )

    monkeypatch.setattr(handler, '_compute_dasha_system', lambda body: chart['modules']['dasha'])
    monkeypatch.setattr(handler, '_compute_yogas_api', lambda body: {})
    monkeypatch.setattr(handler, '_compute_shadbala', lambda body: {})
    monkeypatch.setattr(handler, '_compute_ashtakavarga', lambda body: {})
    monkeypatch.setattr(handler, '_compute_relationship', lambda body: {})
    monkeypatch.setattr(handler, '_compute_career', lambda body: {'summary': 'career ready'})
    monkeypatch.setattr(handler, '_compute_jaimini', lambda body: chart['modules'].get('jaimini', {}))

    result = handler._compute_thematic_report({
        'theme': ['career'],
        'chart_data': {
            **chart,
            'skip_full_reading_for_thematic': True,
        },
        'skip_full_reading_for_thematic': True,
    })

    career = result['themes']['career']
    techniques = {item['technique'] for item in career['evidence']}
    assert result['mode'] == 'derived_chart_evidence'
    assert result['evidence_source']['source'] == 'reused_chart_modules'
    assert result['evidence_source']['chart_modules_reused'] is True
    assert 'D10-Dashamsha-local' in techniques
    assert 'A10-Karma-Pada-local' in techniques
    assert 'Narayana-Dasha-local' in techniques
    assert 'Dashamsha 未提供显著信息' not in career['narrative']


def test_strict_narrative_tolerates_optional_none_contracts() -> None:
    from jyotish_engine import _base_strict_narrative_payload

    payload = _base_strict_narrative_payload(
        '事业',
        {
            'event_judgement': None,
            'adjudication_stages': None,
            'prediction_boundary_contract': None,
        },
        fallback_headline='事业结构可读',
        strengths=[],
        risks=[],
        boundaries=[],
    )

    assert payload['headline'] == '事业结构可读'
    assert 'confidence_cap: unknown' in payload['markdown']


def test_consumer_context_does_not_surface_optional_provider_as_user_limitation() -> None:
    context = _build_consumer_context(
        question='请分析我的事业方向',
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        chart={'success': True},
        rectification={'summary': {'warned': ['D9'], 'disabled': []}},
        machine_evidence_packet={
            'sections': {
                'D1': {'status': 'used'},
                'D10': {'status': 'used'},
                'A10': {'status': 'used'},
                'dasha_boundaries': {'status': 'used'},
                'narayana_dasha': {'status': 'used'},
                'external_oracle_status': {'status': 'official_blocked'},
            },
        },
        vedastro_official={'status': 'blocked'},
    )

    assert context['user_facing_limitation'] is None
    assert context['optional_unavailable_layers'][0]['layer'] == 'vedastro_official_cross_check'


def _relationship_shaped_question() -> str:
    """A question that is unmistakably about relationships and matches no marriage keyword.

    The old keyword list held 感情; this says 情感. That one transposition was enough to lose the
    marriage domain's boundary on a marriage-route answer.
    """

    return '我情感上遇到合适的人了吗现在这个阶段'


def _marriage_route() -> dict:
    return UnifiedConsultationOrchestrator().resolve_route(
        _relationship_shaped_question(), ['marriage'], declared_route='marriage',
    )


def _sections(**overrides: str) -> dict:
    base = {
        'D1': 'used',
        'D2': 'used',
        'D4': 'used',
        'D9': 'used',
        'D10': 'used',
        'A7': 'used',
        'A10': 'used',
        'UL': 'used',
        'ashtakavarga': 'used',
        'dasha_boundaries': 'used',
        'narayana_dasha': 'used',
        'external_oracle_status': 'official_blocked',
    }
    base.update(overrides)
    return {'sections': {name: {'status': status} for name, status in base.items()}}


def _confirmed_birth_time() -> dict:
    return {
        'effective_accuracy': 'minute',
        'lagna_boundary': {'is_sensitive': False},
        'summary': {'warned': ['D7'], 'disabled': ['D30', 'D60']},
    }


def test_every_orchestrator_route_declares_its_own_evidence_gate() -> None:
    """The guard that would have caught this: no route may inherit `general`'s gate by accident.

    `route_requirements` was keyed `relationship`/`finance` while the routes are named
    `marriage`/`wealth`, so `.get(route, general)` silently downgraded both gates and nothing failed.
    """

    routes = set(UnifiedConsultationOrchestrator._ROUTE_DEFINITIONS)
    assert routes <= set(_ROUTE_REQUIRED_LAYERS), sorted(routes - set(_ROUTE_REQUIRED_LAYERS))
    assert set(_ROUTE_DOMAIN_CONTEXT) <= routes, sorted(set(_ROUTE_DOMAIN_CONTEXT) - routes)

    packet = UnifiedConsultationOrchestrator().machine_evidence_packet(
        chart=_base_chart(),
        route_packet={'question_type': 'general', 'primary_theme': 'general'},
        vedastro_official={'status': 'blocked'},
    )
    buildable = set(packet['sections'])
    for route, layers in _ROUTE_REQUIRED_LAYERS.items():
        unbuildable = sorted(set(layers) - buildable)
        assert not unbuildable, f'{route} requires layers the evidence packet never builds: {unbuildable}'


def _frontend_declared_layers() -> dict[str, list[str]]:
    """Read each domain's declared layers out of the frontend registry.

    Parsing TypeScript from a Python test is not pretty, but the alternative is what this pins
    against: two lists describing the same contract with no mechanism keeping them in step.
    """

    registry = os.path.join(
        os.path.dirname(__file__), '..', 'frontend', 'src', 'lib', 'consultation-domain-registry.ts',
    )
    with open(registry, encoding='utf-8') as handle:
        source = handle.read()
    declared: dict[str, list[str]] = {}
    for match in re.finditer(r'\{\s*id:\s*"(\w+)".*?requiredLayers:\s*\[(.*?)\]', source):
        declared[match.group(1)] = [
            value.strip().strip('"') for value in match.group(2).split(',') if value.strip()
        ]
    return declared


def test_evidence_gate_is_never_weaker_than_the_layers_the_product_promises() -> None:
    """Whatever the frontend tells the user it will use must actually gate the answer.

    Only entries naming a real evidence section are compared. The registry also carries human-facing
    labels ('7th house/lord', 'negative holdout gate') and D11, which the engine never builds; those
    cannot be requirements, and demanding them would pin every route to `degraded`.
    """

    declared = _frontend_declared_layers()
    routes = set(UnifiedConsultationOrchestrator._ROUTE_DEFINITIONS)
    # Fail closed: a parse that silently found nothing would make this test vacuously pass.
    assert routes <= set(declared), sorted(routes - set(declared))
    assert all(declared[route] for route in routes)

    packet = UnifiedConsultationOrchestrator().machine_evidence_packet(
        chart=_base_chart(),
        route_packet={'question_type': 'general', 'primary_theme': 'general'},
        vedastro_official={'status': 'blocked'},
    )
    by_lowercase = {name.lower(): name for name in packet['sections']}
    for route in sorted(routes):
        real = [by_lowercase[v.lower()] for v in declared[route] if v.lower() in by_lowercase]
        missing = [name for name in real if name not in _ROUTE_REQUIRED_LAYERS[route]]
        assert not missing, f'{route} promises {missing} but its evidence gate does not check them'


def test_marriage_route_requires_upapada_however_the_question_is_worded() -> None:
    context = _build_consumer_context(
        question=_relationship_shaped_question(),
        route_packet=_marriage_route(),
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(UL='missing'),
        vedastro_official={'status': 'blocked'},
    )

    assert context['route'] == 'marriage'
    assert context['missing_route_layers'] == ['UL']
    assert context['core_status'] == 'degraded'


def test_wealth_route_requires_the_hora_chart() -> None:
    context = _build_consumer_context(
        question='接下来的收入结构会怎么变',
        route_packet={'question_type': 'wealth', 'primary_theme': 'wealth'},
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(D2='missing'),
        vedastro_official={'status': 'blocked'},
    )

    assert context['missing_route_layers'] == ['D2']
    assert context['core_status'] == 'degraded'


def test_domain_boundary_follows_the_route_not_the_wording() -> None:
    context = _build_consumer_context(
        question=_relationship_shaped_question(),
        route_packet=_marriage_route(),
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(),
        vedastro_official={'status': 'blocked'},
    )

    assert 'gender_interpretation_boundary' in context['domain_boundaries']
    assert 'gender interpretation boundary' in context['domain_context_layers']
    # A marriage answer must not pick up unrelated domains' boundaries.
    assert 'health_non_medical_boundary' not in context['domain_boundaries']


def test_birth_time_boundary_reads_rectification_state_instead_of_the_question() -> None:
    uncertain = _build_consumer_context(
        question='我的事业接下来怎么走',
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        chart={'success': True},
        rectification={'effective_accuracy': '1hour', 'lagna_boundary': {'is_sensitive': False}},
        machine_evidence_packet=_sections(),
        vedastro_official={'status': 'blocked'},
    )
    confirmed = _build_consumer_context(
        question='我的事业接下来怎么走',
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(),
        vedastro_official={'status': 'blocked'},
    )

    assert 'birth_time_uncertainty_boundary' in uncertain['domain_boundaries']
    assert 'birth_time_uncertainty_boundary' not in confirmed['domain_boundaries']
    assert 'minute' in _PRECISE_BIRTH_TIME_ACCURACY


def test_unknown_birth_time_precision_counts_as_uncertain() -> None:
    context = _build_consumer_context(
        question='我的事业接下来怎么走',
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        chart={'success': True},
        rectification={},
        machine_evidence_packet=_sections(),
        vedastro_official={'status': 'blocked'},
    )

    assert 'birth_time_uncertainty_boundary' in context['domain_boundaries']


def test_sensitive_lagna_keeps_the_boundary_at_minute_accuracy() -> None:
    context = _build_consumer_context(
        question='我的事业接下来怎么走',
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        chart={'success': True},
        rectification={'effective_accuracy': 'minute', 'lagna_boundary': {'is_sensitive': True}},
        machine_evidence_packet=_sections(),
        vedastro_official={'status': 'blocked'},
    )

    assert 'birth_time_uncertainty_boundary' in context['domain_boundaries']


def test_precise_timing_needs_narayana_dasha_on_a_route_that_does_not_require_it() -> None:
    """`timing_layers_ready` used to read `missing_route_layers`, which never lists a layer the route
    does not require. Precise timing was therefore granted on marriage/wealth without the layer."""

    context = _build_consumer_context(
        question=_relationship_shaped_question(),
        route_packet=_marriage_route(),
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(narayana_dasha='missing'),
        vedastro_official={'status': 'blocked'},
    )

    assert context['missing_route_layers'] == []
    assert context['core_status'] == 'ready'
    assert context['answer_policy']['can_answer_precise_timing'] is False


def test_failed_vedastro_raw_packet_is_not_marked_as_used() -> None:
    packet = UnifiedConsultationOrchestrator().machine_evidence_packet(
        chart=_base_chart(),
        route_packet={'question_type': 'career', 'primary_theme': 'career'},
        vedastro_official={
            'status': 'partial',
            'runtime_truth': {
                'status': 'partial',
                'official_execution_layers': {'chart_core': 'blocked'},
            },
            'raw_response': {
                'sections': {
                    'chart_core': {'Status': 'Fail', 'Payload': {'status': 'python_package_not_installed'}},
                },
            },
        },
    )

    assert packet['sections']['vedastro_official_raw_response']['status'] == 'received_unverified'