From 0be284321c2ef021fbf39844782cc0cbe4b9adfa Mon Sep 17 00:00:00 2001 From: test Date: Sun, 23 Aug 2026 19:10:41 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=E5=B0=86=20Steam=20=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87=E8=BF=81=E7=A7=BB=E5=88=B0=20Timer=20?= =?UTF-8?q?=E7=BB=84=E5=90=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/plugin-api-v3.yml | 6 +- README.md | 74 +-- akashic.plugin.toml | 16 +- context_source.py | 189 ++++++ mcp/run_mcp.py | 42 +- mcp/steam_mcp.py | 37 -- mcp/steam_proactive.py | 338 ---------- mcp/tests/test_context_contract.py | 131 ++-- mcp/tests/test_snapshot_freshness.py | 205 +++--- mcp/tests/test_v3_runtime.py | 69 +- plugin.py | 55 +- pyrightconfig.json | 2 + steam_runtime/__init__.py | 1 + steam_runtime/backend.py | 621 ++++++++++++++++++ .../config.py | 4 +- tests/test_context_source.py | 229 +++++++ tests/test_manager_integration.py | 379 ++++++----- tests/test_plugin.py | 78 ++- 18 files changed, 1656 insertions(+), 820 deletions(-) create mode 100644 context_source.py delete mode 100644 mcp/steam_proactive.py create mode 100644 steam_runtime/__init__.py create mode 100644 steam_runtime/backend.py rename mcp/runtime_config.py => steam_runtime/config.py (94%) create mode 100644 tests/test_context_source.py diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index 8d0e2be..404767a 100644 --- a/.github/workflows/plugin-api-v3.yml +++ b/.github/workflows/plugin-api-v3.yml @@ -36,7 +36,7 @@ jobs: - uses: actions/checkout@v4 with: repository: kachofugetsu09/akashic-agent - ref: 3005f838bcd96e2cbc58616aede46e4f39df4523 + ref: 9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5 path: .akashic-core - uses: actions/setup-python@v5 with: @@ -60,8 +60,8 @@ jobs: - name: Check changed v3 sources env: PYTHONPATH: .akashic-core:mcp - run: mcp/.venv/bin/pyright plugin.py mcp/runtime_config.py mcp/run_mcp.py scripts tests + run: mcp/.venv/bin/pyright plugin.py context_source.py steam_runtime mcp/run_mcp.py mcp/steam_mcp.py scripts tests - name: Compile Python sources - run: python -m compileall -q plugin.py mcp scripts tests + run: python -m compileall -q plugin.py context_source.py steam_runtime mcp scripts tests - name: Check diff formatting run: git diff --check diff --git a/README.md b/README.md index b2fae2f..8cfd815 100644 --- a/README.md +++ b/README.md @@ -1,36 +1,41 @@ # steam-mcp -Akashic Steam plugin. It bundles: +Steam 是 Akashic Plugin API v3 插件。它用现有普通原语组合 current context： -- `steam` MCP server -- `steam-inventory-analyzer` skill - -## Install +```text +Core Timer ──触发──> Steam shared domain ──覆盖──> current presence + │ + └──有意义变化──追加──> game snapshot history -```bash -python main.py plugin-install --source https://github.com/akashic-plugins/steam-mcp --marketplace github +Wake BeforeTurn ──只读 fresh state──> extra_hints +Passive BeforeTurn ──────────────────> 不变 +用户 Turn ──调用 Steam MCP───────────> 主动查询 Steam API ``` -Restart Akashic after install. +## 能力与 owner -## Data directory +- `MCP_SERVERS`：保留用户主动查询工具。MCP 不再包含 context fetch 或手动 + snapshot 特权工具。 +- `TIMERS`：正式稳定 Root 独占一个 one-shot Timer。presence 每 5 分钟刷新； + 网络瞬时失败记录结构化 Incident，并在 60 秒后重试。 +- `turn.context_prepared`：只在 `channel=wake` 且 current presence 仍 fresh 时 + append 一个普通 hint；不会 abort、替换 prompt 或影响 passive Turn。 -Runtime data lives in: +Steam SQLite 是唯一 domain state owner： -```text -/plugin-data/steam-/ -``` +- `current_state` 是可覆盖 singleton，保存 current presence、当前游戏列表、刷新 + deadline 和最近错误。 +- `snapshots`、`snapshot_runs` 保存真实游戏快照历史，不自动裁切。相同快照或空 + 结果只推进 current check time，不制造历史。 +- 旧文件名 `steam_proactive.sqlite3` 为了原位继承正式历史而保留；运行代码、表 + owner 和插件能力已经不依赖旧主动系统。 -Common files: +纯诊断日志固定为 5 MiB、最多 3 个备份。candidate 只在自己的隔离目录完成 MCP +readiness/handshake：不注册 Timer、不访问 Steam 外网、不读取或写入正式 state。 -- `steam_mcp_config.json` -- `steam_user_cache.json` -- `steam_app_cache.json` -- `steam_proactive.sqlite3` +## 配置 -## Config - -Create `steam_mcp_config.json` in the plugin data directory: +在插件 data root 创建 `steam_mcp_config.json`： ```json { @@ -40,25 +45,18 @@ Create `steam_mcp_config.json` in the plugin data directory: } ``` -`get_steam_context` 每次读取实时在线状态，并在历史游戏时长快照超过 -`snapshot_interval_seconds` 时自动刷新。空的最近游玩列表也会记录快照批次，避免重复刷新。 +`snapshot_interval_seconds` 只控制游戏历史快照检查；presence 使用固定 5 分钟 +freshness，避免把历史采样频率和当前状态时效揉成一个概念。 ## v2 data migration -v3 不会在插件加载时隐式复制正式数据。停止 Akashic 后显式执行： - -```bash -PYTHONPATH=/path/to/akashic-agent \ -python scripts/migrate_v2_data.py \ - --workspace /path/to/workspace \ - --marketplace github -``` +显式迁移仍使用 `scripts/migrate_v2_data.py`。它保留 `mcp/steam-mcp` 原文件， +通过 SQLite backup、integrity check 和 hash receipt 发布到 +`plugin-data/steam-/`，不删除历史源。 -迁移保留 `mcp/steam-mcp` 原文件，在 -`plugin-data/steam-/.steam-v2-migration.json` 写入 hash 与 SQLite -完整性证据。进程内失败会回滚本次新增文件；进程崩溃后重跑会清理 staging， -并只接纳已经发布且内容完全相同的文件。 +## 验证 -候选验证使用无凭证、无外网、无数据库的 recording backend；正式 MCP -只从自己的 `plugin-data` 读取 `steam_mcp_config.json`，不读取 ambient -`STEAM_API_KEY` 或 `STEAM_ID`。 +CI 固定 Core `9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5`。测试覆盖真实 +PluginManager + stdio MCP + Timer、candidate 零正式 write set、reload Timer +换班、网络失败恢复、fresh/stale/unknown、Wake/passive 分流、历史保留、日志轮转、 +pyright、compileall、Plugin API contract 和 `git diff --check`。 diff --git a/akashic.plugin.toml b/akashic.plugin.toml index ccef591..42344dd 100644 --- a/akashic.plugin.toml +++ b/akashic.plugin.toml @@ -1,6 +1,6 @@ schema_version = 1 name = "steam" -version = "3.0.0" +version = "3.1.0" api_version = 3 entrypoint = "plugin.py" @@ -13,12 +13,22 @@ exclude_data_paths = [ "steam_user_cache.json", "steam_app_cache.json", "steam_proactive.sqlite3", + "steam_proactive.sqlite3-wal", + "steam_proactive.sqlite3-shm", + "steam_mcp.runtime.log", + "steam_mcp.runtime.log.1", + "steam_mcp.runtime.log.2", + "steam_mcp.runtime.log.3", + "steam_context.runtime.log", + "steam_context.runtime.log.1", + "steam_context.runtime.log.2", + "steam_context.runtime.log.3", ".steam-v2-migration.json", ] [[mcp]] name = "steam" command = ["python", "mcp/run_mcp.py"] -required_tools = ["get_steam_context"] -candidate_read_only_tools = ["get_steam_context"] +required_tools = ["get_player_summaries"] +candidate_read_only_tools = [] candidate_env = {STEAM_BACKEND = "recording"} diff --git a/context_source.py b/context_source.py new file mode 100644 index 0000000..8f66d19 --- /dev/null +++ b/context_source.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import Callable +from datetime import UTC, datetime +from logging.handlers import RotatingFileHandler +from pathlib import Path + +from agent.control.timer import TimerHandle, TimerStatus +from agent.lifecycle.types import BeforeTurnCtx +from agent.plugin_composition import HealthHandle, PluginTimers + +from steam_runtime import backend + + +class SteamContextRuntime: + """用 Timer 刷新 Steam current state，并只为 Wake 追加 fresh hint。""" + + def __init__( + self, + data_root: Path, + timers: PluginTimers, + health: HealthHandle, + report_incident: Callable[[str, str], object], + *, + now: Callable[[], datetime] | None = None, + ) -> None: + self._data_root = data_root + self._timers = timers + self._health = health + self._report_incident = report_incident + self._now = now or (lambda: datetime.now(UTC)) + self._handle: TimerHandle | None = None + self._task: asyncio.Task[None] | None = None + self._closed = False + self._log = logging.Logger("steam-context-source", level=logging.INFO) + self._log.propagate = False + self._log_handler: RotatingFileHandler | None = None + + async def start(self) -> None: + """初始化正式 state，并只注册一个 source Timer。""" + + if self._closed: + raise RuntimeError("Steam Context runtime 已关闭") + if self._handle is not None: + return + self._start_diagnostics() + now = self._aware_now() + await asyncio.to_thread(backend.initialize, self._data_root, now) + deadline = await asyncio.to_thread(backend.next_deadline, self._data_root, now) + self._arm(deadline) + + async def close(self) -> None: + """取消并收束当前等待，不改变 Steam domain state。""" + + if self._closed: + return + self._closed = True + handle = self._handle + task = self._task + self._handle = None + self._task = None + if handle is not None: + _ = await handle.cancel() + if task is not None and task is not asyncio.current_task(): + _ = await asyncio.gather(task, return_exceptions=True) + if handle is not None: + await handle.cleanup() + self._stop_diagnostics() + + def prepare(self, ctx: BeforeTurnCtx) -> None: + """只在 Wake channel 读取 fresh state 并追加一个普通 hint。""" + + if ctx.channel != "wake": + return + current = backend.wake_context(self._data_root, ctx.timestamp) + if current is None: + return + ctx.extra_hints.append( + "Steam current context:\n" + + json.dumps(current, sort_keys=True, separators=(",", ":")) + ) + + def _arm(self, deadline: datetime) -> None: + if self._closed or self._handle is not None: + return + handle = self._timers.schedule(deadline) + self._handle = handle + self._task = asyncio.create_task( + self._wait_refresh_rearm(handle), + name="steam-context-source:refresh", + ) + self._task.add_done_callback(self._observe_task) + + async def _wait_refresh_rearm(self, handle: TimerHandle) -> None: + """消费一次 Timer，明确处理网络重试，再注册下一次。""" + + next_due: datetime | None = None + try: + receipt = await handle.result() + if receipt.status is TimerStatus.CANCELLED or self._closed: + return + now = self._aware_now() + try: + result = await asyncio.to_thread( + backend.refresh, + self._data_root, + now, + ) + except backend.SteamNetworkError as error: + next_due = await asyncio.to_thread( + backend.record_transient_failure, + self._data_root, + now, + error, + ) + _ = self._report_incident("steam_refresh_transient", str(error)) + self._log.warning( + "refresh transient retry_at=%s error=%s", + next_due.isoformat(), + error, + ) + except Exception as error: + reason = f"{type(error).__name__}: {error}" + self._health.degrade(reason) + _ = self._report_incident("steam_refresh_contract", reason) + self._log.exception("refresh stopped by contract failure") + raise + else: + self._health.recover() + next_due = result.next_due + self._log.info( + "refresh committed presence=%s history_appended=%s next_due=%s", + result.presence, + result.history_appended, + result.next_due.isoformat(), + ) + finally: + self._handle = None + self._task = None + await handle.cleanup() + if not self._closed and next_due is not None: + self._arm(next_due) + + def _aware_now(self) -> datetime: + value = self._now() + if value.tzinfo is None: + raise ValueError("Steam Context clock 必须包含时区") + return value.astimezone(UTC) + + def _observe_task(self, task: asyncio.Task[None]) -> None: + """把未被等待的 runtime failure 转为 required health 与 Incident。""" + + if task.cancelled(): + return + error = task.exception() + if error is None or not self._health.healthy: + return + reason = f"{type(error).__name__}: {error}" + self._health.degrade(reason) + _ = self._report_incident("steam_runtime_failure", reason) + + def _start_diagnostics(self) -> None: + if self._log_handler is not None: + return + self._data_root.mkdir(parents=True, exist_ok=True) + handler = RotatingFileHandler( + self._data_root / "steam_context.runtime.log", + maxBytes=5 * 1024 * 1024, + backupCount=3, + encoding="utf-8", + ) + handler.setFormatter( + logging.Formatter( + "%(asctime)s %(levelname)-8s %(name)s | %(message)s" + ) + ) + self._log.addHandler(handler) + self._log_handler = handler + + def _stop_diagnostics(self) -> None: + handler = self._log_handler + if handler is None: + return + self._log.removeHandler(handler) + handler.close() + self._log_handler = None diff --git a/mcp/run_mcp.py b/mcp/run_mcp.py index d3cb30d..992a9a3 100644 --- a/mcp/run_mcp.py +++ b/mcp/run_mcp.py @@ -1,24 +1,58 @@ #!/usr/bin/env python3 +import logging import os import sys +from logging.handlers import RotatingFileHandler from pathlib import Path +def _runtime_dir() -> Path: + raw = os.environ.get("AKA_PLUGIN_DATA_DIR", "").strip() + if not raw: + raise RuntimeError("Steam MCP 缺少 AKA_PLUGIN_DATA_DIR") + return Path(raw).resolve() + + +def _setup_logging(runtime_dir: Path) -> None: + """将诊断写入 stderr 和三个有界本地轮转文件。""" + + runtime_dir.mkdir(parents=True, exist_ok=True) + formatter = logging.Formatter( + "%(asctime)s %(levelname)-8s %(name)s | %(message)s" + ) + file_handler = RotatingFileHandler( + runtime_dir / "steam_mcp.runtime.log", + maxBytes=5 * 1024 * 1024, + backupCount=3, + encoding="utf-8", + ) + file_handler.setFormatter(formatter) + stream_handler = logging.StreamHandler(sys.stderr) + stream_handler.setFormatter(formatter) + root = logging.getLogger() + root.setLevel(logging.INFO) + root.handlers.clear() + root.addHandler(file_handler) + root.addHandler(stream_handler) + + def main() -> None: script_dir = Path(__file__).resolve().parent os.chdir(script_dir) - if str(script_dir) not in sys.path: - sys.path.insert(0, str(script_dir)) + for path in (script_dir.parent, script_dir): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) backend = os.environ.get("STEAM_BACKEND", "formal").strip().lower() if backend not in {"formal", "recording"}: raise RuntimeError(f"未知 STEAM_BACKEND: {backend}") if backend == "formal": - from runtime_config import load_runtime_config + from steam_runtime.config import load_runtime_config - data_root = Path(os.environ["AKA_PLUGIN_DATA_DIR"]).resolve() + data_root = _runtime_dir() _ = load_runtime_config(data_root / "steam_mcp_config.json") + _setup_logging(_runtime_dir()) from steam_mcp import mcp mcp.run(transport="stdio") diff --git a/mcp/steam_mcp.py b/mcp/steam_mcp.py index afe4413..25a1770 100644 --- a/mcp/steam_mcp.py +++ b/mcp/steam_mcp.py @@ -13,7 +13,6 @@ mcp = FastMCP("steam-web-api") RUNTIME_DIR = Path(os.environ.get("AKA_PLUGIN_DATA_DIR", "").strip() or Path.cwd()) RUNTIME_DIR.mkdir(parents=True, exist_ok=True) -BACKEND = os.environ.get("STEAM_BACKEND", "formal").strip().lower() http_client = HttpClient(config_path=str(RUNTIME_DIR / "steam_mcp_config.json")) SUPPORTED_FORMATS = {"json", "xml", "vdf"} MAX_STEAM_IDS_PER_REQUEST = 100 @@ -625,41 +624,5 @@ def _convert_game_playtimes_to_hours(game: dict) -> None: game[key] = round(value / 60, 1) -# --------------------------------------------------------------------------- -# Proactive context tools -# --------------------------------------------------------------------------- - -@mcp.tool() -def get_steam_context() -> dict: - """获取用户 Steam 游戏活动的持久上下文，供 proactive engine 注入 background_context。 - 返回近两周游戏时长、历史对比、当前在线状态等结构化数据。 - """ - if BACKEND == "recording": - return { - "items": [ - { - "presence": "unknown", - "interruptibility": 0.4, - "confidence": 0.0, - "transition": "", - "recording": True, - } - ] - } - import steam_proactive - - return {"items": [steam_proactive.get_context()]} - - -@mcp.tool() -def take_steam_snapshot() -> dict: - """拉取并存储一次 Steam 游戏数据快照。通常由定时任务调用，也可手动触发。""" - if BACKEND == "recording": - raise RuntimeError("recording backend 禁止写 Steam snapshot") - import steam_proactive - - return steam_proactive.take_snapshot() - - if __name__ == "__main__": mcp.run() diff --git a/mcp/steam_proactive.py b/mcp/steam_proactive.py deleted file mode 100644 index 44a49cc..0000000 --- a/mcp/steam_proactive.py +++ /dev/null @@ -1,338 +0,0 @@ -"""steam_proactive.py — Steam proactive context backend. - -维护本地 SQLite 快照，追踪用户近期游戏活动。 -通过 get_context() 向 proactive engine 的 background_context channel 提供持久感知数据。 -""" -from __future__ import annotations - -import json -import logging -import os -import sqlite3 -from datetime import datetime, timezone, timedelta -from pathlib import Path -from typing import Any -from urllib.error import HTTPError -from urllib.parse import urlencode -from urllib.request import urlopen - -logger = logging.getLogger(__name__) - -_SCRIPT_DIR = Path(__file__).parent -_RUNTIME_DIR = Path(os.environ.get("AKA_PLUGIN_DATA_DIR", "").strip() or _SCRIPT_DIR) -_RUNTIME_DIR.mkdir(parents=True, exist_ok=True) -_DB_PATH = _RUNTIME_DIR / "steam_proactive.sqlite3" -_CONFIG_PATH = _RUNTIME_DIR / "steam_mcp_config.json" -_last_wake_presence = "unknown" -_DEFAULT_SNAPSHOT_INTERVAL_SECONDS = 6 * 3600 - - -# --------------------------------------------------------------------------- -# Config -# --------------------------------------------------------------------------- - -def _load_config() -> dict: - if _CONFIG_PATH.exists(): - loaded = json.loads(_CONFIG_PATH.read_text()) - if not isinstance(loaded, dict): - raise ValueError("steam_mcp_config.json 根节点必须是 object") - else: - loaded = {} - return loaded - - -# --------------------------------------------------------------------------- -# DB -# --------------------------------------------------------------------------- - -def _get_conn() -> sqlite3.Connection: - conn = sqlite3.connect(_DB_PATH) - conn.row_factory = sqlite3.Row - conn.executescript(""" - CREATE TABLE IF NOT EXISTS snapshots ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - snapshotted_at TEXT NOT NULL, - game_appid INTEGER NOT NULL, - game_name TEXT NOT NULL, - playtime_2w_mins INTEGER NOT NULL, - playtime_forever_mins INTEGER NOT NULL - ); - CREATE INDEX IF NOT EXISTS idx_snap_time ON snapshots(snapshotted_at); - CREATE TABLE IF NOT EXISTS snapshot_runs ( - snapshotted_at TEXT PRIMARY KEY - ); - """) - conn.execute( - "INSERT OR IGNORE INTO snapshot_runs(snapshotted_at) " - "SELECT DISTINCT snapshotted_at FROM snapshots" - ) - conn.commit() - return conn - - -# --------------------------------------------------------------------------- -# Steam API -# --------------------------------------------------------------------------- - -def _steam_get(path: str, params: dict) -> dict: - cfg = _load_config() - p = dict(params, key=cfg.get("steam_api_key", ""), format="json") - url = f"https://api.steampowered.com/{path}?{urlencode(p)}" - try: - with urlopen(url, timeout=15) as resp: - return json.loads(resp.read().decode()) - except HTTPError as e: - raise RuntimeError(f"Steam API {e.code}: {path}") from e - - -def _fetch_recently_played(steamid: str) -> list[dict]: - data = _steam_get("IPlayerService/GetRecentlyPlayedGames/v0001", {"steamid": steamid, "count": 0}) - return data.get("response", {}).get("games", []) - - -def _fetch_player_summary(steamid: str) -> dict: - data = _steam_get("ISteamUser/GetPlayerSummaries/v0002", {"steamids": steamid}) - players = data.get("response", {}).get("players", []) - return players[0] if players else {} - - -# --------------------------------------------------------------------------- -# Snapshot -# --------------------------------------------------------------------------- - -def take_snapshot() -> dict: - """拉取 Steam API 并存储快照，通常由定时任务调用。""" - cfg = _load_config() - steamid = cfg.get("steam_id", "") - if not steamid: - return {"ok": False, "error": "steam_id 未在 steam_mcp_config.json 中配置"} - - try: - games = _fetch_recently_played(steamid) - except Exception as e: - return {"ok": False, "error": str(e)} - - conn = _get_conn() - now = _now().isoformat() - conn.execute( - "INSERT INTO snapshot_runs(snapshotted_at) VALUES (?)", - (now,), - ) - for g in games: - conn.execute( - "INSERT INTO snapshots (snapshotted_at, game_appid, game_name, " - "playtime_2w_mins, playtime_forever_mins) VALUES (?, ?, ?, ?, ?)", - ( - now, - g["appid"], - g.get("name", f"App {g['appid']}"), - g.get("playtime_2weeks", 0), - g.get("playtime_forever", 0), - ), - ) - conn.commit() - conn.close() - return {"ok": True, "snapshotted_at": now, "game_count": len(games)} - - -def _now() -> datetime: - return datetime.now(timezone.utc) - - -def _snapshot_interval_seconds(cfg: dict) -> int: - raw = cfg.get("snapshot_interval_seconds", _DEFAULT_SNAPSHOT_INTERVAL_SECONDS) - return max(300, int(raw)) - - -def _refresh_snapshot_if_due(cfg: dict) -> str | None: - """快照过期时刷新；失败原因交给主动上下文显式展示。""" - - conn = _get_conn() - try: - row = conn.execute( - "SELECT snapshotted_at FROM snapshot_runs " - "ORDER BY snapshotted_at DESC LIMIT 1" - ).fetchone() - finally: - conn.close() - if row is not None: - latest = datetime.fromisoformat(str(row["snapshotted_at"])) - if (_now() - latest).total_seconds() < _snapshot_interval_seconds(cfg): - return None - result = take_snapshot() - return None if result["ok"] else str(result["error"]) - - -# --------------------------------------------------------------------------- -# Context computation -# --------------------------------------------------------------------------- - -def get_context() -> dict[str, Any]: - """返回结构化游戏上下文，供 proactive engine 注入 background_context。""" - cfg = _load_config() - steamid = cfg.get("steam_id", "") - snapshot_refresh_error = _refresh_snapshot_if_due(cfg) - - # 1. 实时状态（每次调用直接打 API，轻量） - realtime: dict[str, Any] = {} - if steamid: - try: - summary = _fetch_player_summary(steamid) - _PERSONA = {0: "offline", 1: "online", 2: "busy", 3: "away", 4: "snooze"} - realtime = { - "fetched_at": _now().isoformat(), - "online_status": "in-game" if summary.get("gameid") else _PERSONA.get(summary.get("personastate", 0), "offline"), - "currently_playing": summary.get("gameextrainfo"), - } - except Exception as e: - realtime = {"fetched_at": _now().isoformat(), "error": str(e)} - - # 2. 读取快照 - conn = _get_conn() - - latest = conn.execute( - "SELECT snapshotted_at FROM snapshot_runs ORDER BY snapshotted_at DESC LIMIT 1" - ).fetchone() - if not latest: - conn.close() - return _with_wake_contract( - { - "available": False, - "realtime": realtime, - "snapshot_refresh_error": snapshot_refresh_error, - } - ) - - recent_snap_at: str = latest["snapshotted_at"] - recent_rows = conn.execute( - "SELECT * FROM snapshots WHERE snapshotted_at = ?", (recent_snap_at,) - ).fetchall() - - # 上次快照：距今 ≥14 天中最新的一条 - cutoff = (_now() - timedelta(days=14)).isoformat() - prev_meta = conn.execute( - "SELECT snapshotted_at FROM snapshot_runs WHERE snapshotted_at <= ? " - "ORDER BY snapshotted_at DESC LIMIT 1", - (cutoff,), - ).fetchone() - prev_snap_at: str | None = prev_meta["snapshotted_at"] if prev_meta else None - prev_rows = ( - conn.execute("SELECT * FROM snapshots WHERE snapshotted_at = ?", (prev_snap_at,)).fetchall() - if prev_snap_at - else [] - ) - conn.close() - - # 3. 两个快照取并集，任意一侧时长不为 0 的游戏都收录 - recent_by_appid = {r["game_appid"]: r for r in recent_rows} - prev_by_appid = {r["game_appid"]: r for r in prev_rows} - all_appids = set(recent_by_appid) | set(prev_by_appid) - - games: list[dict] = [] - for appid in all_appids: - recent_r = recent_by_appid.get(appid) - prev_r = prev_by_appid.get(appid) - recent_2w_h = round(recent_r["playtime_2w_mins"] / 60, 1) if recent_r else 0 - prev_2w_h = round(prev_r["playtime_2w_mins"] / 60, 1) if prev_r else 0 - if recent_2w_h == 0 and prev_2w_h == 0: - continue - if recent_r is not None: - r = recent_r - elif prev_r is not None: - r = prev_r - else: - raise RuntimeError(f"快照索引缺少 appid={appid}") - all_time_h = round(max(r["playtime_forever_mins"], r["playtime_2w_mins"]) / 60, 1) - games.append({ - "name": r["game_name"], - "recent_2w_hours": recent_2w_h, - "prev_snapshot_2w_hours": prev_2w_h, - "all_time_hours": all_time_h, - }) - games.sort(key=lambda g: g["recent_2w_hours"], reverse=True) - - # 4. 时间元数据 - now_dt = _now() - recent_dt = datetime.fromisoformat(recent_snap_at) - data_freshness_h = round((now_dt - recent_dt).total_seconds() / 3600, 1) - prev_age_days: float | None = None - if prev_snap_at: - prev_age_days = round((now_dt - datetime.fromisoformat(prev_snap_at)).total_seconds() / 86400, 1) - - payload = { - "_hint": { - "recent_snapshot_at": "本次快照时间戳", - "prev_snapshot_at": "对比快照时间戳，null 表示尚无历史数据", - "prev_snapshot_age_days": "对比快照距今天数", - "data_freshness_hours": "本次快照距现在的小时数，越小越新鲜", - "games[].recent_2w_hours": "本次快照记录的近两周时长，0 表示这两周没玩", - "games[].prev_snapshot_2w_hours": "对比快照记录的近两周时长，0 表示当时没玩或无数据", - "games[].all_time_hours": "历史累计时长，反映用户对该游戏的熟悉程度", - "realtime.currently_playing": "非 null 时表示用户此刻正在游戏中", - "realtime.online_status": "in-game/online/away/offline", - }, - "available": True, - "data_freshness_hours": data_freshness_h, - "recent_snapshot_at": recent_snap_at, - "prev_snapshot_at": prev_snap_at, - "prev_snapshot_age_days": prev_age_days, - "games": games, - "realtime": realtime, - "snapshot_refresh_error": snapshot_refresh_error, - } - return _with_wake_contract(payload) - - -def _with_wake_contract( - payload: dict[str, Any], - *, - observed_at: datetime | None = None, -) -> dict[str, Any]: - global _last_wake_presence - realtime = payload.get("realtime") - realtime = realtime if isinstance(realtime, dict) else {} - status = str(realtime.get("online_status") or "unknown").strip().lower() - presence = { - "in-game": "in_game", - "online": "active", - "busy": "active", - "away": "idle", - "snooze": "idle", - "offline": "offline", - }.get(status, "unknown") - interruptibility = { - "in-game": 0.1, - "online": 0.8, - "busy": 0.1, - "away": 0.4, - "snooze": 0.3, - "offline": 0.0, - }.get(status, 0.4) - confidence = 0.9 if presence != "unknown" and not realtime.get("error") else 0.1 - transition = "" - if _last_wake_presence != "unknown" and presence != _last_wake_presence: - transition = f"{_last_wake_presence}->{presence}" - if presence != "unknown": - _last_wake_presence = presence - observed = observed_at or _parse_observed_at(realtime.get("fetched_at")) - original = dict(payload) - return { - **original, - "presence": presence, - "interruptibility": interruptibility, - "confidence": confidence, - "transition": transition, - "observed_at": observed.isoformat(), - "expires_at": (observed + timedelta(minutes=5)).isoformat(), - "payload": original, - } - - -def _parse_observed_at(value: object) -> datetime: - try: - parsed = datetime.fromisoformat(str(value)) - except ValueError: - return datetime.now(timezone.utc) - if parsed.tzinfo is None: - return parsed.replace(tzinfo=timezone.utc) - return parsed.astimezone(timezone.utc) diff --git a/mcp/tests/test_context_contract.py b/mcp/tests/test_context_contract.py index 2e0e0ad..9524e98 100644 --- a/mcp/tests/test_context_contract.py +++ b/mcp/tests/test_context_contract.py @@ -1,56 +1,105 @@ -from datetime import UTC, datetime +from __future__ import annotations -import steam_proactive +import json +from datetime import UTC, datetime, timedelta +from steam_runtime import backend -def test_in_game_context_exposes_wake_contract_and_preserves_payload() -> None: - steam_proactive._last_wake_presence = "unknown" - observed = datetime(2026, 7, 12, 8, tzinfo=UTC) - context = steam_proactive._with_wake_contract( - { - "available": True, - "realtime": { - "online_status": "in-game", - "currently_playing": "Game", - }, - "games": [{"name": "Game"}], - }, - observed_at=observed, +def _config(tmp_path) -> None: + (tmp_path / "steam_mcp_config.json").write_text( + json.dumps( + { + "steam_api_key": "test-key", + "steam_id": "test-user", + "snapshot_interval_seconds": 300, + } + ), + encoding="utf-8", ) - assert context["presence"] == "in_game" - assert context["interruptibility"] == 0.1 - assert context["confidence"] == 0.9 - assert context["transition"] == "" - assert context["payload"]["realtime"]["currently_playing"] == "Game" - assert datetime.fromisoformat(context["expires_at"]) > observed - -def test_steam_owner_emits_generic_transition() -> None: - steam_proactive._last_wake_presence = "in_game" - context = steam_proactive._with_wake_contract( - { - "available": True, - "realtime": {"online_status": "offline"}, +def test_fresh_context_uses_persisted_transition_and_stale_is_absent( + tmp_path, + monkeypatch, +) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + monkeypatch.setattr( + backend, + "_fetch_player_summary", + lambda _: { + "personastate": 1, + "gameid": "1", + "gameextrainfo": "Game", }, - observed_at=datetime(2026, 7, 12, 9, tzinfo=UTC), ) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: []) + _ = backend.refresh(tmp_path, now) + + fresh = backend.wake_context(tmp_path, now + timedelta(minutes=1)) + assert fresh is not None + assert fresh["presence"] == "in_game" + assert fresh["currently_playing"] == "Game" + assert fresh["transition"] == "" + + monkeypatch.setattr( + backend, + "_fetch_player_summary", + lambda _: {"personastate": 0}, + ) + _ = backend.refresh(tmp_path, now + timedelta(minutes=5)) + changed = backend.wake_context(tmp_path, now + timedelta(minutes=6)) + assert changed is not None + assert changed["presence"] == "offline" + assert changed["transition"] == "in_game->offline" + assert backend.wake_context(tmp_path, now + timedelta(minutes=11)) is None - assert context["presence"] == "offline" - assert context["interruptibility"] == 0.0 - assert context["transition"] == "in_game->offline" +def test_unknown_presence_never_becomes_wake_hint(tmp_path, monkeypatch) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + monkeypatch.setattr(backend, "_fetch_player_summary", lambda _: {}) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: []) -def test_realtime_error_yields_low_confidence_unknown_context() -> None: - context = steam_proactive._with_wake_contract( + _ = backend.refresh(tmp_path, now) + + assert backend.wake_context(tmp_path, now) is None + + +def test_fresh_context_keeps_prior_history_visible_when_current_games_are_empty( + tmp_path, + monkeypatch, +) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + games = [ { - "available": False, - "realtime": {"error": "timeout"}, - }, - observed_at=datetime(2026, 7, 12, 9, tzinfo=UTC), + "appid": 1, + "name": "Old Game", + "playtime_2weeks": 120, + "playtime_forever": 600, + } + ] + monkeypatch.setattr( + backend, + "_fetch_player_summary", + lambda _: {"personastate": 1}, ) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: games) + _ = backend.refresh(tmp_path, now - timedelta(days=15)) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: []) + _ = backend.refresh(tmp_path, now) + + current = backend.wake_context(tmp_path, now) - assert context["presence"] == "unknown" - assert context["confidence"] == 0.1 - assert context["payload"]["realtime"]["error"] == "timeout" + assert current is not None + assert current["previous_snapshot_at"] == (now - timedelta(days=15)).isoformat() + assert current["games"] == [ + { + "name": "Old Game", + "recent_2w_hours": 0.0, + "all_time_hours": 10.0, + "previous_snapshot_2w_hours": 2.0, + } + ] diff --git a/mcp/tests/test_snapshot_freshness.py b/mcp/tests/test_snapshot_freshness.py index 16c5076..49da72b 100644 --- a/mcp/tests/test_snapshot_freshness.py +++ b/mcp/tests/test_snapshot_freshness.py @@ -1,114 +1,145 @@ from __future__ import annotations import json +import sqlite3 from datetime import UTC, datetime, timedelta -import steam_proactive +import pytest +from steam_runtime import backend -def _configure(monkeypatch, tmp_path, now: datetime) -> None: - monkeypatch.setattr(steam_proactive, "_DB_PATH", tmp_path / "steam.sqlite3") - monkeypatch.setattr(steam_proactive, "_CONFIG_PATH", tmp_path / "steam.json") - monkeypatch.setattr(steam_proactive, "_now", lambda: now) - (tmp_path / "steam.json").write_text( + +def _config(tmp_path) -> None: + (tmp_path / "steam_mcp_config.json").write_text( json.dumps( { "steam_api_key": "test-key", "steam_id": "test-user", - "snapshot_interval_seconds": 3600, + "snapshot_interval_seconds": 300, } ), encoding="utf-8", ) -def test_context_refreshes_expired_snapshot_once(monkeypatch, tmp_path) -> None: - now = datetime(2026, 7, 13, tzinfo=UTC) - _configure(monkeypatch, tmp_path, now) - recent_calls = 0 - - def recently_played(_: str) -> list[dict]: - nonlocal recent_calls - recent_calls += 1 - return [ - { - "appid": 1, - "name": "Game", - "playtime_2weeks": 120, - "playtime_forever": 600, - } - ] - - monkeypatch.setattr(steam_proactive, "_fetch_recently_played", recently_played) +def test_changed_snapshot_appends_but_same_and_empty_only_advance_current_state( + tmp_path, + monkeypatch, +) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + games = [ + { + "appid": 1, + "name": "Game", + "playtime_2weeks": 120, + "playtime_forever": 600, + } + ] monkeypatch.setattr( - steam_proactive, + backend, "_fetch_player_summary", lambda _: {"personastate": 1}, ) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: games) - first = steam_proactive.get_context() - second = steam_proactive.get_context() - - assert recent_calls == 1 - assert first["recent_snapshot_at"] == now.isoformat() - assert first["games"][0]["recent_2w_hours"] == 2.0 - assert second["snapshot_refresh_error"] is None - - -def test_empty_snapshot_records_fresh_run(monkeypatch, tmp_path) -> None: - now = datetime(2026, 7, 13, tzinfo=UTC) - _configure(monkeypatch, tmp_path, now) - calls = 0 - - def recently_played(_: str) -> list[dict]: - nonlocal calls - calls += 1 - return [] - - monkeypatch.setattr(steam_proactive, "_fetch_recently_played", recently_played) - monkeypatch.setattr(steam_proactive, "_fetch_player_summary", lambda _: {}) - - context = steam_proactive.get_context() - _ = steam_proactive.get_context() - - assert calls == 1 - assert context["available"] is True - assert context["games"] == [] - - -def test_snapshot_refresh_failure_is_visible(monkeypatch, tmp_path) -> None: - now = datetime(2026, 7, 13, tzinfo=UTC) - _configure(monkeypatch, tmp_path, now) - monkeypatch.setattr( - steam_proactive, - "_fetch_recently_played", - lambda _: (_ for _ in ()).throw(OSError("Steam unavailable")), - ) - monkeypatch.setattr(steam_proactive, "_fetch_player_summary", lambda _: {}) - - context = steam_proactive.get_context() + first = backend.refresh(tmp_path, now) + repeated = backend.refresh(tmp_path, now + timedelta(minutes=5)) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: []) + empty = backend.refresh(tmp_path, now + timedelta(minutes=10)) + current = backend.state(tmp_path, now + timedelta(minutes=10)) - assert context["available"] is False - assert context["snapshot_refresh_error"] == "Steam unavailable" + assert first.history_appended is True + assert repeated.history_appended is False + assert empty.history_appended is False + assert current["snapshot_runs"] == 1 + assert current["snapshots"] == 1 + assert json.loads(str(current["current_games_json"])) == [] -def test_recent_snapshot_skips_refresh(monkeypatch, tmp_path) -> None: - now = datetime(2026, 7, 13, tzinfo=UTC) - _configure(monkeypatch, tmp_path, now) - conn = steam_proactive._get_conn() - conn.execute( - "INSERT INTO snapshot_runs(snapshotted_at) VALUES (?)", - ((now - timedelta(minutes=30)).isoformat(),), - ) - conn.commit() - conn.close() +def test_existing_history_is_never_trimmed(tmp_path, monkeypatch) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) monkeypatch.setattr( - steam_proactive, - "_fetch_recently_played", - lambda _: (_ for _ in ()).throw(AssertionError("不应刷新")), + backend, + "_fetch_player_summary", + lambda _: {"personastate": 1}, ) - monkeypatch.setattr(steam_proactive, "_fetch_player_summary", lambda _: {}) - - context = steam_proactive.get_context() - - assert context["snapshot_refresh_error"] is None + games = [ + { + "appid": 1, + "name": "Game", + "playtime_2weeks": 60, + "playtime_forever": 60, + } + ] + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: games) + for index in range(4): + games[0] = {**games[0], "playtime_forever": 60 + index} + _ = backend.refresh(tmp_path, now + timedelta(minutes=5 * index)) + + current = backend.state(tmp_path, now + timedelta(minutes=20)) + assert current["snapshot_runs"] == 4 + assert current["snapshots"] == 4 + + +def test_initialize_adopts_existing_history_without_rewriting_it(tmp_path) -> None: + database = tmp_path / "steam_proactive.sqlite3" + existing = [ + ("2026-07-01T08:00:00+00:00", 1, "Old Game", 120, 600), + ("2026-08-01T08:00:00+00:00", 1, "Old Game", 60, 660), + ] + with sqlite3.connect(database) as connection: + connection.executescript( + """ + CREATE TABLE snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshotted_at TEXT NOT NULL, + game_appid INTEGER NOT NULL, + game_name TEXT NOT NULL, + playtime_2w_mins INTEGER NOT NULL, + playtime_forever_mins INTEGER NOT NULL + ); + CREATE TABLE snapshot_runs (snapshotted_at TEXT PRIMARY KEY); + """ + ) + connection.executemany( + """ + INSERT INTO snapshots( + snapshotted_at, game_appid, game_name, + playtime_2w_mins, playtime_forever_mins + ) VALUES (?, ?, ?, ?, ?) + """, + existing, + ) + connection.executemany( + "INSERT INTO snapshot_runs(snapshotted_at) VALUES (?)", + [(row[0],) for row in existing], + ) + connection.commit() + + backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + + with sqlite3.connect(database) as connection: + adopted = connection.execute( + """ + SELECT snapshotted_at, game_appid, game_name, + playtime_2w_mins, playtime_forever_mins + FROM snapshots ORDER BY id + """ + ).fetchall() + runs = connection.execute( + "SELECT snapshotted_at FROM snapshot_runs ORDER BY snapshotted_at" + ).fetchall() + assert adopted == existing + assert runs == [(row[0],) for row in existing] + + +def test_incompatible_history_schema_fails_loud(tmp_path) -> None: + database = tmp_path / "steam_proactive.sqlite3" + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE snapshots(value TEXT NOT NULL)") + connection.commit() + + with pytest.raises(RuntimeError, match="schema 不兼容"): + backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) diff --git a/mcp/tests/test_v3_runtime.py b/mcp/tests/test_v3_runtime.py index 3b15461..8c17239 100644 --- a/mcp/tests/test_v3_runtime.py +++ b/mcp/tests/test_v3_runtime.py @@ -1,12 +1,19 @@ from __future__ import annotations import importlib +import importlib.util import json +import logging import sys +from logging.handlers import RotatingFileHandler +from pathlib import Path import pytest -from runtime_config import load_runtime_config +from steam_runtime.config import load_runtime_config + + +ROOT = Path(__file__).resolve().parents[2] def test_formal_runtime_config_requires_plugin_data_credentials( @@ -44,32 +51,52 @@ def test_formal_runtime_config_accepts_complete_file(tmp_path) -> None: assert config.snapshot_interval_seconds == 3600 -def test_recording_context_never_reads_formal_config_or_creates_database( +def test_recording_mcp_exposes_only_user_driven_tools( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: - data_root = tmp_path / "candidate-data" - monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(data_root)) + monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(tmp_path)) monkeypatch.setenv("STEAM_BACKEND", "recording") - monkeypatch.setenv("STEAM_API_KEY", "ambient-secret") - monkeypatch.setenv("STEAM_ID", "ambient-user") sys.modules.pop("steam_mcp", None) - sys.modules.pop("steam_proactive", None) module = importlib.import_module("steam_mcp") - result = module.get_steam_context() + names = {tool.name for tool in module.mcp._tool_manager.list_tools()} - assert result == { - "items": [ - { - "presence": "unknown", - "interruptibility": 0.4, - "confidence": 0.0, - "transition": "", - "recording": True, - } - ] + assert names == { + "get_news_for_app", + "get_player_summaries", + "get_owned_games", + "get_recently_played_games", + "get_friend_list", + "get_player_achievements", + "get_user_stats_for_game", + "get_number_of_current_players", + "resolve_app_ids", } - assert "steam_proactive" not in sys.modules - assert not (data_root / "steam_mcp_config.json").exists() - assert not (data_root / "steam_proactive.sqlite3").exists() + assert not (tmp_path / "steam_mcp_config.json").exists() + assert not (tmp_path / "steam_proactive.sqlite3").exists() + + +def test_runner_uses_three_bounded_log_rotations( + tmp_path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + path = ROOT / "mcp" / "run_mcp.py" + spec = importlib.util.spec_from_file_location("steam_test_runner", path) + assert spec is not None and spec.loader is not None + runner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(runner) + monkeypatch.setenv("AKA_PLUGIN_DATA_DIR", str(tmp_path)) + + runner._setup_logging(tmp_path) + + rotating = [ + handler + for handler in logging.getLogger().handlers + if isinstance(handler, RotatingFileHandler) + ] + assert len(rotating) == 1 + assert rotating[0].backupCount == 3 + assert rotating[0].maxBytes == 5 * 1024 * 1024 + for handler in logging.getLogger().handlers: + handler.close() diff --git a/plugin.py b/plugin.py index 5317ea4..7ed254e 100644 --- a/plugin.py +++ b/plugin.py @@ -1,59 +1,64 @@ from __future__ import annotations -from pydantic import BaseModel, Field +from pydantic import BaseModel +from agent.lifecycle.composition import CONTEXT_PREPARED_EVENT from agent.plugin_composition import ( MCP_SERVERS, - PROACTIVE_COMPONENTS, + RUNTIME_STARTED, + RUNTIME_STOPPING, + TIMERS, Context, McpServerDefinition, - ProactiveSourceDefinition, ) - -class SteamProactiveConfig(BaseModel): - enabled: bool = True +from context_source import SteamContextRuntime class SteamConfig(BaseModel): - proactive: SteamProactiveConfig = Field(default_factory=SteamProactiveConfig) + pass api_version = 3 name = "steam" -version = "3.0.0" -desc = "Steam MCP plugin" +version = "3.1.0" +desc = "Timer 刷新的 Steam current context 与用户 MCP" Config = SteamConfig -inject = (MCP_SERVERS, PROACTIVE_COMPONENTS) +inject = (MCP_SERVERS, TIMERS) skill_roots = ("skills",) async def apply(ctx: Context, config: object) -> None: - """声明 Steam MCP 与可选的主动上下文源。""" + """组合用户 MCP、Timer current state 和 Wake context listener。""" if not isinstance(config, SteamConfig): raise TypeError("steam config 必须是 SteamConfig") - # 1. MCP 由 Core staged Python runtime 启动，candidate 仅开放 recording 上下文 + # 1. MCP 只保留用户主动查询；candidate 只完成隔离握手。 await ctx.require(MCP_SERVERS).register( ctx, McpServerDefinition( name="steam", command=("python", "mcp/run_mcp.py"), - required_tools=("get_steam_context",), - candidate_read_only_tools=("get_steam_context",), + required_tools=("get_player_summaries",), + candidate_read_only_tools=(), candidate_env={"STEAM_BACKEND": "recording"}, ), ) - # 2. 主动源只消费明确的 FetchItems/FetchEmpty 结果 - if config.proactive.enabled: - await ctx.require(PROACTIVE_COMPONENTS).register( - ctx, - ProactiveSourceDefinition( - name="presence", - channels=("context",), - mcp_server="steam", - fetch_tool="get_steam_context", - ), - ) + # 2. 正式 Root 独占 Timer 刷新；listener 只读 current state。 + health = await ctx.health("context-refresh", required=True) + runtime = SteamContextRuntime( + ctx.data_root, + ctx.require(TIMERS), + health, + ctx.report_incident, + ) + + def setup() -> object: + return runtime.close + + _ = await ctx.effect(setup, label="steam-context-runtime") + _ = await ctx.on(CONTEXT_PREPARED_EVENT, runtime.prepare) + _ = await ctx.on(RUNTIME_STARTED, lambda _: runtime.start()) + _ = await ctx.on(RUNTIME_STOPPING, lambda _: runtime.close()) diff --git a/pyrightconfig.json b/pyrightconfig.json index 7bf803e..fe06baa 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -1,6 +1,8 @@ { "include": [ "plugin.py", + "context_source.py", + "steam_runtime", "mcp", "scripts", "tests" diff --git a/steam_runtime/__init__.py b/steam_runtime/__init__.py new file mode 100644 index 0000000..5af9a0f --- /dev/null +++ b/steam_runtime/__init__.py @@ -0,0 +1 @@ +"""Steam 插件共享 domain runtime。""" diff --git a/steam_runtime/backend.py b/steam_runtime/backend.py new file mode 100644 index 0000000..a01a155 --- /dev/null +++ b/steam_runtime/backend.py @@ -0,0 +1,621 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any, cast +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import urlopen + +from steam_runtime.config import SteamRuntimeConfig, load_runtime_config + + +_DB_NAME = "steam_proactive.sqlite3" +_PRESENCE_REFRESH_SECONDS = 300 +_TRANSIENT_RETRY_SECONDS = 60 + + +class SteamNetworkError(RuntimeError): + """表示可重试的 Steam 网络失败。""" + + +@dataclass(frozen=True, slots=True) +class RefreshResult: + next_due: datetime + history_appended: bool + presence: str + + +def initialize(data_root: Path, now: datetime) -> None: + """建立并验证 Steam state schema，同时保留全部既有快照。""" + + connection = _connect(data_root) + try: + _ensure_current_row(connection, _aware(now)) + connection.commit() + finally: + connection.close() + + +def next_deadline(data_root: Path, now: datetime) -> datetime: + """返回 source 持久化的下一次刷新时间。""" + + aware_now = _aware(now) + connection = _connect(data_root) + try: + _ensure_current_row(connection, aware_now) + row = connection.execute( + "SELECT next_refresh_at FROM current_state WHERE singleton = 1" + ).fetchone() + if row is None: + raise RuntimeError("Steam current_state singleton 缺失") + return _parse_aware(str(row["next_refresh_at"]), "next_refresh_at") + finally: + connection.close() + + +def refresh(data_root: Path, now: datetime) -> RefreshResult: + """刷新 current presence，并仅在游戏快照有意义变化时追加历史。""" + + aware_now = _aware(now) + config = load_runtime_config(data_root / "steam_mcp_config.json") + state = _read_current(data_root, aware_now) + + # 1. 网络边界先冻结完整结果；部分响应不能写入本地状态。 + summary = _fetch_player_summary(config) + snapshot_due = _snapshot_due(state, config, aware_now) + games = _fetch_recently_played(config) if snapshot_due else None + previous_presence = "unknown" + if state.get("presence_json") is not None: + previous = _json_object(state["presence_json"], "presence_json") + previous_presence = str(previous.get("presence") or "unknown") + presence = _presence(summary, aware_now, previous_presence) + + # 2. 一个事务覆盖 current state，并按 fingerprint 决定是否追加历史。 + connection = _connect(data_root) + try: + _ensure_current_row(connection, aware_now) + history_appended = False + if games is not None: + fingerprint = _games_fingerprint(games) + current = connection.execute( + "SELECT last_history_fingerprint FROM current_state WHERE singleton = 1" + ).fetchone() + if current is None: + raise RuntimeError("Steam current_state singleton 缺失") + previous_fingerprint = current["last_history_fingerprint"] + if games and fingerprint != previous_fingerprint: + _append_snapshot(connection, games, aware_now) + previous_fingerprint = fingerprint + history_appended = True + connection.execute( + """ + UPDATE current_state + SET current_games_json = ?, + last_snapshot_checked_at = ?, + last_history_fingerprint = ? + WHERE singleton = 1 + """, + ( + _json(games), + aware_now.isoformat(), + previous_fingerprint, + ), + ) + next_due = aware_now + timedelta(seconds=_PRESENCE_REFRESH_SECONDS) + connection.execute( + """ + UPDATE current_state + SET presence_json = ?, + presence_observed_at = ?, + presence_expires_at = ?, + last_refresh_attempt_at = ?, + last_refresh_error = NULL, + next_refresh_at = ? + WHERE singleton = 1 + """, + ( + _json(presence), + aware_now.isoformat(), + (aware_now + timedelta(seconds=_PRESENCE_REFRESH_SECONDS)).isoformat(), + aware_now.isoformat(), + next_due.isoformat(), + ), + ) + connection.commit() + return RefreshResult(next_due, history_appended, str(presence["presence"])) + finally: + connection.close() + + +def record_transient_failure( + data_root: Path, + now: datetime, + error: SteamNetworkError, +) -> datetime: + """记录可观察的当前失败，并返回有界重试 deadline。""" + + aware_now = _aware(now) + retry_due = aware_now + timedelta(seconds=_TRANSIENT_RETRY_SECONDS) + connection = _connect(data_root) + try: + _ensure_current_row(connection, aware_now) + connection.execute( + """ + UPDATE current_state + SET last_refresh_attempt_at = ?, + last_refresh_error = ?, + next_refresh_at = ? + WHERE singleton = 1 + """, + (aware_now.isoformat(), str(error), retry_due.isoformat()), + ) + connection.commit() + return retry_due + finally: + connection.close() + + +def wake_context(data_root: Path, now: datetime) -> dict[str, object] | None: + """读取 fresh current state；stale 或 unknown 时不返回 hint。""" + + aware_now = _aware(now) + database = data_root / _DB_NAME + if not database.is_file(): + return None + connection = sqlite3.connect( + f"file:{database.as_posix()}?mode=ro", + uri=True, + timeout=30, + ) + connection.row_factory = sqlite3.Row + try: + _validate_schema(connection) + row = connection.execute( + "SELECT * FROM current_state WHERE singleton = 1" + ).fetchone() + if row is None or row["presence_expires_at"] is None: + return None + expires_at = _parse_aware(str(row["presence_expires_at"]), "presence_expires_at") + presence = _json_object(row["presence_json"], "presence_json") + if expires_at < aware_now or presence.get("presence") == "unknown": + return None + games = _json_list(row["current_games_json"], "current_games_json") + previous_at, previous_games = _previous_snapshot(connection, aware_now) + return { + "observed_at": row["presence_observed_at"], + "expires_at": row["presence_expires_at"], + "presence": presence["presence"], + "online_status": presence["online_status"], + "currently_playing": presence["currently_playing"], + "interruptibility": presence["interruptibility"], + "confidence": presence["confidence"], + "transition": presence["transition"], + "games": _context_games(games, previous_games), + "last_snapshot_checked_at": row["last_snapshot_checked_at"], + "previous_snapshot_at": previous_at, + } + finally: + connection.close() + + +def state(data_root: Path, now: datetime) -> dict[str, object]: + """返回 fixture 和诊断使用的 current state 与历史计数。""" + + aware_now = _aware(now) + connection = _connect(data_root) + try: + _ensure_current_row(connection, aware_now) + row = connection.execute( + "SELECT * FROM current_state WHERE singleton = 1" + ).fetchone() + if row is None: + raise RuntimeError("Steam current_state singleton 缺失") + snapshot_runs = connection.execute( + "SELECT COUNT(*) FROM snapshot_runs" + ).fetchone()[0] + snapshots = connection.execute("SELECT COUNT(*) FROM snapshots").fetchone()[0] + return {**dict(row), "snapshot_runs": snapshot_runs, "snapshots": snapshots} + finally: + connection.close() + + +def _connect(data_root: Path) -> sqlite3.Connection: + data_root.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(data_root / _DB_NAME, timeout=30) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA busy_timeout=30000") + connection.execute("PRAGMA journal_mode=WAL") + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshotted_at TEXT NOT NULL, + game_appid INTEGER NOT NULL, + game_name TEXT NOT NULL, + playtime_2w_mins INTEGER NOT NULL, + playtime_forever_mins INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS snapshot_runs ( + snapshotted_at TEXT PRIMARY KEY + ); + CREATE TABLE IF NOT EXISTS current_state ( + singleton INTEGER PRIMARY KEY CHECK(singleton = 1), + presence_json TEXT, + presence_observed_at TEXT, + presence_expires_at TEXT, + current_games_json TEXT NOT NULL, + last_refresh_attempt_at TEXT, + last_refresh_error TEXT, + last_snapshot_checked_at TEXT, + last_history_fingerprint TEXT, + next_refresh_at TEXT NOT NULL + ); + """ + ) + _validate_schema(connection) + connection.execute( + "CREATE INDEX IF NOT EXISTS idx_snap_time ON snapshots(snapshotted_at)" + ) + connection.execute( + "INSERT OR IGNORE INTO snapshot_runs(snapshotted_at) " + "SELECT DISTINCT snapshotted_at FROM snapshots" + ) + return connection + + +def _validate_schema(connection: sqlite3.Connection) -> None: + required = { + "snapshots": { + "id", "snapshotted_at", "game_appid", "game_name", + "playtime_2w_mins", "playtime_forever_mins", + }, + "snapshot_runs": {"snapshotted_at"}, + "current_state": { + "singleton", "presence_json", "presence_observed_at", + "presence_expires_at", "current_games_json", + "last_refresh_attempt_at", "last_refresh_error", + "last_snapshot_checked_at", "last_history_fingerprint", + "next_refresh_at", + }, + } + for table, expected in required.items(): + columns = { + str(row["name"]) + for row in connection.execute(f"PRAGMA table_info({table})").fetchall() + } + if not expected <= columns: + raise RuntimeError(f"Steam SQLite schema 不兼容: {table}") + + +def _ensure_current_row(connection: sqlite3.Connection, now: datetime) -> None: + latest = connection.execute( + "SELECT snapshotted_at FROM snapshot_runs ORDER BY snapshotted_at DESC LIMIT 1" + ).fetchone() + latest_at = None if latest is None else str(latest["snapshotted_at"]) + fingerprint = _latest_history_fingerprint(connection, latest_at) + connection.execute( + """ + INSERT OR IGNORE INTO current_state( + singleton, current_games_json, last_snapshot_checked_at, + last_history_fingerprint, next_refresh_at + ) VALUES(1, '[]', ?, ?, ?) + """, + (latest_at, fingerprint, now.isoformat()), + ) + + +def _read_current(data_root: Path, now: datetime) -> dict[str, object]: + connection = _connect(data_root) + try: + _ensure_current_row(connection, now) + row = connection.execute( + "SELECT * FROM current_state WHERE singleton = 1" + ).fetchone() + if row is None: + raise RuntimeError("Steam current_state singleton 缺失") + connection.commit() + return dict(row) + finally: + connection.close() + + +def _snapshot_due( + state: Mapping[str, object], + config: SteamRuntimeConfig, + now: datetime, +) -> bool: + checked = state.get("last_snapshot_checked_at") + if checked is None: + return True + elapsed = (now - _parse_aware(str(checked), "last_snapshot_checked_at")).total_seconds() + return elapsed >= config.snapshot_interval_seconds + + +def _fetch_player_summary(config: SteamRuntimeConfig) -> dict[str, object]: + data = _steam_get( + config, + "ISteamUser/GetPlayerSummaries/v0002", + {"steamids": config.steam_id}, + ) + response = _mapping(data.get("response"), "Steam player response") + players = _list(response.get("players"), "Steam players") + if not players: + return {} + return dict(_mapping(players[0], "Steam player")) + + +def _fetch_recently_played(config: SteamRuntimeConfig) -> list[dict[str, object]]: + data = _steam_get( + config, + "IPlayerService/GetRecentlyPlayedGames/v0001", + {"steamid": config.steam_id, "count": 0}, + ) + response = _mapping(data.get("response"), "Steam games response") + raw_games = _list(response.get("games", []), "Steam games") + games: list[dict[str, object]] = [] + for raw in raw_games: + game = _mapping(raw, "Steam game") + appid = game.get("appid") + if not isinstance(appid, int) or isinstance(appid, bool) or appid <= 0: + raise RuntimeError("Steam game appid 无效") + games.append( + { + "appid": appid, + "name": str(game.get("name") or f"App {appid}"), + "playtime_2weeks": _nonnegative_int( + game.get("playtime_2weeks", 0), "playtime_2weeks" + ), + "playtime_forever": _nonnegative_int( + game.get("playtime_forever", 0), "playtime_forever" + ), + } + ) + return sorted(games, key=lambda item: cast(int, item["appid"])) + + +def _steam_get( + config: SteamRuntimeConfig, + path: str, + params: Mapping[str, object], +) -> dict[str, object]: + query = urlencode({**params, "key": config.steam_api_key, "format": "json"}) + url = f"https://api.steampowered.com/{path}?{query}" + try: + with urlopen(url, timeout=15) as response: + payload = json.loads(response.read().decode("utf-8")) + except HTTPError as error: + if error.code in {408, 429} or error.code >= 500: + raise SteamNetworkError(f"Steam API HTTP {error.code}: {path}") from error + raise RuntimeError(f"Steam API 拒绝请求 HTTP {error.code}: {path}") from error + except (URLError, TimeoutError, OSError) as error: + raise SteamNetworkError(f"Steam API 网络失败: {path}: {error}") from error + except json.JSONDecodeError as error: + raise RuntimeError(f"Steam API 返回非法 JSON: {path}") from error + return dict(_mapping(payload, "Steam API payload")) + + +def _presence( + summary: Mapping[str, object], + now: datetime, + previous_presence: str, +) -> dict[str, object]: + status_by_code = {0: "offline", 1: "online", 2: "busy", 3: "away", 4: "snooze"} + if not summary: + status = "unknown" + else: + state = summary.get("personastate", 0) + if not isinstance(state, int) or isinstance(state, bool): + raise RuntimeError("Steam personastate 无效") + status = "in-game" if summary.get("gameid") else status_by_code.get(state, "offline") + presence = { + "in-game": "in_game", + "online": "active", + "busy": "active", + "away": "idle", + "snooze": "idle", + "offline": "offline", + }.get(status, "unknown") + interruptibility = { + "in-game": 0.1, + "online": 0.8, + "busy": 0.1, + "away": 0.4, + "snooze": 0.3, + "offline": 0.0, + }.get(status, 0.4) + transition = "" + if ( + previous_presence != "unknown" + and presence != "unknown" + and presence != previous_presence + ): + transition = f"{previous_presence}->{presence}" + return { + "observed_at": now.isoformat(), + "online_status": status, + "presence": presence, + "currently_playing": summary.get("gameextrainfo"), + "interruptibility": interruptibility, + "confidence": 0.9 if presence != "unknown" else 0.1, + "transition": transition, + } + + +def _append_snapshot( + connection: sqlite3.Connection, + games: list[dict[str, object]], + now: datetime, +) -> None: + timestamp = now.isoformat() + connection.execute( + "INSERT INTO snapshot_runs(snapshotted_at) VALUES (?)", (timestamp,) + ) + connection.executemany( + """ + INSERT INTO snapshots( + snapshotted_at, game_appid, game_name, + playtime_2w_mins, playtime_forever_mins + ) VALUES (?, ?, ?, ?, ?) + """, + [ + ( + timestamp, + game["appid"], + game["name"], + game["playtime_2weeks"], + game["playtime_forever"], + ) + for game in games + ], + ) + + +def _latest_history_fingerprint( + connection: sqlite3.Connection, + snapshot_at: str | None, +) -> str | None: + if snapshot_at is None: + return None + rows = connection.execute( + """ + SELECT game_appid, game_name, playtime_2w_mins, playtime_forever_mins + FROM snapshots WHERE snapshotted_at = ? ORDER BY game_appid + """, + (snapshot_at,), + ).fetchall() + games = [ + { + "appid": row["game_appid"], + "name": row["game_name"], + "playtime_2weeks": row["playtime_2w_mins"], + "playtime_forever": row["playtime_forever_mins"], + } + for row in rows + ] + return _games_fingerprint(games) if games else None + + +def _games_fingerprint(games: list[dict[str, object]]) -> str: + return hashlib.sha256(_json(games).encode("utf-8")).hexdigest() + + +def _previous_snapshot( + connection: sqlite3.Connection, + now: datetime, +) -> tuple[str | None, dict[int, sqlite3.Row]]: + cutoff = (now - timedelta(days=14)).isoformat() + meta = connection.execute( + """ + SELECT snapshotted_at FROM snapshot_runs + WHERE snapshotted_at <= ? ORDER BY snapshotted_at DESC LIMIT 1 + """, + (cutoff,), + ).fetchone() + if meta is None: + return None, {} + snapshot_at = str(meta["snapshotted_at"]) + rows = connection.execute( + "SELECT * FROM snapshots WHERE snapshotted_at = ?", + (snapshot_at,), + ).fetchall() + return snapshot_at, {int(row["game_appid"]): row for row in rows} + + +def _context_games( + games: list[object], + previous: Mapping[int, sqlite3.Row], +) -> list[dict[str, object]]: + context: list[dict[str, object]] = [] + current_appids: set[int] = set() + for raw in games: + game = _mapping(raw, "Steam current game") + appid = _nonnegative_int(game["appid"], "appid") + current_appids.add(appid) + previous_row = previous.get(appid) + context.append( + { + "name": game["name"], + "recent_2w_hours": round( + _nonnegative_int(game["playtime_2weeks"], "playtime_2weeks") / 60, + 1, + ), + "all_time_hours": round( + _nonnegative_int(game["playtime_forever"], "playtime_forever") / 60, + 1, + ), + "previous_snapshot_2w_hours": ( + 0.0 + if previous_row is None + else round(int(previous_row["playtime_2w_mins"]) / 60, 1) + ), + } + ) + for appid, previous_row in previous.items(): + if appid in current_appids or int(previous_row["playtime_2w_mins"]) == 0: + continue + context.append( + { + "name": str(previous_row["game_name"]), + "recent_2w_hours": 0.0, + "all_time_hours": round( + int(previous_row["playtime_forever_mins"]) / 60, + 1, + ), + "previous_snapshot_2w_hours": round( + int(previous_row["playtime_2w_mins"]) / 60, + 1, + ), + } + ) + return sorted(context, key=lambda item: cast(float, item["recent_2w_hours"]), reverse=True) + + +def _json(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def _json_object(value: object, label: str) -> dict[str, object]: + parsed = json.loads(str(value)) + return dict(_mapping(parsed, label)) + + +def _json_list(value: object, label: str) -> list[object]: + parsed = json.loads(str(value)) + return list(_list(parsed, label)) + + +def _mapping(value: object, label: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise RuntimeError(f"{label} 必须是 object") + return cast(Mapping[str, Any], value) + + +def _list(value: object, label: str) -> list[Any]: + if not isinstance(value, list): + raise RuntimeError(f"{label} 必须是 list") + return cast(list[Any], value) + + +def _nonnegative_int(value: object, label: str) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise RuntimeError(f"Steam {label} 无效") + return value + + +def _aware(value: datetime) -> datetime: + if value.tzinfo is None: + raise ValueError("Steam clock 必须包含时区") + return value.astimezone(UTC) + + +def _parse_aware(value: str, label: str) -> datetime: + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + raise RuntimeError(f"Steam {label} 必须包含时区") + return parsed.astimezone(UTC) diff --git a/mcp/runtime_config.py b/steam_runtime/config.py similarity index 94% rename from mcp/runtime_config.py rename to steam_runtime/config.py index 02eac76..b381876 100644 --- a/mcp/runtime_config.py +++ b/steam_runtime/config.py @@ -15,7 +15,7 @@ class SteamRuntimeConfig: def load_runtime_config(path: Path) -> SteamRuntimeConfig: """读取并校验 formal Steam runtime 配置。""" - # 1. 配置只来自 formal plugin-data，不接受 ambient secret + # 1. 配置只来自 formal plugin-data，不接受 ambient secret。 try: raw = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as error: @@ -25,7 +25,7 @@ def load_runtime_config(path: Path) -> SteamRuntimeConfig: if not isinstance(raw, dict): raise RuntimeError("steam_mcp_config.json 根节点必须是 object") - # 2. 启动前建立完整 credential 与用户身份不变量 + # 2. 在配置边界建立 credential、用户身份和刷新间隔不变量。 api_key = raw.get("steam_api_key") steam_id = raw.get("steam_id") if not isinstance(api_key, str) or not api_key.strip(): diff --git a/tests/test_context_source.py b/tests/test_context_source.py new file mode 100644 index 0000000..e1600e1 --- /dev/null +++ b/tests/test_context_source.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from agent.control.timer import TimerReceipt, TimerStatus +from agent.lifecycle.types import BeforeTurnCtx +from agent.plugin_composition import PluginTimers +from context_source import SteamContextRuntime +from steam_runtime import backend + + +class _TimerHandle: + def __init__(self, timer_id: str, deadline: datetime, now: datetime) -> None: + self._id = timer_id + self.deadline = deadline + self.now = now + self.future: asyncio.Future[TimerReceipt] = ( + asyncio.get_running_loop().create_future() + ) + + @property + def id(self) -> str: + return self._id + + async def result(self) -> TimerReceipt: + return await asyncio.shield(self.future) + + async def cancel(self) -> TimerReceipt: + if not self.future.done(): + self.future.set_result(self._receipt(TimerStatus.CANCELLED)) + return await self.future + + async def cleanup(self) -> None: + _ = await self.cancel() + + def fire(self) -> None: + self.future.set_result(self._receipt(TimerStatus.FIRED)) + + def _receipt(self, status: TimerStatus) -> TimerReceipt: + return TimerReceipt(self.id, self.deadline, self.now, status) + + +class _Timer: + def __init__(self, now: datetime) -> None: + self.now = now + self.handles: list[_TimerHandle] = [] + + def schedule(self, deadline: datetime) -> _TimerHandle: + handle = _TimerHandle(f"timer:{len(self.handles)}", deadline, self.now) + self.handles.append(handle) + return handle + + +class _Health: + def __init__(self) -> None: + self.reason: str | None = None + + @property + def healthy(self) -> bool: + return self.reason is None + + def degrade(self, reason: str) -> None: + self.reason = reason + + def recover(self) -> None: + self.reason = None + + +def _config(data_root: Path) -> None: + data_root.mkdir(parents=True, exist_ok=True) + (data_root / "steam_mcp_config.json").write_text( + json.dumps( + { + "steam_api_key": "test-key", + "steam_id": "test-user", + "snapshot_interval_seconds": 300, + } + ), + encoding="utf-8", + ) + + +async def _eventually(predicate) -> None: + for _ in range(200): + if predicate(): + return + await asyncio.sleep(0.01) + raise AssertionError("condition did not settle") + + +def _ctx(now: datetime, channel: str) -> BeforeTurnCtx: + return BeforeTurnCtx( + session_key="session", + channel=channel, + chat_id="chat", + content="hello", + timestamp=now, + retrieved_memory_block="", + retrieval_trace_raw=None, + history_messages=(), + turn_id="turn:1", + ) + + +@pytest.mark.asyncio +async def test_network_incident_retries_and_recovers( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + timer = _Timer(now) + health = _Health() + incidents: list[tuple[str, str]] = [] + calls = 0 + + def refresh(_data_root: Path, attempt: datetime) -> backend.RefreshResult: + nonlocal calls + calls += 1 + if calls == 1: + raise backend.SteamNetworkError("temporary outage") + return backend.RefreshResult(attempt + timedelta(minutes=5), False, "active") + + monkeypatch.setattr(backend, "refresh", refresh) + runtime = SteamContextRuntime( + tmp_path, + PluginTimers(timer), + health, # type: ignore[arg-type] + lambda kind, message: incidents.append((kind, message)), + now=lambda: now, + ) + await runtime.start() + handler = runtime._log_handler # pyright: ignore[reportPrivateUsage] + assert handler is not None + assert handler.maxBytes == 5 * 1024 * 1024 + assert handler.backupCount == 3 + + timer.handles[0].fire() + await _eventually(lambda: len(timer.handles) == 2) + assert incidents == [("steam_refresh_transient", "temporary outage")] + assert "temporary outage" in str(backend.state(tmp_path, now)["last_refresh_error"]) + + timer.handles[1].fire() + await _eventually(lambda: len(timer.handles) == 3) + assert calls == 2 + assert health.reason is None + await runtime.close() + + +def test_context_listener_is_wake_only_fresh_only_and_read_only( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + monkeypatch.setattr( + backend, + "_fetch_player_summary", + lambda _: {"personastate": 1}, + ) + monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: []) + _ = backend.refresh(tmp_path, now) + database = tmp_path / "steam_proactive.sqlite3" + before = hashlib.sha256(database.read_bytes()).hexdigest() + runtime = SteamContextRuntime( + tmp_path, + PluginTimers(None), + _Health(), # type: ignore[arg-type] + lambda _kind, _message: None, + now=lambda: now, + ) + + passive = _ctx(now, "passive") + runtime.prepare(passive) + wake = _ctx(now + timedelta(minutes=1), "wake") + runtime.prepare(wake) + stale = _ctx(now + timedelta(minutes=6), "wake") + runtime.prepare(stale) + + assert passive.extra_hints == [] + assert len(wake.extra_hints) == 1 + assert wake.extra_hints[0].startswith("Steam current context:\n") + assert wake.abort is False + assert stale.extra_hints == [] + assert hashlib.sha256(database.read_bytes()).hexdigest() == before + + +@pytest.mark.asyncio +async def test_contract_failure_degrades_and_does_not_retry( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + _config(tmp_path) + timer = _Timer(now) + health = _Health() + incidents: list[tuple[str, str]] = [] + runtime = SteamContextRuntime( + tmp_path, + PluginTimers(timer), + health, # type: ignore[arg-type] + lambda kind, message: incidents.append((kind, message)), + now=lambda: now, + ) + monkeypatch.setattr( + backend, + "refresh", + lambda *_: (_ for _ in ()).throw(RuntimeError("schema mismatch")), + ) + + await runtime.start() + task = runtime._task # pyright: ignore[reportPrivateUsage] + assert task is not None + timer.handles[0].fire() + with pytest.raises(RuntimeError, match="schema mismatch"): + await task + + assert len(timer.handles) == 1 + assert health.reason == "RuntimeError: schema mismatch" + assert incidents == [ + ("steam_refresh_contract", "RuntimeError: schema mismatch") + ] + await runtime.close() diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py index 05d6c0f..7069c08 100644 --- a/tests/test_manager_integration.py +++ b/tests/test_manager_integration.py @@ -1,98 +1,101 @@ from __future__ import annotations +import asyncio import hashlib import json -import os import shutil +import sqlite3 import sys +from datetime import UTC, datetime from pathlib import Path import pytest -from agent.plugins.artifacts import ArtifactPointer, write_pointers -from agent.plugins.generation_activity_host import ActivityHost -from agent.plugins.generation_proactive_host import ProactiveActivityAdapter -from agent.plugins.manifest import write_plugin_manifest +import agent.plugins.manager as plugin_manager_module +from agent.control.timer import TimerReceipt, TimerStatus +from agent.lifecycle.composition import CONTEXT_PREPARED_EVENT +from agent.lifecycle.types import BeforeTurnCtx from agent.plugins.manager import PluginManager from bus.event_bus import EventBus +from steam_runtime import backend -ROOT = Path(__file__).resolve().parents[1] +ROOT = Path(__file__).resolve().parents[1] -def _stage_installed_plugin(tmp_path: Path) -> Path: - """为 installed candidate 测试准备 stable/latest 两个 immutable artifact。""" - # 1. 复制完整插件 artifact，复用当前测试解释器作为隔离 MCP runtime。 - plugin_base = tmp_path / "home" / "cache" / "github" / "steam" - artifacts = plugin_base / ".artifacts" - runtime = Path(sys.executable).parent.parent - for label in ("stable", "candidate"): - artifact = artifacts / label - shutil.copytree( - ROOT, - artifact, - ignore=shutil.ignore_patterns( - ".git", - ".akashic-core", - ".pytest_cache", - "__pycache__", - ".venv", - ), +class _TimerHandle: + def __init__(self, timer_id: str, deadline: datetime, now: datetime) -> None: + self._id = timer_id + self.deadline = deadline + self.now = now + self.future: asyncio.Future[TimerReceipt] = ( + asyncio.get_running_loop().create_future() ) - (artifact / "mcp" / ".venv").symlink_to(runtime, target_is_directory=True) - if label == "candidate": - (artifact / ".candidate-marker").write_text("candidate\n", encoding="utf-8") - - # 2. stable 与 latest 指向不同 artifact，显式打开 installed 插件。 - write_pointers( - plugin_base, - stable=ArtifactPointer(".artifacts/stable"), - latest=ArtifactPointer(".artifacts/stable"), - ) - write_plugin_manifest( - {"steam@github": True}, - plugins_home=tmp_path / "home", - ) - return plugin_base + + @property + def id(self) -> str: + return self._id + + async def result(self) -> TimerReceipt: + return await asyncio.shield(self.future) + + async def cancel(self) -> TimerReceipt: + if not self.future.done(): + self.future.set_result(self._receipt(TimerStatus.CANCELLED)) + return await self.future + + async def cleanup(self) -> None: + _ = await self.cancel() + + def _receipt(self, status: TimerStatus) -> TimerReceipt: + return TimerReceipt(self.id, self.deadline, self.now, status) + + +class _Timer: + def __init__(self, now: datetime) -> None: + self.now = now + self.handles: list[_TimerHandle] = [] + + def schedule(self, deadline: datetime) -> _TimerHandle: + handle = _TimerHandle(f"timer:{len(self.handles)}", deadline, self.now) + self.handles.append(handle) + return handle + + +async def _eventually(predicate) -> None: + for _ in range(300): + if predicate(): + return + await asyncio.sleep(0.01) + raise AssertionError("condition did not settle") def _stage_plugin(tmp_path: Path) -> Path: - """复制可执行插件，并复用当前测试解释器的依赖环境。""" + """复制真实 Steam 插件并复用当前测试解释器。""" source = tmp_path / "plugins" / "steam" - source.mkdir(parents=True) - for relative in ( - "plugin.py", - "akashic.plugin.toml", - "mcp/requirements.txt", - "mcp/run_mcp.py", - "mcp/runtime_config.py", - "mcp/steam_mcp.py", - "mcp/steam_proactive.py", - "mcp/http_client.py", - ): - target = source / relative - target.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(ROOT / relative, target) - shutil.copytree(ROOT / "skills", source / "skills") + shutil.copytree( + ROOT, + source, + ignore=shutil.ignore_patterns( + ".git", + ".akashic-core", + ".plugin-contracts", + ".pytest_cache", + ".venv", + "__pycache__", + "tests", + ), + ) runtime = Path(sys.executable).parent.parent (source / "mcp" / ".venv").symlink_to(runtime, target_is_directory=True) return source -@pytest.mark.asyncio -async def test_manager_boots_formal_steam_without_network_calls_and_drains( - tmp_path: Path, -) -> None: - """启动真实 stdio handshake，并证明 formal data 与 runtime 精确回收。""" - - # 1. 只提供测试专用 formal 配置；测试不调用任何 Steam tool - plugin_root = _stage_plugin(tmp_path) - workspace = tmp_path / "workspace" - data_root = workspace / "plugin-data" / "steam-builtin" - data_root.mkdir(parents=True) - config = data_root / "steam_mcp_config.json" - config.write_text( +def _config(data_root: Path) -> Path: + data_root.mkdir(parents=True, exist_ok=True) + path = data_root / "steam_mcp_config.json" + path.write_text( json.dumps( { "steam_api_key": "test-only", @@ -102,149 +105,143 @@ async def test_manager_boots_formal_steam_without_network_calls_and_drains( ), encoding="utf-8", ) - config_digest = hashlib.sha256(config.read_bytes()).hexdigest() - - # 2. 走真实 Manager/Host formal publication，只观察 tools/list - manager = PluginManager( - plugin_dirs=[plugin_root.parent], - event_bus=EventBus(), - tool_registry=None, - workspace=workspace, - installed_cache_root=tmp_path / "cache", + return path + + +def _seed_fresh_state(data_root: Path, now: datetime) -> None: + backend.initialize(data_root, now) + database = data_root / "steam_proactive.sqlite3" + presence = { + "observed_at": now.isoformat(), + "online_status": "online", + "presence": "active", + "currently_playing": None, + "interruptibility": 0.8, + "confidence": 0.9, + "transition": "", + } + with sqlite3.connect(database) as connection: + connection.execute( + """ + UPDATE current_state + SET presence_json = ?, presence_observed_at = ?, + presence_expires_at = ?, current_games_json = '[]' + WHERE singleton = 1 + """, + ( + json.dumps(presence, sort_keys=True, separators=(",", ":")), + now.isoformat(), + datetime(2026, 8, 23, 8, 5, tzinfo=UTC).isoformat(), + ), + ) + connection.commit() + + +def _ctx(now: datetime, channel: str) -> BeforeTurnCtx: + return BeforeTurnCtx( + session_key="session", + channel=channel, + chat_id="chat", + content="hello", + timestamp=now, + retrieved_memory_block="", + retrieval_trace_raw=None, + history_messages=(), + turn_id="turn:1", ) - adapter = ProactiveActivityAdapter(manager.composition_generation_host) - activity = ActivityHost((adapter,)) - manager.bind_activity_host(activity) - snapshot = None - generation_id = None - route = None - try: - await manager.load_all() - snapshot = manager.current_snapshot - assert snapshot is not None and snapshot.mcp_server_registry is not None - assert tuple(snapshot.mcp_server_registry) == ("steam",) - generation = next(iter(snapshot.generations.values())) - generation_id = generation.generation_id - runtime = manager.composition_generation_host.get(generation_id) - assert runtime is not None and runtime.mode == "formal" - assert runtime.mcp is not None and runtime.mcp.state == "ready" - server = runtime.mcp.server("steam") - route = server.route() - assert route.mode == "formal" - assert "get_steam_context" in route.tool_names - assert "take_steam_snapshot" in route.tool_names - logs = list(server.logs().stdout) + list(server.logs().stderr) - assert not any("CallToolRequest" in line or '"tools/call"' in line for line in logs) - assert adapter.source_fetch_invocations == 0 - assert hashlib.sha256(config.read_bytes()).hexdigest() == config_digest - finally: - if route is not None: - await route.aclose() - await manager.terminate_all() - - # 3. terminate 后 exact generation、Activity、Root effects 全部释放 - assert activity.active is None - assert manager.composition_generation_host.get(generation_id) is None - assert snapshot is not None and snapshot.composition_root is not None - assert snapshot.composition_root.receipt().effects == () - assert snapshot.composition_root.topology_view().listeners == () @pytest.mark.asyncio -async def test_manager_candidate_publish_switches_formal_steam_and_cleans_validation( +async def test_manager_candidate_context_and_timer_handoff( tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: - """验证 installed Steam candidate 能 formalize、promote 并清理隔离资源。""" + """验证真实 MCP、Wake hint、静默 candidate 与 Timer 换班。""" - # 1. 准备 stable formal 配置与 latest candidate artifact。 - plugin_base = _stage_installed_plugin(tmp_path) + now = datetime(2026, 8, 23, 8, tzinfo=UTC) + timers: list[_Timer] = [] + + def timer_factory() -> _Timer: + timer = _Timer(now) + timers.append(timer) + return timer + + monkeypatch.setattr(plugin_manager_module, "AsyncioOneShotTimer", timer_factory) + plugin_root = _stage_plugin(tmp_path) workspace = tmp_path / "workspace" - data_root = workspace / "plugin-data" / "steam-github" - data_root.mkdir(parents=True) - (data_root / "steam_mcp_config.json").write_text( - json.dumps( - { - "steam_api_key": "test-only", - "steam_id": "76561198000000000", - "snapshot_interval_seconds": 3600, - } - ), - encoding="utf-8", - ) + data_root = workspace / "plugin-data" / "steam-builtin" + config = _config(data_root) + _seed_fresh_state(data_root, now) manager = PluginManager( - plugin_dirs=[tmp_path / "builtin-plugins"], + plugin_dirs=[plugin_root.parent], event_bus=EventBus(), tool_registry=None, workspace=workspace, - installed_cache_root=plugin_base.parent.parent, + installed_cache_root=tmp_path / "cache", ) - adapter = ProactiveActivityAdapter(manager.composition_generation_host) - activity = ActivityHost((adapter,)) - manager.bind_activity_host(activity) - stable_snapshot = None - candidate = None + await manager.load_all() + snapshot = manager.current_snapshot + assert snapshot is not None and snapshot.composition_root is not None + runtime = manager.composition_generation_host.get( + snapshot.generations["steam"].generation_id + ) + assert runtime is not None and runtime.mcp is not None + assert "get_player_summaries" in runtime.mcp.server("steam").tool_names + lifecycle = asyncio.create_task(manager.run_runtime_services()) try: - # 2. 先启动 stable formal runtime，再走 candidate -> latest_ready。 - await manager.load_all() - stable_snapshot = manager.current_snapshot - assert stable_snapshot is not None - stable_root = stable_snapshot.composition_root - assert stable_root is not None - assert stable_snapshot.proactive_component_catalog is not None - assert ( - stable_snapshot.proactive_component_catalog.root_instance_token - is stable_root.instance_token + # 1. 稳定 Root 只注册一个 Timer；listener 不影响 passive。 + await _eventually(lambda: sum(len(timer.handles) for timer in timers) == 1) + formal_timer = next(timer for timer in timers if timer.handles) + passive = _ctx(now, "passive") + wake = _ctx(now, "wake") + _ = await snapshot.composition_root.context.serial( + CONTEXT_PREPARED_EVENT, + passive, ) - write_pointers( - plugin_base, - stable=ArtifactPointer(".artifacts/stable"), - latest=ArtifactPointer(".artifacts/candidate"), + _ = await snapshot.composition_root.context.serial( + CONTEXT_PREPARED_EVENT, + wake, ) - candidate = await manager.prepare_candidate("steam@github") + assert passive.extra_hints == [] + assert len(wake.extra_hints) == 1 + + # 2. candidate 可握手，但没有 Timer、外网或正式 write set。 + database = data_root / "steam_proactive.sqlite3" + formal_hashes = { + "config": hashlib.sha256(config.read_bytes()).hexdigest(), + "database": hashlib.sha256(database.read_bytes()).hexdigest(), + } + with (plugin_root / "plugin.py").open("a", encoding="utf-8") as handle: + handle.write("\n# candidate fixture revision\n") + candidate = await manager.prepare_candidate("steam") assert candidate is not None and candidate.runtime_snapshot is not None - assert candidate.validation_workspace is not None + assert sum(len(timer.handles) for timer in timers) == 1 candidate_root = candidate.runtime_snapshot.composition_root assert candidate_root is not None - assert candidate.runtime_snapshot.proactive_component_catalog is not None - assert ( - candidate.runtime_snapshot.proactive_component_catalog.root_instance_token - is candidate_root.instance_token + candidate_wake = _ctx(now, "wake") + _ = await candidate_root.context.serial( + CONTEXT_PREPARED_EVENT, + candidate_wake, ) - - # 3. publish 重新 formalize 到最终 Root，再 promote 并观察验证目录清理。 - ready_result = await manager.publish_prepared("steam@github") - assert ready_result["publication_state"] == "latest_ready" - ready = manager.ready_candidate - assert ready is not None - assert ready is candidate - ready_snapshot = ready.runtime_snapshot - assert ready_snapshot is not None and ready_snapshot.composition_root is not None - assert ready_snapshot.proactive_component_catalog is not None - assert ( - ready_snapshot.proactive_component_catalog.root_instance_token - is ready_snapshot.composition_root.instance_token - ) - validation_root = candidate.validation_workspace.parent - assert validation_root.exists() - - promoted = await manager.switch_ready("steam@github") - assert promoted["publication_state"] == "promoted" - final_snapshot = manager.current_snapshot - assert final_snapshot is not None and final_snapshot.composition_root is not None - assert final_snapshot.proactive_component_catalog is not None - assert ( - final_snapshot.proactive_component_catalog.root_instance_token - is final_snapshot.composition_root.instance_token - ) - assert not validation_root.exists() - assert manager.ready_candidate is None - assert json.loads((plugin_base / ".pointers.json").read_text()) == { - "latest": ".artifacts/candidate", - "stable": ".artifacts/candidate", - } + assert candidate_wake.extra_hints == [] + assert hashlib.sha256(config.read_bytes()).hexdigest() == formal_hashes["config"] + assert hashlib.sha256(database.read_bytes()).hexdigest() == formal_hashes["database"] + + # 3. 发布先取消旧 Timer，再由新稳定 Root 注册一个 Timer。 + result = await manager.publish_prepared("steam") + assert result["publication_state"] == "committed" + await _eventually(lambda: sum(len(timer.handles) for timer in timers) == 2) + assert (await formal_timer.handles[0].result()).status is TimerStatus.CANCELLED + active = [ + handle + for timer in timers + for handle in timer.handles + if not handle.future.done() + ] + assert len(active) == 1 finally: + lifecycle.cancel() + _ = await asyncio.gather(lifecycle, return_exceptions=True) await manager.terminate_all() - assert candidate is not None - assert activity.active is None - assert manager.composition_generation_host.get(candidate.generation_id) is None + assert all(handle.future.done() for timer in timers for handle in timer.handles) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 57be110..dfbbe5c 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -2,22 +2,24 @@ import inspect from pathlib import Path +from typing import cast import plugin import pytest +from agent.control.timer import OneShotTimer from agent.plugin_composition import ( MCP_SERVERS, - PROACTIVE_COMPONENTS, + TIMERS, CompositionRoot, - PluginProactiveComponents, PluginRuntime, + PluginTimers, ) from agent.plugin_composition.mcp_slots import ( PluginMcpServers, _freeze_plugin_mcp_servers, ) -from agent.plugin_composition.proactive import _freeze_plugin_proactive_components from agent.plugins.composable import ComposablePlugin +from agent.plugins.manager import _copy_validation_data from agent.plugins.static_manifest import load_static_plugin_manifest @@ -27,22 +29,23 @@ def test_pure_v3_exports_and_exact_apply() -> None: assert plugin.api_version == 3 assert plugin.name == "steam" - assert plugin.version == "3.0.0" + assert plugin.version == "3.1.0" assert plugin.skill_roots == ("skills",) assert tuple(inspect.signature(plugin.apply).parameters) == ("ctx", "config") assert ComposablePlugin.from_module(plugin).skill_roots == ("skills",) - assert not hasattr(plugin, "SteamPlugin") @pytest.mark.asyncio -async def test_apply_registers_mcp_and_proactive_without_data_writes( +async def test_apply_registers_user_mcp_and_dormant_context_runtime( tmp_path: Path, ) -> None: root = CompositionRoot("steam:test") servers = PluginMcpServers(root.instance_token) - components = PluginProactiveComponents(root.instance_token) await root.context.provide(MCP_SERVERS, servers) - await root.context.provide(PROACTIVE_COMPONENTS, components) + await root.context.provide( + TIMERS, + PluginTimers(cast(OneShotTimer, object())), + ) data_root = tmp_path / "plugin-data" await root.mount( ComposablePlugin.from_module(plugin), @@ -60,37 +63,52 @@ async def test_apply_registers_mcp_and_proactive_without_data_writes( servers, root.instance_token, )["steam"].definition - source = _freeze_plugin_proactive_components( - components, - root.instance_token, - {"steam": "steam:test"}, - ).source("presence") - assert server.command == ("python", "mcp/run_mcp.py") - assert server.required_tools == ("get_steam_context",) - assert server.candidate_read_only_tools == ("get_steam_context",) + assert server.required_tools == ("get_player_summaries",) + assert server.candidate_read_only_tools == () assert server.candidate_env == {"STEAM_BACKEND": "recording"} - assert source is not None - assert source.definition.mcp_server == "steam" - assert source.definition.fetch_tool == "get_steam_context" assert not data_root.exists() + assert root.topology_view().listeners == ( + "serial:turn.context_prepared:steam", + "serial:runtime.started:steam", + "serial:runtime.stopping:steam", + ) await root.dispose() -def test_static_manifest_matches_module_and_recording_contract() -> None: +def test_static_manifest_excludes_state_and_bounded_logs() -> None: manifest = load_static_plugin_manifest(ROOT) assert manifest.name == plugin.name == "steam" - assert manifest.version == plugin.version == "3.0.0" + assert manifest.version == plugin.version == "3.1.0" assert manifest.api_version == plugin.api_version == 3 assert manifest.requirements == ("mcp/requirements.txt",) - assert manifest.exclude_data_paths == ( - "steam_mcp_config.json", - "steam_user_cache.json", - "steam_app_cache.json", - "steam_proactive.sqlite3", - ".steam-v2-migration.json", - ) + assert "steam_proactive.sqlite3" in manifest.exclude_data_paths + assert "steam_proactive.sqlite3-wal" in manifest.exclude_data_paths + assert "steam_context.runtime.log.3" in manifest.exclude_data_paths + assert "steam_mcp.runtime.log.3" in manifest.exclude_data_paths server = manifest.mcp_servers[0] - assert server.required_tools == ("get_steam_context",) - assert server.candidate_read_only_tools == ("get_steam_context",) + assert server.required_tools == ("get_player_summaries",) + assert server.candidate_read_only_tools == () assert server.candidate_env == (("STEAM_BACKEND", "recording"),) + + +def test_candidate_copy_excludes_formal_state_and_logs(tmp_path: Path) -> None: + manifest = load_static_plugin_manifest(ROOT) + source = tmp_path / "workspace" / "plugin-data" / "steam-builtin" + source.mkdir(parents=True) + for name in manifest.exclude_data_paths: + path = source / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"formal:{name}", encoding="utf-8") + (source / "candidate-visible.txt").write_text("visible", encoding="utf-8") + target = tmp_path / "validation" / "steam" + + inventory = _copy_validation_data( + source, + target, + manifest.exclude_data_paths, + ) + + assert inventory == ("candidate-visible.txt",) + assert (target / "candidate-visible.txt").read_text() == "visible" + assert not any((target / name).exists() for name in manifest.exclude_data_paths) From 4ba37db56e1ae8e1f5aa284a734a90d6298a9c4c Mon Sep 17 00:00:00 2001 From: test Date: Sun, 23 Aug 2026 19:19:23 +0800 Subject: [PATCH 2/4] =?UTF-8?q?fix:=20=E5=8E=9F=E5=AD=90=E9=AA=8C=E8=AF=81?= =?UTF-8?q?=20Steam=20=E5=8E=86=E5=8F=B2=E8=BF=81=E7=A7=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mcp/tests/test_snapshot_freshness.py | 88 ++++++++++++- steam_runtime/backend.py | 189 ++++++++++++++++++++++----- 2 files changed, 237 insertions(+), 40 deletions(-) diff --git a/mcp/tests/test_snapshot_freshness.py b/mcp/tests/test_snapshot_freshness.py index 49da72b..823735b 100644 --- a/mcp/tests/test_snapshot_freshness.py +++ b/mcp/tests/test_snapshot_freshness.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import json import sqlite3 from datetime import UTC, datetime, timedelta @@ -90,7 +91,7 @@ def test_initialize_adopts_existing_history_without_rewriting_it(tmp_path) -> No ("2026-08-01T08:00:00+00:00", 1, "Old Game", 60, 660), ] with sqlite3.connect(database) as connection: - connection.executescript( + connection.execute( """ CREATE TABLE snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -99,10 +100,12 @@ def test_initialize_adopts_existing_history_without_rewriting_it(tmp_path) -> No game_name TEXT NOT NULL, playtime_2w_mins INTEGER NOT NULL, playtime_forever_mins INTEGER NOT NULL - ); - CREATE TABLE snapshot_runs (snapshotted_at TEXT PRIMARY KEY); + ) """ ) + connection.execute( + "CREATE TABLE snapshot_runs (snapshotted_at TEXT PRIMARY KEY)" + ) connection.executemany( """ INSERT INTO snapshots( @@ -135,11 +138,86 @@ def test_initialize_adopts_existing_history_without_rewriting_it(tmp_path) -> No assert runs == [(row[0],) for row in existing] +def test_legacy_schema_install_rolls_back_and_can_be_retried( + tmp_path, + monkeypatch, +) -> None: + database = tmp_path / "steam_proactive.sqlite3" + with sqlite3.connect(database) as connection: + connection.execute( + """ + CREATE TABLE snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshotted_at TEXT NOT NULL, + game_appid INTEGER NOT NULL, + game_name TEXT NOT NULL, + playtime_2w_mins INTEGER NOT NULL, + playtime_forever_mins INTEGER NOT NULL + ) + """ + ) + connection.execute( + """ + INSERT INTO snapshots( + snapshotted_at, game_appid, game_name, + playtime_2w_mins, playtime_forever_mins + ) VALUES ('2026-08-01T08:00:00+00:00', 1, 'Old Game', 60, 600) + """ + ) + connection.commit() + before = database.read_bytes() + + def interrupted(connection: sqlite3.Connection) -> None: + connection.execute( + "CREATE TABLE snapshot_runs (snapshotted_at TEXT PRIMARY KEY)" + ) + raise RuntimeError("fixture interrupted migration") + + original = backend._install_schema + monkeypatch.setattr(backend, "_install_schema", interrupted) + with pytest.raises(RuntimeError, match="interrupted migration"): + backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + + assert database.read_bytes() == before + with sqlite3.connect(database) as connection: + assert connection.execute( + "SELECT name FROM sqlite_schema WHERE type = 'table' ORDER BY name" + ).fetchall() == [("snapshots",), ("sqlite_sequence",)] + assert connection.execute( + "SELECT game_name, playtime_forever_mins FROM snapshots" + ).fetchall() == [("Old Game", 600)] + + monkeypatch.setattr(backend, "_install_schema", original) + backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + current = backend.state(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + assert current["snapshot_runs"] == 1 + assert current["snapshots"] == 1 + + def test_incompatible_history_schema_fails_loud(tmp_path) -> None: database = tmp_path / "steam_proactive.sqlite3" with sqlite3.connect(database) as connection: connection.execute("CREATE TABLE snapshots(value TEXT NOT NULL)") + connection.execute("INSERT INTO snapshots(value) VALUES ('keep-me')") connection.commit() + before_bytes = database.read_bytes() + before_hash = hashlib.sha256(before_bytes).hexdigest() + before_sidecars = sorted(path.name for path in tmp_path.iterdir()) - with pytest.raises(RuntimeError, match="schema 不兼容"): - backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + for _ in range(2): + with pytest.raises(RuntimeError, match="schema 不兼容"): + backend.initialize(tmp_path, datetime(2026, 8, 23, tzinfo=UTC)) + + assert database.read_bytes() == before_bytes + assert hashlib.sha256(database.read_bytes()).hexdigest() == before_hash + assert sorted(path.name for path in tmp_path.iterdir()) == before_sidecars + with sqlite3.connect(database) as connection: + assert connection.execute( + "SELECT name, sql FROM sqlite_schema WHERE type = 'table'" + ).fetchall() == [ + ("snapshots", "CREATE TABLE snapshots(value TEXT NOT NULL)") + ] + assert connection.execute("SELECT value FROM snapshots").fetchall() == [ + ("keep-me",) + ] diff --git a/steam_runtime/backend.py b/steam_runtime/backend.py index a01a155..409d34d 100644 --- a/steam_runtime/backend.py +++ b/steam_runtime/backend.py @@ -18,6 +18,29 @@ _DB_NAME = "steam_proactive.sqlite3" _PRESENCE_REFRESH_SECONDS = 300 _TRANSIENT_RETRY_SECONDS = 60 +_TABLE_SCHEMAS = { + "snapshots": ( + ("id", "INTEGER", 0, None, 1), + ("snapshotted_at", "TEXT", 1, None, 0), + ("game_appid", "INTEGER", 1, None, 0), + ("game_name", "TEXT", 1, None, 0), + ("playtime_2w_mins", "INTEGER", 1, None, 0), + ("playtime_forever_mins", "INTEGER", 1, None, 0), + ), + "snapshot_runs": (("snapshotted_at", "TEXT", 0, None, 1),), + "current_state": ( + ("singleton", "INTEGER", 0, None, 1), + ("presence_json", "TEXT", 0, None, 0), + ("presence_observed_at", "TEXT", 0, None, 0), + ("presence_expires_at", "TEXT", 0, None, 0), + ("current_games_json", "TEXT", 1, None, 0), + ("last_refresh_attempt_at", "TEXT", 0, None, 0), + ("last_refresh_error", "TEXT", 0, None, 0), + ("last_snapshot_checked_at", "TEXT", 0, None, 0), + ("last_history_fingerprint", "TEXT", 0, None, 0), + ("next_refresh_at", "TEXT", 1, None, 0), + ), +} class SteamNetworkError(RuntimeError): @@ -227,11 +250,129 @@ def state(data_root: Path, now: datetime) -> dict[str, object]: def _connect(data_root: Path) -> sqlite3.Connection: data_root.mkdir(parents=True, exist_ok=True) - connection = sqlite3.connect(data_root / _DB_NAME, timeout=30) + database = data_root / _DB_NAME + if database.exists(): + _validate_existing_database(database) + connection = sqlite3.connect(database, timeout=30) connection.row_factory = sqlite3.Row connection.execute("PRAGMA busy_timeout=30000") + try: + # 1. 只在只读盘点通过后，用一个事务安装或补齐兼容 schema。 + connection.execute("BEGIN IMMEDIATE") + _install_schema(connection) + _validate_schema(connection) + connection.execute( + "INSERT OR IGNORE INTO snapshot_runs(snapshotted_at) " + "SELECT DISTINCT snapshotted_at FROM snapshots" + ) + connection.commit() + except BaseException: + connection.rollback() + connection.close() + raise connection.execute("PRAGMA journal_mode=WAL") - connection.executescript( + return connection + + +def _validate_schema(connection: sqlite3.Connection) -> None: + _validate_table_inventory(connection, require_all=True) + + +def _validate_existing_database(database: Path) -> None: + """只读盘点既有数据库，保证失败不会留下迁移痕迹。""" + + connection = sqlite3.connect( + f"file:{database.as_posix()}?mode=ro", + uri=True, + timeout=30, + ) + connection.row_factory = sqlite3.Row + try: + _validate_table_inventory(connection, require_all=False) + finally: + connection.close() + + +def _validate_table_inventory( + connection: sqlite3.Connection, + *, + require_all: bool, +) -> None: + existing = { + str(row["name"]) + for row in connection.execute( + """ + SELECT name FROM sqlite_schema + WHERE type = 'table' AND name NOT LIKE 'sqlite_%' + """ + ).fetchall() + } + unknown = existing - _TABLE_SCHEMAS.keys() + if unknown: + raise RuntimeError( + "Steam SQLite schema 不兼容: 未知表 " + ", ".join(sorted(unknown)) + ) + if require_all and existing != _TABLE_SCHEMAS.keys(): + missing = _TABLE_SCHEMAS.keys() - existing + raise RuntimeError( + "Steam SQLite schema 不兼容: 缺少表 " + ", ".join(sorted(missing)) + ) + for table in sorted(existing): + actual = tuple( + ( + str(row["name"]), + str(row["type"]).upper(), + int(row["notnull"]), + row["dflt_value"], + int(row["pk"]), + ) + for row in connection.execute(f"PRAGMA table_info({table})").fetchall() + ) + if actual != _TABLE_SCHEMAS[table]: + raise RuntimeError(f"Steam SQLite schema 不兼容: {table}") + _validate_table_constraints(connection, table) + _validate_snapshot_index(connection, required=require_all) + + +def _validate_table_constraints( + connection: sqlite3.Connection, + table: str, +) -> None: + row = connection.execute( + "SELECT sql FROM sqlite_schema WHERE type = 'table' AND name = ?", + (table,), + ).fetchone() + if row is None or row["sql"] is None: + raise RuntimeError(f"Steam SQLite schema 不兼容: {table}") + compact = "".join(str(row["sql"]).lower().split()) + if table == "snapshots" and "integerprimarykeyautoincrement" not in compact: + raise RuntimeError("Steam SQLite schema 不兼容: snapshots 缺少 AUTOINCREMENT") + if table == "current_state" and "check(singleton=1)" not in compact: + raise RuntimeError("Steam SQLite schema 不兼容: current_state 缺少 singleton CHECK") + + +def _validate_snapshot_index( + connection: sqlite3.Connection, + *, + required: bool, +) -> None: + row = connection.execute( + "SELECT name FROM sqlite_schema WHERE type = 'index' AND name = 'idx_snap_time'" + ).fetchone() + if row is None: + if required: + raise RuntimeError("Steam SQLite schema 不兼容: 缺少 idx_snap_time") + return + columns = tuple( + str(item["name"]) + for item in connection.execute("PRAGMA index_info(idx_snap_time)").fetchall() + ) + if columns != ("snapshotted_at",): + raise RuntimeError("Steam SQLite schema 不兼容: idx_snap_time") + + +def _install_schema(connection: sqlite3.Connection) -> None: + connection.execute( """ CREATE TABLE IF NOT EXISTS snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -240,10 +381,18 @@ def _connect(data_root: Path) -> sqlite3.Connection: game_name TEXT NOT NULL, playtime_2w_mins INTEGER NOT NULL, playtime_forever_mins INTEGER NOT NULL - ); + ) + """ + ) + connection.execute( + """ CREATE TABLE IF NOT EXISTS snapshot_runs ( snapshotted_at TEXT PRIMARY KEY - ); + ) + """ + ) + connection.execute( + """ CREATE TABLE IF NOT EXISTS current_state ( singleton INTEGER PRIMARY KEY CHECK(singleton = 1), presence_json TEXT, @@ -255,42 +404,12 @@ def _connect(data_root: Path) -> sqlite3.Connection: last_snapshot_checked_at TEXT, last_history_fingerprint TEXT, next_refresh_at TEXT NOT NULL - ); + ) """ ) - _validate_schema(connection) connection.execute( "CREATE INDEX IF NOT EXISTS idx_snap_time ON snapshots(snapshotted_at)" ) - connection.execute( - "INSERT OR IGNORE INTO snapshot_runs(snapshotted_at) " - "SELECT DISTINCT snapshotted_at FROM snapshots" - ) - return connection - - -def _validate_schema(connection: sqlite3.Connection) -> None: - required = { - "snapshots": { - "id", "snapshotted_at", "game_appid", "game_name", - "playtime_2w_mins", "playtime_forever_mins", - }, - "snapshot_runs": {"snapshotted_at"}, - "current_state": { - "singleton", "presence_json", "presence_observed_at", - "presence_expires_at", "current_games_json", - "last_refresh_attempt_at", "last_refresh_error", - "last_snapshot_checked_at", "last_history_fingerprint", - "next_refresh_at", - }, - } - for table, expected in required.items(): - columns = { - str(row["name"]) - for row in connection.execute(f"PRAGMA table_info({table})").fetchall() - } - if not expected <= columns: - raise RuntimeError(f"Steam SQLite schema 不兼容: {table}") def _ensure_current_row(connection: sqlite3.Connection, now: datetime) -> None: From 982fcb4bc34a9c06aa4c7998e58070d9eab5bae6 Mon Sep 17 00:00:00 2001 From: test Date: Mon, 24 Aug 2026 02:36:46 +0800 Subject: [PATCH 3/4] =?UTF-8?q?test:=20=E6=98=BE=E5=BC=8F=E5=A3=B0?= =?UTF-8?q?=E6=98=8E=20Steam=20fixture=20=E8=A7=A3=E9=87=8A=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/plugin-api-v3.yml | 1 + tests/test_manager_integration.py | 34 ++++++++++++++++++++++++++--- 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index 404767a..5962e90 100644 --- a/.github/workflows/plugin-api-v3.yml +++ b/.github/workflows/plugin-api-v3.yml @@ -55,6 +55,7 @@ jobs: - name: Verify Steam v3 composition and migration env: AKASHIC_AGENT_ROOT: .akashic-core + AKASHIC_PLUGIN_FIXTURE_PYTHON: ${{ github.workspace }}/mcp/.venv/bin/python PYTHONPATH: .akashic-core:mcp:mcp/.venv/lib/python3.13/site-packages run: mcp/.venv/bin/python -m pytest -q mcp/tests tests - name: Check changed v3 sources diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py index 7069c08..ba8486f 100644 --- a/tests/test_manager_integration.py +++ b/tests/test_manager_integration.py @@ -3,9 +3,9 @@ import asyncio import hashlib import json +import os import shutil import sqlite3 -import sys from datetime import UTC, datetime from pathlib import Path @@ -71,8 +71,9 @@ async def _eventually(predicate) -> None: def _stage_plugin(tmp_path: Path) -> Path: - """复制真实 Steam 插件并复用当前测试解释器。""" + """复制真实 Steam 插件并链接调用方声明的 artifact 运行时。""" + runtime = Path(os.environ["AKASHIC_PLUGIN_FIXTURE_PYTHON"]).parent.parent source = tmp_path / "plugins" / "steam" shutil.copytree( ROOT, @@ -87,11 +88,38 @@ def _stage_plugin(tmp_path: Path) -> Path: "tests", ), ) - runtime = Path(sys.executable).parent.parent (source / "mcp" / ".venv").symlink_to(runtime, target_is_directory=True) return source +def test_stage_plugin_links_declared_fixture_runtime( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime = tmp_path / "artifact" / ".venv" + fixture_python = runtime / "bin" / "python" + fixture_python.parent.mkdir(parents=True) + fixture_python.touch() + monkeypatch.setenv("AKASHIC_PLUGIN_FIXTURE_PYTHON", str(fixture_python)) + + source = _stage_plugin(tmp_path) + + assert (source / "mcp" / ".venv").readlink() == runtime + + +def test_stage_plugin_requires_fixture_python( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("AKASHIC_PLUGIN_FIXTURE_PYTHON", raising=False) + + with pytest.raises(KeyError) as error: + _stage_plugin(tmp_path) + + assert error.value.args == ("AKASHIC_PLUGIN_FIXTURE_PYTHON",) + assert not (tmp_path / "plugins").exists() + + def _config(data_root: Path) -> Path: data_root.mkdir(parents=True, exist_ok=True) path = data_root / "steam_mcp_config.json" From 4eeebbc7b8c0f6299ab8d86a998b2c86f7ef6183 Mon Sep 17 00:00:00 2001 From: test Date: Mon, 24 Aug 2026 20:46:58 +0800 Subject: [PATCH 4/4] fix: load Steam runtime as a package --- context_source.py | 2 +- plugin.py | 2 +- steam_runtime/backend.py | 2 +- tests/conftest.py | 12 ++++++++++++ tests/test_context_source.py | 4 ++-- tests/test_plugin.py | 2 +- 6 files changed, 18 insertions(+), 6 deletions(-) create mode 100644 tests/conftest.py diff --git a/context_source.py b/context_source.py index 8f66d19..15d72a4 100644 --- a/context_source.py +++ b/context_source.py @@ -12,7 +12,7 @@ from agent.lifecycle.types import BeforeTurnCtx from agent.plugin_composition import HealthHandle, PluginTimers -from steam_runtime import backend +from .steam_runtime import backend class SteamContextRuntime: diff --git a/plugin.py b/plugin.py index 7ed254e..3f4a1a0 100644 --- a/plugin.py +++ b/plugin.py @@ -12,7 +12,7 @@ McpServerDefinition, ) -from context_source import SteamContextRuntime +from .context_source import SteamContextRuntime class SteamConfig(BaseModel): diff --git a/steam_runtime/backend.py b/steam_runtime/backend.py index 409d34d..394e461 100644 --- a/steam_runtime/backend.py +++ b/steam_runtime/backend.py @@ -12,7 +12,7 @@ from urllib.parse import urlencode from urllib.request import urlopen -from steam_runtime.config import SteamRuntimeConfig, load_runtime_config +from .config import SteamRuntimeConfig, load_runtime_config _DB_NAME = "steam_proactive.sqlite3" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..be3139c --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +import sys +from pathlib import Path +from types import ModuleType + + +repo_root = Path(__file__).resolve().parents[1] +package = ModuleType("steam_test_plugin") +package.__path__ = [str(repo_root)] +package.__package__ = "steam_test_plugin" +sys.modules["steam_test_plugin"] = package diff --git a/tests/test_context_source.py b/tests/test_context_source.py index e1600e1..91e9d4d 100644 --- a/tests/test_context_source.py +++ b/tests/test_context_source.py @@ -11,8 +11,8 @@ from agent.control.timer import TimerReceipt, TimerStatus from agent.lifecycle.types import BeforeTurnCtx from agent.plugin_composition import PluginTimers -from context_source import SteamContextRuntime -from steam_runtime import backend +from steam_test_plugin.context_source import SteamContextRuntime # pyright: ignore[reportMissingImports] +from steam_test_plugin.steam_runtime import backend # pyright: ignore[reportMissingImports] class _TimerHandle: diff --git a/tests/test_plugin.py b/tests/test_plugin.py index dfbbe5c..2f0af97 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -4,8 +4,8 @@ from pathlib import Path from typing import cast -import plugin import pytest +from steam_test_plugin import plugin # pyright: ignore[reportMissingImports] from agent.control.timer import OneShotTimer from agent.plugin_composition import ( MCP_SERVERS,