fix: require VedAstro raw response for official closure
This commit is contained in:
@@ -55,6 +55,7 @@ For large architecture or release work, also read:
|
||||
| ERR-022 | Technique Audit Table can show VedAstro cloud state but omit whether archived official raw evidence is actually auditable. | mitigated 2026-07-08 | Keep `VedAstro Raw Archive Manifest` as a first-class Technique Audit Table row immediately after `VedAstro Cloud State`. |
|
||||
| ERR-023 | `professional_reading` can require a Technique Audit Table while omitting the user-visible VedAstro raw archive row. | mitigated 2026-07-08 | Keep `VedAstro Raw Archive Manifest` in `professional_reading.technique_audit_table_required_rows`. |
|
||||
| ERR-024 | VedAstro gateway status can choose `local_fallback` before `.env` is loaded, even when official endpoint/network settings are present. | mitigated 2026-07-08 | `gateway_status()` must load official readiness before resolving active backend; `run_gateway_packet()` must expose `official_closure_state` separately from legacy `status`. |
|
||||
| ERR-025 | VedAstro gateway can report legacy `status=ok` from catalog availability even when no official raw response is present. | mitigated 2026-07-08 | `official_closure_state=official_verified` requires `official_raw_response`; otherwise expose `official_closure_reason=official_raw_response_missing`. |
|
||||
|
||||
## Fragment Sweep Command Set
|
||||
|
||||
|
||||
@@ -295,15 +295,34 @@ def _status_from_report(gateway: dict[str, Any], report: dict[str, Any]) -> str:
|
||||
|
||||
|
||||
def _official_closure_state_from_report(gateway: dict[str, Any], report: dict[str, Any]) -> str:
|
||||
catalog = report.get("official_capability_catalog") if isinstance(report, dict) else {}
|
||||
active_backend = gateway.get("active_backend") or "local_fallback"
|
||||
if active_backend == "local_fallback":
|
||||
return "local_fallback"
|
||||
if (catalog or {}).get("available"):
|
||||
if _official_raw_response_from_report(report):
|
||||
return "official_verified"
|
||||
return "official_blocked"
|
||||
|
||||
|
||||
def _official_raw_response_from_report(report: dict[str, Any]) -> dict[str, Any]:
|
||||
if not isinstance(report, dict):
|
||||
return {}
|
||||
raw = (
|
||||
report.get("official_raw_response")
|
||||
or report.get("raw_response")
|
||||
or report.get("vedastro_official_raw_response")
|
||||
)
|
||||
return raw if isinstance(raw, dict) else {}
|
||||
|
||||
|
||||
def _official_closure_reason_from_report(gateway: dict[str, Any], report: dict[str, Any]) -> str:
|
||||
active_backend = gateway.get("active_backend") or "local_fallback"
|
||||
if active_backend == "local_fallback":
|
||||
return "local_fallback_backend"
|
||||
if _official_raw_response_from_report(report):
|
||||
return "official_raw_response_present"
|
||||
return "official_raw_response_missing"
|
||||
|
||||
|
||||
def run_gateway_packet(
|
||||
case: dict[str, Any],
|
||||
question: str = "",
|
||||
@@ -316,11 +335,14 @@ def run_gateway_packet(
|
||||
args = _entrypoint_args(case, question, themes, reference_date)
|
||||
report = build_report(args)
|
||||
catalog = report.get("official_capability_catalog") or {}
|
||||
official_raw_response = _official_raw_response_from_report(report)
|
||||
return {
|
||||
"scope": "vedastro_gateway_run",
|
||||
"schema_version": 1,
|
||||
"status": _status_from_report(gateway, report),
|
||||
"official_closure_state": _official_closure_state_from_report(gateway, report),
|
||||
"official_closure_reason": _official_closure_reason_from_report(gateway, report),
|
||||
**({"official_raw_response": official_raw_response} if official_raw_response else {}),
|
||||
"gateway_status": gateway,
|
||||
"input": report.get("input") or {},
|
||||
"runtime_mode": report.get("runtime_mode") or {},
|
||||
|
||||
@@ -93,6 +93,57 @@ def test_gateway_run_packet_uses_user_entrypoint_and_marks_not_all_641(monkeypat
|
||||
assert packet["user_visibility"]["mainland_cn_safe"] is True
|
||||
|
||||
|
||||
def test_gateway_official_closure_requires_raw_response(monkeypatch):
|
||||
from scripts import vedastro_gateway, vedastro_user_entrypoint
|
||||
|
||||
monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1")
|
||||
monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api")
|
||||
monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1")
|
||||
monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "20")
|
||||
|
||||
def fake_report(_args):
|
||||
return {
|
||||
"input": {},
|
||||
"runtime_mode": {},
|
||||
"official_capability_catalog": {"status": "partial", "available": True, "summary": {"catalog_method_count": 641}},
|
||||
"cache_and_queue": {},
|
||||
"strict_workflow": {},
|
||||
"honesty_boundary": {},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(vedastro_user_entrypoint, "build_report", fake_report)
|
||||
packet = vedastro_gateway.run_gateway_packet({"year": 1955}, question="x")
|
||||
|
||||
assert packet["official_closure_state"] == "official_blocked"
|
||||
assert packet["official_closure_reason"] == "official_raw_response_missing"
|
||||
|
||||
|
||||
def test_gateway_official_closure_verified_when_raw_response_present(monkeypatch):
|
||||
from scripts import vedastro_gateway, vedastro_user_entrypoint
|
||||
|
||||
monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1")
|
||||
monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api")
|
||||
monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1")
|
||||
monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "20")
|
||||
|
||||
def fake_report(_args):
|
||||
return {
|
||||
"input": {},
|
||||
"runtime_mode": {},
|
||||
"official_capability_catalog": {"status": "partial", "available": True, "summary": {"catalog_method_count": 641}},
|
||||
"official_raw_response": {"Status": "Pass"},
|
||||
"cache_and_queue": {},
|
||||
"strict_workflow": {},
|
||||
"honesty_boundary": {},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(vedastro_user_entrypoint, "build_report", fake_report)
|
||||
packet = vedastro_gateway.run_gateway_packet({"year": 1955}, question="x")
|
||||
|
||||
assert packet["official_closure_state"] == "official_verified"
|
||||
assert packet["official_raw_response"] == {"Status": "Pass"}
|
||||
|
||||
|
||||
def test_gateway_queue_lifecycle_uses_file_job_store(monkeypatch, tmp_path):
|
||||
from scripts import vedastro_gateway
|
||||
|
||||
|
||||
Reference in New Issue
Block a user