From b336bdd9ac75ee5071ff8dd43178a5405117eb53 Mon Sep 17 00:00:00 2001
From: 732642856 <732642856@qq.com>
Date: Sun, 12 Jul 2026 13:40:26 +0800
Subject: [PATCH] harden consultation reliability and rectification flow
---
docs/research/pre_work_error_ledger.md | 1 +
scripts/candidate_time_sensitivity_scan.py | 106 +++++++++++++++++
scripts/external_oracle_raw_import.py | 65 +++++++++++
scripts/jyotish_api_server.py | 108 +++++++++++++++++-
tests/test_api_async_job_contract.py | 34 ++++++
tests/test_candidate_time_sensitivity_scan.py | 21 ++++
tests/test_external_oracle_raw_import.py | 35 ++++++
web/index.html | 9 ++
web/rectification.html | 8 +-
9 files changed, 382 insertions(+), 5 deletions(-)
create mode 100644 scripts/candidate_time_sensitivity_scan.py
create mode 100644 scripts/external_oracle_raw_import.py
create mode 100644 tests/test_candidate_time_sensitivity_scan.py
create mode 100644 tests/test_external_oracle_raw_import.py
create mode 100644 web/index.html
diff --git a/docs/research/pre_work_error_ledger.md b/docs/research/pre_work_error_ledger.md
index b7dd94ae..1d5be933 100644
--- a/docs/research/pre_work_error_ledger.md
+++ b/docs/research/pre_work_error_ledger.md
@@ -78,6 +78,7 @@ For large architecture or release work, also read:
| ERR-045 | Three-engine readiness was mistaken for completed same-chart parity. Public replay on 2026-07-11 captured PyJHora and jyotishganit raw, but VedAstro returned `official_snapshot_budget_exhausted` with no raw response. | active external blocker | Keep `three_engine_parity_runner.py`; status remains `blocked`/`partial` until all required raw artifacts are normalized into comparison rows. |
| ERR-046 | Report-renderer SSRF/file PoC could not run because the Playwright Chromium binary was absent and installation exceeded the desktop outer timeout. | blocked environment | Keep route/JS-denial tests; rerun isolated HTTP/file PoC only after a verified Chromium installation, then update this ledger with the measured request count. |
| ERR-047 | Initial `slow` marker partition for `test_api_server_security.py` still exceeded the 120-second desktop budget; heavy paths extend beyond VedAstro/high-rigor prefix groups. | active profiling blocker | Profile test node IDs in bounded subprocess batches, mark only measured heavy tests, and keep fast-security acceptance separate from long CI integration coverage. |
+| ERR-048 | Candidate-time scanner assumed all documented D4/D24/D30 divisions were exposed by `jyotish_engine.py varga`; actual `--d4` failed at runtime. | mitigated 2026-07-12 | Candidate scans must record unsupported Varga flags as `unavailable_vargas`; only successfully computed D1/D9/D10 fields may drive local sensitivity output until a unified Varga contract exists. |
## Fragment Sweep Command Set
diff --git a/scripts/candidate_time_sensitivity_scan.py b/scripts/candidate_time_sensitivity_scan.py
new file mode 100644
index 00000000..cbe40f4c
--- /dev/null
+++ b/scripts/candidate_time_sensitivity_scan.py
@@ -0,0 +1,106 @@
+#!/usr/bin/env python3
+"""Scan actual local-chart differences across a birth-time candidate range."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+from collections import Counter
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ENGINE = ROOT / "scripts" / "jyotish_engine.py"
+_VARGAS = ("d4", "d9", "d10", "d24", "d30")
+
+
+def _engine_json(command: str, payload: dict[str, Any], *, timeout: int = 20) -> dict[str, Any]:
+ args = ["python3", str(ENGINE), command]
+ for key in ("year", "month", "day", "hour", "minute", "lat", "lon", "tz"):
+ args.extend([f"--{key}", str(payload[key])])
+ if command == "varga":
+ args.append(f"--{payload['varga']}")
+ completed = subprocess.run(args, cwd=ROOT, capture_output=True, text=True, timeout=timeout, check=True)
+ return json.loads(completed.stdout)
+
+
+def _varga_ascendant(payload: dict[str, Any], varga: str) -> str | None:
+ try:
+ raw = _engine_json("varga", {**payload, "varga": varga})
+ except subprocess.CalledProcessError:
+ return None
+ charts = raw.get("divisional_charts", {})
+ for chart in charts.values():
+ if isinstance(chart, dict):
+ return chart.get("ascendant")
+ return None
+
+
+def scan_candidate_times(payload: dict[str, Any], *, uncertainty_minutes: int = 30, step_minutes: int = 1) -> dict[str, Any]:
+ required = ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")
+ missing = [key for key in required if payload.get(key) is None]
+ if missing:
+ raise ValueError(f"missing candidate scan fields: {', '.join(missing)}")
+ center = datetime(int(payload["year"]), int(payload["month"]), int(payload["day"]), int(payload["hour"]), int(payload["minute"]))
+ step_minutes = max(int(step_minutes), 1)
+ uncertainty_minutes = max(int(uncertainty_minutes), 1)
+ rows: list[dict[str, Any]] = []
+ for offset in range(-uncertainty_minutes, uncertainty_minutes + 1, step_minutes):
+ moment = center + timedelta(minutes=offset)
+ point = {**payload, "year": moment.year, "month": moment.month, "day": moment.day, "hour": moment.hour, "minute": moment.minute}
+ chart = _engine_json("chart", point)
+ asc = chart.get("ascendant", {})
+ divisional = {varga.upper(): _varga_ascendant(point, varga) for varga in _VARGAS}
+ rows.append({
+ "time": moment.strftime("%Y-%m-%d %H:%M"),
+ "offset_minutes": offset,
+ "d1_ascendant": asc.get("sign"),
+ "d1_degree_in_sign": asc.get("degree_in_sign"),
+ "divisional_ascendants": divisional,
+ })
+ signatures = [tuple([row["d1_ascendant"], *row["divisional_ascendants"].values()]) for row in rows]
+ modal = Counter(signatures).most_common(1)[0][0]
+ for row, signature in zip(rows, signatures):
+ row["sensitivity_count"] = sum(left != right for left, right in zip(signature, modal))
+ row["sensitive_layers"] = [
+ name for name, current, typical in zip(("D1", "D4", "D9", "D10", "D24", "D30"), signature, modal)
+ if current != typical
+ ]
+ unavailable_vargas = [varga.upper() for varga in _VARGAS if all(row["divisional_ascendants"][varga.upper()] is None for row in rows)]
+ transitions = []
+ for previous, current in zip(rows, rows[1:]):
+ changed = [name for name in ("d1_ascendant", "divisional_ascendants") if previous[name] != current[name]]
+ if changed:
+ transitions.append({"between": [previous["time"], current["time"]], "changed": changed})
+ return {
+ "scope": "candidate_time_sensitivity_scan",
+ "status": "local_computed",
+ "engine": "local_jyotish_engine",
+ "candidate_count": len(rows),
+ "center_time": center.strftime("%Y-%m-%d %H:%M"),
+ "uncertainty_minutes": uncertainty_minutes,
+ "step_minutes": step_minutes,
+ "rows": rows,
+ "transitions": transitions,
+ "unavailable_vargas": unavailable_vargas,
+ "pending_layers": ["UL", "A7", "A10", "KP_cusp"],
+ "boundary": "Actual local D1/Varga differences only. Unsupported Varga CLI flags are explicitly unavailable. Event answers still require an explicit event-to-candidate adjudication model before minute-level rectification.",
+ }
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ for field, cast in (("year", int), ("month", int), ("day", int), ("hour", int), ("minute", int), ("lat", float), ("lon", float), ("tz", float)):
+ parser.add_argument(f"--{field}", required=True, type=cast)
+ parser.add_argument("--uncertainty-minutes", type=int, default=30)
+ parser.add_argument("--step-minutes", type=int, default=1)
+ args = parser.parse_args()
+ print(json.dumps(scan_candidate_times(vars(args), uncertainty_minutes=args.uncertainty_minutes, step_minutes=args.step_minutes), ensure_ascii=False, indent=2))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/external_oracle_raw_import.py b/scripts/external_oracle_raw_import.py
new file mode 100644
index 00000000..e20889a6
--- /dev/null
+++ b/scripts/external_oracle_raw_import.py
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Validate a reviewable external raw-oracle artifact before parity replay."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+
+SUPPORTED_ENGINES = {"VedAstro", "PyJHora_JHora", "jyotishganit"}
+
+
+def sha256_file(path: Path) -> str:
+ digest = hashlib.sha256()
+ with path.open("rb") as handle:
+ for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+
+
+def build_raw_oracle_import(engine: str, artifact_path: str | Path, metadata: dict[str, Any]) -> dict[str, Any]:
+ if engine not in SUPPORTED_ENGINES:
+ raise ValueError(f"unsupported oracle engine: {engine}")
+ path = Path(artifact_path).expanduser().resolve()
+ if not path.is_file():
+ raise ValueError("source artifact does not exist")
+ required = ("case_id", "license_boundary", "collection_method", "birth_data_policy")
+ missing = [key for key in required if not metadata.get(key)]
+ if missing:
+ raise ValueError(f"missing raw-oracle metadata: {', '.join(missing)}")
+ if metadata["birth_data_policy"] != "public_case_only":
+ raise ValueError("raw-oracle imports require public_case_only birth data")
+ return {
+ "scope": "external_raw_oracle_import",
+ "schema_version": 1,
+ "engine": engine,
+ "status": "raw_imported_uncompared",
+ "source_artifact": str(path),
+ "source_artifact_sha256": sha256_file(path),
+ "metadata": {
+ key: metadata[key]
+ for key in (*required, "engine_version", "ayanamsa", "node_mode", "captured_at")
+ if metadata.get(key) is not None
+ },
+ "comparison_ready": False,
+ "boundary": "Import integrity only. Parity is external_verified only after normalized field comparison passes.",
+ }
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--engine", required=True, choices=sorted(SUPPORTED_ENGINES))
+ parser.add_argument("--artifact", required=True)
+ parser.add_argument("--metadata-json", required=True, help="JSON file containing import metadata")
+ args = parser.parse_args()
+ metadata = json.loads(Path(args.metadata_json).read_text(encoding="utf-8"))
+ print(json.dumps(build_raw_oracle_import(args.engine, args.artifact, metadata), ensure_ascii=False, indent=2))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py
index ac3645d2..d805157a 100644
--- a/scripts/jyotish_api_server.py
+++ b/scripts/jyotish_api_server.py
@@ -44,6 +44,10 @@ except ModuleNotFoundError: # pragma: no cover - script execution path
score_rectification_answers,
summarize_execution_status,
)
+try:
+ from scripts.candidate_time_sensitivity_scan import scan_candidate_times
+except ModuleNotFoundError: # pragma: no cover - script execution path
+ from candidate_time_sensitivity_scan import scan_candidate_times
try:
from scripts.western_oracle_adapter import build_packet_from_oracle_payload
except ModuleNotFoundError: # pragma: no cover - script execution path
@@ -64,6 +68,8 @@ _ASYNC_JOB_EXECUTOR = ThreadPoolExecutor(
thread_name_prefix='jyotish-job',
)
_ASYNC_JOB_CAPACITY = threading.BoundedSemaphore(_ASYNC_JOB_WORKERS + _ASYNC_JOB_QUEUE_SIZE)
+_RATE_LIMIT_LOCK = threading.Lock()
+_RATE_LIMIT_BUCKETS: dict[str, tuple[float, int]] = {}
def build_evidence_packet_view(job_record: dict | None) -> dict:
@@ -82,6 +88,45 @@ def build_evidence_packet_view(job_record: dict | None) -> dict:
}
+def _rate_limit_per_minute() -> int:
+ raw = str(os.environ.get('JYOTISH_API_RATE_LIMIT_PER_MINUTE', '120')).strip()
+ try:
+ return max(int(raw), 0)
+ except ValueError:
+ return 120
+
+
+def enforce_rate_limit(client_id: str, *, now: float | None = None) -> None:
+ limit = _rate_limit_per_minute()
+ if limit == 0:
+ return
+ now = time.time() if now is None else now
+ with _RATE_LIMIT_LOCK:
+ window, count = _RATE_LIMIT_BUCKETS.get(client_id, (now, 0))
+ if now - window >= 60:
+ window, count = now, 0
+ if count >= limit:
+ raise RateLimited('Rate limit exceeded')
+ _RATE_LIMIT_BUCKETS[client_id] = (window, count + 1)
+
+
+def async_job_runtime_status() -> dict:
+ scopes = (_HIGH_RIGOR_JOB_SCOPE, _API_CHART_CACHE_SCOPE)
+ return {
+ 'scope': 'async_job_runtime_status',
+ 'storage': 'local_file_single_host',
+ 'worker_count': _ASYNC_JOB_WORKERS,
+ 'queue_size': _ASYNC_JOB_QUEUE_SIZE,
+ 'ttl_seconds': _async_job_ttl_seconds(),
+ 'record_counts': {
+ scope: len(list(_async_job_dir(scope).glob('*.json')))
+ if _async_job_dir(scope).is_dir() else 0
+ for scope in scopes
+ },
+ 'boundary': 'No cross-process queue, restart recovery, or distributed worker guarantee.',
+ }
+
+
def _submit_background_job(callback):
if not _ASYNC_JOB_CAPACITY.acquire(blocking=False):
raise JobQueueFull('Async job queue is full')
@@ -524,6 +569,27 @@ def _async_job_ttl_seconds() -> float:
return 3600.0
+def prune_expired_async_jobs() -> dict:
+ """Best-effort startup cleanup for local job records; never reads payloads."""
+ removed = 0
+ scanned = 0
+ for scope in (_HIGH_RIGOR_JOB_SCOPE, _API_CHART_CACHE_SCOPE):
+ directory = _async_job_dir(scope)
+ if not directory.is_dir():
+ continue
+ for path in directory.glob('*.json'):
+ scanned += 1
+ try:
+ record = json.loads(path.read_text(encoding='utf-8'))
+ expires_at = record.get('expires_at_unix') if isinstance(record, dict) else None
+ if isinstance(expires_at, (int, float)) and time.time() >= float(expires_at):
+ path.unlink()
+ removed += 1
+ except (OSError, json.JSONDecodeError):
+ continue
+ return {'scope': 'async_job_cleanup', 'scanned': scanned, 'removed': removed}
+
+
def _new_async_job_identity(prefix: str) -> dict:
return {
'job_id': f'{prefix}_{secrets.token_hex(16)}',
@@ -911,6 +977,10 @@ class JobQueueFull(RuntimeError):
"""Bounded async worker queue has no remaining capacity."""
+class RateLimited(RuntimeError):
+ """Client exceeded the local fixed-window request budget."""
+
+
class JyotishAPIHandler(BaseHTTPRequestHandler):
server_version = 'JyotishAPI/6.9.14'
@@ -957,6 +1027,9 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
content_type = (self.headers.get('Content-Type') or '').split(';', 1)[0].strip().lower()
if content_type != 'application/json':
raise UnsupportedMediaType('Content-Type must be application/json')
+ if urlparse(self.path).path.startswith('/api/'):
+ client = getattr(self, 'client_address', ('unknown',))[0]
+ enforce_rate_limit(str(client))
def _job_access_token(self):
authorization = self.headers.get('Authorization') or ''
@@ -1011,6 +1084,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
try:
self._enforce_request_security()
self._json({})
+ except RateLimited as exc:
+ self._error_json(str(exc), 429, 'ERR_RATE_LIMITED')
except Forbidden as exc:
self._error_json(str(exc), 403, 'ERR_FORBIDDEN')
@@ -1018,7 +1093,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
path = urlparse(self.path).path
try:
self._enforce_request_security()
- if path == '/evidence':
+ if path == '/':
+ page = Path(REPO_ROOT) / 'web' / 'index.html'
+ if not page.is_file():
+ self._error_json('Home page unavailable', 404, 'ERR_NOT_FOUND')
+ else:
+ self._html(page.read_text(encoding='utf-8'))
+ elif path == '/evidence':
page = Path(REPO_ROOT) / 'web' / 'evidence_packet.html'
if not page.is_file():
self._error_json('Evidence Packet page unavailable', 404, 'ERR_NOT_FOUND')
@@ -1047,6 +1128,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
'swisseph_version': swisseph_version,
'ayanamsa_default': 'lahiri',
'modules': 'Chart/KP/Synastry/Prashna/Remedies/Dasha/Varga/Jaimini/Ashtakavarga/Shadbala/Yoga/Aspects/Tajika/Muhurta/BhavaChalit/BhavaBala/Sudarshana/Nakshatra/Transit/RectificationGate/CaseValidation/DivisionalYoga/Kakshya',
+ 'async_job_runtime': async_job_runtime_status(),
})
elif path == '/api/cities':
self._json(list(CITY_DB.keys()))
@@ -1099,6 +1181,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
self._json(self._real_case_revalidation())
else:
self._error_json('Not found', 404, 'ERR_NOT_FOUND')
+ except RateLimited as exc:
+ self._error_json(str(exc), 429, 'ERR_RATE_LIMITED')
except (Forbidden, JobAccessDenied) as exc:
self._error_json(str(exc), 403, 'ERR_FORBIDDEN')
except Exception:
@@ -1111,7 +1195,17 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
try:
self._enforce_request_security(require_json=True)
body = self._read_json_body()
- if path == '/api/chart':
+ if path == '/api/location/resolve':
+ city = str(body.get('city') or '').strip()
+ city_aliases = {'beijing': '北京', 'shanghai': '上海', 'guangzhou': '广州', 'shenzhen': '深圳'}
+ query = city_aliases.get(city.casefold(), city)
+ matched = next((name for name in CITY_DB if name.casefold() == query.casefold()), None)
+ if not matched:
+ self._error_json('City not found in local city database', 404, 'ERR_CITY_NOT_FOUND')
+ else:
+ lat, lon, tz = CITY_DB[matched]
+ self._json({'status': 'local_city_match', 'city': matched, 'lat': lat, 'lon': lon, 'tz': tz})
+ elif path == '/api/chart':
result = self._compute_chart(body)
self._json(result)
elif path == '/api/remedies':
@@ -1213,6 +1307,12 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
self._json(result)
elif path == '/api/rectification/questionnaire':
self._json(build_rectification_questionnaire(body))
+ elif path == '/api/rectification/sensitivity_scan':
+ self._json(scan_candidate_times(
+ body,
+ uncertainty_minutes=int(body.get('time_uncertainty_minutes') or 30),
+ step_minutes=int(body.get('step_minutes') or 1),
+ ))
elif path == '/api/rectification/answers':
questionnaire = body.get('questionnaire')
answers = body.get('answers')
@@ -1254,6 +1354,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
self._json(result)
else:
self._error_json(f'Unknown endpoint: {path}', 404, 'ERR_NOT_FOUND')
+ except RateLimited as exc:
+ self._error_json(str(exc), 429, 'ERR_RATE_LIMITED')
except BadRequest as e:
self._error_json(str(e), 400, 'ERR_BAD_REQUEST')
except Forbidden as exc:
@@ -7641,10 +7743,12 @@ def _parse_allowed_origins(value):
def start_server(port=5200, host='127.0.0.1', allowed_origins=None):
+ cleanup = prune_expired_async_jobs()
server = ThreadingHTTPServer((host, port), JyotishAPIHandler)
server.daemon_threads = True
server.allowed_origins = allowed_origins or DEFAULT_ALLOWED_ORIGINS
print(f'Jyotish API v6.9.14 running on http://{host}:{port}')
+ print(f" Async job cleanup: scanned={cleanup['scanned']}, removed={cleanup['removed']}")
print(f' CORS origins: {", ".join(sorted(server.allowed_origins))}')
print(f' POST /api/chart — 完整星盘计算')
print(f' POST /api/remedies — 补救建议')
diff --git a/tests/test_api_async_job_contract.py b/tests/test_api_async_job_contract.py
index 4013344d..8a075469 100644
--- a/tests/test_api_async_job_contract.py
+++ b/tests/test_api_async_job_contract.py
@@ -1,6 +1,9 @@
import json
+import time
from pathlib import Path
+import pytest
+
from scripts import jyotish_api_server as api
@@ -57,3 +60,34 @@ def test_rectification_page_uses_choice_questionnaire_contract():
assert "/api/rectification/questionnaire" in source
assert "/api/rectification/answers" in source
assert "候选簇排序" in source
+
+
+def test_home_page_keeps_location_confirmation_local():
+ page = Path(api.REPO_ROOT) / "web" / "index.html"
+ source = page.read_text(encoding="utf-8")
+
+ assert "/api/location/resolve" in source
+ assert "第三方地理服务" in source
+
+
+def test_startup_cleanup_removes_only_expired_job_records(monkeypatch, tmp_path):
+ expired = tmp_path / "expired.json"
+ active = tmp_path / "active.json"
+ expired.write_text(json.dumps({"expires_at_unix": time.time() - 1}), encoding="utf-8")
+ active.write_text(json.dumps({"expires_at_unix": time.time() + 60}), encoding="utf-8")
+ monkeypatch.setattr(api, "_async_job_dir", lambda scope: tmp_path)
+
+ result = api.prune_expired_async_jobs()
+
+ assert result["removed"] == 1
+ assert not expired.exists()
+ assert active.exists()
+
+
+def test_rate_limit_is_configurable_and_rejects_over_budget(monkeypatch):
+ api._RATE_LIMIT_BUCKETS.clear()
+ monkeypatch.setenv("JYOTISH_API_RATE_LIMIT_PER_MINUTE", "1")
+
+ api.enforce_rate_limit("test-client", now=0)
+ with pytest.raises(api.RateLimited):
+ api.enforce_rate_limit("test-client", now=1)
diff --git a/tests/test_candidate_time_sensitivity_scan.py b/tests/test_candidate_time_sensitivity_scan.py
new file mode 100644
index 00000000..6076a07c
--- /dev/null
+++ b/tests/test_candidate_time_sensitivity_scan.py
@@ -0,0 +1,21 @@
+from scripts import candidate_time_sensitivity_scan as scanner
+
+
+def test_scanner_reports_real_divisional_transitions(monkeypatch):
+ def fake_engine(command, payload, timeout=20):
+ minute = payload["minute"]
+ if command == "chart":
+ return {"ascendant": {"sign": "Leo", "degree_in_sign": 10 + minute / 100}}
+ ascendant = "Aries" if minute % 2 else "Taurus"
+ return {"divisional_charts": {"D": {"ascendant": ascendant}}}
+
+ monkeypatch.setattr(scanner, "_engine_json", fake_engine)
+ report = scanner.scan_candidate_times(
+ {"year": 2000, "month": 1, "day": 1, "hour": 12, "minute": 1, "lat": 1, "lon": 1, "tz": 0},
+ uncertainty_minutes=1,
+ )
+
+ assert report["candidate_count"] == 3
+ assert report["transitions"]
+ assert report["pending_layers"] == ["UL", "A7", "A10", "KP_cusp"]
+ assert report["rows"][0]["divisional_ascendants"]["D9"] in {"Aries", "Taurus"}
diff --git a/tests/test_external_oracle_raw_import.py b/tests/test_external_oracle_raw_import.py
new file mode 100644
index 00000000..58fb6e74
--- /dev/null
+++ b/tests/test_external_oracle_raw_import.py
@@ -0,0 +1,35 @@
+from pathlib import Path
+
+import pytest
+
+from scripts.external_oracle_raw_import import build_raw_oracle_import
+
+
+def _metadata(**overrides):
+ value = {
+ "case_id": "public_case",
+ "license_boundary": "external benchmark only",
+ "collection_method": "manual export",
+ "birth_data_policy": "public_case_only",
+ }
+ value.update(overrides)
+ return value
+
+
+def test_raw_import_requires_reviewable_public_case_artifact(tmp_path: Path):
+ artifact = tmp_path / "oracle.json"
+ artifact.write_text('{"raw": true}', encoding="utf-8")
+
+ result = build_raw_oracle_import("VedAstro", artifact, _metadata())
+
+ assert result["status"] == "raw_imported_uncompared"
+ assert result["source_artifact_sha256"]
+ assert result["comparison_ready"] is False
+
+
+def test_raw_import_rejects_non_public_birth_policy(tmp_path: Path):
+ artifact = tmp_path / "oracle.json"
+ artifact.write_text("{}", encoding="utf-8")
+
+ with pytest.raises(ValueError, match="public_case_only"):
+ build_raw_oracle_import("VedAstro", artifact, _metadata(birth_data_policy="private"))
diff --git a/web/index.html b/web/index.html
new file mode 100644
index 00000000..27aa93f9
--- /dev/null
+++ b/web/index.html
@@ -0,0 +1,9 @@
+
+
+Jyotish Consultation
+
+Jyotish Consultation
先确认出生资料，再选择直接排盘或主动问询式生时校正。外部引擎状态将在证据包中明示。
+
+
+
+
diff --git a/web/rectification.html b/web/rectification.html
index a5c0036b..ea77b72f 100644
--- a/web/rectification.html
+++ b/web/rectification.html
@@ -3,11 +3,13 @@
主动问询式生时校正
主动问询式生时校正
先扫描候选时间，再回答选择题。结果只缩小候选簇，不宣称已经精确到分钟。
-
+