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
from datetime import datetime

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

from domain_calculation_service import compute_vimshottari_timeline  # noqa: E402
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,
    _compact_vedastro_cross_check,
    _compact_western_spectrum,
)
from unified_consultation_orchestrator import UnifiedConsultationOrchestrator  # noqa: E402


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


_BIRTH = {'year': 1995, 'month': 8, 'day': 18, 'hour': 12, 'minute': 0}
_REFERENCE_DATE = '2026-07-14'


def _canonical_dasha(moon_lon: float) -> dict:
    """The mahadasha periods a real chart carries, from the engine the server uses.

    The sub-period layer is cut out of these periods, so a fixture that only names a current lord
    would exercise nothing. Deriving them keeps the fixture from drifting away from the contract.
    """

    timeline = compute_vimshottari_timeline(
        birth_dt=datetime(_BIRTH['year'], _BIRTH['month'], _BIRTH['day'], _BIRTH['hour'], _BIRTH['minute']),
        moon_lon=moon_lon,
    )
    return {
        'current_md': timeline['birth_balance']['lord'],
        'periods': timeline['periods'],
        'birth_balance': timeline['birth_balance'],
    }


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': _canonical_dasha(longitudes['Moon']),
        'modules': {'dasha': _canonical_dasha(longitudes['Moon'])},
    }


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'
    assert packet['sections']['D11']['status'] == 'used'
    assert packet['sections']['D60']['status'] == 'used'
    assert packet['sections']['varga_spectrum']['status'] == 'used'
    assert modules['functional_benefic_malefic']['status'] == 'used'
    assert modules['functional_benefic_malefic']['functional_benefics']
    assert modules['shadbala']
    assert modules['kakshya']['status'] == 'observation_only'
    assert 'functional_benefic_malefic' in chart['local_consultation_layers']['available']
    assert 'shadbala' in chart['local_consultation_layers']['available']
    assert 'kakshya' in chart['local_consultation_layers']['available']


def test_local_consultation_layers_include_yogas_transits_and_route_chara() -> None:
    career = _attach_local_consultation_layers(
        _handler(),
        _base_chart(),
        dict(_BIRTH),
        {'current_date': _REFERENCE_DATE, 'strict_workflow_route': 'career'},
    )
    yogas = career['modules']['yogas']
    assert yogas['status'] == 'executed'
    assert yogas['yogas'][0]['name']
    chara = career['modules']['chara_dasha']
    assert chara['status'] == 'executed'
    assert chara['current']['sign']
    assert chara['sequence'][0]['lord']
    transits = career['modules']['transits']
    assert transits['status'] == 'executed'
    assert transits['sade_sati']['moon_sign']
    assert transits['sade_sati']['saturn_sign']
    assert transits['sade_sati']['phase'] in {'none', 'rising', 'peak', 'setting'}
    assert transits['search_period'] == {'start': '2026-07-14', 'end': '2026-10-12'}
    assert isinstance(transits['triggers'], list)
    assert transits['trigger_count'] == len(transits['triggers'])
    assert 'yogas' in career['local_consultation_layers']['available']
    assert 'chara_dasha' in career['local_consultation_layers']['available']
    assert 'transits' in career['local_consultation_layers']['available']

    wealth = _attach_local_consultation_layers(
        _handler(),
        _base_chart(),
        dict(_BIRTH),
        {'current_date': _REFERENCE_DATE, 'strict_workflow_route': 'wealth'},
    )
    assert wealth['modules']['chara_dasha']['status'] == 'not_applicable'
    assert 'chara_dasha' not in wealth['local_consultation_layers']['available']


def test_consultation_computes_transit_trigger_windows(monkeypatch) -> None:
    class FakeTransit:
        @staticmethod
        def search_transit_triggers(planet, target_longitude, start_date, end_date, orb=1.0, natal_planets=None, ayanamsa_name='lahiri'):
            if planet != 'Saturn' or abs(float(target_longitude) - 31.4) > 0.2:
                return []
            return [{
                'date': datetime(2026, 8, 21),
                'planet': 'Saturn',
                'sensitive_point': 'Moon',
                'type': 'exact_hit',
                'orb': 0.08,
                'event': 'Saturn exact on Moon',
            }]

    import jyotish_api_server as api
    real_load = api._load_local_module

    def load(name):
        if name == 'transit_trigger':
            return FakeTransit
        return real_load(name)

    monkeypatch.setattr(api, '_load_local_module', load)
    api._LOCAL_MODULE_CACHE.pop('transit_trigger', None)
    chart = _attach_local_consultation_layers(
        _handler(),
        _base_chart(),
        dict(_BIRTH),
        {'current_date': _REFERENCE_DATE},
    )
    transits = chart['modules']['transits']
    assert transits['search_period'] == {'start': '2026-07-14', 'end': '2026-10-12'}
    assert transits['triggers'][0]['date'] == '2026-08-21'
    assert transits['triggers'][0]['planet'] == 'Saturn'
    assert transits['triggers'][0]['target'] == 'Moon'
    assert transits['triggers'][0]['kind'] == 'exact_hit'


