From 09dcfb9bccc88247c7edb2da10a819b637f7b142 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Wed, 8 Jul 2026 20:35:42 +0800 Subject: [PATCH] feat: list VedAstro official raw archives --- docs/research/pre_work_error_ledger.md | 1 + scripts/jyotish_api_server.py | 7 +++++++ scripts/vedastro_gateway.py | 29 ++++++++++++++++++++++++++ tests/test_api_server_security.py | 21 +++++++++++++++++++ tests/test_vedastro_gateway.py | 20 ++++++++++++++++++ 5 files changed, 78 insertions(+) diff --git a/docs/research/pre_work_error_ledger.md b/docs/research/pre_work_error_ledger.md index 9355ab6a..142366aa 100644 --- a/docs/research/pre_work_error_ledger.md +++ b/docs/research/pre_work_error_ledger.md @@ -50,6 +50,7 @@ For large architecture or release work, also read: | ERR-017 | Pre-work gate could pass without checking the older Round 25 fragment sweep or aggregate external-engine adapter diagnostics. | mitigated 2026-07-05 | `scripts/pre_work_check.py` must require both whole-machine sweep docs and run `scripts/diagnose_external_engine_adapters.py --json` before substantial work. | | ERR-018 | External engine blockers can be described verbally but not carried into diagnostics. | mitigated 2026-07-05 | `diagnose_external_engine_adapters.py` must expose VedAstro closure plan and PyJHora/JHora install/license/ephemeris boundary; keep `docs/research/external_engine_blocker_research_2026_07_05.md` current. | | ERR-019 | WorkBuddy/cloud/local acceptance summaries can invent pass counts, stale asset counts, non-existent error docs, or wrong dasha windows. | mitigated 2026-07-06 | Read `docs/research/user_invocation_acceptance_error_log_2026_07_06.md`; run `scripts/user_invocation_acceptance_check.py` and `tests/test_user_invocation_acceptance_contract.py` before accepting ordinary-user skill invocation validation claims. | +| ERR-020 | VedAstro official raw responses can be archived but hard to audit if no manifest/API listing exposes them. | mitigated 2026-07-08 | Keep `list_official_raw_response_archives()` and `GET /api/vedastro_gateway/archives`; tests must prove archived official raw responses are enumerable. | ## Fragment Sweep Command Set diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 9e952d5a..12d017ae 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -877,6 +877,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): self._json(self._vedastro_status()) elif path == '/api/vedastro_gateway/status': self._json(self._compute_vedastro_gateway_status()) + elif path == '/api/vedastro_gateway/archives': + self._json(self._compute_vedastro_gateway_archives()) elif path.startswith('/api/vedastro_gateway/jobs/'): job_id = path.rsplit('/', 1)[-1] result = self._compute_vedastro_gateway_job(job_id) @@ -1766,6 +1768,11 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): return gateway_status() + def _compute_vedastro_gateway_archives(self): + from scripts.vedastro_gateway import list_official_raw_response_archives + + return list_official_raw_response_archives() + def _compute_vedastro_gateway_job(self, job_id): from scripts.vedastro_gateway import get_gateway_job diff --git a/scripts/vedastro_gateway.py b/scripts/vedastro_gateway.py index 5e82b688..8d53e51f 100644 --- a/scripts/vedastro_gateway.py +++ b/scripts/vedastro_gateway.py @@ -169,6 +169,35 @@ def complete_gateway_job(job_id: str, result: dict[str, Any]) -> dict[str, Any]: return _write_job(job) +def list_official_raw_response_archives() -> dict[str, Any]: + archives: list[dict[str, Any]] = [] + queue_dir = _queue_dir() + if queue_dir.exists(): + for path in sorted(queue_dir.glob("*.json")): + if path.name.endswith(".official_raw_response.json"): + continue + try: + job = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + continue + archive = job.get("raw_response_archive") if isinstance(job, dict) else {} + if not isinstance(archive, dict) or not archive.get("official_raw_response_available"): + continue + archives.append( + { + "job_id": job.get("job_id"), + "status": archive.get("status"), + "official_raw_response_available": True, + "official_raw_response_path": archive.get("official_raw_response_path"), + } + ) + return { + "scope": "vedastro_official_raw_response_archive_manifest", + "archive_count": len(archives), + "archives": archives, + } + + def run_gateway_job(job_id: str) -> dict[str, Any] | None: job = get_gateway_job(job_id) if job is None: diff --git a/tests/test_api_server_security.py b/tests/test_api_server_security.py index a04c297a..908100da 100644 --- a/tests/test_api_server_security.py +++ b/tests/test_api_server_security.py @@ -347,6 +347,27 @@ def test_vedastro_gateway_poll_rejects_invalid_job_id(monkeypatch, tmp_path) -> assert poller.payload()['error_code'] == 'ERR_NOT_FOUND' +def test_vedastro_gateway_archive_manifest_route(monkeypatch, tmp_path) -> None: + from scripts import vedastro_gateway + + monkeypatch.setenv('VEDASTRO_GATEWAY_QUEUE_DIR', str(tmp_path)) + job = vedastro_gateway.enqueue_gateway_job({'year': 1955}, question='x') + vedastro_gateway.complete_gateway_job( + job['job_id'], + {'status': 'ok', 'official_raw_response': {'source': 'vedastro_official'}}, + ) + + reader = _ResponseCaptureHandler() + reader.path = '/api/vedastro_gateway/archives' + reader.do_GET() + + assert reader.status_code == 200 + payload = reader.payload() + assert payload['scope'] == 'vedastro_official_raw_response_archive_manifest' + assert payload['archive_count'] == 1 + assert payload['archives'][0]['job_id'] == job['job_id'] + + def test_vedastro_gateway_job_run_route_executes_worker(monkeypatch, tmp_path) -> None: from scripts import vedastro_gateway diff --git a/tests/test_vedastro_gateway.py b/tests/test_vedastro_gateway.py index 3e23c046..1c1a5fbb 100644 --- a/tests/test_vedastro_gateway.py +++ b/tests/test_vedastro_gateway.py @@ -138,6 +138,26 @@ def test_gateway_completion_archives_official_raw_response(monkeypatch, tmp_path assert '"vedastro_official"' in path.read_text(encoding="utf-8") +def test_gateway_lists_official_raw_response_archives(monkeypatch, tmp_path): + from scripts import vedastro_gateway + + monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_DIR", str(tmp_path)) + empty = vedastro_gateway.list_official_raw_response_archives() + assert empty["archive_count"] == 0 + + job = vedastro_gateway.enqueue_gateway_job({"year": 1955}, question="x") + vedastro_gateway.complete_gateway_job( + job["job_id"], + {"status": "ok", "official_raw_response": {"source": "vedastro_official"}}, + ) + manifest = vedastro_gateway.list_official_raw_response_archives() + + assert manifest["scope"] == "vedastro_official_raw_response_archive_manifest" + assert manifest["archive_count"] == 1 + assert manifest["archives"][0]["job_id"] == job["job_id"] + assert manifest["archives"][0]["official_raw_response_available"] is True + + def test_gateway_run_job_executes_queued_request(monkeypatch, tmp_path): from scripts import vedastro_gateway