refactor(plugin): 归还 KVCache 命令所有权

README.md

@@ -2,6 +2,8 @@
 
 Akashic 可观测性插件，负责采集 Turn、检索、记忆写入和全局错误遥测。
 
+插件同时拥有 `/kvcache`（兼容别名 `/cache_status`）只读命令。命令只读取当前会话在 `observe.db` 中的缓存统计；Core 只负责挂载 BeforeTurn module，不理解 Observe schema。
+
 ## 移动端
 
 插件自带一个移动端 Observe 入口，并在同一看板内提供两个任务视图：

kvcache_command.py

@@ -0,0 +1,212 @@
+from __future__ import annotations
+
+import logging
+import re
+import sqlite3
+from datetime import datetime
+from pathlib import Path
+from typing import Protocol, cast, final
+from zoneinfo import ZoneInfo
+
+from agent.lifecycle.types import BeforeTurnCtx, TurnState
+from agent.prompting import is_context_frame
+
+logger = logging.getLogger("plugin.observe.kvcache_command")
+
+_SESSION_SLOT = "session:session"
+_CTX_SLOT = "session:ctx"
+_TS_PATTERN = re.compile(r"(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2})")
+_BEIJING_TZ = ZoneInfo("Asia/Shanghai")
+
+
+class _BeforeTurnFrame(Protocol):
+    input: TurnState
+    slots: dict[str, object]
+
+
+@final
+class KVCacheCommandModule:
+    """把 Observe 自有的 KV Cache 数据渲染成短路命令回复。"""
+
+    slot = "observe.kvcache_command"
+    requires = ("before_turn.acquire_session", _SESSION_SLOT)
+    produces = (_CTX_SLOT,)
+
+    def __init__(self, db_path: Path | None) -> None:
+        self._db_path = db_path
+
+    async def run(self, frame: object) -> object:
+        typed_frame = cast(_BeforeTurnFrame, frame)
+        if _CTX_SLOT in typed_frame.slots:
+            return frame
+        state = typed_frame.input
+        if _normalize_command(state.msg.content) not in {"/kvcache", "/cache_status"}:
+            return frame
+        typed_frame.slots[_CTX_SLOT] = _abort_ctx(state, self._build_reply(state))
+        return frame
+
+    def _build_reply(self, state: TurnState) -> str:
+        """读取当前会话快照并保持既有命令结果。"""
+
+        # 1. 将缺少 Observe 状态表达为有效的空诊断结果
+        db_path = self._db_path
+        if db_path is None or not db_path.exists():
+            return "暂无 KVCache 数据（observe 数据库不存在）。"
+        limit = _command_limit(state.msg.content)
+
+        # 2. 只查询当前 Session 的 Observe 自有数据
+        try:
+            with sqlite3.connect(str(db_path)) as connection:
+                raw_rows = connection.execute(
+                    """
+                    SELECT llm_output, ts,
+                           react_cache_prompt_tokens, react_cache_hit_tokens
+                    FROM turns
+                    WHERE session_key=? AND react_cache_prompt_tokens IS NOT NULL
+                    ORDER BY id DESC LIMIT ?
+                    """,
+                    (state.session_key, limit),
+                ).fetchall()
+                rows = cast(
+                    list[tuple[object, object, object, object]],
+                    raw_rows,
+                )
+        except sqlite3.Error:
+            logger.exception("KVCache 查询失败")
+            return "KVCache 查询失败。"
+
+        # 3. 保持既有用户结果，不向 Core 泄露 schema
+        if not rows:
+            return "暂无 KVCache 数据。"
+        return _format_reply(rows)
+
+
+def _normalize_command(content: str) -> str:
+    parts = (content or "").strip().split(maxsplit=1)
+    if not parts:
+        return ""
+    return parts[0].lower().split("@", 1)[0]
+
+
+def _command_limit(content: str) -> int:
+    args = (content or "").strip().split()
+    if len(args) <= 1:
+        return 5
+    try:
+        return max(1, min(30, int(args[1])))
+    except ValueError:
+        return 5
+
+
+def _format_reply(
+    rows: list[tuple[object, object, object, object]],
+) -> str:
+    overall_prompt = sum(_db_int(row[2]) for row in rows)
+    overall_hit = sum(_db_int(row[3]) for row in rows)
+    overall_pct = (overall_hit / overall_prompt * 100) if overall_prompt > 0 else 0.0
+    lines = [
+        f"⚡ KVCache · 最近 {len(rows)} 轮",
+        "",
+        f"命中率  {overall_pct:.1f}%  {_pct_bar(overall_pct)}",
+        f"Token  {overall_hit:,} / {overall_prompt:,}",
+    ]
+    for llm_output, ts, prompt_tokens, hit_tokens in rows:
+        content = _content_to_text(_db_text(llm_output))
+        if is_context_frame(content):
+            content = ""
+        preview = _preview_text(content, limit=72)
+        hit = _db_int(hit_tokens)
+        prompt = _db_int(prompt_tokens)
+        pct = (hit / prompt * 100) if prompt > 0 else 0.0
+        lines.extend(
+            [
+                "",
+                "",
+                f"{_format_ts(_db_text(ts))}   {_pct_emoji(pct)} {pct:.1f}%  {_pct_bar(pct)}",
+                f"    {hit:,} / {prompt:,} tokens",
+            ]
+        )
+        if preview:
+            lines.append(f"    {preview}")
+    return "\n".join(lines)
+
+
+def _db_int(value: object) -> int:
+    if value is None:
+        return 0
+    if not isinstance(value, int) or isinstance(value, bool):
+        raise RuntimeError("observe.db KV Cache token 字段不是整数")
+    return value
+
+
+def _db_text(value: object) -> str:
+    if value is None:
+        return ""
+    if not isinstance(value, str):
+        raise RuntimeError("observe.db KV Cache 文本字段不是字符串")
+    return value
+
+
+def _abort_ctx(state: TurnState, reply: str) -> BeforeTurnCtx:
+    return BeforeTurnCtx(
+        session_key=state.session_key,
+        channel=state.msg.channel,
+        chat_id=state.msg.chat_id,
+        content=state.msg.content,
+        timestamp=state.msg.timestamp,
+        skill_names=[],
+        retrieved_memory_block="",
+        retrieval_trace_raw=None,
+        history_messages=(),
+        abort=True,
+        abort_reply=reply,
+    )
+
+
+def _format_ts(ts: str) -> str:
+    try:
+        parsed = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+        if parsed.tzinfo is not None:
+            parsed = parsed.astimezone(_BEIJING_TZ)
+        return f"{parsed.month}-{parsed.day} {parsed.hour:02d}:{parsed.minute:02d}"
+    except ValueError:
+        match = _TS_PATTERN.search(ts)
+        if match:
+            return f"{int(match.group(2))}-{int(match.group(3))} {match.group(4)}:{match.group(5)}"
+        return ts
+
+
+def _content_to_text(content: object) -> str:
+    if isinstance(content, str):
+        return content.strip()
+    if isinstance(content, list):
+        raw_items = cast(list[object], content)
+        parts: list[str] = []
+        for item in raw_items:
+            if not isinstance(item, dict):
+                continue
+            mapping = cast(dict[object, object], item)
+            if mapping.get("type") == "text":
+                parts.append(str(mapping.get("text", "")).strip())
+        return "\n".join(part for part in parts if part).strip()
+    return str(content).strip()
+
+
+def _preview_text(text: str, limit: int = 80) -> str:
+    normalized = " ".join(text.split())
+    if len(normalized) <= limit:
+        return normalized
+    return normalized[: limit - 1] + "…"
+
+
+def _pct_bar(pct: float, width: int = 10) -> str:
+    filled = max(0, min(width, round(pct / 100 * width)))
+    return "█" * filled + "░" * (width - filled)
+
+
+def _pct_emoji(pct: float) -> str:
+    if pct >= 80:
+        return "🟢"
+    if pct >= 40:
+        return "🟡"
+    return "🔴"