def test_compact_western_spectrum_keeps_windows_and_drops_clocks() -> None:
    compact = _compact_western_spectrum({
        'status': 'executed',
        'native_chart': {
            'zodiac': 'tropical',
            'natal': {
                'planets': {'sun': {'sign': 'Aries'}, 'moon': {'sign': 'Taurus'}},
                'ascendant': {'sign': 'Leo'},
            },
        },
        'timing_techniques': {
            'transits': {
                'status': 'used',
                'target_date': '2026-07-14',
                'aspects': [{'transit_planet': 'saturn', 'natal_point': 'moon', 'aspect': 'square', 'orb': 0.4}],
                'transit_planets': {'saturn': {'longitude': 25.03}},
                'boundary': 'snapshot',
            },
            'solar_return': {
                'status': 'used',
                'target_year': 2026,
                'return_local_time': '2026-04-17T08:32:11+08:00',
                'natal_sun_longitude': 25.03,
                'boundary': 'exact return',
            },
            'transit_duration_scan': {
                'status': 'used',
                'start_date': '2026-07-14',
                'end_date': '2026-07-27',
                'windows': [{
                    'transit_planet': 'jupiter',
                    'natal_point': 'ascendant',
                    'aspect': 'trine',
                    'start_date': '2026-07-14',
                    'end_date': '2026-07-20',
                    'min_orb': 0.2,
                }],
                'daily_hits': [{'date': '2026-07-14'}],
                'boundary': 'windows',
            },
        },
    })
    transits = compact['techniques']['transits']
    assert transits['target_date'] == '2026-07-14'
    assert transits['aspects'][0]['transit_planet'] == 'saturn'
    assert 'transit_planets' not in transits
    solar = compact['techniques']['solar_return']
    assert solar['target_year'] == 2026
    assert solar['return_date'] == '2026-04-17'
    assert 'return_local_time' not in solar
    assert '08:32' not in str(compact)
    assert '25.03' not in str(compact)
    windows = compact['techniques']['transit_duration_scan']['windows']
    assert windows[0]['start_date'] == '2026-07-14'
    assert windows[0]['end_date'] == '2026-07-20'


def test_consultation_computes_formal_and_research_varga_spectrum() -> None:
    chart = _attach_local_consultation_layers(
        _handler(),
        _base_chart(),
        dict(_BIRTH),
        {'current_date': _REFERENCE_DATE},
    )

    spectrum = chart['modules']['varga_spectrum']
    assert spectrum['counts']['formal'] == 20
    assert spectrum['counts']['research_dn'] == 40
    assert 'D9' in spectrum['formal']
    assert spectrum['formal']['D9']['lagna']
    assert 'D60' in spectrum['formal']
    assert 'D13' in spectrum['research_dn']
    assert spectrum['research_dn']['D13']['boundary']
    assert spectrum['blocked'] == []
    assert 'varga_spectrum' in chart['local_consultation_layers']['available']


def test_sub_period_boundaries_are_cut_out_of_the_periods_the_packet_shows() -> None:
    """The model never received antardasha boundaries: it read `modules.dasha_boundaries`, a key the
    engine never wrote. Mahadashas run six to twenty years, so every answer had to say sub-periods
    were not calculated while the receipt still reported precise timing as allowed (BUG-279).

    The boundaries must also come from the mahadasha the packet already shows, or the model would
    hold two sets of dasha dates and could quote either.
    """

    chart = _attach_local_consultation_layers(
        _handler(), _base_chart(), dict(_BIRTH), {'current_date': _REFERENCE_DATE},
    )

    layer = chart['modules']['dasha_sub_periods']
    reference = datetime.fromisoformat(_REFERENCE_DATE)
    current_md = layer['current']['mahadasha']
    current_ad = layer['current']['antardasha']

    assert layer['boundary_count'] == 9
    assert current_md in [
        {'lord': period['lord'], 'start': period['start'], 'end': period['end']}
        for period in chart['dasha']['periods']
    ]
    assert datetime.fromisoformat(current_md['start']) <= reference < datetime.fromisoformat(current_md['end'])
    assert datetime.fromisoformat(current_ad['start']) <= reference < datetime.fromisoformat(current_ad['end'])
    assert layer['boundaries'][0]['start'] == current_md['start']
    assert layer['boundaries'][-1]['end'] == current_md['end']
    assert chart['local_consultation_layers']['status'] == 'ready'
    assert 'dasha_sub_periods' in chart['local_consultation_layers']['available']

    packet = UnifiedConsultationOrchestrator().machine_evidence_packet(
        chart=chart,
        route_packet={'question_type': 'timing', 'primary_theme': 'timing'},
        vedastro_official={'status': 'blocked'},
    )
    assert packet['sections']['dasha_sub_periods']['status'] == 'used'


