diff --git a/.github/workflows/plugin-api-v2.yml b/.github/workflows/plugin-api-v2.yml deleted file mode 100644 index 7c1876b..0000000 --- a/.github/workflows/plugin-api-v2.yml +++ /dev/null @@ -1,28 +0,0 @@ -name: plugin-api-v2 - -on: - pull_request: - push: - branches: - - main - -permissions: - contents: read - -jobs: - contract: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/checkout@v4 - with: - repository: akashic-plugins/plugin-contracts - ref: 24543445c7b99ca63fcd90b5828f754a148b184c - path: .plugin-contracts - - uses: actions/setup-python@v5 - with: - python-version: "3.13" - - name: Check Plugin API v2 - env: - PYTHONPATH: .plugin-contracts - run: python -m akashic_plugin_contracts check plugin.py diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml new file mode 100644 index 0000000..9784e39 --- /dev/null +++ b/.github/workflows/plugin-api-v3.yml @@ -0,0 +1,67 @@ +name: plugin-api-v3 + +on: + pull_request: + push: + branches: + - main + workflow_dispatch: + +permissions: + contents: read + +jobs: + contract: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: akashic-plugins/plugin-contracts + ref: 4dd69dd621e029e51e99aa428443fa3a4ec1f6cf + path: .plugin-contracts + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Check Plugin API v3 + env: + PYTHONPATH: .plugin-contracts + run: python -m akashic_plugin_contracts check plugin.py + + composition-parity: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: kachofugetsu09/akashic-agent + ref: 3005f838bcd96e2cbc58616aede46e4f39df4523 + path: .akashic-core + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: .akashic-core/requirements.txt + - name: Install exact Core runtime + run: | + python -m venv .venv + .venv/bin/python -m pip install \ + -r .akashic-core/requirements.txt \ + -r .akashic-core/requirements-dev.txt \ + pytest pytest-asyncio + - name: Verify Observe v3 composition + env: + AKASHIC_AGENT_ROOT: ${{ github.workspace }}/.akashic-core + PYTHONPATH: ${{ github.workspace }}/.akashic-core + run: cd tests && ../.venv/bin/python -m pytest -q . + - name: Verify Mobile panel behavior + run: node --test tests/test_mobile_panel.mjs + - name: Check v3 source types + env: + PYTHONPATH: .akashic-core + run: .venv/bin/pyright --pythonpath .venv/bin/python --level error plugin.py collector.py dashboard.py + - name: Compile Python sources + run: python -m compileall -q . + - name: Check diff formatting + run: git diff --check diff --git a/README.md b/README.md index e629ab8..29a4620 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,25 @@ # observe -Akashic 可观测性插件，负责采集 Turn、检索、记忆写入和全局错误遥测。 +Akashic 可观测性插件（Plugin API v3），负责采集已提交 Turn 和全局错误遥测， +并提供 Dashboard 与移动端只读投影。 + +插件通过模块级 `api_version = 3` / `apply(ctx, config)` 接入 Core： + +- `turn.after_turn.committed`：在 Core 提交 `TurnCommitted` 后写入 Observe 数据库； +- `core.ui_slots`：发布移动端静态资源和只读 query； +- `workspace_roots = ("observe",)`：数据库、retention marker 和候选副本都由 Core 分配。 + +Observe 不再导出 v2 `Plugin` class、`activate()`、`terminate()`、`mobile_ui()` 或 +`mobile_ui_query()` ABI。Dashboard 使用 `register(app, DashboardContext)`，只读取当前 +generation 的声明式 workspace root。 + +历史 `rag_queries` 与 `memory_writes` 表保留供兼容读取；Core +`5f2c8fb5c64496897475bd3226812b2e17fcf37e` 提供的 +`proactive.finished`、`memory.retrieval.completed` 和 `memory.written` typed +Observe seam 都直接转换为既有 Observe 表结构，不复制或伪造领域 DTO。 + +插件测试与 CI 固定使用上述 Core commit；candidate 验证只写 Core 分配的临时 +workspace，formal publish 后才继续写正式 `observe` workspace。 ## 移动端 diff --git a/akashic.plugin.toml b/akashic.plugin.toml new file mode 100644 index 0000000..82badc4 --- /dev/null +++ b/akashic.plugin.toml @@ -0,0 +1,5 @@ +schema_version = 1 +name = "observe" +version = "1.3.0" +api_version = 3 +entrypoint = "plugin.py" diff --git a/collector.py b/collector.py index a56ae7c..e17db1c 100644 --- a/collector.py +++ b/collector.py @@ -15,7 +15,7 @@ import threading import traceback import types -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path @@ -84,11 +84,8 @@ class GlobalErrorCollector: def __init__( self, writer: _Emitter, - *, - create_task: Callable[..., asyncio.Task[Any]] = asyncio.create_task, ) -> None: self._writer = writer - self._create_task = create_task self._lock = threading.Lock() self._buckets: dict[tuple[str, str], _BucketAgg] = {} self._flush_task: asyncio.Task[None] | None = None @@ -102,34 +99,50 @@ def __init__( # ── 生命周期 ───────────────────────────────── - def install(self) -> None: + async def install( + self, + *, + spawn_task: Callable[..., Awaitable[asyncio.Task[Any]]] | None = None, + ) -> None: if self._installed: return self._installed = True - # 1. root logging handler（level >= ERROR） - handler = _CollectorLogHandler(self) - logging.getLogger().addHandler(handler) - self._log_handler = handler - # 2. 同步未捕获异常 - self._prev_excepthook = sys.excepthook - sys.excepthook = self._on_sys_except - # 3. 线程崩溃 - self._prev_threadhook = threading.excepthook - threading.excepthook = self._on_thread_except - # 4. asyncio 任务异常 + flush task try: - loop = asyncio.get_running_loop() - except RuntimeError: - loop = None - if loop is not None: - self._loop = loop - self._prev_loop_handler = loop.get_exception_handler() - loop.set_exception_handler(self._on_loop_except) - self._flush_task = self._create_task( - self._flush_loop(), - name="observe_error_flush", - ) - logger.info("global error collector installed") + # 1. root logging handler（level >= ERROR） + handler = _CollectorLogHandler(self) + logging.getLogger().addHandler(handler) + self._log_handler = handler + # 2. 同步未捕获异常 + self._prev_excepthook = sys.excepthook + sys.excepthook = self._on_sys_except + # 3. 线程崩溃 + self._prev_threadhook = threading.excepthook + threading.excepthook = self._on_thread_except + # 4. asyncio 任务异常 + flush task + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + if loop is not None: + self._loop = loop + self._prev_loop_handler = loop.get_exception_handler() + loop.set_exception_handler(self._on_loop_except) + flush = self._flush_loop() + if spawn_task is None: + self._flush_task = asyncio.create_task( + flush, + name="observe_error_flush", + ) + else: + self._flush_task = await spawn_task( + flush, + name="observe_error_flush", + ) + logger.info("global error collector installed") + except BaseException: + # 安装是事务性的：任务准入失败时不得遗留进程级 hook。 + await self.uninstall() + raise async def uninstall(self) -> None: if not self._installed: diff --git a/dashboard.py b/dashboard.py index 35dddbb..e65935c 100644 --- a/dashboard.py +++ b/dashboard.py @@ -10,6 +10,8 @@ from fastapi import FastAPI +from agent.plugin_composition import DashboardContext + from .db import open_db # Observe monitoring dashboard: aggregates the agent-loop telemetry written to @@ -38,8 +40,8 @@ def _resolve_range(range_token: str) -> tuple[str | None, int]: class ObserveDashboardReader: - def __init__(self, workspace: Path) -> None: - self.db_path = workspace / "observe" / "observe.db" + def __init__(self, observe_root: Path) -> None: + self.db_path = observe_root / "observe.db" self._lock = threading.RLock() # Aggregate the metric-card figures over the selected window. @@ -356,8 +358,12 @@ def _global_occurrences( return out -def register(app: FastAPI, plugin_dir: Path, workspace: Path) -> None: - reader = ObserveDashboardReader(workspace) +def register(app: FastAPI, context: DashboardContext) -> None: + """Register read-only Observe routes against the generation-owned workspace.""" + + # 1. Dashboard only receives the declared Observe root, never the full workspace. + observe_root = context.workspace_root("observe") + reader = ObserveDashboardReader(observe_root) @app.get("/api/dashboard/observe/overview") def observe_overview(range: str = "24h") -> dict[str, Any]: diff --git a/mobile_kvcache.py b/mobile_kvcache.py index ae0dfa9..d9e3b5d 100644 --- a/mobile_kvcache.py +++ b/mobile_kvcache.py @@ -10,8 +10,8 @@ class KVCacheDashboardReader: """从 observe 数据库读取桌面与移动端共用的 KV Cache 投影。""" - def __init__(self, workspace: Path) -> None: - self.db_path = workspace / "observe" / "observe.db" + def __init__(self, observe_root: Path) -> None: + self.db_path = observe_root / "observe.db" self._lock = threading.RLock() def get_summary(self) -> dict[str, Any]: diff --git a/plugin.py b/plugin.py index 5dc655f..e7da2fb 100644 --- a/plugin.py +++ b/plugin.py @@ -1,15 +1,24 @@ from __future__ import annotations -import asyncio import json import logging from collections.abc import Mapping -from contextlib import suppress from pathlib import Path -from typing import Literal, Protocol, cast, runtime_checkable - -from agent.plugins import MobileUiContribution, MobileUiNavigation, Plugin -from agent.plugins.mobile_ui import MobileUiRpcInvalidRequest +from typing import Literal, Protocol, cast + +from agent.plugin_composition import ( + Context, + MobileUiDefinition, + MobileUiNavigation, + MobileUiRpcInvalidRequest, + UI_SLOTS, +) +from agent.turn_events.after_turn import AFTER_TURN_COMMITTED +from agent.turn_events.observe import ( + MEMORY_WRITTEN, + PROACTIVE_FINISHED, + RETRIEVAL_COMPLETED, +) from bus.events_lifecycle import ProactiveFinished, TurnCommitted from core.memory.events import MemoryWritten, RetrievalCompleted @@ -17,174 +26,170 @@ from .dashboard import ObserveDashboardReader from .mobile_kvcache import KVCacheDashboardReader from .retention import run_retention_if_needed +from .events import TurnTrace as TurnTraceEvent from .writer import TraceWriter logger = logging.getLogger("plugin.observe") +api_version = 3 +name = "observe" +version = "1.3.0" +inject = (UI_SLOTS,) +workspace_roots = ("observe",) +dashboard_module = "dashboard.py" + -@runtime_checkable class _ObserveWriter(Protocol): - def emit(self, event: object) -> None: ... + def emit(self, event: TurnTraceEvent) -> None: ... -class ObservePlugin(Plugin): - api_version = 2 - @classmethod - def dashboard_module(cls) -> str | None: - return "dashboard.py" +async def apply(ctx: Context, config: object) -> None: + """登记 Observe 的 committed 事件、写入任务和只读投影。""" - @classmethod - def mobile_ui(cls) -> MobileUiContribution: - return MobileUiContribution( - module="mobile_panel.js", - stylesheet="mobile_panel.css", - navigation=MobileUiNavigation( - label="Observe", - description="缓存效率与运行健康", - ), - slots=("turn.after_answer",), - ) - - name = "observe" - version = "1.2.0" + # 1. 所有运行时文件都落在 Core 分配的声明式 Observe workspace root。 + del config + observe_root = ctx.workspace_root("observe") + db_path = observe_root / "observe.db" + writer = TraceWriter(db_path) + _ = await ctx.spawn(writer.run(), name="observe_writer") + _ = await ctx.spawn( + run_retention_if_needed(db_path), + name="observe_retention", + ) - def activate(self) -> None: - workspace = self.context.workspace - if workspace is None: - logger.warning("observe 插件缺少 workspace，跳过加载") - return + # 2. 全局错误出口由一个可逆 Effect 持有，flush task 也归当前 Fiber。 + collector = GlobalErrorCollector(writer) - self._writer = TraceWriter(workspace / "observe" / "observe.db") - self._writer_task = self.context.create_task( - self._writer.run(), - name="observe_writer", - ) - self._retention_task = self.context.create_task( - run_retention_if_needed(workspace / "observe" / "observe.db"), - name="observe_retention", - ) - self._collector = GlobalErrorCollector( - self._writer, - create_task=self.context.create_task, - ) - self._collector.install() - self.context.event_bus.on(TurnCommitted, self._observe_turn_committed) - self.context.event_bus.on(ProactiveFinished, self._observe_proactive_finished) - self.context.event_bus.on(RetrievalCompleted, self._observe_retrieval) - self.context.event_bus.on(MemoryWritten, self._observe_memory_written) - - async def terminate(self) -> None: - collector = getattr(self, "_collector", None) - if collector is not None: + async def setup_collector() -> object: + try: + await collector.install(spawn_task=ctx.spawn) + except BaseException: await collector.uninstall() - for task in ( - getattr(self, "_retention_task", None), - getattr(self, "_writer_task", None), - ): - if task is None: - continue - _ = task.cancel() - with suppress(asyncio.CancelledError): - await task - - def mobile_ui_query( - self, + raise + return collector.uninstall + + _ = await ctx.effect(setup_collector, label="observe_global_errors") + + # 3. TurnCommitted 使用 request-bound composition Root 的 typed emit。 + def observe_turn_committed(event: TurnCommitted) -> None: + _emit_turn_trace(writer, event) + + _ = await ctx.on(AFTER_TURN_COMMITTED, observe_turn_committed) + + def observe_proactive_finished(event: ProactiveFinished) -> None: + writer.emit(_to_proactive_turn_trace(event)) + + def observe_retrieval(event: RetrievalCompleted) -> None: + writer.emit(_to_rag_query_log(event)) + + def observe_memory_written(event: MemoryWritten) -> None: + writer.emit(_to_memory_write_trace(event)) + + _ = await ctx.on(PROACTIVE_FINISHED, observe_proactive_finished) + _ = await ctx.on(RETRIEVAL_COMPLETED, observe_retrieval) + _ = await ctx.on(MEMORY_WRITTEN, observe_memory_written) + + # 4. Mobile 只读查询和静态资源绑定同一 generation Effect。 + def mobile_query( method: str, payload: dict[str, object], *, session_id: str | None, turn_id: str | None, ) -> dict[str, object]: - """返回 Observe 自有的移动端只读投影。""" - - # 1. 在插件 RPC 边界校验方法与查询参数 - if method not in { - "kvcache.bootstrap", - "kvcache.message_usage", - "health.snapshot", - "health.error_detail", - }: - raise MobileUiRpcInvalidRequest(f"未知 observe 移动方法: {method}") - workspace = self.context.workspace - if workspace is None: - raise RuntimeError("observe 移动看板缺少 workspace") - if method.startswith("health."): - return self._mobile_health_query(method, payload, workspace) - reader = KVCacheDashboardReader(workspace) - if method == "kvcache.bootstrap": - return cast("dict[str, object]", reader.get_bootstrap()) - if method == "kvcache.message_usage": - message_id = _required_mobile_string(payload, "message_id") - if session_id is None: - raise MobileUiRpcInvalidRequest("kvcache.message_usage 缺少 session_id") - usage = reader.get_message_usage( - message_id=message_id, - session_key=session_id, - ) - return {"usage": usage} - raise AssertionError(f"未处理的 observe 移动方法: {method}") + return _mobile_ui_query( + observe_root, + method, + payload, + session_id=session_id, + turn_id=turn_id, + ) - def _mobile_health_query( - self, - method: str, - payload: dict[str, object], - workspace: Path, - ) -> dict[str, object]: - """把 Observe 错误聚合裁成手机排障所需的只读投影。""" - - # 1. 复用桌面聚合 owner，只在 RPC 边界限制时间范围和载荷体积 - range_token = _mobile_range_value(payload) - reader = ObserveDashboardReader(workspace) - if method == "health.snapshot": - result = reader.get_mobile_global_health(range_token, limit=50) - raw_items = cast("list[dict[str, object]]", result["items"]) - return { - "range": range_token, - "items": cast( - "list[object]", - [_mobile_error_summary(item) for item in raw_items], - ), - "types": int(result["types"]), - "total": int(result["total"]), - "new_types": int(result["new_types"]), - "spiking_types": int(result["spiking_types"]), - } - - # 2. 详情按用户展开时再读取，列表不搬运 traceback 和 occurrence - fingerprint = _required_mobile_string(payload, "fingerprint") - detail = reader.get_mobile_global_detail(fingerprint, range_token) - if not detail: - return {"error": None} - return {"error": _mobile_error_detail(detail)} - - def _observe_turn_committed(self, event: TurnCommitted) -> None: - writer = getattr(self, "_writer", None) - if not isinstance(writer, _ObserveWriter): - return - _emit_turn_trace(writer, event) + await ctx.require(UI_SLOTS).register_mobile( + ctx, + MobileUiDefinition( + module="mobile_panel.js", + stylesheet="mobile_panel.css", + navigation=MobileUiNavigation( + label="Observe", + description="缓存效率与运行健康", + ), + slots=("turn.after_answer",), + ), + query=mobile_query, + ) - def _observe_retrieval(self, event: RetrievalCompleted) -> None: - writer = getattr(self, "_writer", None) - if not isinstance(writer, _ObserveWriter): - return - writer.emit(_to_rag_query_log(event)) - def _observe_proactive_finished(self, event: ProactiveFinished) -> None: - writer = getattr(self, "_writer", None) - if not isinstance(writer, _ObserveWriter): - return - writer.emit(_to_proactive_turn_trace(event)) +def _mobile_ui_query( + observe_root: Path, + method: str, + payload: dict[str, object], + *, + session_id: str | None, + turn_id: str | None, +) -> dict[str, object]: + """返回 Observe 自有的移动端只读投影。""" + + # 1. 在插件 RPC 边界校验方法与查询参数。 + _ = turn_id + if method not in { + "kvcache.bootstrap", + "kvcache.message_usage", + "health.snapshot", + "health.error_detail", + }: + raise MobileUiRpcInvalidRequest(f"未知 observe 移动方法: {method}") + if method.startswith("health."): + return _mobile_health_query(method, payload, observe_root) + reader = KVCacheDashboardReader(observe_root) + if method == "kvcache.bootstrap": + return cast("dict[str, object]", reader.get_bootstrap()) + if method == "kvcache.message_usage": + message_id = _required_mobile_string(payload, "message_id") + if session_id is None: + raise MobileUiRpcInvalidRequest("kvcache.message_usage 缺少 session_id") + usage = reader.get_message_usage( + message_id=message_id, + session_key=session_id, + ) + return {"usage": usage} + raise AssertionError(f"未处理的 observe 移动方法: {method}") + + +def _mobile_health_query( + method: str, + payload: dict[str, object], + observe_root: Path, +) -> dict[str, object]: + """把 Observe 错误聚合裁成手机排障所需的只读投影。""" + + # 1. 复用桌面聚合 owner，只在 RPC 边界限制时间范围和载荷体积。 + range_token = _mobile_range_value(payload) + reader = ObserveDashboardReader(observe_root) + if method == "health.snapshot": + result = reader.get_mobile_global_health(range_token, limit=50) + raw_items = cast("list[dict[str, object]]", result["items"]) + return { + "range": range_token, + "items": cast( + "list[object]", + [_mobile_error_summary(item) for item in raw_items], + ), + "types": int(result["types"]), + "total": int(result["total"]), + "new_types": int(result["new_types"]), + "spiking_types": int(result["spiking_types"]), + } - def _observe_memory_written(self, event: MemoryWritten) -> None: - writer = getattr(self, "_writer", None) - if not isinstance(writer, _ObserveWriter): - return - writer.emit(_to_memory_write_trace(event)) + # 2. 详情按用户展开时再读取，列表不搬运 traceback 和 occurrence。 + fingerprint = _required_mobile_string(payload, "fingerprint") + detail = reader.get_mobile_global_detail(fingerprint, range_token) + if not detail: + return {"error": None} + return {"error": _mobile_error_detail(detail)} def _emit_turn_trace(writer: _ObserveWriter, event: TurnCommitted) -> None: - from .events import TurnTrace as TurnTraceEvent - post_reply_budget = event.post_reply_budget react_stats = event.react_stats tool_chain = event.tool_chain_raw @@ -280,9 +285,7 @@ def _model_usage_int(model_usage: Mapping[str, object], name: str) -> int | None return value if isinstance(value, int) and not isinstance(value, bool) else None -def _to_proactive_turn_trace(event: ProactiveFinished): - from .events import TurnTrace as TurnTraceEvent - +def _to_proactive_turn_trace(event: ProactiveFinished) -> TurnTraceEvent: summary = event.final_message or event.skip_reason or event.gate_exit or "" return TurnTraceEvent( source=event.mode, diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 56d0aee..57969c0 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,37 +1,45 @@ from __future__ import annotations import asyncio +import hashlib import importlib.util +import json +import shutil import sqlite3 import sys import threading -from dataclasses import replace from datetime import datetime, timedelta, timezone from pathlib import Path -from types import SimpleNamespace -from typing import Any -from unittest.mock import Mock +from typing import Any, cast import pytest - -from agent.core.runtime_support import TurnRunResult -from agent.lifecycle.phase import Phase -from agent.lifecycle.phases.after_reasoning import ( - AfterReasoningFrame, - default_after_reasoning_modules, +from fastapi import FastAPI + +from agent.plugin_composition import ( + CompositionRoot, + DashboardContext, + PluginRuntime, + PluginUiSlots, + UI_SLOTS, ) -from agent.lifecycle.phases.after_turn import ( - AfterTurnFrame, - default_after_turn_modules, +from agent.plugins.composable import ComposablePlugin +from agent.plugins.dashboard_host import DashboardBinding, PluginDashboardHost +from agent.plugins.manager import PluginManager +from agent.plugins.mobile_ui import PluginMobileUiProvider +from agent.plugins.static_manifest import load_static_plugin_manifest +from agent.turn_events.after_turn import AFTER_TURN_COMMITTED +from agent.turn_events.observe import ( + MEMORY_WRITTEN, + PROACTIVE_FINISHED, + RETRIEVAL_COMPLETED, ) -from agent.lifecycle.types import AfterReasoningInput, TurnSnapshot, TurnState -from agent.looping.ports import SessionServices -from agent.plugins.context import PluginContext, PluginKVStore -from agent.plugins.scope import PluginScope, ScopedEventBus from bus.event_bus import EventBus -from bus.events import InboundMessage -from bus.events_lifecycle import TurnCommitted -from session.manager import SessionManager +from bus.events_lifecycle import ProactiveFinished, TurnCommitted +from core.memory.events import ( + MemoryWritten, + RetrievalCompleted, + RetrievalHitSummary, +) def _load_plugin_module(): @@ -50,7 +58,6 @@ def _load_plugin_module(): module = _load_plugin_module() -ObservePlugin = module.ObservePlugin GlobalErrorCollector = module.GlobalErrorCollector @@ -62,149 +69,278 @@ def emit(self, event: object) -> None: self.events.append(event) -class _DiscardOutbound: - async def dispatch(self, outbound: object) -> bool: - return True +def _turn_event( + *, + assistant_message_id: str | None = "mobile:demo:2", + turn_id: str = "turn-1", + model_usage: dict[str, object] | None = None, +) -> TurnCommitted: + return TurnCommitted( + session_key="mobile:demo", + channel="mobile", + chat_id="demo", + input_message="hi", + persisted_user_message="hi", + assistant_response="hello", + tools_used=[], + turn_id=turn_id, + assistant_message_id=assistant_message_id, + model_usage=model_usage + if model_usage is not None + else {"coverage": "exact", "output_tokens": 321}, + react_stats={ + "cache_prompt_tokens": 100, + "cache_hit_tokens": 80, + }, + ) -async def _run_mobile_turn_observe_seam( - root: Path, - *, - clear_assistant_message_id: bool, -) -> dict[str, object]: - """运行真实 mobile 持久化与 Observe 查询缝隙。""" - - workspace = root / "workspace" - event_bus = EventBus() - scope = PluginScope("observe") - plugin = ObservePlugin() - plugin.context = PluginContext( - event_bus=ScopedEventBus(event_bus, scope), +async def _mount_observe(tmp_path: Path) -> tuple[CompositionRoot, Path]: + workspace = tmp_path / "workspace" + workspace.mkdir() + root = CompositionRoot("observe-test") + ui_slots = PluginUiSlots() + _ = await root.context.provide(UI_SLOTS, ui_slots) + plugin_dir = Path(module.__file__ or "").resolve().parent + composable = ComposablePlugin.from_module(module) + await root.mount( + composable.apply, + name="observe", + inject=module.inject, + runtime=PluginRuntime( + plugin_id="observe", + plugin_dir=plugin_dir, + data_dir=tmp_path / "plugin-data", + workspace=workspace, + config={}, + workspace_roots=module.workspace_roots, + ), + ) + return root, workspace + + +def _manager_for_observe(tmp_path: Path) -> tuple[PluginManager, Path]: + workspace = tmp_path / "workspace" + plugin_root = tmp_path / "plugins" / "observe" + shutil.copytree( + Path(module.__file__ or "").resolve().parent, + plugin_root, + ignore=shutil.ignore_patterns( + ".git", ".akashic-core", ".pytest_cache", "__pycache__" + ), + ) + manager = PluginManager( + plugin_dirs=[plugin_root.parent], + event_bus=EventBus(), tool_registry=None, - plugin_id="observe", - plugin_dir=Path(__file__).parents[1], - data_dir=workspace / "plugin-data/observe-builtin", - kv_store=PluginKVStore(workspace / "plugin-data/observe-builtin/.kv.json"), workspace=workspace, - scope=scope, + installed_cache_root=tmp_path / "cache", + ) + return manager, workspace + + +async def _wait_for_turn_row(db_path: Path) -> None: + for _ in range(50): + if db_path.exists(): + conn = sqlite3.connect(db_path) + try: + if conn.execute("SELECT COUNT(*) FROM turns").fetchone()[0] > 0: + return + finally: + conn.close() + await asyncio.sleep(0.01) + raise AssertionError(f"Observe 没有写入 Turn row: {db_path}") + + +async def _wait_for_table_rows( + db_path: Path, + table: str, + expected: int, +) -> None: + for _ in range(50): + if db_path.exists(): + conn = sqlite3.connect(db_path) + try: + count = conn.execute( + f"SELECT COUNT(*) FROM {table}" + ).fetchone()[0] + finally: + conn.close() + if count >= expected: + return + await asyncio.sleep(0.01) + raise AssertionError( + f"Observe 没有写入 {table} rows={expected}: {db_path}" ) - manager = SessionManager(workspace) - committed: list[TurnCommitted] = [] - event_bus.on(TurnCommitted, committed.append) - try: - # 1. 用真实 after-reasoning 持久化生成 assistant message ID - plugin.activate() - session = manager.get_or_create("mobile:observe-seam") - message = InboundMessage( - channel="mobile", - sender="device:observe-seam", - chat_id="observe-seam", - content="hi", - metadata={"client_message_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV"}, - ) - state = TurnState( - msg=message, - session_key=session.key, - dispatch_outbound=False, - session=session, - ) - after_reasoning = Phase( - default_after_reasoning_modules( - event_bus, - SessionServices(session_manager=manager), - ), - frame_factory=AfterReasoningFrame, - ) - reasoning_result = await after_reasoning.run( - AfterReasoningInput( - state=state, - turn_result=TurnRunResult( - reply="hello", - context_retry={ - "react_stats": { - "model_usage": { - "coverage": "exact", - "output_tokens": 321, - } - } - }, - ), - ) - ) - message_id = reasoning_result.outbound.session_message_id - assert message_id is not None - outbound = ( - replace(reasoning_result.outbound, session_message_id=None) - if clear_assistant_message_id - else reasoning_result.outbound - ) - # 2. 用真实 after-turn 发布 TurnCommitted，等待 Observe 落库 - context = Mock() - context.render = Mock(return_value=SimpleNamespace(messages=[])) - context.last_debug_breakdown = [] - after_turn = Phase( - default_after_turn_modules( - event_bus, - _DiscardOutbound(), - context, - ), - frame_factory=AfterTurnFrame, - ) - await after_turn.run( - TurnSnapshot( - state=state, - outbound=outbound, - ctx=reasoning_result.ctx, + +def _proactive_finished_event() -> ProactiveFinished: + return ProactiveFinished( + session_key="proactive:demo", + tick_id="tick-1", + mode="proactive", + terminal_action="deliver", + gate_exit=None, + skip_reason="", + steps_taken=2, + alert_count=1, + content_count=1, + context_count=1, + final_message="proactive result", + llm_call_count=3, + cache_prompt_tokens=120, + cache_hit_tokens=90, + ) + + +def _retrieval_completed_event() -> RetrievalCompleted: + return RetrievalCompleted( + session_key="mobile:demo", + channel="mobile", + chat_id="demo", + query="rewritten query", + orig_query="original query", + hits=[ + RetrievalHitSummary( + item_id="memory-1", + memory_type="event", + score=0.91, + summary="retrieved summary", + injected=True, + confidence_label="certain", + forced=True, + metadata={"forced": True}, ) - ) - await plugin._writer.drain() + ], + injected_count=1, + route_decision="RETRIEVE", + aux_queries=["hypothesis"], + ) + + +def _memory_written_event() -> MemoryWritten: + return MemoryWritten( + session_key="mobile:demo", + channel="mobile", + chat_id="demo", + action="supersede", + source_ref="mobile:demo@post_response", + superseded_ids=["memory-old"], + ) - # 3. 从移动端公开查询读取同一条 assistant 消息 - result = plugin.mobile_ui_query( + +@pytest.mark.asyncio +async def test_v3_apply_emits_turn_trace_and_disposes_all_effects(tmp_path: Path) -> None: + root, workspace = await _mount_observe(tmp_path) + try: + root.context.emit(AFTER_TURN_COMMITTED, _turn_event()) + db_path = workspace / "observe" / "observe.db" + await _wait_for_turn_row(db_path) + result = module._mobile_ui_query( + workspace / "observe", "kvcache.message_usage", - {"message_id": message_id}, - session_id=session.key, + {"message_id": "mobile:demo:2"}, + session_id="mobile:demo", turn_id=None, ) - assert len(committed) == 1 - return { - "persisted_message_id": message_id, - "committed_message_id": committed[0].assistant_message_id, - "query": result, - } + assert result == {"usage": {"output_tokens": 321}} + assert root.receipt().ready + assert root.topology_view().listeners finally: - await plugin.terminate() - _ = await scope.aclose() - manager.close() + await root.dispose() + assert root.receipt().effects == () + assert root.topology_view().listeners == () @pytest.mark.asyncio -async def test_observe_plugin_activate_and_terminate(tmp_path: Path) -> None: - plugin = ObservePlugin() - scope = PluginScope("observe") - plugin.context = PluginContext( - event_bus=ScopedEventBus(EventBus(), scope), - tool_registry=None, - plugin_id="observe", - plugin_dir=tmp_path, - data_dir=tmp_path, - kv_store=PluginKVStore(tmp_path / ".kv.json"), - workspace=tmp_path, - scope=scope, - ) - plugin.activate() - await asyncio.sleep(0.05) - await plugin.terminate() - assert await scope.aclose() == [] - db_path = tmp_path / "observe" / "observe.db" - assert db_path.exists() - conn = sqlite3.connect(db_path) +async def test_v3_observes_core_domain_events_without_duplicate_rows( + tmp_path: Path, +) -> None: + root, workspace = await _mount_observe(tmp_path) + proactive = _proactive_finished_event() + retrieval = _retrieval_completed_event() + memory = _memory_written_event() + db_path = workspace / "observe" / "observe.db" try: - tables = {row[0] for row in conn.execute("select name from sqlite_master where type='table'")} + rag_projection = module._to_rag_query_log(retrieval) + assert rag_projection.caller == "passive" + assert rag_projection.hits[0].confidence_label == "certain" + assert rag_projection.hits[0].forced is True + assert module._to_memory_write_trace(memory).superseded_ids == [ + "memory-old" + ] + await root.context.observe(PROACTIVE_FINISHED, proactive) + await root.context.observe(RETRIEVAL_COMPLETED, retrieval) + await root.context.observe(MEMORY_WRITTEN, memory) + await _wait_for_table_rows(db_path, "turns", 1) + await _wait_for_table_rows(db_path, "rag_queries", 1) + await _wait_for_table_rows(db_path, "memory_writes", 1) + + conn = sqlite3.connect(db_path) + try: + proactive_row = conn.execute( + """ + SELECT source, session_key, llm_output, react_iteration_count, + react_cache_prompt_tokens, react_cache_hit_tokens + FROM turns + """ + ).fetchone() + retrieval_row = conn.execute( + """ + SELECT caller, session_key, query, orig_query, aux_queries, + hits_json, injected_count, route_decision, error + FROM rag_queries + """ + ).fetchone() + memory_row = conn.execute( + """ + SELECT session_key, source_ref, action, memory_type, item_id, + summary, superseded_ids, error + FROM memory_writes + """ + ).fetchone() + assert proactive_row == ( + "proactive", + "proactive:demo", + "proactive result", + 3, + 120, + 90, + ) + assert retrieval_row[0:5] == ( + "passive", + "mobile:demo", + "rewritten query", + "original query", + '["hypothesis"]', + ) + assert json.loads(retrieval_row[5]) == [ + { + "id": "memory-1", + "type": "event", + "score": 0.91, + "summary": "retrieved summary", + "injected": 1, + } + ] + assert retrieval_row[6:] == (1, "RETRIEVE", None) + assert memory_row == ( + "mobile:demo", + "mobile:demo@post_response", + "supersede", + None, + None, + None, + '["memory-old"]', + None, + ) + assert conn.execute("SELECT COUNT(*) FROM turns").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM rag_queries").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM memory_writes").fetchone()[0] == 1 + finally: + conn.close() finally: - conn.close() - assert "turns" in tables + await root.dispose() @pytest.mark.asyncio @@ -216,14 +352,12 @@ async def test_global_hooks_survive_overlapping_generations() -> None: first = GlobalErrorCollector(_Emitter()) second = GlobalErrorCollector(_Emitter()) try: - first.install() - second.install() + await first.install() + await second.install() await first.uninstall() - assert sys.excepthook == second._on_sys_except assert threading.excepthook == second._on_thread_except assert loop.get_exception_handler() == second._on_loop_except - await second.uninstall() assert sys.excepthook == original_sys assert threading.excepthook == original_thread @@ -233,32 +367,39 @@ async def test_global_hooks_survive_overlapping_generations() -> None: await first.uninstall() +@pytest.mark.asyncio +async def test_global_hooks_roll_back_when_flush_task_cannot_spawn() -> None: + original_sys = sys.excepthook + original_thread = threading.excepthook + loop = asyncio.get_running_loop() + original_loop = loop.get_exception_handler() + collector = GlobalErrorCollector(_Emitter()) + + async def reject_spawn(coroutine: Any, *, name: str) -> asyncio.Task[Any]: + del name + coroutine.close() + raise RuntimeError("flush task rejected") + + with pytest.raises(RuntimeError, match="flush task rejected"): + await collector.install(spawn_task=reject_spawn) + assert sys.excepthook == original_sys + assert threading.excepthook == original_thread + assert loop.get_exception_handler() == original_loop + await collector.uninstall() + + def test_turn_trace_keeps_message_identity_and_output_tokens(tmp_path: Path) -> None: emitter = _Emitter() - module._emit_turn_trace( - emitter, - TurnCommitted( - session_key="mobile:demo", - channel="mobile", - chat_id="demo", - input_message="hi", - persisted_user_message="hi", - assistant_response="hello", - tools_used=[], - turn_id="turn-1", - assistant_message_id="mobile:demo:2", - model_usage={"output_tokens": 321, "coverage": "exact"}, - ), - ) - + module._emit_turn_trace(emitter, _turn_event()) trace = emitter.events[0] assert trace.turn_id == "turn-1" assert trace.assistant_message_id == "mobile:demo:2" assert trace.model_output_tokens == 321 db_module = sys.modules[f"{module.__name__}.db"] - conn = db_module.open_db(tmp_path / "observe.db") + db_path = tmp_path / "observe.db" + conn = db_module.open_db(db_path) try: - module.TraceWriter(tmp_path / "observe.db")._write_one(conn, trace) + module.TraceWriter(db_path)._write_one(conn, trace) row = conn.execute( "SELECT turn_id, assistant_message_id, model_output_tokens FROM turns" ).fetchone() @@ -267,76 +408,23 @@ def test_turn_trace_keeps_message_identity_and_output_tokens(tmp_path: Path) -> assert row == ("turn-1", "mobile:demo:2", 321) -@pytest.mark.asyncio -async def test_mobile_turn_identity_reaches_observe_usage_query(tmp_path: Path) -> None: - result = await _run_mobile_turn_observe_seam( - tmp_path / "healthy", - clear_assistant_message_id=False, - ) - - assert result == { - "persisted_message_id": "mobile:observe-seam:1", - "committed_message_id": "mobile:observe-seam:1", - "query": {"usage": {"output_tokens": 321}}, - } - - -@pytest.mark.asyncio -async def test_mobile_turn_identity_mutant_is_killed(tmp_path: Path) -> None: - result = await _run_mobile_turn_observe_seam( - tmp_path / "mutant", - clear_assistant_message_id=True, - ) - - with pytest.raises(AssertionError): - assert result == { - "persisted_message_id": "mobile:observe-seam:1", - "committed_message_id": "mobile:observe-seam:1", - "query": {"usage": {"output_tokens": 321}}, - } - assert result["committed_message_id"] is None - assert result["query"] == {"usage": None} - - -def test_partial_usage_and_empty_turn_ids_do_not_claim_complete_output( - tmp_path: Path, -) -> None: +def test_partial_usage_and_empty_turn_ids_do_not_claim_complete_output() -> None: emitter = _Emitter() for index in range(2): - module._emit_turn_trace( - emitter, - TurnCommitted( - session_key="mobile:demo", - channel="mobile", - chat_id="demo", - input_message="hi", - persisted_user_message="hi", - assistant_response="hello", - tools_used=[], - assistant_message_id=f"mobile:demo:{index + 2}", - model_usage={"output_tokens": 100, "coverage": "partial"}, - ), + event = _turn_event( + assistant_message_id=f"mobile:demo:{index + 2}", + turn_id="", + model_usage={"coverage": "partial", "output_tokens": 100}, ) - + module._emit_turn_trace(emitter, event) assert all(event.turn_id is None for event in emitter.events) assert all(event.model_output_tokens is None for event in emitter.events) - db_module = sys.modules[f"{module.__name__}.db"] - conn = db_module.open_db(tmp_path / "observe.db") - try: - writer = module.TraceWriter(tmp_path / "observe.db") - for event in emitter.events: - writer._write_one(conn, event) - count = conn.execute("SELECT COUNT(*) FROM turns").fetchone()[0] - finally: - conn.close() - assert count == 2 def test_mobile_message_usage_returns_true_output_tokens(tmp_path: Path) -> None: - plugin = ObservePlugin() - plugin.context = SimpleNamespace(workspace=tmp_path) db_module = sys.modules[f"{module.__name__}.db"] - conn = db_module.open_db(tmp_path / "observe" / "observe.db") + db_path = tmp_path / "observe" / "observe.db" + conn = db_module.open_db(db_path) try: conn.execute( """ @@ -359,27 +447,20 @@ def test_mobile_message_usage_returns_true_output_tokens(tmp_path: Path) -> None conn.commit() finally: conn.close() - - result = plugin.mobile_ui_query( + result = module._mobile_ui_query( + tmp_path / "observe", "kvcache.message_usage", {"message_id": "mobile:demo:2"}, session_id="mobile:demo", turn_id=None, ) - assert result == {"usage": {"output_tokens": 321}} - contribution = ObservePlugin.mobile_ui() - assert contribution.module == "mobile_panel.js" - assert contribution.stylesheet == "mobile_panel.css" - assert contribution.navigation.label == "Observe" - assert contribution.slots == ("turn.after_answer",) def test_mobile_health_reuses_global_error_projection(tmp_path: Path) -> None: - plugin = ObservePlugin() - plugin.context = SimpleNamespace(workspace=tmp_path) db_module = sys.modules[f"{module.__name__}.db"] - conn = db_module.open_db(tmp_path / "observe" / "observe.db") + db_path = tmp_path / "observe" / "observe.db" + conn = db_module.open_db(db_path) now = datetime.now(timezone.utc) traceback_text = "Traceback\n" + ("failure detail\n" * 500) try: @@ -429,7 +510,7 @@ def test_mobile_health_reuses_global_error_projection(tmp_path: Path) -> None: now.isoformat(), now.isoformat(), 20, - '[]', + "[]", "active", ), ) @@ -437,19 +518,20 @@ def test_mobile_health_reuses_global_error_projection(tmp_path: Path) -> None: finally: conn.close() - snapshot = plugin.mobile_ui_query( + snapshot = module._mobile_ui_query( + tmp_path / "observe", "health.snapshot", {"range": "24h"}, session_id=None, turn_id=None, ) - detail = plugin.mobile_ui_query( + detail = module._mobile_ui_query( + tmp_path / "observe", "health.error_detail", {"range": "24h", "fingerprint": "fp-mobile-health"}, session_id=None, turn_id=None, ) - assert snapshot["total"] == 27 assert snapshot["types"] == 2 assert snapshot["spiking_types"] == 1 @@ -461,13 +543,14 @@ def test_mobile_health_reuses_global_error_projection(tmp_path: Path) -> None: assert len(detail["error"]["traceback"]) == 4000 assert "occurrences" not in detail["error"] - conn = db_module.open_db(tmp_path / "observe" / "observe.db") + conn = db_module.open_db(db_path) try: conn.execute("UPDATE global_errors SET status = 'ignored'") conn.commit() finally: conn.close() - ignored = plugin.mobile_ui_query( + ignored = module._mobile_ui_query( + tmp_path / "observe", "health.snapshot", {"range": "24h"}, session_id=None, @@ -481,10 +564,10 @@ def test_mobile_health_reuses_global_error_projection(tmp_path: Path) -> None: "new_types": 0, "spiking_types": 0, } - for invalid_range in ("all", [], {}, True, None): with pytest.raises(ValueError, match="range 只支持"): - plugin.mobile_ui_query( + module._mobile_ui_query( + tmp_path / "observe", "health.snapshot", {"range": invalid_range}, session_id=None, @@ -492,6 +575,45 @@ def test_mobile_health_reuses_global_error_projection(tmp_path: Path) -> None: ) +def test_static_manifest_and_module_exports_match() -> None: + plugin_dir = Path(module.__file__ or "").resolve().parent + manifest = load_static_plugin_manifest(plugin_dir) + composable = ComposablePlugin.from_module(module) + assert manifest.name == composable.name == "observe" + assert manifest.version == composable.version == "1.3.0" + assert manifest.api_version == composable.api_version == 3 + assert manifest.entrypoint == "plugin.py" + assert composable.dashboard_module == "dashboard.py" + assert composable.workspace_roots == ("observe",) + + +def test_dashboard_uses_declared_generation_root(tmp_path: Path) -> None: + dashboard = importlib.util.spec_from_file_location( + f"{module.__name__}.dashboard", + Path(module.__file__ or "").resolve().parent / "dashboard.py", + ) + assert dashboard is not None and dashboard.loader is not None + dashboard_module = importlib.util.module_from_spec(dashboard) + sys.modules[dashboard.name] = dashboard_module + dashboard.loader.exec_module(dashboard_module) + app = FastAPI() + declared = tmp_path / "workspace" / "observe" + dashboard_module.register( + app, + DashboardContext( + plugin_id="observe", + plugin_dir=Path(module.__file__ or "").resolve().parent, + data_root=tmp_path / "plugin-data", + validation=True, + _workspace_roots=(("observe", declared),), + ), + ) + assert any( + getattr(route, "path", None) == "/api/dashboard/observe/overview" + for route in app.routes + ) + + def test_open_db_removes_legacy_unique_turn_id_index(tmp_path: Path) -> None: db_path = tmp_path / "observe.db" conn = sqlite3.connect(db_path) @@ -512,7 +634,6 @@ def test_open_db_removes_legacy_unique_turn_id_index(tmp_path: Path) -> None: conn.commit() finally: conn.close() - db_module = sys.modules[f"{module.__name__}.db"] migrated = db_module.open_db(db_path) try: @@ -524,57 +645,6 @@ def test_open_db_removes_legacy_unique_turn_id_index(tmp_path: Path) -> None: assert legacy_index is None -def test_kvcache_bootstrap_uses_incremental_projection_and_one_snapshot( - tmp_path: Path, -) -> None: - db_module = sys.modules[f"{module.__name__}.db"] - events_module = sys.modules[f"{module.__name__}.events"] - db_path = tmp_path / "observe" / "observe.db" - conn = db_module.open_db(db_path) - writer = module.TraceWriter(db_path) - try: - writer._write_one( - conn, - events_module.TurnTrace( - source="agent", - session_key="mobile:demo", - user_msg="passive", - llm_output="ok", - react_cache_prompt_tokens=100, - react_cache_hit_tokens=80, - ), - ) - writer._write_one( - conn, - events_module.TurnTrace( - source="proactive", - session_key="proactive:demo", - user_msg="proactive", - llm_output="ok", - react_cache_prompt_tokens=50, - react_cache_hit_tokens=20, - ), - ) - finally: - conn.close() - - plugin = ObservePlugin() - plugin.context = SimpleNamespace(workspace=tmp_path) - bootstrap = plugin.mobile_ui_query( - "kvcache.bootstrap", - {}, - session_id=None, - turn_id=None, - ) - - assert bootstrap["snapshot_turn_id"] == 2 - assert bootstrap["projection_through_turn_id"] == 2 - assert bootstrap["overview"]["tracked_turn_count"] == 2 - assert bootstrap["overview"]["hit_tokens"] == 100 - assert bootstrap["recent"]["total"] == 2 - assert bootstrap["recent_agent"]["total"] == 1 - - def test_kvcache_bootstrap_fails_loudly_on_projection_drift(tmp_path: Path) -> None: db_module = sys.modules[f"{module.__name__}.db"] db_path = tmp_path / "observe" / "observe.db" @@ -587,11 +657,9 @@ def test_kvcache_bootstrap_fails_loudly_on_projection_drift(tmp_path: Path) -> N conn.commit() finally: conn.close() - - plugin = ObservePlugin() - plugin.context = SimpleNamespace(workspace=tmp_path) with pytest.raises(RuntimeError, match="投影水位不一致"): - plugin.mobile_ui_query( + module._mobile_ui_query( + tmp_path / "observe", "kvcache.bootstrap", {}, session_id=None, @@ -599,61 +667,227 @@ def test_kvcache_bootstrap_fails_loudly_on_projection_drift(tmp_path: Path) -> N ) -def test_turn_and_projection_update_roll_back_together(tmp_path: Path) -> None: - db_module = sys.modules[f"{module.__name__}.db"] - events_module = sys.modules[f"{module.__name__}.events"] - db_path = tmp_path / "observe.db" - conn = db_module.open_db(db_path) +@pytest.mark.asyncio +async def test_real_manager_publishes_v3_observe_mobile_query_and_candidate( + tmp_path: Path, +) -> None: + workspace = tmp_path / "workspace" + plugin_root = tmp_path / "plugins" / "observe" + shutil.copytree( + Path(module.__file__ or "").resolve().parent, + plugin_root, + ignore=shutil.ignore_patterns( + ".git", ".akashic-core", ".pytest_cache", "__pycache__" + ), + ) + manager = PluginManager( + plugin_dirs=[plugin_root.parent], + event_bus=EventBus(), + tool_registry=None, + workspace=workspace, + installed_cache_root=tmp_path / "cache", + ) try: - conn.execute("DELETE FROM kv_cache_totals") - conn.commit() - with pytest.raises(RuntimeError, match="singleton"): - module.TraceWriter(db_path)._write_one( - conn, - events_module.TurnTrace( - source="agent", - session_key="mobile:demo", - user_msg="must roll back", - llm_output="ok", - react_cache_prompt_tokens=10, - react_cache_hit_tokens=5, - ), - ) - assert conn.execute("SELECT COUNT(*) FROM turns").fetchone()[0] == 0 + await manager.load_all() + snapshot = manager.current_snapshot + assert snapshot is not None + assert snapshot.mobile_ui_registry is not None + dashboard_host = PluginDashboardHost(core_routes=()) + dashboard_host.prepare_initial_snapshot(snapshot) + manager.bind_dashboard_preparer( + dashboard_host.prepare_snapshot, + validation_releaser=dashboard_host.release_validation, + ) + assert snapshot.dashboard_bindings + assert isinstance(snapshot.dashboard_bindings[0], DashboardBinding) + provider = PluginMobileUiProvider(manager) + catalog = cast(list[dict[str, object]], provider.catalog()["items"]) + item = next(value for value in catalog if value["id"] == "observe") + assert (await provider.query( + "observe", + cast(str, item["revision"]), + "health.snapshot", + {}, + session_id=None, + turn_id=None, + ))["total"] == 0 + stable_root = snapshot.composition_root + assert stable_root is not None + stable_db = workspace / "observe" / "observe.db" + before = ( + hashlib.sha256(stable_db.read_bytes()).hexdigest() + if stable_db.exists() + else None + ) + + candidate = await manager.prepare_candidate("observe") + assert candidate is not None + assert manager.current_snapshot is snapshot + candidate_snapshot = candidate.runtime_snapshot + assert candidate_snapshot is not None + candidate_root = candidate_snapshot.composition_root + assert candidate_root is not None + dashboard_host.prepare_snapshot(candidate_snapshot) + assert len(candidate_snapshot.dashboard_bindings) == 1 + candidate_dashboard = candidate_snapshot.dashboard_bindings[0] + assert isinstance(candidate_dashboard, DashboardBinding) + assert candidate_dashboard.validation is True + candidate_mobile = candidate_snapshot.mobile_ui_registry.binding( + "observe" + ) + assert candidate_mobile is not None + assert candidate_mobile.is_live() + await manager.discard_prepared("observe") + assert candidate_root.receipt().effects == () + assert candidate_root.topology_view().listeners == () + assert candidate_snapshot.dashboard_bindings == () + assert not candidate_mobile.is_live() + assert all( + not binding.validation for binding in dashboard_host._bindings.values() + ) + assert provider.catalog()["items"] + after = ( + hashlib.sha256(stable_db.read_bytes()).hexdigest() + if stable_db.exists() + else None + ) + assert after == before + await manager.terminate_all() + assert stable_root.receipt().effects == () + assert stable_root.topology_view().listeners == () + assert not dashboard_host._bindings + assert provider.catalog()["items"] == [] finally: - conn.close() + if manager.current_snapshot is not None: + await manager.terminate_all() -def test_retention_rebuilds_projection_in_the_delete_transaction(tmp_path: Path) -> None: - db_module = sys.modules[f"{module.__name__}.db"] - retention_module = sys.modules[f"{module.__name__}.retention"] - db_path = tmp_path / "observe" / "observe.db" - conn = db_module.open_db(db_path) +@pytest.mark.asyncio +async def test_manager_candidate_domain_observe_isolated_and_cleaned( + tmp_path: Path, +) -> None: + manager, workspace = _manager_for_observe(tmp_path) + original_sys = sys.excepthook + original_thread = threading.excepthook + original_loop = asyncio.get_running_loop().get_exception_handler() + candidate_root: CompositionRoot | None = None + candidate_workspace: Path | None = None try: - conn.executemany( - """ - INSERT INTO turns( - ts, source, session_key, user_msg, llm_output, - react_cache_prompt_tokens, react_cache_hit_tokens - ) VALUES (?, 'agent', 'mobile:demo', ?, 'ok', ?, ?) - """, - [ - ("2020-01-01T00:00:00+00:00", "expired", 100, 80), - (datetime.now(timezone.utc).isoformat(), "current", 50, 20), - ], + await manager.load_all() + stable_db = workspace / "observe" / "observe.db" + stable_hook = sys.excepthook + candidate = await manager.prepare_candidate("observe") + assert candidate is not None + candidate_snapshot = candidate.runtime_snapshot + assert candidate_snapshot is not None + candidate_root = candidate_snapshot.composition_root + assert candidate_root is not None + runtime = candidate_root.root_fiber.children[0].runtime + assert runtime is not None + candidate_workspace = runtime.workspace + candidate_db = candidate_workspace / "observe" / "observe.db" + + assert sys.excepthook != stable_hook + await candidate_root.context.observe( + MEMORY_WRITTEN, + _memory_written_event(), ) - conn.commit() + await _wait_for_table_rows(candidate_db, "memory_writes", 1) + if stable_db.exists(): + with sqlite3.connect(stable_db) as conn: + assert conn.execute( + "SELECT COUNT(*) FROM memory_writes" + ).fetchone()[0] == 0 + + await manager.discard_prepared("observe") + assert candidate_root.receipt().effects == () + assert candidate_root.topology_view().listeners == () + assert candidate_workspace is not None + assert not candidate_workspace.parent.exists() + assert sys.excepthook == stable_hook finally: - conn.close() - migrated = db_module.open_db(db_path) - migrated.close() + if manager.current_snapshot is not None: + await manager.terminate_all() + assert sys.excepthook == original_sys + assert threading.excepthook == original_thread + assert asyncio.get_running_loop().get_exception_handler() == original_loop - retention_module._run_cleanup(db_path) - bootstrap = sys.modules[f"{module.__name__}.mobile_kvcache"].KVCacheDashboardReader( - tmp_path - ).get_bootstrap() - assert bootstrap["overview"]["tracked_turn_count"] == 1 - assert bootstrap["overview"]["prompt_tokens"] == 50 - assert bootstrap["recent"]["items"][0]["user_preview"] == "current" - assert bootstrap["snapshot_turn_id"] == bootstrap["projection_through_turn_id"] == 2 +@pytest.mark.asyncio +async def test_manager_formal_publish_drops_candidate_rows_and_keeps_observe_live( + tmp_path: Path, +) -> None: + manager, workspace = _manager_for_observe(tmp_path) + original_sys = sys.excepthook + original_thread = threading.excepthook + original_loop = asyncio.get_running_loop().get_exception_handler() + stable_root: CompositionRoot | None = None + formal_root: CompositionRoot | None = None + candidate_workspace: Path | None = None + try: + 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 + candidate = await manager.prepare_candidate("observe") + assert candidate is not None + candidate_snapshot = candidate.runtime_snapshot + assert candidate_snapshot is not None + candidate_root = candidate_snapshot.composition_root + assert candidate_root is not None + runtime = candidate_root.root_fiber.children[0].runtime + assert runtime is not None + candidate_workspace = runtime.workspace + candidate_db = candidate_workspace / "observe" / "observe.db" + + await candidate_root.context.observe( + RETRIEVAL_COMPLETED, + _retrieval_completed_event(), + ) + await _wait_for_table_rows(candidate_db, "rag_queries", 1) + stable_db = workspace / "observe" / "observe.db" + if stable_db.exists(): + with sqlite3.connect(stable_db) as conn: + assert conn.execute( + "SELECT COUNT(*) FROM rag_queries" + ).fetchone()[0] == 0 + + publication = await manager.publish_prepared("observe") + assert publication["publication_state"] == "committed" + formal_snapshot = manager.current_snapshot + assert formal_snapshot is not None + formal_root = formal_snapshot.composition_root + assert formal_root is not None + assert formal_root is not candidate_root + assert candidate_workspace is not None + assert not candidate_workspace.parent.exists() + + formal_db = workspace / "observe" / "observe.db" + await _wait_for_table_rows(formal_db, "rag_queries", 0) + with sqlite3.connect(formal_db) as conn: + assert conn.execute( + "SELECT COUNT(*) FROM rag_queries" + ).fetchone()[0] == 0 + + await formal_root.context.observe( + MEMORY_WRITTEN, + _memory_written_event(), + ) + await _wait_for_table_rows(formal_db, "memory_writes", 1) + with sqlite3.connect(formal_db) as conn: + assert conn.execute( + "SELECT COUNT(*) FROM memory_writes" + ).fetchone()[0] == 1 + finally: + if manager.current_snapshot is not None: + await manager.terminate_all() + assert stable_root is not None + assert stable_root.receipt().effects == () + assert stable_root.topology_view().listeners == () + assert formal_root is not None + assert formal_root.receipt().effects == () + assert formal_root.topology_view().listeners == () + assert sys.excepthook == original_sys + assert threading.excepthook == original_thread + assert asyncio.get_running_loop().get_exception_handler() == original_loop