diff --git a/SKILL.md b/SKILL.md index 25c92818..95f28c6d 100644 --- a/SKILL.md +++ b/SKILL.md @@ -94,6 +94,12 @@ adapter available 解释为已完成 VedAstro、PyJHora/JHora 或 jyotishganit r - `official_blocked`：官方请求失败、额度/网络/超时受阻； - `local_fallback`：本地计算继续可用，但不能称为官方云端闭环。 +### Web/API 任务存储 + +默认 `JYOTISH_ASYNC_JOB_BACKEND=file` 使用本机受限权限的临时任务文件。单机部署可设 +`JYOTISH_ASYNC_JOB_BACKEND=sqlite`，使用 `scratch/local/async_jobs.sqlite3` 保存 token-hash +与 TTL 任务记录。两种后端都不是 Redis、多节点队列或跨主机 worker；不得把它们描述为分布式恢复能力。 + **强制工作流**（完整规范 → `references/ai-reading-workflow-prompt.md` v5.1.0）： 0. **阶段负一**：问题类型路由（事业/婚恋/财务/应期/历史验证/综合解盘）→ 必须先读 `references/strict-workflow-router.md`，按对应 strict checklist 执行；用户不需要主动点名高级技法。 diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 4b9199a9..3cfe9012 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -16,6 +16,7 @@ import importlib.util import hashlib import re import secrets +import sqlite3 import threading import time from concurrent.futures import ThreadPoolExecutor @@ -112,18 +113,28 @@ def enforce_rate_limit(client_id: str, *, now: float | None = None) -> None: 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': { + if _async_job_backend() == "sqlite": + with _sqlite_job_connection() as connection: + counts = { + scope: connection.execute("SELECT COUNT(*) FROM async_jobs WHERE scope = ?", (scope,)).fetchone()[0] + for scope in scopes + } + storage = "sqlite_single_host" + else: + 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.', + } + storage = "local_file_single_host" + return { + 'scope': 'async_job_runtime_status', + 'storage': storage, + 'worker_count': _ASYNC_JOB_WORKERS, + 'queue_size': _ASYNC_JOB_QUEUE_SIZE, + 'ttl_seconds': _async_job_ttl_seconds(), + 'record_counts': counts, + 'boundary': 'SQLite supports single-host persistence. No distributed queue or multi-node worker guarantee.', } @@ -569,10 +580,39 @@ def _async_job_ttl_seconds() -> float: return 3600.0 +def _async_job_backend() -> str: + return "sqlite" if os.environ.get("JYOTISH_ASYNC_JOB_BACKEND", "file").strip().lower() == "sqlite" else "file" + + +def _sqlite_job_db_path() -> Path: + return Path(REPO_ROOT) / "scratch" / "local" / "async_jobs.sqlite3" + + +def _sqlite_job_connection() -> sqlite3.Connection: + path = _sqlite_job_db_path() + path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(path, timeout=10) + connection.execute( + "CREATE TABLE IF NOT EXISTS async_jobs (scope TEXT NOT NULL, job_id TEXT NOT NULL, expires_at REAL, payload TEXT NOT NULL, PRIMARY KEY (scope, job_id))" + ) + try: + os.chmod(path, 0o600) + except OSError: + pass + return connection + + def prune_expired_async_jobs() -> dict: """Best-effort startup cleanup for local job records; never reads payloads.""" removed = 0 scanned = 0 + if _async_job_backend() == "sqlite": + with _sqlite_job_connection() as connection: + scanned = connection.execute("SELECT COUNT(*) FROM async_jobs").fetchone()[0] + removed = connection.execute( + "DELETE FROM async_jobs WHERE expires_at IS NOT NULL AND expires_at <= ?", (time.time(),) + ).rowcount + return {'scope': 'async_job_cleanup', 'scanned': scanned, 'removed': removed} for scope in (_HIGH_RIGOR_JOB_SCOPE, _API_CHART_CACHE_SCOPE): directory = _async_job_dir(scope) if not directory.is_dir(): @@ -614,19 +654,36 @@ def _write_high_rigor_job_record(job_id: str, payload: dict) -> dict: def _load_async_job_record(scope: str, job_id: str, *, access_token: str = '') -> dict | None: - path = _async_job_path(scope, job_id) - if not path.exists(): - return None - try: - record = json.loads(path.read_text(encoding='utf-8')) - except (OSError, json.JSONDecodeError): - return None + path = None + if _async_job_backend() == "sqlite": + with _sqlite_job_connection() as connection: + row = connection.execute( + "SELECT payload FROM async_jobs WHERE scope = ? AND job_id = ?", (scope, job_id) + ).fetchone() + if row is None: + return None + try: + record = json.loads(row[0]) + except json.JSONDecodeError: + return None + else: + path = _async_job_path(scope, job_id) + if not path.exists(): + return None + try: + record = json.loads(path.read_text(encoding='utf-8')) + except (OSError, json.JSONDecodeError): + return None expires_at = record.get('expires_at_unix') if isinstance(expires_at, (int, float)) and time.time() >= float(expires_at): - try: - path.unlink() - except OSError: - pass + if _async_job_backend() == "sqlite": + with _sqlite_job_connection() as connection: + connection.execute("DELETE FROM async_jobs WHERE scope = ? AND job_id = ?", (scope, job_id)) + elif path is not None: + try: + path.unlink() + except OSError: + pass return None expected = record.get('access_token_hash') if not isinstance(expected, str) or not access_token: @@ -637,6 +694,13 @@ def _load_async_job_record(scope: str, job_id: str, *, access_token: str = '') - def _write_async_job_record(scope: str, job_id: str, payload: dict) -> dict: + if _async_job_backend() == "sqlite": + with _sqlite_job_connection() as connection: + connection.execute( + "INSERT OR REPLACE INTO async_jobs (scope, job_id, expires_at, payload) VALUES (?, ?, ?, ?)", + (scope, job_id, payload.get("expires_at_unix"), json.dumps(payload, ensure_ascii=False, sort_keys=True)), + ) + return payload path = _async_job_path(scope, job_id) temp_path = path.with_suffix(f'.{secrets.token_hex(8)}.tmp') temp_path.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True), encoding='utf-8') diff --git a/tests/test_api_async_job_contract.py b/tests/test_api_async_job_contract.py index 8a075469..994dcaf3 100644 --- a/tests/test_api_async_job_contract.py +++ b/tests/test_api_async_job_contract.py @@ -91,3 +91,15 @@ def test_rate_limit_is_configurable_and_rejects_over_budget(monkeypatch): api.enforce_rate_limit("test-client", now=0) with pytest.raises(api.RateLimited): api.enforce_rate_limit("test-client", now=1) + + +def test_sqlite_async_job_backend_preserves_token_and_ttl(monkeypatch, tmp_path): + monkeypatch.setenv("JYOTISH_ASYNC_JOB_BACKEND", "sqlite") + monkeypatch.setattr(api, "_sqlite_job_db_path", lambda: tmp_path / "jobs.sqlite3") + record = {"job_id": "job_sqlite", "access_token_hash": api._access_token_hash("secret"), "expires_at_unix": time.time() + 60} + + api._write_async_job_record("test_scope", "job_sqlite", record) + + assert api._load_async_job_record("test_scope", "job_sqlite", access_token="secret")["job_id"] == "job_sqlite" + with pytest.raises(api.JobAccessDenied): + api._load_async_job_record("test_scope", "job_sqlite", access_token="wrong")