From 9bce83edc8379e72774d352afa4e647528f608ab Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Thu, 9 Jul 2026 09:54:08 +0800 Subject: [PATCH] feat: propagate VedAstro official raw snapshots --- docs/research/pre_work_error_ledger.md | 1 + scripts/vedastro_user_entrypoint.py | 77 +++++++++++++++++++++++- tests/test_vedastro_gateway.py | 39 +++++++++++++ tests/test_vedastro_user_entrypoint.py | 81 ++++++++++++++++++++++++++ 4 files changed, 195 insertions(+), 3 deletions(-) diff --git a/docs/research/pre_work_error_ledger.md b/docs/research/pre_work_error_ledger.md index 9a3bfdcb..a3cd6569 100644 --- a/docs/research/pre_work_error_ledger.md +++ b/docs/research/pre_work_error_ledger.md @@ -56,6 +56,7 @@ For large architecture or release work, also read: | 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`. | +| ERR-026 | VedAstro service adapter can obtain an official full-snapshot raw response while the user entrypoint drops it, leaving gateway official closure permanently blocked. | mitigated 2026-07-09 | `vedastro_user_entrypoint` must expose `vedastro_official_full_snapshot.raw_response_available` and root `official_raw_response` when explicitly requested; gateway tests must prove raw propagation reaches `official_verified`. | ## Fragment Sweep Command Set diff --git a/scripts/vedastro_user_entrypoint.py b/scripts/vedastro_user_entrypoint.py index fb616659..9a740f50 100644 --- a/scripts/vedastro_user_entrypoint.py +++ b/scripts/vedastro_user_entrypoint.py @@ -134,6 +134,66 @@ def _run_capability_catalog(case: dict[str, Any]) -> dict[str, Any]: } +def _official_raw_requested(args: argparse.Namespace) -> bool: + return ( + bool(getattr(args, "require_official_raw_response", False)) + or _bool_env("VEDASTRO_REQUIRE_OFFICIAL_RAW_RESPONSE") + or _bool_env("VEDASTRO_GATEWAY_REQUIRE_OFFICIAL_RAW_RESPONSE") + ) + + +def _run_official_full_snapshot(case: dict[str, Any]) -> dict[str, Any]: + stub = os.environ.get("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_STUB", "").strip() + if stub: + try: + parsed = json.loads(stub) + except json.JSONDecodeError as exc: + return { + "available": False, + "status": "official_full_snapshot_stub_invalid_json", + "error": {"type": type(exc).__name__, "message": str(exc)}, + } + return parsed if isinstance(parsed, dict) else {"available": False, "status": "official_full_snapshot_stub_not_object"} + + try: + from scripts.vedastro_service_adapter import run_official_full_snapshot_for_case + except ModuleNotFoundError: # pragma: no cover - direct script execution + from vedastro_service_adapter import run_official_full_snapshot_for_case + + try: + result = run_official_full_snapshot_for_case(case, case_id="user_chart") + except Exception as exc: # pragma: no cover - runtime boundary + return { + "available": False, + "status": "official_full_snapshot_runtime_error", + "error": {"type": type(exc).__name__, "message": str(exc)}, + } + return result if isinstance(result, dict) else {"available": False, "status": "official_full_snapshot_invalid_result"} + + +def _extract_official_raw_response(snapshot: dict[str, Any]) -> dict[str, Any]: + raw = ( + snapshot.get("official_raw_response") + or snapshot.get("raw_response") + or snapshot.get("vedastro_official_raw_response") + ) + return raw if isinstance(raw, dict) else {} + + +def _snapshot_summary(snapshot: dict[str, Any], raw_response: dict[str, Any]) -> dict[str, Any]: + metadata = snapshot.get("source_metadata") if isinstance(snapshot.get("source_metadata"), dict) else {} + return { + "status": snapshot.get("status") or "not_requested", + "available": bool(snapshot.get("available")) or bool(raw_response), + "operation": snapshot.get("operation") or "official_full_snapshot", + "raw_response_available": bool(raw_response), + "artifact_path": metadata.get("artifact_path"), + "reason": snapshot.get("reason"), + "error": snapshot.get("error") if isinstance(snapshot.get("error"), dict) else {}, + "boundary": "official_raw_response must be present before claiming VedAstro official cloud closure.", + } + + def _strict_workflow_summary(route: str, catalog: dict[str, Any]) -> dict[str, Any]: dynamic_selection = catalog.get("dynamic_selection") if isinstance(catalog.get("dynamic_selection"), dict) else {} theme_for_route = {"relationship": "marriage", "finance": "wealth"}.get(route, route) @@ -179,8 +239,10 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: case = _case_from_args(args, themes) runtime = build_runtime_mode_report() catalog = _run_capability_catalog(case) + snapshot = _run_official_full_snapshot(case) if _official_raw_requested(args) else {"status": "not_requested"} + raw_response = _extract_official_raw_response(snapshot) route = _primary_route(args.question, themes) - return { + report = { "scope": "vedastro_user_entrypoint", "schema_version": 1, "input": { @@ -190,6 +252,7 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: "reference_date": args.reference_date, }, "runtime_mode": runtime, + "vedastro_official_full_snapshot": _snapshot_summary(snapshot, raw_response), "official_capability_catalog": { "status": catalog.get("status") or "blocked", "available": bool(catalog.get("available")), @@ -213,8 +276,9 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: "json": ( "python3 scripts/vedastro_user_entrypoint.py --year YYYY --month MM --day DD " "--hour HH --minute MM --lat LAT --lon LON --tz TZ --question '...' " - "--themes career,marriage,wealth --reference-date YYYY-MM-DD --format json " - "# includes official_full_capability_catalog" + "--themes career,marriage,wealth --reference-date YYYY-MM-DD " + "--require-official-raw-response --format json " + "# includes official_full_capability_catalog and official_raw_response when verified" ), "markdown": ( "python3 scripts/vedastro_user_entrypoint.py --year YYYY --month MM --day DD " @@ -222,10 +286,14 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: ), }, } + if raw_response: + report["official_raw_response"] = raw_response + return report def render_markdown(report: dict[str, Any]) -> str: runtime = report["runtime_mode"] + snapshot = report.get("vedastro_official_full_snapshot") or {} catalog = report["official_capability_catalog"] cache = report["cache_and_queue"] strict = report["strict_workflow"] @@ -234,6 +302,8 @@ def render_markdown(report: dict[str, Any]) -> str: "# VedAstro 用户级入口", "", f"- runtime_mode: `{runtime.get('mode')}`", + f"- official_full_snapshot: `{snapshot.get('status')}`", + f"- official_raw_response_available: `{str(bool(snapshot.get('raw_response_available'))).lower()}`", f"- official_ready: `{str(runtime.get('official_ready')).lower()}`", f"- catalog_status: `{catalog.get('status')}`", f"- catalog_method_count: `{summary.get('catalog_method_count', 0)}`", @@ -273,6 +343,7 @@ def main() -> int: parser.add_argument("--reference-date", required=True) parser.add_argument("--ayanamsa", default="lahiri") parser.add_argument("--node-mode", default="mean") + parser.add_argument("--require-official-raw-response", action="store_true") parser.add_argument("--format", choices=["json", "markdown"], default="markdown") args = parser.parse_args() diff --git a/tests/test_vedastro_gateway.py b/tests/test_vedastro_gateway.py index 20315eac..8bee7be4 100644 --- a/tests/test_vedastro_gateway.py +++ b/tests/test_vedastro_gateway.py @@ -144,6 +144,45 @@ def test_gateway_official_closure_verified_when_raw_response_present(monkeypatch assert packet["official_raw_response"] == {"Status": "Pass"} +def test_gateway_requires_and_propagates_entrypoint_official_raw_response(monkeypatch): + from scripts import vedastro_gateway, vedastro_user_entrypoint + + raw_response = { + "source": "vedastro_official_full_snapshot", + "sections": {"chart_core": {"Status": "Pass"}}, + } + 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_GATEWAY_REQUIRE_OFFICIAL_RAW_RESPONSE", "1") + monkeypatch.setattr( + vedastro_user_entrypoint, + "build_runtime_mode_report", + lambda: {"mode": "official_extended", "official_ready": True, "readiness_blockers": []}, + ) + monkeypatch.setattr( + vedastro_user_entrypoint, + "_run_capability_catalog", + lambda _case: {"status": "ok", "available": True, "summary": {"catalog_method_count": 1}}, + ) + monkeypatch.setattr( + vedastro_user_entrypoint, + "_run_official_full_snapshot", + lambda _case: {"status": "ok", "available": True, "raw_response": raw_response}, + ) + + packet = vedastro_gateway.run_gateway_packet( + {"year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15, "lat": 37.7749, "lon": -122.4194, "tz": 8}, + question="事业机会什么时候出现", + themes=["career"], + reference_date="2026-07-02", + ) + + assert packet["official_closure_state"] == "official_verified" + assert packet["official_closure_reason"] == "official_raw_response_present" + assert packet["official_raw_response"] == raw_response + + def test_gateway_queue_lifecycle_uses_file_job_store(monkeypatch, tmp_path): from scripts import vedastro_gateway diff --git a/tests/test_vedastro_user_entrypoint.py b/tests/test_vedastro_user_entrypoint.py index fae48940..1a097354 100644 --- a/tests/test_vedastro_user_entrypoint.py +++ b/tests/test_vedastro_user_entrypoint.py @@ -132,6 +132,87 @@ def test_user_entrypoint_runs_catalog_and_strict_workflow_contract() -> None: assert "official_full_capability_catalog" in report["user_commands"]["json"] +def test_user_entrypoint_emits_official_raw_response_when_requested() -> None: + catalog_stub = { + "available": True, + "status": "ok", + "capabilities": [ + { + "method": "AllPlanetData", + "signature": "(birthTime)", + "bucket": "chart_core", + "parameter_names": ["birthTime"], + "callable": True, + } + ], + "buckets": {"chart_core": {"count": 1, "examples": ["AllPlanetData"]}}, + } + raw_response = { + "source": "vedastro_official_full_snapshot", + "sections": {"chart_core": {"Status": "Pass"}}, + "section_statuses": {"chart_core": "ok"}, + } + snapshot_stub = { + "status": "ok", + "available": True, + "operation": "official_full_snapshot", + "raw_response": raw_response, + "source_metadata": {"artifact_path": "scratch/local/vedastro_adapter/stub.json"}, + } + + completed = subprocess.run( + [ + sys.executable, + "scripts/vedastro_user_entrypoint.py", + "--year", + "1955", + "--month", + "2", + "--day", + "24", + "--hour", + "19", + "--minute", + "15", + "--lat", + "37.7749", + "--lon", + "-122.4194", + "--tz", + "8", + "--question", + "事业机会什么时候出现", + "--themes", + "career", + "--reference-date", + "2026-07-02", + "--require-official-raw-response", + "--format", + "json", + ], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + env={ + **os.environ, + "JYOTISH_SKIP_LOCAL_ENV": "1", + "VEDASTRO_API_ENDPOINT": "https://api.vedastro.org/api", + "VEDASTRO_ENABLE_NETWORK": "1", + "VEDASTRO_TIMEOUT_SECONDS": "20", + "VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB": json.dumps(catalog_stub), + "VEDASTRO_OFFICIAL_FULL_SNAPSHOT_STUB": json.dumps(snapshot_stub), + }, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + assert report["vedastro_official_full_snapshot"]["status"] == "ok" + assert report["vedastro_official_full_snapshot"]["raw_response_available"] is True + assert report["official_raw_response"] == raw_response + + def test_user_entrypoint_markdown_documents_boundaries() -> None: completed = subprocess.run( [