jesse-ux
e61535f464
BUG-727：同日 VedAstro 快照零等待，跨日先用旧的并后台刷新；join 超时必须 cancel，budget 不超过 2×join。BUG-728：western_evidence_packet 无读取点，默认不再进咨询响应。jyotish_api_server.py 未增长（11334→11291）。
351 lines
13 KiB
Python
351 lines
13 KiB
Python
"""BUG-727 / BUG-728: foreground VedAstro snapshot cache, cancel, and on-demand western packet."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts import jyotish_api_server
|
|
from scripts import vedastro_foreground as foreground
|
|
from scripts.jyotish_api_server import JyotishAPIHandler
|
|
from scripts.vedastro_snapshot_cache import (
|
|
lookup_snapshot,
|
|
snapshot_cache_dir,
|
|
snapshot_cache_key,
|
|
snapshot_cache_path,
|
|
store_snapshot,
|
|
)
|
|
|
|
|
|
def _handler() -> JyotishAPIHandler:
|
|
return JyotishAPIHandler.__new__(JyotishAPIHandler)
|
|
|
|
|
|
def _verified_gateway(*, request_id: str, reference_date: str) -> dict:
|
|
return {
|
|
"scope": "vedastro_gateway_run",
|
|
"status": "official_verified",
|
|
"official_closure_state": "official_verified",
|
|
"official_raw_response": {
|
|
"request_id": request_id,
|
|
"natal": {"sun": "Leo", "moon": "Taurus", "ascendant": "Cancer"},
|
|
"reference_date": reference_date,
|
|
},
|
|
}
|
|
|
|
|
|
def _fake_chart() -> dict:
|
|
return {
|
|
"success": True,
|
|
"birth_info": {"date": "1997-08-08", "time": "05:00", "tz": 8},
|
|
"ascendant": {"lon": 92.0, "sign": "Cancer", "sign_idx": 3},
|
|
"planets": {},
|
|
"dasha": {"periods": [{"lord": "Sun", "start": "2026-01-01", "end": "2027-01-01"}]},
|
|
"modules": {
|
|
"varga_full": {"D9": {}, "D10": {}},
|
|
"arudha_padas": {"A10": {}, "UL": {}},
|
|
"narayana_dasha": {"periods": []},
|
|
"ashtakavarga": {"sav": []},
|
|
"kp_cusps": {"houses": []},
|
|
},
|
|
"special_lagnas": {"precision": "sunrise_correct"},
|
|
}
|
|
|
|
|
|
def _stub_local(monkeypatch, handler: JyotishAPIHandler, gateway) -> None:
|
|
monkeypatch.setattr(handler, "_compute_chart", lambda body: _fake_chart())
|
|
monkeypatch.setattr(
|
|
handler,
|
|
"_compute_rectification_gate",
|
|
lambda body: {
|
|
"success": True,
|
|
"summary": {"recommended_events": [], "warned": [], "disabled": []},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
handler,
|
|
"_compute_thematic_report",
|
|
lambda body: {"success": True, "endpoint": "thematic_report", "themes": {}},
|
|
)
|
|
monkeypatch.setattr(handler, "_compute_vedastro_gateway_run", gateway)
|
|
|
|
|
|
def _body(*, reference_date: str, extra: dict | None = None) -> dict:
|
|
payload = {
|
|
"entry_mode": "direct_chart",
|
|
"question": "请直接排盘并重点看事业",
|
|
"theme": ["career"],
|
|
"year": 1997,
|
|
"month": 8,
|
|
"day": 8,
|
|
"hour": 5,
|
|
"minute": 0,
|
|
"lat": 36.420487,
|
|
"lon": 114.209936,
|
|
"tz": 8,
|
|
"reference_date": reference_date,
|
|
"today": reference_date,
|
|
"current_date": reference_date,
|
|
"western_mode": False,
|
|
"defer_optional_external_evidence": True,
|
|
}
|
|
if extra:
|
|
payload.update(extra)
|
|
return payload
|
|
|
|
|
|
@pytest.fixture
|
|
def snapshot_cache(tmp_path, monkeypatch):
|
|
cache_dir = tmp_path / "vedastro_snapshot_cache"
|
|
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(cache_dir))
|
|
return cache_dir
|
|
|
|
|
|
def test_snapshot_cache_dir_is_not_api_chart_cache_dir(snapshot_cache) -> None:
|
|
chart_dir = jyotish_api_server._api_chart_cache_dir()
|
|
snap_dir = snapshot_cache_dir()
|
|
assert snap_dir.resolve() != chart_dir.resolve()
|
|
assert snap_dir.name == "vedastro_snapshot_cache"
|
|
assert chart_dir.name == "api_chart_cache"
|
|
assert snapshot_cache_key is not jyotish_api_server._api_chart_cache_key
|
|
|
|
|
|
def test_snapshot_cache_key_differs_from_api_chart_cache_key() -> None:
|
|
body = _body(reference_date="2026-09-15")
|
|
snap_key = snapshot_cache_key(body, reference_date="2026-09-15")
|
|
chart_key = jyotish_api_server._api_chart_cache_key(
|
|
jyotish_api_server._build_api_chart_cache_payload(body)
|
|
)
|
|
assert snap_key != chart_key
|
|
assert len(snap_key) == 64
|
|
assert "1997" not in snap_key
|
|
assert "36.420487" not in snap_key
|
|
assert snapshot_cache_path(snap_key).name == f"{snap_key}.json"
|
|
|
|
|
|
def test_cache_file_strips_identity_and_keeps_hash_filename(snapshot_cache) -> None:
|
|
body = _body(reference_date="2026-09-15")
|
|
packet = _verified_gateway(request_id="keep-me", reference_date="2026-09-15")
|
|
packet["name"] = "Secret Person"
|
|
packet["email"] = "user@example.com"
|
|
packet["user_id"] = "usr_123"
|
|
stored = store_snapshot(body, packet, reference_date="2026-09-15")
|
|
assert stored is not None
|
|
path = snapshot_cache_path(stored["cache_key"])
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "Secret Person" not in text
|
|
assert "user@example.com" not in text
|
|
assert "usr_123" not in text
|
|
assert "keep-me" in text
|
|
assert path.name.endswith(".json")
|
|
assert "1997" not in path.name
|
|
|
|
|
|
def test_second_same_day_consultation_skips_gateway_and_urlopen(snapshot_cache, monkeypatch) -> None:
|
|
handler = _handler()
|
|
calls = {"gateway": 0, "urlopen": 0}
|
|
|
|
def counting_gateway(body):
|
|
calls["gateway"] += 1
|
|
return _verified_gateway(request_id="same-day", reference_date=body.get("reference_date"))
|
|
|
|
def counting_urlopen(*_args, **_kwargs):
|
|
calls["urlopen"] += 1
|
|
raise AssertionError("urlopen must not run when the gateway is stubbed")
|
|
|
|
_stub_local(monkeypatch, handler, counting_gateway)
|
|
monkeypatch.setattr("urllib.request.urlopen", counting_urlopen)
|
|
|
|
body = _body(reference_date="2026-09-15")
|
|
first = handler._compute_consultation_workflow(body)
|
|
second = handler._compute_consultation_workflow(body)
|
|
|
|
assert calls["gateway"] == 1
|
|
assert calls["urlopen"] == 0
|
|
assert first["vedastro_gateway"] == second["vedastro_gateway"]
|
|
assert first["vedastro_snapshot_cache"]["freshness"] == "miss"
|
|
assert second["vedastro_snapshot_cache"]["freshness"] == "fresh"
|
|
|
|
|
|
def test_next_day_serves_stale_and_refreshes_without_waiting(snapshot_cache, monkeypatch) -> None:
|
|
handler = _handler()
|
|
yesterday = "2026-09-14"
|
|
today = "2026-09-15"
|
|
store_snapshot(
|
|
_body(reference_date=yesterday),
|
|
_verified_gateway(request_id="yesterday", reference_date=yesterday),
|
|
reference_date=yesterday,
|
|
)
|
|
release = threading.Event()
|
|
calls = {"gateway": 0}
|
|
|
|
def slow_today(body):
|
|
calls["gateway"] += 1
|
|
release.wait(timeout=2.0)
|
|
return _verified_gateway(request_id="today", reference_date=today)
|
|
|
|
_stub_local(monkeypatch, handler, slow_today)
|
|
started = time.monotonic()
|
|
result = handler._compute_consultation_workflow(_body(reference_date=today))
|
|
elapsed = time.monotonic() - started
|
|
release.set()
|
|
|
|
assert elapsed < 1.0
|
|
assert result["vedastro_snapshot_cache"]["freshness"] == "stale"
|
|
assert result["vedastro_snapshot_cache"]["reference_date"] == yesterday
|
|
assert result["vedastro_gateway"]["snapshot_reference_date"] == yesterday
|
|
assert result["vedastro_gateway"]["official_closure_state"] == "official_verified"
|
|
assert result["vedastro_gateway"]["official_raw_response"]["request_id"] == "yesterday"
|
|
deadline = time.monotonic() + 2.0
|
|
while calls["gateway"] < 1 and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
assert calls["gateway"] >= 1
|
|
|
|
|
|
def test_daily_starlanguage_does_not_eat_yesterday_cache(snapshot_cache, monkeypatch) -> None:
|
|
handler = _handler()
|
|
yesterday = "2026-09-14"
|
|
today = "2026-09-15"
|
|
store_snapshot(
|
|
_body(reference_date=yesterday),
|
|
_verified_gateway(request_id="yesterday", reference_date=yesterday),
|
|
reference_date=yesterday,
|
|
)
|
|
calls = {"gateway": 0}
|
|
|
|
def today_gateway(body):
|
|
calls["gateway"] += 1
|
|
return _verified_gateway(request_id="today-live", reference_date=today)
|
|
|
|
_stub_local(monkeypatch, handler, today_gateway)
|
|
result = handler._compute_consultation_workflow(
|
|
_body(reference_date=today, extra={"entrypoint": "daily_starlanguage"})
|
|
)
|
|
|
|
assert calls["gateway"] == 1
|
|
assert result["vedastro_snapshot_cache"]["freshness"] == "miss"
|
|
assert result["vedastro_gateway"]["official_raw_response"]["request_id"] == "today-live"
|
|
assert lookup_snapshot(
|
|
_body(reference_date=today, extra={"entrypoint": "daily_starlanguage"}),
|
|
today=today,
|
|
)["freshness"] == "fresh"
|
|
|
|
|
|
def test_nth_plus_one_wait_stays_within_join_and_queued_work_is_cancelled(
|
|
snapshot_cache, monkeypatch
|
|
) -> None:
|
|
join = 0.2
|
|
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", str(join))
|
|
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "8")
|
|
handler = _handler()
|
|
started: list[int] = []
|
|
lock = threading.Lock()
|
|
release = threading.Event()
|
|
|
|
def slow_gateway(body):
|
|
with lock:
|
|
started.append(1)
|
|
release.wait(timeout=8)
|
|
return _verified_gateway(request_id="slow", reference_date="2026-09-15")
|
|
|
|
_stub_local(monkeypatch, handler, slow_gateway)
|
|
workers = foreground._FOREGROUND_VEDASTRO_WORKERS
|
|
body = _body(reference_date="2026-09-15")
|
|
elapsed: list[float] = []
|
|
|
|
def run_one() -> None:
|
|
t0 = time.monotonic()
|
|
result = handler._compute_consultation_workflow(body)
|
|
elapsed.append(time.monotonic() - t0)
|
|
assert result["vedastro_gateway"]["official_closure_reason"] == "foreground_optional_evidence_timeout"
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=workers + 3) as pool:
|
|
futs = [pool.submit(run_one) for _ in range(workers + 3)]
|
|
for fut in futs:
|
|
fut.result(timeout=8)
|
|
assert max(elapsed) <= join + 2.5
|
|
assert len(started) <= workers
|
|
finally:
|
|
release.set()
|
|
|
|
|
|
def test_join_budget_workers_declared_together_and_budget_respects_ratio(monkeypatch) -> None:
|
|
source = Path("scripts/vedastro_foreground.py").read_text(encoding="utf-8")
|
|
join_idx = source.index("FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT")
|
|
ratio_idx = source.index("FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO")
|
|
workers_idx = source.index("FOREGROUND_VEDASTRO_WORKERS_DEFAULT")
|
|
span = max(join_idx, ratio_idx, workers_idx) - min(join_idx, ratio_idx, workers_idx)
|
|
assert span < 400
|
|
assert "change together" in source
|
|
|
|
monkeypatch.delenv("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", raising=False)
|
|
monkeypatch.delenv("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", raising=False)
|
|
join = foreground._foreground_vedastro_join_seconds()
|
|
budget = foreground._foreground_vedastro_budget_seconds()
|
|
assert budget <= foreground.FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO * join + 1e-9
|
|
|
|
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", "1.5")
|
|
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "8")
|
|
assert foreground._foreground_vedastro_budget_seconds() <= 3.0 + 1e-9
|
|
|
|
|
|
def test_default_consultation_omits_western_packet_and_schema_required_keys_remain(
|
|
snapshot_cache, monkeypatch
|
|
) -> None:
|
|
handler = _handler()
|
|
_stub_local(
|
|
monkeypatch,
|
|
handler,
|
|
lambda body: _verified_gateway(request_id="omit-western", reference_date="2026-09-15"),
|
|
)
|
|
omitted = handler._compute_consultation_workflow(_body(reference_date="2026-09-15"))
|
|
included = handler._compute_consultation_workflow(
|
|
_body(reference_date="2026-09-15", extra={"include_western_evidence_packet": True})
|
|
)
|
|
|
|
assert "western_evidence_packet" not in omitted
|
|
assert "western_spectrum" in omitted["consumer_context"]
|
|
assert omitted["success"] is True
|
|
assert isinstance(omitted["chart"], dict)
|
|
assert isinstance(omitted["routing"], dict)
|
|
assert isinstance(omitted["consumer_context"], dict)
|
|
assert "western_evidence_packet" in included
|
|
|
|
workflow_ts = Path("frontend/src/mastra/consultation-workflow.ts").read_text(encoding="utf-8")
|
|
schema = workflow_ts.split("export const consultationWorkflowResponseSchema", 1)[1]
|
|
assert "success: z.boolean()" in schema
|
|
assert "chart: z.record(z.unknown())" in schema
|
|
assert "routing: z.record(z.unknown())" in schema
|
|
assert "consumer_context:" in schema
|
|
assert ").passthrough()" in schema
|
|
|
|
omitted_chars = len(json.dumps(omitted, ensure_ascii=False))
|
|
included_chars = len(json.dumps(included, ensure_ascii=False))
|
|
assert included_chars >= omitted_chars
|
|
|
|
|
|
def test_western_oracle_payload_still_returns_packet(snapshot_cache, monkeypatch) -> None:
|
|
handler = _handler()
|
|
_stub_local(
|
|
monkeypatch,
|
|
handler,
|
|
lambda body: _verified_gateway(request_id="oracle", reference_date="2026-09-15"),
|
|
)
|
|
result = handler._compute_consultation_workflow(
|
|
_body(
|
|
reference_date="2026-09-15",
|
|
extra={
|
|
"western_oracle_payload": {
|
|
"source_engine": "kerykeion_external_json",
|
|
"natal": {"ascendant": "Virgo", "mc": "Gemini"},
|
|
}
|
|
},
|
|
)
|
|
)
|
|
assert result["western_evidence_packet"]["source_engine"] == "kerykeion_external_json"
|