def test_every_module_the_model_packet_reads_is_a_module_the_server_writes() -> None:
    """The guard that would have caught this: `local_layers` read `modules.dasha_boundaries`, which
    nothing in this repository ever assigned, so the field was permanently undefined and no test on
    either side of the boundary could see it. Reading a key the engine does not write is the failure
    mode, so compare the names directly against a chart the server actually built.
    """

    workflow = os.path.join(
        os.path.dirname(__file__), '..', 'frontend', 'src', 'mastra', 'consultation-workflow.ts',
    )
    with open(workflow, encoding='utf-8') as handle:
        source = handle.read()
    block = re.search(r'local_layers:\s*\{(.*?)\n\s{4}\},', source, re.DOTALL)
    assert block, 'local_layers block not found in consultation-workflow.ts'
    code = re.sub(r'//[^\n]*', '', block.group(1))
    read_keys = set(re.findall(r'modules\.(\w+)', code))
    # Fail closed: a regex that matched nothing would make this vacuously pass.
    assert len(read_keys) >= 3, sorted(read_keys)

    chart = _attach_local_consultation_layers(
        _handler(), _base_chart(), dict(_BIRTH), {'current_date': _REFERENCE_DATE},
    )
    unwritten = sorted(read_keys - set(chart['modules']))
    assert not unwritten, f'the model packet reads modules the engine never writes: {unwritten}'


def test_a_chart_without_sub_periods_is_not_granted_precise_timing() -> None:
    """`dasha_boundaries` is the mahadasha list, so on its own it can place a decade, never a month.

    Granting precise timing from it is how the receipt came to say `preciseTiming: allowed` for runs
    whose answers had to admit the sub-periods were missing.
    """

    context = _build_consumer_context(
        question='具体哪几个月适合行动',
        route_packet={'question_type': 'timing', 'primary_theme': 'timing'},
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(dasha_sub_periods='missing'),
        vedastro_official={'status': 'blocked'},
    )

    assert context['answer_policy']['can_answer_precise_timing'] is False
    assert context['answer_policy']['should_lead_with_limitations'] is True


def test_precise_timing_is_granted_once_the_sub_period_boundaries_are_there() -> None:
    context = _build_consumer_context(
        question='具体哪几个月适合行动',
        route_packet={'question_type': 'timing', 'primary_theme': 'timing'},
        chart={'success': True},
        rectification=_confirmed_birth_time(),
        machine_evidence_packet=_sections(),
        vedastro_official={'status': 'blocked'},
    )

    assert context['answer_policy']['can_answer_precise_timing'] is True


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',
        'D6': 'used',
        'D7': 'used',
        'D8': 'used',
        'D9': 'used',
        'D10': 'used',
        'D11': 'used',
        'D12': 'used',
        'D24': 'used',
        'A7': 'used',
        'A10': 'used',
        'UL': 'used',
        'ashtakavarga': 'used',
        'dasha_boundaries': 'used',
        'dasha_sub_periods': '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') which cannot be requirements. Formal Vargas
    the engine now materializes are required wherever the product already promises them.
    """

    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'


def test_compact_vedastro_cross_check_keeps_signs_and_drops_coordinates() -> None:
    compact = _compact_vedastro_cross_check(
        {'official_closure_state': 'official_verified', 'status': 'ok'},
        {
            'official_raw_response': {
                'natal': {'sun': 'Leo', 'moon': 'Taurus', 'ascendant': 'Cancer'},
                'lat': 25.03,
                'lon': 121.56,
                'hour': 3,
                'minute': 4,
            },
        },
    )
    assert compact['status'] == 'executed'
    assert compact['natal'] == {'sun': 'Leo', 'moon': 'Taurus', 'ascendant': 'Cancer'}
    assert 'lat' not in compact
    assert 'lon' not in compact
    assert 'hour' not in compact


def test_compact_vedastro_cross_check_marks_timeout_blocked() -> None:
    compact = _compact_vedastro_cross_check(
        {},
        {
            'official_closure_state': 'official_blocked',
            'official_closure_reason': 'foreground_optional_evidence_timeout',
        },
    )
    assert compact['status'] == 'blocked'
    assert compact['official_closure_reason'] == 'foreground_optional_evidence_timeout'