plugin.py

@@ -15,6 +15,7 @@
 
 from .collector import GlobalErrorCollector
 from .dashboard import ObserveDashboardReader
+from .kvcache_command import KVCacheCommandModule
 from .mobile_kvcache import KVCacheDashboardReader
 from .retention import run_retention_if_needed
 from .writer import TraceWriter
@@ -46,7 +47,17 @@ def mobile_ui(cls) -> MobileUiContribution:
         )
 
     name = "observe"
-    version = "1.2.0"
+    version = "1.3.0"
+
+    def telegram_bot_commands(self) -> list[tuple[str, str]]:
+        return [("kvcache", "查看 KVCache 状态")]
+
+    def before_turn_modules(self) -> list[object]:
+        workspace = self.context.workspace
+        db_path = (
+            None if workspace is None else workspace / "observe" / "observe.db"
+        )
+        return [KVCacheCommandModule(db_path)]
 
     def activate(self) -> None:
         workspace = self.context.workspace

tests/test_plugin.py

@@ -52,6 +52,7 @@ def _load_plugin_module():
 module = _load_plugin_module()
 ObservePlugin = module.ObservePlugin
 GlobalErrorCollector = module.GlobalErrorCollector
+KVCacheCommandModule = module.KVCacheCommandModule
 
 
 class _Emitter:
