root/Jyotisha
2
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat: propagate VedAstro official raw snapshots

Browse Source
This commit is contained in:
732642856
2026-07-09 09:54:08 +08:00
parent 27aa2db40c
commit 9bce83edc8
4 changed files with 195 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/research/pre_work_error_ledger.md
+1
View File
@@ -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
scripts/vedastro_user_entrypoint.py
+74 -3
View File
@@ -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()
tests/test_vedastro_gateway.py
+39
View File
@@ -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
tests/test_vedastro_user_entrypoint.py
+81
View File
@@ -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(
[