Files
Jyotisha/tests/test_consultation_consumer_context.py
T
Jesse_Chen 10ae149c58
Independent Staging Quality Gate / validate (push) Successful in 7m41s
Independent Staging Quality Gate / publish (push) Successful in 9m39s
fix(consult): check the evidence gate against the route the answer is on
七条路由的证据门都不是自己的:route_requirements 的键写成 relationship/finance,而
路由名是 marriage/wealth,另有 5 条路由压根没有条目,全部静默落到 general 的门。
missingLayers: [] 因此不表示证据齐备,只表示没检查过——婚姻的 UL 与财富的 D2 从未
进入检查。

同一函数另有两处判据也没接到权威来源。7 块正则用问题文本重猜领域,而领域早已由模型
声明并写进 route_packet,一句写作「情感」而非表里「感情」的提问在 marriage 路由上完全
拿不到性别解读边界。timing_layers_ready 读的是 missing_route_layers,该列表只装本路由
要求的层,于是对任何不要求 narayana_dasha 的路由恒为真,精确应期在该层根本没算出来时
也照样放行。三处的失败方向都是静默放宽,因此没有任何人报错。

三处都接回权威来源:10 条路由逐条显式列出必需层(层名限定为证据包真实构建的 section,
所以 wealth 不要求引擎不产出的 D11)、领域边界按 route 查表、出生时间边界从矫正闸门的
effective_accuracy 与 Lagna 敏感度派生、就绪判断直接读 section 状态。唯一保留文本探测
的是「用户有没有要一个具体日期」——服务端对此没有权威来源,改为 timing/annual 路由结构
性携带、文本仅作叠加,一次措辞漏判不再能把信号清零。

另外把 skillReferenceReadCount 暴露为回执的 skill.referenceReads(必填)与可观测日志的
skillReferenceReads。它此前数完即丢,而 skill_read 按设计不记成 runtime step,因此「模型
有没有真的翻开方法文档」在运行结束后无处可查。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-18 11:50:28 +08:00

463 lines
17 KiB
Python

#!/usr/bin/env python3
"""Chat-facing consultation contract regression tests."""
from __future__ import annotations
import os
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',
'A10': 'used',
'UL': '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 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'