@@ -67,6 +68,90 @@ async def dispatch(self, outbound: object) -> bool:
         return True
 
 
+@pytest.mark.asyncio
+async def test_observe_owns_kvcache_command_and_existing_reply(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    db_path = workspace / "observe" / "observe.db"
+    db_path.parent.mkdir(parents=True)
+    connection = sqlite3.connect(db_path)
+    try:
+        connection.execute(
+            """
+            CREATE TABLE turns(
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                ts TEXT NOT NULL,
+                source TEXT NOT NULL,
+                session_key TEXT NOT NULL,
+                user_msg TEXT,
+                llm_output TEXT NOT NULL DEFAULT '',
+                react_cache_prompt_tokens INTEGER,
+                react_cache_hit_tokens INTEGER
+            )
+            """
+        )
+        connection.execute(
+            """
+            INSERT INTO turns(
+                ts, source, session_key, user_msg, llm_output,
+                react_cache_prompt_tokens, react_cache_hit_tokens
+            ) VALUES(?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                "2026-04-19T03:20:00+00:00",
+                "agent",
+                "telegram:100",
+                "again",
+                "ok",
+                300,
+                260,
+            ),
+        )
+        connection.commit()
+    finally:
+        connection.close()
+
+    plugin = ObservePlugin()
+    plugin.context = SimpleNamespace(workspace=workspace)
+    state = SimpleNamespace(
+        session_key="telegram:100",
+        msg=SimpleNamespace(
+            content="/kvcache",
+            channel="telegram",
+            chat_id="100",
+            timestamp=datetime.now(timezone.utc),
+        ),
+    )
+    frame = SimpleNamespace(input=state, slots={"session:session": object()})
+
+    await plugin.before_turn_modules()[0].run(frame)
+
+    reply = frame.slots["session:ctx"].abort_reply
+    assert plugin.telegram_bot_commands() == [("kvcache", "查看 KVCache 状态")]
+    assert frame.slots["session:ctx"].abort is True
+    assert "KVCache" in reply
+    assert "260 / 300" in reply
+
+
+@pytest.mark.asyncio
+async def test_kvcache_command_preserves_alias_and_missing_data_result() -> None:
+    state = SimpleNamespace(
+        session_key="telegram:100",
+        msg=SimpleNamespace(
+            content="/cache_status",
+            channel="telegram",
+            chat_id="100",
+            timestamp=datetime.now(timezone.utc),
+        ),
+    )
+    frame = SimpleNamespace(input=state, slots={"session:session": object()})
+
+    await KVCacheCommandModule(None).run(frame)
+
+    assert frame.slots["session:ctx"].abort_reply == (
+        "暂无 KVCache 数据（observe 数据库不存在）。"
+    )
+
+
 async def _run_mobile_turn_observe_seam(
     root: Path,
     *,
@@ -87,6 +172,7 @@ async def _run_mobile_turn_observe_seam(
         kv_store=PluginKVStore(workspace / "plugin-data/observe-builtin/.kv.json"),
         workspace=workspace,
         scope=scope,
+        _can_start_tasks=lambda: True,
     )
     manager = SessionManager(workspace)
     committed: list[TurnCommitted] = []
@@ -192,6 +278,7 @@ async def test_observe_plugin_activate_and_terminate(tmp_path: Path) -> None:
         kv_store=PluginKVStore(tmp_path / ".kv.json"),
         workspace=tmp_path,
         scope=scope,
+        _can_start_tasks=lambda: True,
     )
     plugin.activate()
     await asyncio.sleep(0.05)