From 95f8e2ac9bcd1a9534348bfe5641651fb029001d Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 17:48:17 +0800 Subject: [PATCH 1/9] =?UTF-8?q?feat:=20=E7=94=A8=20Content=20=E7=BB=84?= =?UTF-8?q?=E5=90=88=20Fitbit=20=E4=B8=BB=E5=8A=A8=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- akashic.plugin.toml | 9 +- plugin.py | 294 ++++++-------------------- src/content_adapter.py | 269 +++++++++++++++++++++++ src/mcp_bridge.py | 332 +++-------------------------- src/mobile_reader.py | 185 ++++++++++++++++ src/sleep_context.py | 153 ++++++++++++++ tests/test_content_adapter.py | 341 ++++++++++++++++++++++++++++++ tests/test_context_contract.py | 55 ----- tests/test_manager_integration.py | 42 +--- tests/test_mcp_v3_runtime.py | 82 ++++--- tests/test_mobile_dashboard.py | 4 +- tests/test_plugin.py | 116 ++++------ tests/test_sleep_context.py | 69 ++++++ 13 files changed, 1206 insertions(+), 745 deletions(-) create mode 100644 src/content_adapter.py create mode 100644 src/mobile_reader.py create mode 100644 src/sleep_context.py create mode 100644 tests/test_content_adapter.py delete mode 100644 tests/test_context_contract.py create mode 100644 tests/test_sleep_context.py diff --git a/akashic.plugin.toml b/akashic.plugin.toml index 3887d38..0ba1801 100644 --- a/akashic.plugin.toml +++ b/akashic.plugin.toml @@ -1,6 +1,6 @@ schema_version = 1 name = "fitbit" -version = "3.0.0" +version = "3.1.0" api_version = 3 entrypoint = "plugin.py" @@ -37,10 +37,9 @@ startup_timeout_seconds = 15.0 name = "fitbit" command = ["python", "run_mcp.py"] required_tools = [ - "get_proactive_events", - "get_sleep_context", - "acknowledge_events", + "fitbit_health_snapshot", + "fitbit_sleep_report", ] -candidate_read_only_tools = ["get_proactive_events", "get_sleep_context"] +candidate_read_only_tools = ["fitbit_health_snapshot", "fitbit_sleep_report"] endpoint_env = [{env = "FITBIT_MONITOR_PORT", process = "monitor"}] candidate_env = {FITBIT_BACKEND = "recording"} diff --git a/plugin.py b/plugin.py index 3427df3..0660dd8 100644 --- a/plugin.py +++ b/plugin.py @@ -1,15 +1,17 @@ from __future__ import annotations -from collections.abc import Mapping -from pathlib import Path +from datetime import UTC, datetime, timedelta +from typing import Protocol -import requests -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field +from agent.lifecycle.composition import CONTEXT_PREPARED_EVENT from agent.plugin_composition import ( MANAGED_PROCESSES, MCP_SERVERS, - PROACTIVE_COMPONENTS, + RUNTIME_STARTED, + RUNTIME_STOPPING, + TIMERS, UI_SLOTS, Context, EndpointEnv, @@ -17,201 +19,50 @@ McpServerDefinition, MobileUiDefinition, MobileUiNavigation, - MobileUiRpcInvalidRequest, - ProactiveSourceDefinition, + ServiceKey, ) +from src.content_adapter import ( + BoundContentSource, + FitbitContentRuntime, + FitbitMonitorClient, +) +from src.mobile_reader import mobile_ui_query +from src.sleep_context import FitbitAdapterStore, SleepContextAppender -_MONITOR_URL = "http://127.0.0.1:18765" - - -class FitbitMobileDashboardReader: - """读取 monitor，并生成稳定的移动健康总览。""" - - def get_current(self) -> dict[str, object]: - """投影当前健康快照与最近睡眠节律。""" - - # 1. 在本地 HTTP 边界取得当前快照 - snapshot = self._get_json("/api/tool/fitbit_health_snapshot") - - # 2. 只投影手机快速判断需要的字段 - return { - "available": _boolean(snapshot, "available"), - "freshness": { - "last_updated": _optional_string(snapshot, "last_updated"), - "data_lag_min": _optional_number(snapshot, "data_lag_min"), - "spo2_lag_min": _optional_number(snapshot, "spo2_lag_min"), - }, - "current": { - "heart_rate": _optional_number(snapshot, "heart_rate"), - "spo2": _optional_number(snapshot, "spo2"), - "steps": _optional_number(snapshot, "steps"), - "sleep_state": _optional_string(snapshot, "sleep_state") or "unknown", - "sleep_prob": _optional_number(snapshot, "sleep_prob"), - }, - "sleep_24h": _sleep_segments(snapshot), - } - - def get_sleep_history(self) -> dict[str, object]: - """投影七天睡眠摘要与逐日记录。""" - - # 1. 只读后台轮询维护的本地投影，不触发 OAuth 或 Fitbit API - report = self._get_json("/api/mobile/sleep_projection") - if not _boolean(report, "available"): - return { - "available": False, - "reason": _optional_string(report, "reason") or "projection_not_ready", - "freshness": _mapping(report, "freshness"), - "sleep_summary": { - "days_with_data": 0, - "avg_duration_min": None, - "avg_efficiency": None, - "avg_deep_min": None, - }, - "sleep_days": [], - } - - # 2. 只投影移动端历史浏览需要的字段 - summary = _mapping(report, "summary") - days = _list_of_mappings(report, "days") - return { - "available": True, - "reason": None, - "freshness": _mapping(report, "freshness"), - "sleep_summary": { - "days_with_data": _optional_number(summary, "days_with_data"), - "avg_duration_min": _optional_number(summary, "avg_duration_min"), - "avg_efficiency": _optional_number(summary, "avg_efficiency"), - "avg_deep_min": _optional_number(summary, "avg_deep_min"), - }, - "sleep_days": [_sleep_day(day) for day in reversed(days)], - } - - def _get_json( - self, - path: str, - *, - params: dict[str, str | int | float] | None = None, - ) -> Mapping[str, object]: - response = requests.get(f"{_MONITOR_URL}{path}", params=params, timeout=8) - response.raise_for_status() - payload = response.json() - if not isinstance(payload, Mapping): - raise TypeError(f"Fitbit monitor 返回非对象: {path}") - return payload - - -def _sleep_segments(payload: Mapping[str, object]) -> list[dict[str, object]]: - raw = payload.get("sleep_24h") - if not isinstance(raw, Mapping): - raise TypeError("Fitbit monitor sleep_24h 必须是对象") - segments: list[dict[str, object]] = [] - for time_range, state in raw.items(): - if not isinstance(time_range, str) or not isinstance(state, str): - raise TypeError("Fitbit monitor sleep_24h 条目无效") - if state not in {"sleeping", "awake", "unknown"}: - raise TypeError(f"Fitbit monitor sleep_24h 状态无效: {state}") - segments.append( - { - "range": time_range, - "state": state, - "duration_min": _range_duration_minutes(time_range), - } - ) - return segments - - -def _range_duration_minutes(value: str) -> int: - try: - start, end = value.split("-", maxsplit=1) - start_hour, start_minute = (int(part) for part in start.split(":")) - end_hour, end_minute = (int(part) for part in end.split(":")) - except (TypeError, ValueError) as error: - raise ValueError(f"Fitbit monitor 睡眠时间段无效: {value}") from error - if not ( - 0 <= start_hour < 24 - and 0 <= end_hour < 24 - and 0 <= start_minute < 60 - and 0 <= end_minute < 60 - ): - raise ValueError(f"Fitbit monitor 睡眠时间段无效: {value}") - start_total = start_hour * 60 + start_minute - end_total = end_hour * 60 + end_minute - duration = (end_total - start_total) % (24 * 60) - if duration == 0: - return 1 - return duration - - -def _sleep_day(payload: Mapping[str, object]) -> dict[str, object]: - return { - "date": _optional_string(payload, "date"), - "duration_min": _optional_number(payload, "duration_min"), - "efficiency": _optional_number(payload, "efficiency"), - "deep_min": _optional_number(payload, "deep_min"), - "no_data": _boolean(payload, "no_data"), - } - - -def _mapping(payload: Mapping[str, object], name: str) -> Mapping[str, object]: - value = payload.get(name) - if not isinstance(value, Mapping): - raise TypeError(f"Fitbit monitor {name} 必须是对象") - return value - - -def _list_of_mappings(payload: Mapping[str, object], name: str) -> list[Mapping[str, object]]: - value = payload.get(name) - if not isinstance(value, list) or any(not isinstance(item, Mapping) for item in value): - raise TypeError(f"Fitbit monitor {name} 必须是对象数组") - return value - - -def _boolean(payload: Mapping[str, object], name: str) -> bool: - value = payload.get(name) - if not isinstance(value, bool): - raise TypeError(f"Fitbit monitor {name} 必须是布尔值") - return value +class ContentSourceServices(Protocol): + def bind(self, source_id: str) -> BoundContentSource: ... -def _optional_number(payload: Mapping[str, object], name: str) -> int | float | None: - value = payload.get(name) - if value is None: - return None - if isinstance(value, bool) or not isinstance(value, int | float): - raise TypeError(f"Fitbit monitor {name} 必须是数字或 null") - return value +CONTENT_SOURCE = ServiceKey[ContentSourceServices]("content.source.v1") -def _optional_string(payload: Mapping[str, object], name: str) -> str | None: - value = payload.get(name) - if value is None: - return None - if not isinstance(value, str): - raise TypeError(f"Fitbit monitor {name} 必须是字符串或 null") - return value +class FitbitContentConfig(BaseModel): + model_config = ConfigDict(extra="forbid") -class FitbitProactiveConfig(BaseModel): - enabled: bool = True + poll_interval_seconds: int = Field(default=300, ge=1) + sleep_ttl_seconds: int = Field(default=600, ge=1) class FitbitConfig(BaseModel): - proactive: FitbitProactiveConfig = Field(default_factory=FitbitProactiveConfig) + model_config = ConfigDict(extra="forbid") + + content: FitbitContentConfig = Field(default_factory=FitbitContentConfig) api_version = 3 name = "fitbit" -version = "3.0.0" -desc = "Fitbit health monitor and sleep model" +version = "3.1.0" +desc = "Fitbit health monitor, Content source, and sleep context" Config = FitbitConfig -inject = (MANAGED_PROCESSES, MCP_SERVERS, PROACTIVE_COMPONENTS, UI_SLOTS) +inject = (MANAGED_PROCESSES, MCP_SERVERS, TIMERS, CONTENT_SOURCE, UI_SLOTS) dashboard_module = "dashboard.py" async def apply(ctx: Context, config: FitbitConfig) -> None: - """登记 Fitbit 进程、MCP、主动源和移动端只读投影。""" + """装配 monitor、工具、Content 采集、睡眠上下文与移动界面。""" - # 1. Core 独占 monitor 端口、进程健康和 MCP endpoint 投影。 + # 1. 登记现有 monitor 与用户显式调用的普通 MCP 工具 await ctx.require(MANAGED_PROCESSES).register( ctx, ManagedProcessDefinition( @@ -229,44 +80,44 @@ async def apply(ctx: Context, config: FitbitConfig) -> None: McpServerDefinition( name="fitbit", command=("python", "run_mcp.py"), - required_tools=( - "get_proactive_events", - "get_sleep_context", - "acknowledge_events", - ), + required_tools=("fitbit_health_snapshot", "fitbit_sleep_report"), candidate_read_only_tools=( - "get_proactive_events", - "get_sleep_context", + "fitbit_health_snapshot", + "fitbit_sleep_report", ), endpoint_env=(EndpointEnv("FITBIT_MONITOR_PORT", "monitor"),), candidate_env={"FITBIT_BACKEND": "recording"}, ), ) - # 2. 主动源只消费 typed fetch/ack，不直接持有 monitor 或进程。 - if config.proactive.enabled: - proactive = ctx.require(PROACTIVE_COMPONENTS) - await proactive.register( - ctx, - ProactiveSourceDefinition( - name="health_alerts", - channels=("alert",), - mcp_server="fitbit", - fetch_tool="get_proactive_events", - ack_tool="acknowledge_events", - ), - ) - await proactive.register( - ctx, - ProactiveSourceDefinition( - name="sleep_context", - channels=("context",), - mcp_server="fitbit", - fetch_tool="get_sleep_context", - ), - ) + # 2. 绑定唯一正式来源；candidate Root 不会收到 STARTED + store = FitbitAdapterStore(ctx.data_root / "adapter.sqlite3") + store.initialize(datetime.now(UTC)) + runtime = FitbitContentRuntime( + store, + ctx.require(TIMERS), + ctx.require(CONTENT_SOURCE).bind("fitbit-health-alerts"), + FitbitMonitorClient(), + poll_interval=timedelta(seconds=config.content.poll_interval_seconds), + sleep_ttl=timedelta(seconds=config.content.sleep_ttl_seconds), + ) - # 3. 静态资产与同步只读查询绑定当前 exact Root。 + def setup() -> object: + return runtime.close + + _ = await ctx.effect(setup, label="fitbit-content-runtime") + + async def start(_event: object) -> None: + await runtime.start() + + async def stop(_event: object) -> None: + await runtime.close() + + _ = await ctx.on(RUNTIME_STARTED, start) + _ = await ctx.on(RUNTIME_STOPPING, stop) + _ = await ctx.on(CONTEXT_PREPARED_EVENT, SleepContextAppender(store).prepare) + + # 3. 在同一个 exact Root 上保留现有移动投影 await ctx.require(UI_SLOTS).register_mobile( ctx, MobileUiDefinition( @@ -277,28 +128,5 @@ async def apply(ctx: Context, config: FitbitConfig) -> None: description="查看当前心率、血氧、步数和最近睡眠节律", ), ), - query=_mobile_ui_query, + query=mobile_ui_query, ) - - -def _mobile_ui_query( - method: str, - payload: dict[str, object], - *, - session_id: str | None, - turn_id: str | None, -) -> dict[str, object]: - """按数据源独立返回当前健康或睡眠历史投影。""" - - # 1. 插件边界只暴露两种只读投影。 - _ = payload, session_id, turn_id - readers = { - "fitbit.current": FitbitMobileDashboardReader.get_current, - "fitbit.sleep_history": FitbitMobileDashboardReader.get_sleep_history, - } - reader_method = readers.get(method) - if reader_method is None: - raise MobileUiRpcInvalidRequest(f"未知 fitbit 移动方法: {method}") - - # 2. Core 调度器会把同步查询隔离到专用线程池。 - return reader_method(FitbitMobileDashboardReader()) diff --git a/src/content_adapter.py b/src/content_adapter.py new file mode 100644 index 0000000..47b27ce --- /dev/null +++ b/src/content_adapter.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +from collections.abc import Callable, Mapping, Sequence +from datetime import UTC, datetime, timedelta +from typing import Protocol, cast +from urllib.parse import quote + +import requests + +from agent.control.timer import TimerHandle, TimerStatus +from agent.plugin_composition import PluginTimers +from src.sleep_context import FitbitAdapterStore + + +class BoundContentSource(Protocol): + def submit( + self, batch_id: str, items: Sequence[Mapping[str, object]] + ) -> Mapping[str, object]: ... + + def unsettled(self, limit: int = 100) -> tuple[Mapping[str, object], ...]: ... + + def ack(self, settlement_ref: str) -> Mapping[str, object]: ... + + +class FitbitMonitorClient: + """读取 monitor 快照，并以目标状态完成 ACK。""" + + def __init__(self, base_url: str = "http://127.0.0.1:18765") -> None: + self._base_url = base_url.rstrip("/") + + def snapshot(self) -> Mapping[str, object]: + response = requests.get(f"{self._base_url}/api/agent", timeout=8) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, Mapping): + raise TypeError("Fitbit monitor /api/agent 必须返回对象") + return cast(Mapping[str, object], payload) + + def ensure_not_pending(self, event_id: str) -> None: + """确保事件已离开 monitor 队列，并允许 ACK 重放。""" + + response = requests.post( + f"{self._base_url}/api/agent/acknowledge/{quote(event_id, safe='')}", + timeout=8, + ) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, Mapping) or not isinstance( + payload.get("acknowledged"), bool + ): + raise TypeError("Fitbit monitor ACK 必须返回 acknowledged bool") + pending = self.snapshot().get("health_events") + if not isinstance(pending, list): + raise TypeError("Fitbit monitor health_events 必须是数组") + if any(_event_id(item) == event_id for item in pending): + raise RuntimeError(f"Fitbit event ACK 后仍在 pending 队列: {event_id}") + + +class FitbitContentRuntime: + """结算已投递 ACK，提交一次 monitor 快照，再登记一个 Timer。""" + + def __init__( + self, + store: FitbitAdapterStore, + timers: PluginTimers, + content: BoundContentSource, + monitor: FitbitMonitorClient, + *, + poll_interval: timedelta, + sleep_ttl: timedelta, + now: Callable[[], datetime] = lambda: datetime.now(UTC), + after_provider_ack: Callable[[], None] | None = None, + ) -> None: + self._store = store + self._timers = timers + self._content = content + self._monitor = monitor + self._poll_interval = poll_interval + self._sleep_ttl = sleep_ttl + self._now = now + self._after_provider_ack = after_provider_ack + self._handle: TimerHandle | None = None + self._task: asyncio.Task[None] | None = None + self._closed = False + + async def start(self) -> None: + """恢复来源 deadline，并登记唯一正式 Timer。""" + + if self._closed: + raise RuntimeError("Fitbit Content runtime 已关闭") + if self._handle is not None: + return + self._arm(self._store.next_due()) + + async def close(self) -> None: + """取消自有等待，不改写 Content 或来源事实。""" + + 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() + + 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_tick_rearm(handle), name="fitbit:content-poll" + ) + + async def _wait_tick_rearm(self, handle: TimerHandle) -> None: + """Timer 到点后执行一次采集，再从持久状态重新登记。""" + + try: + receipt = await handle.result() + if receipt.status is TimerStatus.CANCELLED or self._closed: + return + await asyncio.to_thread(self.tick) + finally: + self._handle = None + self._task = None + await handle.cleanup() + if not self._closed: + self._arm(self._store.next_due()) + + def tick(self) -> None: + """先结算历史投递，再发布当前 monitor 快照。""" + + # 1. 先完成外部 ACK，再结算 Content + self._drain_unsettled() + + # 2. 只拉取一次，独立归一化健康事件，并优先提交 Content + snapshot = self._monitor.snapshot() + items = normalize_health_events(snapshot) + batch_id = stable_batch_id(items) + _ = self._content.submit(batch_id, items) + + # 3. Content 接受批次后，才推进私有缓存和 deadline + now = _aware(self._now()) + sleep = normalize_sleep(snapshot) + self._store.commit_snapshot( + sleep, + observed_at=now, + expires_at=now + self._sleep_ttl, + next_due=now + self._poll_interval, + ) + + def _drain_unsettled(self) -> None: + while True: + rows = self._content.unsettled(limit=100) + for row in rows: + payload = _mapping(row, "payload") + event_id = _string(payload, "upstream_event_id") + settlement_ref = _string(row, "settlement_ref") + self._monitor.ensure_not_pending(event_id) + if self._after_provider_ack is not None: + self._after_provider_ack() + settled = self._content.ack(settlement_ref) + if settled.get("settled") is not True: + raise RuntimeError( + f"Fitbit Content ACK 未结算: {dict(settled)!r}" + ) + if len(rows) < 100: + return + + +def normalize_health_events( + snapshot: Mapping[str, object], +) -> tuple[Mapping[str, object], ...]: + """把 monitor 队列转换成身份稳定的 Content revisions。""" + + raw_events = snapshot.get("health_events") + if not isinstance(raw_events, list): + raise TypeError("Fitbit monitor health_events 必须是数组") + items: list[Mapping[str, object]] = [] + for raw in raw_events: + if not isinstance(raw, Mapping): + raise TypeError("Fitbit health event 必须是对象") + event_id = _event_id(raw) + payload: dict[str, object] = { + "upstream_event_id": event_id, + "source_type": "health_event", + "source_name": "fitbit", + "title": _string(raw, "type"), + "content": _string(raw, "message"), + "severity": _string(raw, "severity"), + "published_at": raw.get("created_at"), + "suggested_tone": raw.get("suggested_tone", ""), + "metrics": raw.get("metrics", {}), + } + revision = hashlib.sha256(_canonical(payload).encode("utf-8")).hexdigest() + items.append( + { + "item_id": event_id, + "revision": revision, + "payload": payload, + "not_before": None, + "requires_ack": True, + } + ) + return tuple(items) + + +def normalize_sleep(snapshot: Mapping[str, object]) -> Mapping[str, object]: + sleep = snapshot.get("sleep") + if not isinstance(sleep, Mapping): + raise TypeError("Fitbit monitor sleep 必须是对象") + return { + "state": _string(sleep, "state"), + "prob": sleep.get("prob"), + "prob_source": sleep.get("prob_source"), + "data_lag_min": sleep.get("data_lag_min"), + "sleep_24h": snapshot.get("sleep_24h", {}), + "last_updated": snapshot.get("last_updated"), + } + + +def stable_batch_id(items: Sequence[Mapping[str, object]]) -> str: + identity = sorted( + ( + {"item_id": item["item_id"], "revision": item["revision"]} + for item in items + ), + key=lambda item: (str(item["item_id"]), str(item["revision"])), + ) + return "fitbit-monitor:" + hashlib.sha256( + _canonical(identity).encode("utf-8") + ).hexdigest() + + +def _event_id(value: object) -> str: + if not isinstance(value, Mapping): + raise TypeError("Fitbit health event 必须是对象") + return _string(cast(Mapping[str, object], value), "id") + + +def _mapping(payload: Mapping[str, object], name: str) -> Mapping[str, object]: + value = payload.get(name) + if not isinstance(value, Mapping): + raise TypeError(f"Fitbit {name} 必须是对象") + return cast(Mapping[str, object], value) + + +def _string(payload: Mapping[str, object], name: str) -> str: + value = payload.get(name) + if not isinstance(value, str) or not value: + raise ValueError(f"Fitbit {name} 必须是非空字符串") + return value + + +def _canonical(payload: object) -> str: + return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def _aware(value: datetime) -> datetime: + if value.tzinfo is None: + raise ValueError("Fitbit Content clock 必须带时区") + return value.astimezone(UTC) diff --git a/src/mcp_bridge.py b/src/mcp_bridge.py index 068b07e..92b8aa8 100644 --- a/src/mcp_bridge.py +++ b/src/mcp_bridge.py @@ -1,345 +1,71 @@ -""" -fitbit-mcp — Fitbit 健康事件 MCP 服务。 - -对接 fitbit-monitor REST API，以标准 ProactiveEvent schema 暴露告警事件。 -约定 schema（alert 通道）： - event_id str 上游事件 ID，用于 ack - kind str 固定值 "alert" - source_type str 固定值 "health_event" - source_name str 固定值 "fitbit" - title str 事件类型（hr_elevated_rest / sleep_spo2 / ...） - content str 人类可读告警消息 - severity str "high" | "medium" - published_at str|None 事件创建时间（ISO 格式） - suggested_tone str LLM 建议语气（可选附加字段） -""" +"""供用户显式查询健康与睡眠的 Fitbit MCP 工具。""" from __future__ import annotations import json import logging import os -from datetime import datetime, timedelta, timezone -from typing import Any import requests from mcp.server.fastmcp import FastMCP -logger = logging.getLogger(__name__) +logger = logging.getLogger(__name__) HOST = os.getenv("FITBIT_MONITOR_HOST", "127.0.0.1") PORT = os.getenv("FITBIT_MONITOR_PORT", "18765") BASE_URL = f"http://{HOST}:{PORT}" -_last_wake_presence = "unknown" - - -def _recording_backend() -> bool: - return os.environ.get("FITBIT_BACKEND", "").strip().lower() == "recording" - - -def _monitor_available(timeout: float = 1.0) -> bool: - try: - resp = requests.get(f"{BASE_URL}/api/data", timeout=timeout) - resp.raise_for_status() - return True - except Exception: - return False - - -def _to_standard_event(raw: dict[str, Any]) -> dict[str, Any]: - """把 fitbit-monitor 的原始事件 dict 转换为标准 ProactiveEvent schema。""" - created_at = raw.get("created_at") - published_at = None - if created_at: - try: - published_at = datetime.strptime(created_at, "%Y-%m-%d %H:%M").isoformat() - except Exception: - published_at = created_at - - return { - "event_id": raw.get("id", ""), - "kind": "alert", - "source_type": "health_event", - "source_name": "fitbit", - "title": raw.get("type", ""), - "content": raw.get("message", ""), - "severity": raw.get("severity", ""), - "published_at": published_at, - "suggested_tone": raw.get("suggested_tone", ""), - "metrics": raw.get("metrics", {}), - } - - -def _fetch_agent_payload(timeout: int = 5) -> dict[str, Any]: - resp = requests.get(f"{BASE_URL}/api/agent", timeout=timeout) - resp.raise_for_status() - data = resp.json() - return data if isinstance(data, dict) else {} - - -def _build_sleep_context(data: dict[str, Any]) -> dict[str, Any]: - sleep = data.get("sleep", {}) or {} - state = str(sleep.get("state", "unknown") or "unknown") - prob = sleep.get("prob") - lag = sleep.get("data_lag_min") - prob_source = str(sleep.get("prob_source", "unavailable") or "unavailable") - - state_text = { - "sleeping": "用户当前可能已经睡着", - "awake": "用户当前更可能醒着", - "uncertain": "用户当前是否睡着还不确定", - "unknown": "暂时无法判断用户当前是否睡着", - }.get(state, "暂时无法判断用户当前是否睡着") - - summary = state_text - if prob is not None: - summary += f"（概率 {prob:.2f}）" - if lag is not None: - summary += f"，数据延迟约 {lag} 分钟" - summary += "。这是对用户当前睡眠状态的概率判断，不保证 100% 准确。" - - payload = { - "topic": "Fitbit 睡眠状态判断", - "summary": summary, - "hint": ( - "这是对用户当前是否睡着的概率判断，不是事实确认，不能据此断言用户一定睡着或一定醒着。" - "当判断用户可能正在睡觉时，应结合最近的聊天内容及时间戳，适当克制主动打扰，" - "减少普通强度、可推可不推的内容。" - "但如果出现你判断用户很可能会非常感兴趣的内容，仍然应该推送，" - "不要因为“可能在睡觉”而一律压掉。" - "拿不准时，默认更保守；但对明显强兴趣、高相关的内容，应优先保留发送机会。" - ), - "available": True, - "sleep": { - "state": state, - "prob": prob, - "prob_source": prob_source, - "data_lag_min": lag, - }, - "health_event_count": len(data.get("health_events") or []), - } - return _with_wake_contract(payload, state=state, probability=prob) - - -def _with_wake_contract( - payload: dict[str, Any], - *, - state: str, - probability: object, - observed_at: datetime | None = None, -) -> dict[str, Any]: - global _last_wake_presence - presence = { - "sleeping": "sleeping", - "awake": "active", - }.get(state, "unknown") - probability_value = _bounded_probability(probability) - confidence = ( - probability_value - if presence == "sleeping" - else 1.0 - probability_value - if presence == "active" - else 0.0 - ) - interruptibility = { - "sleeping": 0.0, - "active": 0.85, - "unknown": 0.4, - }[presence] - 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 datetime.now(timezone.utc) - original = dict(payload) - return { - **original, - "presence": presence, - "interruptibility": interruptibility, - "confidence": round(confidence, 3), - "transition": transition, - "observed_at": observed.isoformat(), - "expires_at": (observed + timedelta(minutes=10)).isoformat(), - "payload": original, - } - - -def _bounded_probability(value: object) -> float: - try: - return min(1.0, max(0.0, float(str(value)))) - except (TypeError, ValueError): - return 0.5 - - -def _unavailable_sleep_context(hint: str) -> dict[str, Any]: - payload = { - "available": False, - "topic": "", - "summary": "", - "hint": hint, - "sleep": { - "state": "unknown", - "prob": None, - "prob_source": "unavailable", - "data_lag_min": None, - }, - } - return _with_wake_contract(payload, state="unknown", probability=None) def create_mcp_server() -> FastMCP: - mcp = FastMCP("fitbit-mcp") - - @mcp.tool() - def get_proactive_events() -> str: - """获取 Fitbit 未处理的健康告警事件列表。 - - 返回标准 ProactiveEvent alert schema 的 JSON 数组。 - 空数组表示当前无待处理告警。 - """ - if _recording_backend(): - return json.dumps({"status": "empty"}, ensure_ascii=False) - try: - data = _fetch_agent_payload(timeout=5) - raw_events = data.get("health_events") or [] - events = [_to_standard_event(e) for e in raw_events] - payload = ( - {"status": "items", "items": events} - if events - else {"status": "empty"} - ) - return json.dumps(payload, ensure_ascii=False) - except requests.exceptions.ConnectionError as error: - logger.warning("fitbit-monitor 未运行 (%s)", BASE_URL) - return json.dumps( - {"status": "failure", "error": str(error), "retryable": True}, - ensure_ascii=False, - ) - except Exception as e: - logger.error("get_events 失败: %s", e) - return json.dumps( - {"status": "failure", "error": str(e), "retryable": True}, - ensure_ascii=False, - ) + """暴露两个由用户调用的普通 Fitbit 只读工具。""" - @mcp.tool() - def get_sleep_context() -> str: - """获取 Fitbit 睡眠判断上下文，供 proactive 作为 context 注入。""" - if _recording_backend(): - return json.dumps({"status": "empty"}, ensure_ascii=False) - try: - data = _fetch_agent_payload(timeout=5) - return json.dumps( - {"status": "items", "items": [_build_sleep_context(data)]}, - ensure_ascii=False, - ) - except requests.exceptions.ConnectionError: - logger.warning("fitbit-monitor 未运行 (%s)", BASE_URL) - return json.dumps( - { - "status": "items", - "items": [ - _unavailable_sleep_context( - "Fitbit 睡眠判断当前不可用；即使可用，它也只是概率判断，不保证 100% 准确。" - ) - ], - }, - ensure_ascii=False, - ) - except Exception as e: - logger.error("get_sleep_context 失败: %s", e) - return json.dumps( - { - "status": "items", - "items": [ - _unavailable_sleep_context(f"Fitbit 睡眠判断拉取失败: {e}") - ], - }, - ensure_ascii=False, - ) + mcp = FastMCP("fitbit-mcp") @mcp.tool() def fitbit_health_snapshot() -> str: """获取当前 Fitbit 健康状态快照。""" try: - resp = requests.get( + response = requests.get( f"{BASE_URL}/api/tool/fitbit_health_snapshot", timeout=5, ) - resp.raise_for_status() - return json.dumps(resp.json(), ensure_ascii=False) - except requests.exceptions.ConnectionError as e: + response.raise_for_status() + return json.dumps(response.json(), ensure_ascii=False) + except requests.exceptions.ConnectionError as error: logger.warning("fitbit-monitor 未运行 (%s)", BASE_URL) - return json.dumps({"error": f"无法连接 Fitbit monitor：{e}"}, ensure_ascii=False) - except Exception as e: - logger.error("fitbit_health_snapshot 失败: %s", e) - return json.dumps({"error": str(e)}, ensure_ascii=False) + return json.dumps( + {"error": f"无法连接 Fitbit monitor：{error}"}, + ensure_ascii=False, + ) + except requests.RequestException as error: + logger.error("fitbit_health_snapshot 失败: %s", error) + return json.dumps({"error": str(error)}, ensure_ascii=False) @mcp.tool() def fitbit_sleep_report(days: int = 7) -> str: """获取最近 N 天 Fitbit 睡眠质量报告。""" + bounded_days = max(1, min(int(days), 30)) try: - days = max(1, min(int(days), 30)) - resp = requests.get( + response = requests.get( f"{BASE_URL}/api/sleep_report", - params={"days": days}, + params={"days": bounded_days}, timeout=10, ) - if resp.status_code == 401: + if response.status_code == 401: return json.dumps( {"error": "Fitbit 未授权，请先完成 OAuth 授权。"}, ensure_ascii=False, ) - resp.raise_for_status() - return json.dumps(resp.json(), ensure_ascii=False) - except requests.exceptions.ConnectionError as e: + response.raise_for_status() + return json.dumps(response.json(), ensure_ascii=False) + except requests.exceptions.ConnectionError as error: logger.warning("fitbit-monitor 未运行 (%s)", BASE_URL) - return json.dumps({"error": f"无法连接 Fitbit monitor：{e}"}, ensure_ascii=False) - except Exception as e: - logger.error("fitbit_sleep_report 失败: %s", e) - return json.dumps({"error": str(e)}, ensure_ascii=False) - - @mcp.tool() - def acknowledge_events(event_ids: list[str]) -> str: - """标记健康告警事件为已处理，防止重复触发。 - - Args: - event_ids: 要 ack 的事件 ID 列表（来自 get_events 返回的 event_id 字段）。 - - Returns: - JSON 对象，包含每个 ID 的处理结果。 - """ - if not event_ids: - return json.dumps({"status": "skipped", "reason": "no_ids"}) - if _recording_backend(): - raise RuntimeError("fitbit recording backend 不允许确认事件") - - acknowledged = [] - failed = [] - for eid in event_ids: - try: - resp = requests.post( - f"{BASE_URL}/api/agent/acknowledge/{eid}", timeout=5 - ) - if resp.status_code == 200 and resp.json().get("acknowledged"): - acknowledged.append(eid) - else: - failed.append(eid) - except Exception as e: - logger.error("acknowledge %s 失败: %s", eid, e) - failed.append(eid) - - payload = ( - {"status": "committed", "ids": acknowledged} - if not failed and acknowledged == event_ids - else { - "status": "failure", - "error": "Fitbit 事件未完整确认", - "retryable": True, - "failed_ids": failed, - } - ) - return json.dumps(payload, ensure_ascii=False) + return json.dumps( + {"error": f"无法连接 Fitbit monitor：{error}"}, + ensure_ascii=False, + ) + except requests.RequestException as error: + logger.error("fitbit_sleep_report 失败: %s", error) + return json.dumps({"error": str(error)}, ensure_ascii=False) return mcp diff --git a/src/mobile_reader.py b/src/mobile_reader.py new file mode 100644 index 0000000..0197e78 --- /dev/null +++ b/src/mobile_reader.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +from collections.abc import Mapping + +import requests + +from agent.plugin_composition import MobileUiRpcInvalidRequest + + +_MONITOR_URL = "http://127.0.0.1:18765" + + +class FitbitMobileDashboardReader: + """读取 monitor，并生成稳定的移动健康总览。""" + + def get_current(self) -> dict[str, object]: + snapshot = self._get_json("/api/tool/fitbit_health_snapshot") + return { + "available": _boolean(snapshot, "available"), + "freshness": { + "last_updated": _optional_string(snapshot, "last_updated"), + "data_lag_min": _optional_number(snapshot, "data_lag_min"), + "spo2_lag_min": _optional_number(snapshot, "spo2_lag_min"), + }, + "current": { + "heart_rate": _optional_number(snapshot, "heart_rate"), + "spo2": _optional_number(snapshot, "spo2"), + "steps": _optional_number(snapshot, "steps"), + "sleep_state": _optional_string(snapshot, "sleep_state") or "unknown", + "sleep_prob": _optional_number(snapshot, "sleep_prob"), + }, + "sleep_24h": _sleep_segments(snapshot), + } + + def get_sleep_history(self) -> dict[str, object]: + report = self._get_json("/api/mobile/sleep_projection") + if not _boolean(report, "available"): + return { + "available": False, + "reason": _optional_string(report, "reason") or "projection_not_ready", + "freshness": _mapping(report, "freshness"), + "sleep_summary": { + "days_with_data": 0, + "avg_duration_min": None, + "avg_efficiency": None, + "avg_deep_min": None, + }, + "sleep_days": [], + } + summary = _mapping(report, "summary") + days = _list_of_mappings(report, "days") + return { + "available": True, + "reason": None, + "freshness": _mapping(report, "freshness"), + "sleep_summary": { + "days_with_data": _optional_number(summary, "days_with_data"), + "avg_duration_min": _optional_number(summary, "avg_duration_min"), + "avg_efficiency": _optional_number(summary, "avg_efficiency"), + "avg_deep_min": _optional_number(summary, "avg_deep_min"), + }, + "sleep_days": [_sleep_day(day) for day in reversed(days)], + } + + def _get_json( + self, + path: str, + *, + params: dict[str, str | int | float] | None = None, + ) -> Mapping[str, object]: + response = requests.get(f"{_MONITOR_URL}{path}", params=params, timeout=8) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, Mapping): + raise TypeError(f"Fitbit monitor 返回非对象: {path}") + return payload + + +def mobile_ui_query( + method: str, + payload: dict[str, object], + *, + session_id: str | None, + turn_id: str | None, +) -> dict[str, object]: + """按数据源独立返回当前健康或睡眠历史投影。""" + + _ = payload, session_id, turn_id + readers = { + "fitbit.current": FitbitMobileDashboardReader.get_current, + "fitbit.sleep_history": FitbitMobileDashboardReader.get_sleep_history, + } + reader_method = readers.get(method) + if reader_method is None: + raise MobileUiRpcInvalidRequest(f"未知 fitbit 移动方法: {method}") + return reader_method(FitbitMobileDashboardReader()) + + +def _sleep_segments(payload: Mapping[str, object]) -> list[dict[str, object]]: + raw = payload.get("sleep_24h") + if not isinstance(raw, Mapping): + raise TypeError("Fitbit monitor sleep_24h 必须是对象") + segments: list[dict[str, object]] = [] + for time_range, state in raw.items(): + if not isinstance(time_range, str) or not isinstance(state, str): + raise TypeError("Fitbit monitor sleep_24h 条目无效") + if state not in {"sleeping", "awake", "unknown"}: + raise TypeError(f"Fitbit monitor sleep_24h 状态无效: {state}") + segments.append( + { + "range": time_range, + "state": state, + "duration_min": _range_duration_minutes(time_range), + } + ) + return segments + + +def _range_duration_minutes(value: str) -> int: + try: + start, end = value.split("-", maxsplit=1) + start_hour, start_minute = (int(part) for part in start.split(":")) + end_hour, end_minute = (int(part) for part in end.split(":")) + except (TypeError, ValueError) as error: + raise ValueError(f"Fitbit monitor 睡眠时间段无效: {value}") from error + if not ( + 0 <= start_hour < 24 + and 0 <= end_hour < 24 + and 0 <= start_minute < 60 + and 0 <= end_minute < 60 + ): + raise ValueError(f"Fitbit monitor 睡眠时间段无效: {value}") + duration = ((end_hour * 60 + end_minute) - (start_hour * 60 + start_minute)) % ( + 24 * 60 + ) + return 1 if duration == 0 else duration + + +def _sleep_day(payload: Mapping[str, object]) -> dict[str, object]: + return { + "date": _optional_string(payload, "date"), + "duration_min": _optional_number(payload, "duration_min"), + "efficiency": _optional_number(payload, "efficiency"), + "deep_min": _optional_number(payload, "deep_min"), + "no_data": _boolean(payload, "no_data"), + } + + +def _mapping(payload: Mapping[str, object], name: str) -> Mapping[str, object]: + value = payload.get(name) + if not isinstance(value, Mapping): + raise TypeError(f"Fitbit monitor {name} 必须是对象") + return value + + +def _list_of_mappings(payload: Mapping[str, object], name: str) -> list[Mapping[str, object]]: + value = payload.get(name) + if not isinstance(value, list) or any(not isinstance(item, Mapping) for item in value): + raise TypeError(f"Fitbit monitor {name} 必须是对象数组") + return value + + +def _boolean(payload: Mapping[str, object], name: str) -> bool: + value = payload.get(name) + if not isinstance(value, bool): + raise TypeError(f"Fitbit monitor {name} 必须是布尔值") + return value + + +def _optional_number(payload: Mapping[str, object], name: str) -> int | float | None: + value = payload.get(name) + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int | float): + raise TypeError(f"Fitbit monitor {name} 必须是数字或 null") + return value + + +def _optional_string(payload: Mapping[str, object], name: str) -> str | None: + value = payload.get(name) + if value is None: + return None + if not isinstance(value, str): + raise TypeError(f"Fitbit monitor {name} 必须是字符串或 null") + return value diff --git a/src/sleep_context.py b/src/sleep_context.py new file mode 100644 index 0000000..cf59107 --- /dev/null +++ b/src/sleep_context.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Generator, Mapping +from contextlib import contextmanager +from datetime import UTC, datetime +from pathlib import Path +from typing import cast + +from agent.lifecycle.types import BeforeTurnCtx + + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS source_state( + singleton INTEGER PRIMARY KEY CHECK(singleton = 1), + next_due TEXT NOT NULL, + sleep_json TEXT, + sleep_observed_at TEXT, + sleep_expires_at TEXT +); +""" + + +class FitbitAdapterStore: + """在一个私有文件中持久化来源 deadline 与当前睡眠上下文。""" + + def __init__(self, path: Path) -> None: + self.path = path + + def initialize(self, now: datetime) -> None: + with self._transaction(write=True) as connection: + connection.executescript(_SCHEMA) + connection.execute( + """ + INSERT OR IGNORE INTO source_state( + singleton, next_due, sleep_json, + sleep_observed_at, sleep_expires_at + ) VALUES(1, ?, NULL, NULL, NULL) + """, + (_aware_utc(now),), + ) + + def next_due(self) -> datetime: + with self._transaction(write=False) as connection: + row = connection.execute( + "SELECT next_due FROM source_state WHERE singleton = 1" + ).fetchone() + if row is None: + raise RuntimeError("Fitbit adapter state 尚未初始化") + return datetime.fromisoformat(str(row["next_due"])) + + def commit_snapshot( + self, + sleep: Mapping[str, object], + *, + observed_at: datetime, + expires_at: datetime, + next_due: datetime, + ) -> None: + """原子提交睡眠缓存与下一个来源 deadline。""" + + payload = json.dumps(sleep, sort_keys=True, separators=(",", ":")) + with self._transaction(write=True) as connection: + changed = connection.execute( + """ + UPDATE source_state + SET next_due = ?, sleep_json = ?, + sleep_observed_at = ?, sleep_expires_at = ? + WHERE singleton = 1 + """, + ( + _aware_utc(next_due), + payload, + _aware_utc(observed_at), + _aware_utc(expires_at), + ), + ) + if changed.rowcount != 1: + raise RuntimeError("Fitbit adapter state 尚未初始化") + + def current_sleep(self, now: datetime) -> Mapping[str, object] | None: + with self._transaction(write=False) as connection: + row = connection.execute( + """ + SELECT sleep_json, sleep_expires_at + FROM source_state WHERE singleton = 1 + """ + ).fetchone() + if row is None or row["sleep_json"] is None: + return None + expires_at = datetime.fromisoformat(str(row["sleep_expires_at"])) + if expires_at <= _aware(now): + return None + payload = json.loads(str(row["sleep_json"])) + if not isinstance(payload, Mapping): + raise TypeError("Fitbit sleep cache 必须是对象") + return cast(Mapping[str, object], payload) + + @contextmanager + def _transaction(self, *, write: bool) -> Generator[sqlite3.Connection]: + if write: + self.path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(self.path) + connection.row_factory = sqlite3.Row + try: + connection.execute("BEGIN IMMEDIATE" if write else "BEGIN") + yield connection + connection.commit() + except BaseException: + connection.rollback() + raise + finally: + connection.close() + + +class SleepContextAppender: + """只为 Wake Turn 追加未过期的 Fitbit 睡眠提示。""" + + def __init__(self, store: FitbitAdapterStore) -> None: + self._store = store + + async def prepare(self, ctx: BeforeTurnCtx) -> None: + if ctx.channel != "wake": + return + sleep = self._store.current_sleep(ctx.timestamp) + if sleep is None: + return + state = _string(sleep, "state") + probability = sleep.get("prob") + lag = sleep.get("data_lag_min") + ctx.extra_hints.append( + "Fitbit 睡眠上下文（概率判断，不是事实）：" + f"state={state}, probability={probability}, data_lag_min={lag}。" + "若可能正在睡觉，普通内容应克制打扰；明显高兴趣或高相关内容仍可发送。" + ) + + +def _string(payload: Mapping[str, object], name: str) -> str: + value = payload.get(name) + if not isinstance(value, str) or not value: + raise ValueError(f"Fitbit sleep {name} 必须是非空字符串") + return value + + +def _aware(value: datetime) -> datetime: + if value.tzinfo is None: + raise ValueError("Fitbit adapter 时间必须带时区") + return value.astimezone(UTC) + + +def _aware_utc(value: datetime) -> str: + return _aware(value).isoformat() diff --git a/tests/test_content_adapter.py b/tests/test_content_adapter.py new file mode 100644 index 0000000..5addf9c --- /dev/null +++ b/tests/test_content_adapter.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import asyncio +import json +import os +import signal +import subprocess +import sys +import threading +from collections.abc import Generator, Mapping, Sequence +from datetime import UTC, datetime, timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import cast +from urllib.parse import unquote + +import pytest +from agent.control.timer import AsyncioOneShotTimer +from agent.plugin_composition import PluginTimers +from plugins.content import plugin as content_plugin +from plugins.content.store import ContentStore + +from src.content_adapter import ( + BoundContentSource, + FitbitContentRuntime, + FitbitMonitorClient, + normalize_health_events, + stable_batch_id, +) +from src.sleep_context import FitbitAdapterStore + + +NOW = datetime(2026, 8, 23, 8, tzinfo=UTC) +SNAPSHOT: dict[str, object] = { + "health_events": [ + { + "id": "fitbit:event-1", + "type": "hr_elevated_rest", + "message": "静息心率持续偏高", + "severity": "high", + "created_at": "2026-08-23 08:00", + "suggested_tone": "先询问感受", + "metrics": {"heart_rate": 105}, + } + ], + "sleep": { + "state": "sleeping", + "prob": 0.92, + "prob_source": "model", + "data_lag_min": 3, + }, + "sleep_24h": {"00:00-08:00": "sleeping"}, + "last_updated": NOW.isoformat(), +} + + +class RecordingMonitor: + def __init__(self, snapshot: Mapping[str, object]) -> None: + self.current = dict(snapshot) + self.acknowledged: list[str] = [] + + def snapshot(self) -> Mapping[str, object]: + return self.current + + def ensure_not_pending(self, event_id: str) -> None: + self.acknowledged.append(event_id) + + +def _bound(store: ContentStore, changed=lambda: None) -> BoundContentSource: + return content_plugin._SourceServices(store, changed).bind( + "fitbit-health-alerts" + ) + + +def _runtime( + tmp_path: Path, + content: BoundContentSource, + monitor: object, +) -> tuple[FitbitContentRuntime, FitbitAdapterStore]: + store = FitbitAdapterStore(tmp_path / "adapter.sqlite3") + store.initialize(NOW) + runtime = FitbitContentRuntime( + store, + PluginTimers.candidate_validation(), + content, + cast(FitbitMonitorClient, monitor), + poll_interval=timedelta(minutes=5), + sleep_ttl=timedelta(minutes=10), + now=lambda: NOW, + ) + return runtime, store + + +def test_submit_commits_before_private_deadline_and_replay_is_idempotent( + tmp_path: Path, +) -> None: + content_store = ContentStore(tmp_path / "content.sqlite3") + content_store.initialize() + calls = 0 + + def fail_after_commit() -> None: + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("post-commit hint failed") + + first, adapter = _runtime( + tmp_path, + _bound(content_store, fail_after_commit), + RecordingMonitor(SNAPSHOT), + ) + with pytest.raises(RuntimeError, match="post-commit hint failed"): + first.tick() + + assert content_store.state_counts() == {"pending": 1} + assert adapter.next_due() == NOW + assert adapter.current_sleep(NOW) is None + + second, adapter = _runtime( + tmp_path, + _bound(content_store), + RecordingMonitor(SNAPSHOT), + ) + second.tick() + assert content_store.state_counts() == {"pending": 1} + assert adapter.next_due() == NOW + timedelta(minutes=5) + sleep = adapter.current_sleep(NOW) + assert sleep is not None and sleep["state"] == "sleeping" + + +def test_health_content_and_sleep_cache_have_separate_owners(tmp_path: Path) -> None: + content_store = ContentStore(tmp_path / "content.sqlite3") + content_store.initialize() + runtime, adapter = _runtime( + tmp_path, + _bound(content_store), + RecordingMonitor(SNAPSHOT), + ) + + runtime.tick() + + snapshot = content_store.snapshot(NOW) + snapshot_items = cast(Sequence[Mapping[str, object]], snapshot["items"]) + assert len(snapshot_items) == 1 + payload = cast(Mapping[str, object], snapshot_items[0]["payload"]) + assert payload["upstream_event_id"] == "fitbit:event-1" + assert "sleep" not in payload + sleep = adapter.current_sleep(NOW) + assert sleep is not None and sleep["state"] == "sleeping" + + +def test_normalization_has_stable_batch_and_revision() -> None: + first = normalize_health_events(SNAPSHOT) + second = normalize_health_events(json.loads(json.dumps(SNAPSHOT))) + assert first == second + assert stable_batch_id(first) == stable_batch_id(second) + assert first[0]["requires_ack"] is True + + +def test_batch_identity_does_not_depend_on_monitor_queue_order() -> None: + first = normalize_health_events(SNAPSHOT)[0] + second = { + **first, + "item_id": "fitbit:event-2", + "revision": "revision-2", + } + assert stable_batch_id((first, second)) == stable_batch_id((second, first)) + + +@pytest.mark.asyncio +async def test_reload_has_only_one_real_timer_wait(tmp_path: Path) -> None: + active = 0 + maximum = 0 + entered = asyncio.Event() + + async def sleeper(_delay: float) -> None: + nonlocal active, maximum + active += 1 + maximum = max(maximum, active) + entered.set() + try: + await asyncio.Future() + finally: + active -= 1 + + content_store = ContentStore(tmp_path / "content.sqlite3") + content_store.initialize() + adapter = FitbitAdapterStore(tmp_path / "adapter.sqlite3") + adapter.initialize(NOW) + timers = PluginTimers(AsyncioOneShotTimer(clock=lambda: NOW, sleeper=sleeper)) + + def make_runtime() -> FitbitContentRuntime: + return FitbitContentRuntime( + adapter, + timers, + _bound(content_store), + cast(FitbitMonitorClient, RecordingMonitor(SNAPSHOT)), + poll_interval=timedelta(minutes=5), + sleep_ttl=timedelta(minutes=10), + now=lambda: NOW, + ) + + first = make_runtime() + await first.start() + await entered.wait() + assert active == 1 + await first.close() + assert active == 0 + + entered.clear() + second = make_runtime() + await second.start() + await entered.wait() + assert active == 1 + assert maximum == 1 + await second.close() + + +class _MonitorState: + def __init__(self) -> None: + self.pending = [dict(cast(list[Mapping[str, object]], SNAPSHOT["health_events"])[0])] + + +class _MonitorHandler(BaseHTTPRequestHandler): + state: _MonitorState + + def do_GET(self) -> None: # noqa: N802 + if self.path != "/api/agent": + self.send_error(404) + return + self._json({**SNAPSHOT, "health_events": self.state.pending}) + + def do_POST(self) -> None: # noqa: N802 + prefix = "/api/agent/acknowledge/" + if not self.path.startswith(prefix): + self.send_error(404) + return + event_id = unquote(self.path.removeprefix(prefix)) + before = len(self.state.pending) + self.state.pending = [row for row in self.state.pending if row["id"] != event_id] + self._json({"acknowledged": len(self.state.pending) < before}) + + def log_message(self, format: str, *args: object) -> None: + _ = format, args + + def _json(self, payload: Mapping[str, object]) -> None: + body = json.dumps(payload).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +@pytest.fixture +def monitor_server() -> Generator[tuple[str, _MonitorState], None, None]: + state = _MonitorState() + handler = type("MonitorHandler", (_MonitorHandler,), {"state": state}) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host = str(server.server_address[0]) + port = int(server.server_address[1]) + yield f"http://{host}:{port}", state + finally: + server.shutdown() + server.server_close() + thread.join() + + +def _delivered_content(path: Path) -> None: + store = ContentStore(path) + store.initialize() + items = normalize_health_events(SNAPSHOT) + store.submit("fitbit-health-alerts", stable_batch_id(items), items) + snapshot = store.snapshot(NOW) + snapshot_items = cast(Sequence[Mapping[str, object]], snapshot["items"]) + selected = store.select( + cast(Mapping[str, object], snapshot_items[0]["ref"]), + cast(int, snapshot["snapshot_seq"]), + {"session_id": "wake:fitbit", "turn_id": "turn-1"}, + NOW, + ) + token = cast(str, selected["selection_token"]) + store.transition(token, "ready_for_delivery") + store.transition(token, "delivered", settlement_ref="delivery:fitbit:1") + + +def test_ack_response_sigkill_recovers_by_desired_state( + tmp_path: Path, + monitor_server: tuple[str, _MonitorState], +) -> None: + base_url, state = monitor_server + content_path = tmp_path / "content.sqlite3" + _delivered_content(content_path) + script = """ +import os, signal, sys +from pathlib import Path +from agent.plugin_composition import PluginTimers +from plugins.content import plugin as content_plugin +from plugins.content.store import ContentStore +from src.content_adapter import FitbitContentRuntime, FitbitMonitorClient +from src.sleep_context import FitbitAdapterStore +from datetime import UTC, datetime, timedelta + +content_store = ContentStore(Path(sys.argv[1])) +content_store.initialize() +bound = content_plugin._SourceServices(content_store, lambda: None).bind('fitbit-health-alerts') +adapter = FitbitAdapterStore(Path(sys.argv[2])) +adapter.initialize(datetime.now(UTC)) +runtime = FitbitContentRuntime( + adapter, PluginTimers.candidate_validation(), bound, FitbitMonitorClient(sys.argv[3]), + poll_interval=timedelta(minutes=5), sleep_ttl=timedelta(minutes=10), + after_provider_ack=lambda: os.kill(os.getpid(), signal.SIGKILL), +) +runtime._drain_unsettled() +""" + result = subprocess.run( + [ + sys.executable, + "-c", + script, + str(content_path), + str(tmp_path / "child-adapter.sqlite3"), + base_url, + ], + check=False, + ) + assert result.returncode == -signal.SIGKILL + assert state.pending == [] + assert ContentStore(content_path).state_counts() == {"delivered": 1} + + content_store = ContentStore(content_path) + runtime, _adapter = _runtime( + tmp_path, + _bound(content_store), + FitbitMonitorClient(base_url), + ) + runtime._drain_unsettled() + assert ContentStore(content_path).state_counts() == {"settled": 1} diff --git a/tests/test_context_contract.py b/tests/test_context_contract.py deleted file mode 100644 index 8e74c11..0000000 --- a/tests/test_context_contract.py +++ /dev/null @@ -1,55 +0,0 @@ -from datetime import UTC, datetime - -from src import mcp_bridge - - -def test_sleep_context_exposes_wake_contract_and_preserves_payload() -> None: - mcp_bridge._last_wake_presence = "unknown" - - context = mcp_bridge._build_sleep_context( - { - "sleep": { - "state": "sleeping", - "prob": 0.92, - "prob_source": "model", - "data_lag_min": 3, - }, - "health_events": [{"id": "a"}], - } - ) - - assert context["presence"] == "sleeping" - assert context["interruptibility"] == 0.0 - assert context["confidence"] == 0.92 - assert context["transition"] == "" - assert datetime.fromisoformat(context["expires_at"]) > datetime.fromisoformat( - context["observed_at"] - ) - assert context["payload"]["sleep"]["state"] == "sleeping" - assert context["payload"]["health_event_count"] == 1 - - -def test_sleep_owner_emits_generic_transition() -> None: - mcp_bridge._last_wake_presence = "sleeping" - observed = datetime(2026, 7, 12, 8, tzinfo=UTC) - context = mcp_bridge._with_wake_contract( - {"available": True, "sleep": {"state": "awake", "prob": 0.1}}, - state="awake", - probability=0.1, - observed_at=observed, - ) - - assert context["presence"] == "active" - assert context["interruptibility"] == 0.85 - assert context["confidence"] == 0.9 - assert context["transition"] == "sleeping->active" - assert context["observed_at"] == observed.isoformat() - - -def test_unavailable_context_still_has_complete_contract() -> None: - context = mcp_bridge._unavailable_sleep_context("offline") - - assert context["available"] is False - assert context["presence"] == "unknown" - assert context["confidence"] == 0.0 - assert context["payload"]["hint"] == "offline" diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py index 5206f46..c206bef 100644 --- a/tests/test_manager_integration.py +++ b/tests/test_manager_integration.py @@ -1,17 +1,15 @@ from __future__ import annotations -import json import shutil import sys from pathlib import Path import pytest -from agent.plugins.generation_activity_host import ActivityHost -from agent.plugins.generation_proactive_host import ProactiveActivityAdapter from agent.plugins.generation import PluginGeneration from agent.plugins.manager import PluginManager from agent.plugins.snapshot import RuntimeSnapshot from bus.event_bus import EventBus +from plugins.content import plugin as content_plugin ROOT = Path(__file__).resolve().parents[1] @@ -37,6 +35,9 @@ def _stage_plugin(tmp_path: Path) -> Path: Path(sys.executable).parent.parent, target_is_directory=True, ) + content_source = Path(content_plugin.__file__).resolve().parent + content_target = source.parent / "content" + shutil.copytree(content_source, content_target) return source @@ -56,10 +57,6 @@ async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( workspace=tmp_path / "workspace", installed_cache_root=tmp_path / "home" / "cache", ) - activity = ActivityHost( - (ProactiveActivityAdapter(manager.composition_generation_host),) - ) - manager.bind_activity_host(activity) stable_snapshot = None validation_root = None try: @@ -69,9 +66,8 @@ async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( assert stable_snapshot.composition_root is not None assert stable_snapshot.mcp_server_registry is not None assert stable_snapshot.managed_process_registry is not None - assert stable_snapshot.proactive_component_catalog is not None assert stable_snapshot.mobile_ui_registry is not None - stable_generation = next(iter(stable_snapshot.generations.values())) + stable_generation = stable_snapshot.generations["fitbit"] stable_runtime = manager.composition_generation_host.get( stable_generation.generation_id ) @@ -87,7 +83,7 @@ async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( for relative in ("plugin.py", "akashic.plugin.toml"): path = plugin_root / relative path.write_text( - path.read_text(encoding="utf-8").replace("3.0.0", "3.0.1"), + path.read_text(encoding="utf-8").replace("3.1.0", "3.1.1"), encoding="utf-8", ) candidate = await manager.prepare_candidate("fitbit") @@ -96,11 +92,6 @@ async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( validation_root = candidate.validation_workspace.parent candidate_snapshot = candidate.runtime_snapshot assert candidate_snapshot.composition_root is not None - assert candidate_snapshot.proactive_component_catalog is not None - assert ( - candidate_snapshot.proactive_component_catalog.root_instance_token - is candidate_snapshot.composition_root.instance_token - ) original_invariants = manager._post_publish_invariants # pyright: ignore[reportPrivateUsage] candidate_checked = False @@ -117,18 +108,9 @@ async def inspect_candidate_runtime( assert candidate_runtime.mcp is not None async with candidate_runtime.mcp.route("fitbit") as candidate_route: assert set(candidate_route.tool_names) == { - "get_proactive_events", - "get_sleep_context", + "fitbit_health_snapshot", + "fitbit_sleep_report", } - proactive = await candidate_route.call("get_proactive_events", {}) - sleep = await candidate_route.call("get_sleep_context", {}) - assert json.loads(proactive.output) == {"status": "empty"} - assert json.loads(sleep.output) == {"status": "empty"} - with pytest.raises(PermissionError, match="未获 allowlist 授权"): - _ = await candidate_route.call( - "acknowledge_events", - {"event_ids": ["event-1"]}, - ) candidate_checked = True await original_invariants(generation, snapshot) @@ -143,17 +125,11 @@ async def inspect_candidate_runtime( final_snapshot = manager.current_snapshot assert final_snapshot is not None and final_snapshot.composition_root is not None assert final_snapshot.composition_root is not candidate_snapshot.composition_root - 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() finally: await manager.terminate_all() - # 3. Manager 终止后进程、MCP、Activity 与 Root effects 全部归零。 - assert activity.active is None + # 3. Manager 终止后进程、MCP 与 Root effects 全部归零。 assert stable_snapshot is not None and stable_snapshot.composition_root is not None assert stable_snapshot.composition_root.receipt().effects == () assert stable_snapshot.composition_root.topology_view().listeners == () diff --git a/tests/test_mcp_v3_runtime.py b/tests/test_mcp_v3_runtime.py index 4f278d1..adae5ad 100644 --- a/tests/test_mcp_v3_runtime.py +++ b/tests/test_mcp_v3_runtime.py @@ -4,7 +4,6 @@ from typing import cast import pytest -from mcp.server.fastmcp.exceptions import ToolError from src import mcp_bridge @@ -19,61 +18,58 @@ async def _call(name: str, arguments: dict[str, object]) -> dict[str, object]: @pytest.mark.asyncio -async def test_recording_backend_is_typed_empty_without_monitor_access( +async def test_mcp_exposes_only_ordinary_fitbit_read_tools() -> None: + tools = await mcp_bridge.create_mcp_server().list_tools() + assert [tool.name for tool in tools] == [ + "fitbit_health_snapshot", + "fitbit_sleep_report", + ] + + +@pytest.mark.asyncio +async def test_health_snapshot_uses_monitor_read_endpoint( monkeypatch: pytest.MonkeyPatch, ) -> None: - monkeypatch.setenv("FITBIT_BACKEND", "recording") + calls: list[tuple[str, object]] = [] + + class Response: + status_code = 200 + + @staticmethod + def raise_for_status() -> None: + return None - def forbidden(*args: object, **kwargs: object) -> object: - raise AssertionError((args, kwargs)) + @staticmethod + def json() -> dict[str, object]: + return {"available": True, "heart_rate": 72} - monkeypatch.setattr(mcp_bridge.requests, "get", forbidden) - monkeypatch.setattr(mcp_bridge.requests, "post", forbidden) + def get(url: str, **kwargs: object) -> Response: + calls.append((url, kwargs)) + return Response() - assert await _call("get_proactive_events", {}) == {"status": "empty"} - assert await _call("get_sleep_context", {}) == {"status": "empty"} - assert await _call("acknowledge_events", {"event_ids": []}) == { - "status": "skipped", - "reason": "no_ids", + monkeypatch.setattr(mcp_bridge.requests, "get", get) + assert await _call("fitbit_health_snapshot", {}) == { + "available": True, + "heart_rate": 72, } - with pytest.raises(ToolError, match="recording backend 不允许确认事件"): - _ = await _call("acknowledge_events", {"event_ids": ["event-1"]}) + assert calls[0][0].endswith("/api/tool/fitbit_health_snapshot") @pytest.mark.asyncio -async def test_formal_fetch_and_ack_encode_explicit_results( +async def test_sleep_report_bounds_days_and_preserves_unauthorized_result( monkeypatch: pytest.MonkeyPatch, ) -> None: - monkeypatch.delenv("FITBIT_BACKEND", raising=False) - monkeypatch.setattr( - mcp_bridge, - "_fetch_agent_payload", - lambda timeout: { - "health_events": [ - { - "id": "event-1", - "type": "high_hr", - "message": "心率偏高", - "severity": "high", - } - ] - }, - ) + calls: list[dict[str, object]] = [] class Response: - status_code = 200 - - @staticmethod - def json() -> dict[str, object]: - return {"acknowledged": True} + status_code = 401 - monkeypatch.setattr(mcp_bridge.requests, "post", lambda *args, **kwargs: Response()) + def get(_url: str, **kwargs: object) -> Response: + calls.append(kwargs) + return Response() - fetched = await _call("get_proactive_events", {}) - assert fetched["status"] == "items" - items = cast(list[dict[str, object]], fetched["items"]) - assert [item["event_id"] for item in items] == ["event-1"] - assert await _call("acknowledge_events", {"event_ids": ["event-1"]}) == { - "status": "committed", - "ids": ["event-1"], + monkeypatch.setattr(mcp_bridge.requests, "get", get) + assert await _call("fitbit_sleep_report", {"days": 100}) == { + "error": "Fitbit 未授权，请先完成 OAuth 授权。" } + assert calls == [{"params": {"days": 30}, "timeout": 10}] diff --git a/tests/test_mobile_dashboard.py b/tests/test_mobile_dashboard.py index 383436b..a353dda 100644 --- a/tests/test_mobile_dashboard.py +++ b/tests/test_mobile_dashboard.py @@ -5,8 +5,8 @@ import pytest -import plugin -from plugin import FitbitMobileDashboardReader +from src import mobile_reader as plugin +from src.mobile_reader import FitbitMobileDashboardReader class Response: diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 6e95ed3..724842e 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -7,54 +7,74 @@ from agent.plugin_composition import ( MANAGED_PROCESSES, MCP_SERVERS, - PROACTIVE_COMPONENTS, + TIMERS, UI_SLOTS, CompositionRoot, - PluginProactiveComponents, PluginRuntime, + PluginTimers, PluginUiSlots, ) from agent.plugin_composition.mcp_slots import ( PluginMcpServers, _freeze_plugin_mcp_servers, ) -from agent.plugin_composition.proactive import _freeze_plugin_proactive_components from agent.plugin_composition.process_slots import ( PluginManagedProcesses, _freeze_plugin_managed_processes, ) -from agent.plugins.composable import ComposablePlugin from agent.plugins import manager as manager_module +from agent.plugins.composable import ComposablePlugin from agent.plugins.static_manifest import load_static_plugin_manifest +from plugins.content import plugin as content_plugin import plugin as plugin_module -from plugin import FitbitConfig, _mobile_ui_query +from plugin import FitbitConfig +from src import mobile_reader +from src.mobile_reader import mobile_ui_query ROOT = Path(__file__).resolve().parents[1] +CORE_ROOT = Path(content_plugin.__file__).resolve().parents[2] + + +async def _mount_services(root: CompositionRoot, tmp_path: Path) -> None: + await root.context.provide(TIMERS, PluginTimers.candidate_validation()) + await root.mount( + ComposablePlugin.from_module(content_plugin), + name="content", + runtime=PluginRuntime( + plugin_id="content", + plugin_dir=CORE_ROOT / "plugins/content", + data_dir=tmp_path / "content-data", + workspace=tmp_path / "workspace", + config=object(), + ), + ) + def test_pure_v3_exports_and_exact_apply() -> None: assert plugin_module.api_version == 3 assert plugin_module.name == "fitbit" - assert plugin_module.version == "3.0.0" + assert plugin_module.version == "3.1.0" assert tuple(inspect.signature(plugin_module.apply).parameters) == ("ctx", "config") assert ComposablePlugin.from_module(plugin_module).dashboard_module == "dashboard.py" + assert "PROACTIVE_COMPONENTS" not in ROOT.joinpath("plugin.py").read_text() @pytest.mark.asyncio -async def test_apply_registers_exact_runtime_sources_and_mobile_ui( +async def test_apply_registers_content_runtime_tools_and_mobile_ui( tmp_path: Path, ) -> None: root = CompositionRoot("fitbit:test") processes = PluginManagedProcesses(root.instance_token) servers = PluginMcpServers(root.instance_token) - components = PluginProactiveComponents(root.instance_token) ui_slots = PluginUiSlots() await root.context.provide(MANAGED_PROCESSES, processes) await root.context.provide(MCP_SERVERS, servers) - await root.context.provide(PROACTIVE_COMPONENTS, components) await root.context.provide(UI_SLOTS, ui_slots) + await _mount_services(root, tmp_path) data_dir = tmp_path / "plugin-data" + await root.mount( ComposablePlugin.from_module(plugin_module), name="fitbit", @@ -75,62 +95,27 @@ async def test_apply_registers_exact_runtime_sources_and_mobile_ui( servers, root.instance_token, )["fitbit"].definition - proactive = _freeze_plugin_proactive_components( - components, - root.instance_token, - {"fitbit": "fitbit:test"}, - ) mobile = ui_slots.freeze()["fitbit"] assert process.cwd == "." assert process.port_env == "FITBIT_MONITOR_PORT" + assert mcp.required_tools == ("fitbit_health_snapshot", "fitbit_sleep_report") assert mcp.candidate_env == {"FITBIT_BACKEND": "recording"} - assert [item.definition.name for item in proactive.sources.values()] == [ - "health_alerts", - "sleep_context", - ] assert mobile.descriptor.navigation_label == "健康状态" - assert not data_dir.exists() - await root.dispose() - - -@pytest.mark.asyncio -async def test_disabled_proactive_omits_sources(tmp_path: Path) -> None: - root = CompositionRoot("fitbit:disabled") - processes = PluginManagedProcesses(root.instance_token) - servers = PluginMcpServers(root.instance_token) - components = PluginProactiveComponents(root.instance_token) - ui_slots = PluginUiSlots() - await root.context.provide(MANAGED_PROCESSES, processes) - await root.context.provide(MCP_SERVERS, servers) - await root.context.provide(PROACTIVE_COMPONENTS, components) - await root.context.provide(UI_SLOTS, ui_slots) - await root.mount( - ComposablePlugin.from_module(plugin_module), - name="fitbit", - runtime=PluginRuntime( - plugin_id="fitbit", - plugin_dir=ROOT, - data_dir=tmp_path / "plugin-data", - workspace=tmp_path / "workspace", - config=FitbitConfig.model_validate({"proactive": {"enabled": False}}), - ), - ) - catalog = _freeze_plugin_proactive_components( - components, - root.instance_token, - {"fitbit": "fitbit:disabled"}, - ) - assert catalog.sources == {} + assert data_dir.joinpath("adapter.sqlite3").is_file() await root.dispose() def test_static_manifest_freezes_runtime_and_candidate_exclusions() -> None: manifest = load_static_plugin_manifest(ROOT) assert manifest.name == "fitbit" - assert manifest.version == "3.0.0" + assert manifest.version == "3.1.0" assert manifest.requirements == ("requirements.txt",) assert len(manifest.managed_processes) == 1 assert manifest.managed_processes[0].formal_port == 18765 + assert manifest.mcp_servers[0].required_tools == ( + "fitbit_health_snapshot", + "fitbit_sleep_report", + ) assert manifest.mcp_servers[0].candidate_env == (("FITBIT_BACKEND", "recording"),) assert "tokens.json" in manifest.exclude_data_paths assert "monitor.config.local.toml" in manifest.exclude_data_paths @@ -173,25 +158,14 @@ def get_current(self) -> dict[str, object]: def get_sleep_history(self) -> dict[str, object]: return history - monkeypatch.setattr(plugin_module, "FitbitMobileDashboardReader", Reader) - current_result = _mobile_ui_query( - "fitbit.current", - {}, - session_id=None, - turn_id=None, - ) - history_result = _mobile_ui_query( - "fitbit.sleep_history", - {}, - session_id=None, - turn_id=None, - ) - assert current_result == current - assert history_result == history + monkeypatch.setattr(mobile_reader, "FitbitMobileDashboardReader", Reader) + assert mobile_ui_query( + "fitbit.current", {}, session_id=None, turn_id=None + ) == current + assert mobile_ui_query( + "fitbit.sleep_history", {}, session_id=None, turn_id=None + ) == history with pytest.raises(ValueError, match="未知 fitbit 移动方法"): - _mobile_ui_query( - "fitbit.write", - {}, - session_id=None, - turn_id=None, + mobile_ui_query( + "fitbit.write", {}, session_id=None, turn_id=None ) diff --git a/tests/test_sleep_context.py b/tests/test_sleep_context.py new file mode 100644 index 0000000..77a1326 --- /dev/null +++ b/tests/test_sleep_context.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from agent.lifecycle.types import BeforeTurnCtx + +from src.sleep_context import FitbitAdapterStore, SleepContextAppender + + +NOW = datetime(2026, 8, 23, 8, tzinfo=UTC) + + +def _ctx(channel: str, now: datetime) -> BeforeTurnCtx: + return BeforeTurnCtx( + session_key=f"{channel}:fitbit", + channel=channel, + chat_id="fitbit", + content="continuation", + timestamp=now, + retrieved_memory_block="", + retrieval_trace_raw=None, + history_messages=(), + ) + + +def _store(path: Path) -> FitbitAdapterStore: + store = FitbitAdapterStore(path) + store.initialize(NOW) + store.commit_snapshot( + { + "state": "sleeping", + "prob": 0.92, + "prob_source": "model", + "data_lag_min": 3, + }, + observed_at=NOW, + expires_at=NOW + timedelta(minutes=10), + next_due=NOW + timedelta(minutes=5), + ) + return store + + +@pytest.mark.asyncio +async def test_wake_duty_and_fresh_sleep_hint_coexist(tmp_path: Path) -> None: + ctx = _ctx("wake", NOW + timedelta(minutes=2)) + ctx.extra_hints.append("Content duty: Fitbit health alert") + + await SleepContextAppender(_store(tmp_path / "adapter.sqlite3")).prepare(ctx) + + assert ctx.extra_hints[0] == "Content duty: Fitbit health alert" + assert "state=sleeping" in ctx.extra_hints[1] + assert ctx.abort is False + assert ctx.content == "continuation" + + +@pytest.mark.asyncio +async def test_passive_turn_has_no_sleep_hint(tmp_path: Path) -> None: + ctx = _ctx("mobile", NOW + timedelta(minutes=2)) + await SleepContextAppender(_store(tmp_path / "adapter.sqlite3")).prepare(ctx) + assert ctx.extra_hints == [] + + +@pytest.mark.asyncio +async def test_stale_sleep_cache_has_no_wake_hint(tmp_path: Path) -> None: + ctx = _ctx("wake", NOW + timedelta(minutes=11)) + await SleepContextAppender(_store(tmp_path / "adapter.sqlite3")).prepare(ctx) + assert ctx.extra_hints == [] From b51c247ca04a2229c6714d8cb328590f4f14d2ce Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 17:48:23 +0800 Subject: [PATCH 2/9] =?UTF-8?q?docs:=20=E8=AE=B0=E5=BD=95=20Fitbit=20Conte?= =?UTF-8?q?nt=20=E7=BB=84=E5=90=88=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/content-context-v3.md | 47 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 docs/content-context-v3.md diff --git a/docs/content-context-v3.md b/docs/content-context-v3.md new file mode 100644 index 0000000..ea8dc5a --- /dev/null +++ b/docs/content-context-v3.md @@ -0,0 +1,47 @@ +# Fitbit Content 与睡眠上下文 + +Fitbit v3.1 不再登记 proactive source，也不再用 MCP 工具搬运主动事件。插件只组合已有能力： + +```text +Fitbit 公网 + │ + ▼ +monitor（唯一采集 owner） + │ 本机 /api/agent，一次快照 + ▼ +FitbitContentRuntime + ├── health events ──▶ Content.submit ──▶ Wake / Delivery + │ │ + │ ▼ + │ Content.unsettled ◀── delivered + │ │ + │ ▼ + │ monitor desired-state ACK + │ │ + │ ▼ + │ Content.ack + │ + └── sleep ──▶ adapter.sqlite3 current cache + │ + ▼ + before Turn：仅 channel=wake 且未过期时追加 hint +``` + +## 不变量 + +1. `TIMERS` 只等待一次；插件在成功 tick 后持久化 `next_due` 并重新登记。 +2. monitor 仍是 Fitbit 公网采集的唯一 owner；adapter 只读本机 `/api/agent`。 +3. 健康事件先提交 Content，随后才原子更新插件私有的睡眠缓存与 `next_due`。 +4. 睡眠不进入 Content，也不因状态变化单独唤醒；它只给现有 Wake Turn 增加上下文。 +5. 外部 ACK 以“不再 pending”为成功事实。即使 ACK HTTP 返回后进程崩溃，下一轮也会确认事件已不在队列，再执行 `Content.ack`。 +6. candidate 可以启动隔离 monitor 并完成 MCP handshake，但不会收到 `RUNTIME_STARTED`，因此不会登记 Timer、轮询 `/api/agent`、ACK 或写正式数据。 + +## 私有持久状态 + +`adapter.sqlite3/source_state` 只有一行： + +- `next_due`：下一次本地 monitor 采集时间； +- `sleep_json`：最近一次睡眠判断； +- `sleep_observed_at` 与 `sleep_expires_at`：上下文新鲜度。 + +该文件不会复制 Content 的 item、delivery 或 ACK ledger。Content 继续独占这些权威事实。 From c2b7d27ea83f426e472d7534a5f183793faa2795 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 17:51:14 +0800 Subject: [PATCH 3/9] =?UTF-8?q?test:=20=E5=9B=BA=E5=AE=9A=20Fitbit=20?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E4=BF=9D=E7=95=99=E8=AF=AD=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/content-context-v3.md | 12 ++++++++++++ tests/test_content_adapter.py | 9 +++++++++ tests/test_sleep_context.py | 18 ++++++++++++++++++ 3 files changed, 39 insertions(+) diff --git a/docs/content-context-v3.md b/docs/content-context-v3.md index ea8dc5a..02fb4a2 100644 --- a/docs/content-context-v3.md +++ b/docs/content-context-v3.md @@ -45,3 +45,15 @@ FitbitContentRuntime - `sleep_observed_at` 与 `sleep_expires_at`：上下文新鲜度。 该文件不会复制 Content 的 item、delivery 或 ACK ledger。Content 继续独占这些权威事实。 + +## 保留与减少合同 + +| 对象 | owner | 正常增加 | 允许原位更新 | 逻辑失效 | 物理减少条件 | 恢复证据 | +|---|---|---|---|---|---|---| +| 已提交的健康事件、delivery 与 source ACK 完成事实 | Content | 新 item/revision 与 submission receipt 只追加 | 状态推进、selection、`settlement_ref` | `invalidated`、`abandoned`、`expired` 等 Content 状态 | 本插件没有物理减少协议 | Content row、原 payload、`settlement_ref` 与 `settled` 状态 | +| monitor pending / acked-id 队列投影 | monitor | 检测到事件时加入 pending；ACK 后加入 acked-id | monitor 现有检测状态与 pending 内容 | 事件过期或 ACK 后不再可投递 | ACK/过期可移除 pending；acked-id 超过既有固定上限可轮转 | 尚未提交前依赖 monitor 现有 state；提交后由 Content 成为唯一全量历史 owner | +| Session、Message 与 Turn | Core | 按 Core 协议追加 | Core 已批准的 metadata/terminal 状态 | 用户撤销或 Core 已定义的失效状态 | 只允许用户主动删除会话等 Core 已批准路径 | `sessions.db`、Message 与 Turn ledger | +| sleep current cache | Fitbit adapter | 首次建立 singleton | 每次成功 snapshot 覆盖 current payload 与时间 | `sleep_expires_at` 到期后不再注入 | 新 current snapshot 可以覆盖旧投影；没有历史裁切任务 | `sleep_observed_at`、`sleep_expires_at` 与当前 JSON | +| 纯诊断日志 | 各产生日志的 owner | 本实现不新增持久诊断日志 | 不适用 | 不适用 | 未来若新增，只能按固定数量轮转 | 固定轮转配置与当前日志文件 | + +因此，“外部 ACK 成功”不会删除已发生的健康事实：monitor 的 pending 项可以消失，但 Content 的 settled row 仍保留原 payload 与 settlement receipt。adapter 不另造第二份历史或 ACK ledger。 diff --git a/tests/test_content_adapter.py b/tests/test_content_adapter.py index 5addf9c..98cef06 100644 --- a/tests/test_content_adapter.py +++ b/tests/test_content_adapter.py @@ -4,6 +4,7 @@ import json import os import signal +import sqlite3 import subprocess import sys import threading @@ -339,3 +340,11 @@ def test_ack_response_sigkill_recovers_by_desired_state( ) runtime._drain_unsettled() assert ContentStore(content_path).state_counts() == {"settled": 1} + with sqlite3.connect(content_path) as connection: + row = connection.execute( + "SELECT status, settlement_ref, payload_json FROM items" + ).fetchone() + assert row is not None + assert row[0] == "settled" + assert row[1] == "delivery:fitbit:1" + assert json.loads(row[2])["upstream_event_id"] == "fitbit:event-1" diff --git a/tests/test_sleep_context.py b/tests/test_sleep_context.py index 77a1326..bf99ab0 100644 --- a/tests/test_sleep_context.py +++ b/tests/test_sleep_context.py @@ -42,6 +42,24 @@ def _store(path: Path) -> FitbitAdapterStore: return store +def test_sleep_cache_overwrites_one_current_projection(tmp_path: Path) -> None: + store = _store(tmp_path / "adapter.sqlite3") + store.commit_snapshot( + { + "state": "awake", + "prob": 0.05, + "prob_source": "model", + "data_lag_min": 1, + }, + observed_at=NOW + timedelta(minutes=5), + expires_at=NOW + timedelta(minutes=15), + next_due=NOW + timedelta(minutes=10), + ) + + current = store.current_sleep(NOW + timedelta(minutes=6)) + assert current is not None and current["state"] == "awake" + + @pytest.mark.asyncio async def test_wake_duty_and_fresh_sleep_hint_coexist(tmp_path: Path) -> None: ctx = _ctx("wake", NOW + timedelta(minutes=2)) From 509d16b82c5df5c9f91309f181ebe8e2191f7a62 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 17:53:22 +0800 Subject: [PATCH 4/9] =?UTF-8?q?test:=20=E8=AF=81=E6=98=8E=20Fitbit=20candi?= =?UTF-8?q?date=20=E9=9A=94=E7=A6=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_manager_integration.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py index c206bef..7dba436 100644 --- a/tests/test_manager_integration.py +++ b/tests/test_manager_integration.py @@ -1,10 +1,12 @@ from __future__ import annotations +import hashlib import shutil import sys from pathlib import Path import pytest +from agent.plugin_composition import TIMERS from agent.plugins.generation import PluginGeneration from agent.plugins.manager import PluginManager from agent.plugins.snapshot import RuntimeSnapshot @@ -15,6 +17,14 @@ ROOT = Path(__file__).resolve().parents[1] +def _tree_digest(root: Path) -> str: + digest = hashlib.sha256() + for path in sorted(item for item in root.rglob("*") if item.is_file()): + digest.update(path.relative_to(root).as_posix().encode()) + digest.update(path.read_bytes()) + return digest.hexdigest() + + def _stage_plugin(tmp_path: Path) -> Path: """复制可执行 artifact，并复用当前测试解释器的依赖环境。""" @@ -78,6 +88,8 @@ async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( stable_route = stable_runtime.mcp.server("fitbit").route() assert stable_route.mode == "formal" await stable_route.aclose() + formal_data = tmp_path / "workspace/plugin-data/fitbit-builtin" + formal_digest = _tree_digest(formal_data) # 2. 新版本先在隔离 Root 中验证，再重建 formal Root。 for relative in ("plugin.py", "akashic.plugin.toml"): @@ -92,6 +104,9 @@ async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( validation_root = candidate.validation_workspace.parent candidate_snapshot = candidate.runtime_snapshot assert candidate_snapshot.composition_root is not None + assert candidate_snapshot.composition_root.context.require(TIMERS).formal is False + assert candidate.validation_workspace != tmp_path / "workspace" + assert _tree_digest(formal_data) == formal_digest original_invariants = manager._post_publish_invariants # pyright: ignore[reportPrivateUsage] candidate_checked = False From 455220401d00486006b85f23aeff4b2f7932acb2 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 18:21:08 +0800 Subject: [PATCH 5/9] fix(fitbit): close content retry and log bounds --- .github/workflows/plugin-api-v3.yml | 9 ++- monitor/runtime_env.py | 75 ++++++++++++++++++ monitor/server.py | 4 +- plugin.py | 3 +- src/content_adapter.py | 87 ++++++++++++-------- tests/test_content_adapter.py | 118 +++++++++++++++++++++++++++- tests/test_runtime_env.py | 31 +++++++- 7 files changed, 283 insertions(+), 44 deletions(-) diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index 36e67fb..932c9bb 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: 78e50d4dfb3f4348fff37d55d9c9bdd0e002164d + ref: 9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5 path: .akashic-core - uses: actions/setup-python@v5 with: @@ -70,11 +70,12 @@ jobs: PYTHONPATH: .akashic-core run: >- .venv/bin/pyright --pythonpath .venv/bin/python --level error - plugin.py dashboard.py src/mcp_bridge.py + plugin.py dashboard.py src/content_adapter.py src/mcp_bridge.py + monitor/runtime_env.py scripts/migrate_v2_data.py - tests/test_plugin.py tests/test_dashboard.py + tests/test_plugin.py tests/test_content_adapter.py tests/test_dashboard.py tests/test_manager_integration.py tests/test_mcp_v3_runtime.py - tests/test_migrate_v2_data.py + tests/test_migrate_v2_data.py tests/test_runtime_env.py - name: Compile Python sources run: python -m compileall -q plugin.py dashboard.py src monitor scripts tests - name: Check diff formatting diff --git a/monitor/runtime_env.py b/monitor/runtime_env.py index 62e8c29..8a31a13 100644 --- a/monitor/runtime_env.py +++ b/monitor/runtime_env.py @@ -1,6 +1,81 @@ from __future__ import annotations import os +from pathlib import Path +from threading import Lock +from typing import TextIO + + +RUNTIME_LOG_MAX_BYTES = 1_048_576 +RUNTIME_LOG_BACKUPS = 3 + + +class RotatingTextLog: + """Keep one diagnostic text log within fixed byte and generation limits.""" + + def __init__( + self, + path: Path, + *, + max_bytes: int = RUNTIME_LOG_MAX_BYTES, + backups: int = RUNTIME_LOG_BACKUPS, + ) -> None: + if max_bytes <= 0 or backups < 0: + raise ValueError("runtime log rotation limits 无效") + self.path = path + self._max_bytes = max_bytes + self._backups = backups + self._lock = Lock() + path.parent.mkdir(parents=True, exist_ok=True) + self._stream = self._open() + self._size = path.stat().st_size if path.exists() else 0 + if self._size >= self._max_bytes: + self._rotate() + + def write(self, text: str) -> int: + input_length = len(text) + data = text.encode("utf-8") + with self._lock: + if self._size and self._size + len(data) > self._max_bytes: + self._rotate() + if len(data) > self._max_bytes: + data = data[-self._max_bytes :] + while data and (data[0] & 0xC0) == 0x80: + data = data[1:] + text = data.decode("utf-8") + self._stream.write(text) + self._size += len(data) + return input_length + + def flush(self) -> None: + with self._lock: + self._stream.flush() + + def close(self) -> None: + with self._lock: + self._stream.close() + + def _rotate(self) -> None: + self._stream.close() + if self._backups > 0: + oldest = self.path.with_name(f"{self.path.name}.{self._backups}") + oldest.unlink(missing_ok=True) + for index in range(self._backups - 1, 0, -1): + source = self.path.with_name(f"{self.path.name}.{index}") + if source.exists(): + os.replace( + source, + self.path.with_name(f"{self.path.name}.{index + 1}"), + ) + if self.path.exists(): + os.replace(self.path, self.path.with_name(f"{self.path.name}.1")) + else: + self.path.unlink(missing_ok=True) + self._stream = self._open() + self._size = 0 + + def _open(self) -> TextIO: + return self.path.open("a", encoding="utf-8", buffering=1) def resolve_server_port(configured_port: int) -> int: diff --git a/monitor/server.py b/monitor/server.py index 3b98c49..8611f87 100644 --- a/monitor/server.py +++ b/monitor/server.py @@ -16,7 +16,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse import uvicorn -from runtime_env import resolve_server_port +from runtime_env import RotatingTextLog, resolve_server_port import sleep_model import retrain_guard import build_sleep_diff_report @@ -86,7 +86,7 @@ def _install_runtime_log_mirror() -> None: return try: RUNTIME_LOG_FILE.parent.mkdir(parents=True, exist_ok=True) - log_f = RUNTIME_LOG_FILE.open("a", encoding="utf-8", buffering=1) + log_f = RotatingTextLog(RUNTIME_LOG_FILE) except Exception: return log_f.write( diff --git a/plugin.py b/plugin.py index 0660dd8..bd304b5 100644 --- a/plugin.py +++ b/plugin.py @@ -106,9 +106,10 @@ def setup() -> object: return runtime.close _ = await ctx.effect(setup, label="fitbit-content-runtime") + poll_health = await ctx.health("fitbit-content-poll") async def start(_event: object) -> None: - await runtime.start() + await runtime.start(ctx, poll_health) async def stop(_event: object) -> None: await runtime.close() diff --git a/src/content_adapter.py b/src/content_adapter.py index 47b27ce..e5828b1 100644 --- a/src/content_adapter.py +++ b/src/content_adapter.py @@ -11,7 +11,7 @@ import requests from agent.control.timer import TimerHandle, TimerStatus -from agent.plugin_composition import PluginTimers +from agent.plugin_composition import Context, HealthHandle, PluginTimers from src.sleep_context import FitbitAdapterStore @@ -25,6 +25,10 @@ def unsettled(self, limit: int = 100) -> tuple[Mapping[str, object], ...]: ... def ack(self, settlement_ref: str) -> Mapping[str, object]: ... +class FitbitMonitorTransientError(RuntimeError): + """表示 monitor HTTP/IO 边界可在下一次 Timer 重试。""" + + class FitbitMonitorClient: """读取 monitor 快照，并以目标状态完成 ACK。""" @@ -86,14 +90,15 @@ def __init__( self._task: asyncio.Task[None] | None = None self._closed = False - async def start(self) -> None: - """恢复来源 deadline，并登记唯一正式 Timer。""" + async def start(self, ctx: Context, health: HealthHandle) -> None: + """恢复来源 deadline，并启动唯一 Fiber-owned 采集循环。""" if self._closed: raise RuntimeError("Fitbit Content runtime 已关闭") - if self._handle is not None: - return - self._arm(self._store.next_due()) + if self._task is None: + self._task = await ctx.spawn( + self._run(ctx, health), name="fitbit-content-poll" + ) async def close(self) -> None: """取消自有等待，不改写 Content 或来源事实。""" @@ -110,29 +115,30 @@ async def close(self) -> None: if handle is not None: await handle.cleanup() - 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_tick_rearm(handle), name="fitbit:content-poll" - ) - - async def _wait_tick_rearm(self, handle: TimerHandle) -> None: - """Timer 到点后执行一次采集，再从持久状态重新登记。""" - - try: - receipt = await handle.result() - if receipt.status is TimerStatus.CANCELLED or self._closed: - return - await asyncio.to_thread(self.tick) - finally: - self._handle = None - self._task = None - await handle.cleanup() - if not self._closed: - self._arm(self._store.next_due()) + async def _run(self, ctx: Context, health: HealthHandle) -> None: + """轮询、显式记录临时失败，并只在可恢复结果后重臂。""" + + deadline = self._store.next_due() + while not self._closed: + handle = self._timers.schedule(deadline) + self._handle = handle + try: + receipt = await handle.result() + if receipt.status is TimerStatus.CANCELLED or self._closed: + return + try: + await asyncio.to_thread(self.tick) + except FitbitMonitorTransientError as error: + reason = f"{type(error).__name__}: {error}" + health.degrade(reason) + _ = ctx.report_incident("fitbit_content_retry", reason) + deadline = _aware(self._now()) + self._poll_interval + else: + health.recover() + deadline = self._store.next_due() + finally: + self._handle = None + await handle.cleanup() def tick(self) -> None: """先结算历史投递，再发布当前 monitor 快照。""" @@ -141,7 +147,7 @@ def tick(self) -> None: self._drain_unsettled() # 2. 只拉取一次，独立归一化健康事件，并优先提交 Content - snapshot = self._monitor.snapshot() + snapshot = self._monitor_snapshot() items = normalize_health_events(snapshot) batch_id = stable_batch_id(items) _ = self._content.submit(batch_id, items) @@ -163,7 +169,7 @@ def _drain_unsettled(self) -> None: payload = _mapping(row, "payload") event_id = _string(payload, "upstream_event_id") settlement_ref = _string(row, "settlement_ref") - self._monitor.ensure_not_pending(event_id) + self._ensure_not_pending(event_id) if self._after_provider_ack is not None: self._after_provider_ack() settled = self._content.ack(settlement_ref) @@ -174,6 +180,18 @@ def _drain_unsettled(self) -> None: if len(rows) < 100: return + def _monitor_snapshot(self) -> Mapping[str, object]: + try: + return self._monitor.snapshot() + except (OSError, requests.RequestException) as error: + raise FitbitMonitorTransientError(str(error)) from error + + def _ensure_not_pending(self, event_id: str) -> None: + try: + self._monitor.ensure_not_pending(event_id) + except (OSError, requests.RequestException) as error: + raise FitbitMonitorTransientError(str(error)) from error + def normalize_health_events( snapshot: Mapping[str, object], @@ -209,7 +227,12 @@ def normalize_health_events( "requires_ack": True, } ) - return tuple(items) + return tuple( + sorted( + items, + key=lambda item: (str(item["item_id"]), str(item["revision"])), + ) + ) def normalize_sleep(snapshot: Mapping[str, object]) -> Mapping[str, object]: diff --git a/tests/test_content_adapter.py b/tests/test_content_adapter.py index 98cef06..72b3884 100644 --- a/tests/test_content_adapter.py +++ b/tests/test_content_adapter.py @@ -17,9 +17,9 @@ import pytest from agent.control.timer import AsyncioOneShotTimer -from agent.plugin_composition import PluginTimers +from agent.plugin_composition import CompositionRoot, PluginTimers from plugins.content import plugin as content_plugin -from plugins.content.store import ContentStore +from plugins.content.store import ContentIdentityConflict, ContentStore from src.content_adapter import ( BoundContentSource, @@ -168,6 +168,49 @@ def test_batch_identity_does_not_depend_on_monitor_queue_order() -> None: assert stable_batch_id((first, second)) == stable_batch_id((second, first)) +def test_reordered_monitor_batch_replays_identical_content_sequence(tmp_path) -> None: + second_event = { + "id": "fitbit:event-2", + "type": "spo2_low", + "message": "血氧偏低", + "severity": "high", + "created_at": "2026-08-23 08:01", + "suggested_tone": "先确认状态", + "metrics": {"spo2": 89}, + } + first_snapshot = { + **SNAPSHOT, + "health_events": [ + cast(list[Mapping[str, object]], SNAPSHOT["health_events"])[0], + second_event, + ], + } + reordered = { + **first_snapshot, + "health_events": list(reversed(first_snapshot["health_events"])), + } + first_items = normalize_health_events(first_snapshot) + second_items = normalize_health_events(reordered) + assert second_items == first_items + + control_store = ContentStore(tmp_path / "conflict-control.sqlite3") + control_store.initialize() + control = _bound(control_store) + batch_id = stable_batch_id(first_items) + _ = control.submit(batch_id, first_items) + with pytest.raises(ContentIdentityConflict): + _ = control.submit(batch_id, tuple(reversed(first_items))) + + content_store = ContentStore(tmp_path / "content.sqlite3") + content_store.initialize() + bound = _bound(content_store) + first_receipt = bound.submit(batch_id, first_items) + replay_receipt = bound.submit(batch_id, second_items) + + assert replay_receipt == first_receipt + assert content_store.state_counts() == {"pending": 2} + + @pytest.mark.asyncio async def test_reload_has_only_one_real_timer_wait(tmp_path: Path) -> None: active = 0 @@ -202,19 +245,86 @@ def make_runtime() -> FitbitContentRuntime: ) first = make_runtime() - await first.start() + first_root = CompositionRoot("fitbit-reload:first") + first_health = await first_root.context.health("fitbit-content-poll") + await first.start(first_root.context, first_health) await entered.wait() assert active == 1 await first.close() assert active == 0 + await first_root.dispose() entered.clear() second = make_runtime() - await second.start() + second_root = CompositionRoot("fitbit-reload:second") + second_health = await second_root.context.health("fitbit-content-poll") + await second.start(second_root.context, second_health) await entered.wait() assert active == 1 assert maximum == 1 await second.close() + await second_root.dispose() + + +@pytest.mark.asyncio +async def test_transient_oserror_rearms_and_recovers_without_partial_state( + tmp_path: Path, +) -> None: + sleeper_calls = 0 + third_wait = asyncio.Event() + + async def sleeper(_delay: float) -> None: + nonlocal sleeper_calls + sleeper_calls += 1 + if sleeper_calls <= 2: + return + third_wait.set() + await asyncio.Future() + + class FailOnceMonitor(RecordingMonitor): + def __init__(self) -> None: + super().__init__(SNAPSHOT) + self.snapshot_calls = 0 + + def snapshot(self) -> Mapping[str, object]: + self.snapshot_calls += 1 + if self.snapshot_calls == 1: + raise OSError("temporary monitor read failure") + return super().snapshot() + + content_store = ContentStore(tmp_path / "content.sqlite3") + content_store.initialize() + adapter = FitbitAdapterStore(tmp_path / "adapter.sqlite3") + adapter.initialize(NOW) + monitor = FailOnceMonitor() + runtime = FitbitContentRuntime( + adapter, + PluginTimers(AsyncioOneShotTimer(clock=lambda: NOW, sleeper=sleeper)), + _bound(content_store), + cast(FitbitMonitorClient, monitor), + poll_interval=timedelta(minutes=5), + sleep_ttl=timedelta(minutes=10), + now=lambda: NOW, + ) + root = CompositionRoot("fitbit-transient-retry") + health = await root.context.health("fitbit-content-poll") + + await runtime.start(root.context, health) + await third_wait.wait() + + assert sleeper_calls == 3 + assert monitor.snapshot_calls == 2 + assert health.healthy + assert content_store.state_counts() == {"pending": 1} + assert adapter.next_due() == NOW + timedelta(minutes=5) + assert monitor.acknowledged == [] + assert any( + incident.kind == "fitbit_content_retry" + and "temporary monitor read failure" in incident.message + for incident in root.receipt().incidents + ) + await runtime.close() + await root.dispose() class _MonitorState: diff --git a/tests/test_runtime_env.py b/tests/test_runtime_env.py index f0e04d8..004d89d 100644 --- a/tests/test_runtime_env.py +++ b/tests/test_runtime_env.py @@ -1,8 +1,10 @@ from __future__ import annotations +from pathlib import Path + import pytest -from monitor.runtime_env import resolve_server_port +from monitor.runtime_env import RotatingTextLog, resolve_server_port def test_server_port_uses_config_without_runtime_override( @@ -30,3 +32,30 @@ def test_server_port_rejects_invalid_runtime_override( with pytest.raises(ValueError): resolve_server_port(18765) + + +def test_runtime_log_rotates_without_touching_authoritative_state(tmp_path: Path) -> None: + log_path = tmp_path / "monitor.runtime.log" + protected = { + tmp_path / "stat_events.json": b"pending-events", + tmp_path / "content.sqlite3": b"content-fact", + tmp_path / "sessions.db": b"session-fact", + } + for path, payload in protected.items(): + path.write_bytes(payload) + + log = RotatingTextLog(log_path, max_bytes=12, backups=2) + for payload in ("A" * 10, "B" * 10, "C" * 10, "D" * 10): + _ = log.write(payload) + log.flush() + log.close() + + assert log_path.read_text(encoding="utf-8") == "D" * 10 + assert log_path.with_name("monitor.runtime.log.1").read_text() == "C" * 10 + assert log_path.with_name("monitor.runtime.log.2").read_text() == "B" * 10 + assert not log_path.with_name("monitor.runtime.log.3").exists() + assert sum( + path.stat().st_size + for path in tmp_path.glob("monitor.runtime.log*") + ) <= 36 + assert {path: path.read_bytes() for path in protected} == protected From a7ba644d87712727c0e5e6c8047f0783a1d65dc1 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 23 Aug 2026 18:34:39 +0800 Subject: [PATCH 6/9] fix(fitbit): rotate inherited runtime log streams --- monitor/server.py | 63 +++++++++++++++++++++++++++++---------- tests/test_runtime_env.py | 55 +++++++++++++++++++++++++++++++++- 2 files changed, 102 insertions(+), 16 deletions(-) diff --git a/monitor/server.py b/monitor/server.py index 8611f87..76bdb4f 100644 --- a/monitor/server.py +++ b/monitor/server.py @@ -10,6 +10,7 @@ from datetime import datetime, date, timedelta from threading import Thread, Lock, Event from pathlib import Path +from typing import TextIO import tomllib import requests as req from fastapi import FastAPI, WebSocket, WebSocketDisconnect @@ -64,7 +65,7 @@ def writable(self): return True -def _stream_points_to(path: Path, stream) -> bool: +def _stream_points_to(path: Path, stream: TextIO) -> bool: try: stream_fd = stream.fileno() stream_stat = os.fstat(stream_fd) @@ -77,26 +78,58 @@ def _stream_points_to(path: Path, stream) -> bool: ) +def _runtime_log_stream_fds(path: Path, streams: tuple[TextIO, ...]) -> set[int]: + """Collect inherited descriptors that write directly to the runtime log.""" + + redirected_fds: set[int] = set() + for stream in streams: + if not _stream_points_to(path, stream): + continue + stream.flush() + redirected_fds.add(stream.fileno()) + return redirected_fds + + +def _detach_stream_fds(stream_fds: set[int]) -> None: + """Detach direct writers so all later text passes through the rotator.""" + + for stream_fd in stream_fds: + devnull_fd = os.open(os.devnull, os.O_WRONLY) + try: + os.dup2(devnull_fd, stream_fd) + finally: + os.close(devnull_fd) + + def _install_runtime_log_mirror() -> None: - """ - 保证无论通过何种方式启动，stdout/stderr 都会写入 monitor.runtime.log。 - 若上层已重定向到同一个文件，则不重复包裹，避免双写。 - """ - if isinstance(sys.stdout, _TeeTextIO) or isinstance(sys.stderr, _TeeTextIO): - return - try: - RUNTIME_LOG_FILE.parent.mkdir(parents=True, exist_ok=True) - log_f = RotatingTextLog(RUNTIME_LOG_FILE) - except Exception: + """Route stdout/stderr through the bounded runtime diagnostic log.""" + + # 1. 重复导入保持同一份 rotator；部分安装属于内部合同错误 + installed = ( + isinstance(sys.stdout, _TeeTextIO), + isinstance(sys.stderr, _TeeTextIO), + ) + if installed == (True, True): return + if any(installed): + raise RuntimeError("runtime log mirror 处于部分安装状态") + + # 2. 先记住旧 inode 的直接写入者，再建立 rotator 并解除它们 + RUNTIME_LOG_FILE.parent.mkdir(parents=True, exist_ok=True) + redirected_fds = _runtime_log_stream_fds( + RUNTIME_LOG_FILE, + (sys.stdout, sys.stderr), + ) + log_f = RotatingTextLog(RUNTIME_LOG_FILE) + _detach_stream_fds(redirected_fds) log_f.write( f"\n===== fitbit-monitor start {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} " f"pid={os.getpid()} =====\n" ) - if not _stream_points_to(RUNTIME_LOG_FILE, sys.stdout): - sys.stdout = _TeeTextIO(sys.stdout, log_f) - if not _stream_points_to(RUNTIME_LOG_FILE, sys.stderr): - sys.stderr = _TeeTextIO(sys.stderr, log_f) + + # 3. 两个流共享唯一 rotator；原始流继续承担终端或父进程可观察性 + sys.stdout = _TeeTextIO(sys.stdout, log_f) + sys.stderr = _TeeTextIO(sys.stderr, log_f) atexit.register(log_f.close) diff --git a/tests/test_runtime_env.py b/tests/test_runtime_env.py index 004d89d..82309f0 100644 --- a/tests/test_runtime_env.py +++ b/tests/test_runtime_env.py @@ -1,10 +1,22 @@ from __future__ import annotations +import os from pathlib import Path +import subprocess +import sys import pytest -from monitor.runtime_env import RotatingTextLog, resolve_server_port +from monitor.runtime_env import ( + RUNTIME_LOG_BACKUPS, + RUNTIME_LOG_MAX_BYTES, + RotatingTextLog, + resolve_server_port, +) + + +ROOT = Path(__file__).resolve().parents[1] +MONITOR_DIR = ROOT / "monitor" def test_server_port_uses_config_without_runtime_override( @@ -59,3 +71,44 @@ def test_runtime_log_rotates_without_touching_authoritative_state(tmp_path: Path for path in tmp_path.glob("monitor.runtime.log*") ) <= 36 assert {path: path.read_bytes() for path in protected} == protected + + +def test_preredirected_runtime_log_uses_bounded_rotator(tmp_path: Path) -> None: + runtime_log = tmp_path / "monitor.runtime.log" + protected = { + tmp_path / "stat_events.json": b'{"events": [], "last_event_time": {}}', + tmp_path / "stat_events_v2.json": b'{"pending": [], "acked_ids": []}', + tmp_path / "content.sqlite3": b"content-fact", + tmp_path / "sessions.db": b"session-fact", + } + for path, payload in protected.items(): + path.write_bytes(payload) + runtime_log.write_bytes(b"P" * RUNTIME_LOG_MAX_BYTES) + + script = """ +import sys +sys.path.insert(0, sys.argv[1]) +import server +for index in range(6): + print(f"runtime-line-{index}:" + "X" * 700_000, flush=True) +""" + env = os.environ.copy() + env["AKA_PLUGIN_DATA_DIR"] = str(tmp_path) + with runtime_log.open("ab", buffering=0) as redirected: + subprocess.run( + [sys.executable, "-c", script, str(MONITOR_DIR)], + check=True, + env=env, + stdout=redirected, + stderr=redirected, + ) + + backups = sorted(tmp_path.glob("monitor.runtime.log.*")) + assert runtime_log.stat().st_size <= RUNTIME_LOG_MAX_BYTES + assert backups + assert len(backups) <= RUNTIME_LOG_BACKUPS + assert all(path.stat().st_size <= RUNTIME_LOG_MAX_BYTES for path in backups) + assert not runtime_log.with_name( + f"monitor.runtime.log.{RUNTIME_LOG_BACKUPS + 1}" + ).exists() + assert {path: path.read_bytes() for path in protected} == protected From 93d69547b77904c397294c43b35e21f6c8568acc Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Mon, 24 Aug 2026 02:35:52 +0800 Subject: [PATCH 7/9] test(fitbit): require fixture runtime python --- tests/test_manager_integration.py | 32 +++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py index 7dba436..e275171 100644 --- a/tests/test_manager_integration.py +++ b/tests/test_manager_integration.py @@ -1,8 +1,8 @@ from __future__ import annotations import hashlib +import os import shutil -import sys from pathlib import Path import pytest @@ -26,7 +26,7 @@ def _tree_digest(root: Path) -> str: def _stage_plugin(tmp_path: Path) -> Path: - """复制可执行 artifact，并复用当前测试解释器的依赖环境。""" + """复制可执行 artifact，并挂载调用方明确选择的依赖环境。""" source = tmp_path / "plugins" / "fitbit" shutil.copytree( @@ -41,16 +41,36 @@ def _stage_plugin(tmp_path: Path) -> Path: "node_modules", ), ) - (source / ".venv").symlink_to( - Path(sys.executable).parent.parent, - target_is_directory=True, - ) + fixture_python = Path(os.environ["AKASHIC_PLUGIN_FIXTURE_PYTHON"]) + (source / ".venv").symlink_to(fixture_python.parent.parent, target_is_directory=True) content_source = Path(content_plugin.__file__).resolve().parent content_target = source.parent / "content" shutil.copytree(content_source, content_target) return source +def test_stage_plugin_uses_explicit_fixture_python( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + artifact_python = tmp_path / "artifact" / ".venv" / "bin" / "python" + monkeypatch.setenv("AKASHIC_PLUGIN_FIXTURE_PYTHON", str(artifact_python)) + + plugin_root = _stage_plugin(tmp_path / "stage") + + assert (plugin_root / ".venv").readlink() == artifact_python.parent.parent + + +def test_stage_plugin_requires_explicit_fixture_python( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("AKASHIC_PLUGIN_FIXTURE_PYTHON", raising=False) + + with pytest.raises(KeyError, match="AKASHIC_PLUGIN_FIXTURE_PYTHON"): + _stage_plugin(tmp_path) + + @pytest.mark.asyncio async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( tmp_path: Path, From b81978474a630a13cb8d9bd6f2fa17afb28b2d11 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Mon, 24 Aug 2026 02:42:35 +0800 Subject: [PATCH 8/9] test(fitbit): bind fixture runtime before staging --- .github/workflows/plugin-api-v3.yml | 1 + tests/test_manager_integration.py | 18 +++++++++++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index 932c9bb..e535709 100644 --- a/.github/workflows/plugin-api-v3.yml +++ b/.github/workflows/plugin-api-v3.yml @@ -55,6 +55,7 @@ jobs: - name: Verify Fitbit v3 composition env: AKASHIC_AGENT_ROOT: .akashic-core + AKASHIC_PLUGIN_FIXTURE_PYTHON: ${{ github.workspace }}/.venv/bin/python PYTHONPATH: .akashic-core run: .venv/bin/python -m pytest -q tests/ - uses: actions/setup-node@v4 diff --git a/tests/test_manager_integration.py b/tests/test_manager_integration.py index e275171..e9ab705 100644 --- a/tests/test_manager_integration.py +++ b/tests/test_manager_integration.py @@ -28,6 +28,7 @@ def _tree_digest(root: Path) -> str: def _stage_plugin(tmp_path: Path) -> Path: """复制可执行 artifact，并挂载调用方明确选择的依赖环境。""" + fixture_python = Path(os.environ["AKASHIC_PLUGIN_FIXTURE_PYTHON"]) source = tmp_path / "plugins" / "fitbit" shutil.copytree( ROOT, @@ -41,7 +42,6 @@ def _stage_plugin(tmp_path: Path) -> Path: "node_modules", ), ) - fixture_python = Path(os.environ["AKASHIC_PLUGIN_FIXTURE_PYTHON"]) (source / ".venv").symlink_to(fixture_python.parent.parent, target_is_directory=True) content_source = Path(content_plugin.__file__).resolve().parent content_target = source.parent / "content" @@ -70,6 +70,22 @@ def test_stage_plugin_requires_explicit_fixture_python( with pytest.raises(KeyError, match="AKASHIC_PLUGIN_FIXTURE_PYTHON"): _stage_plugin(tmp_path) + assert not (tmp_path / "plugins").exists() + + +def test_ci_creates_and_exports_absolute_fixture_python_before_pytest() -> None: + workflow = (ROOT / ".github/workflows/plugin-api-v3.yml").read_text( + encoding="utf-8" + ) + + create_runtime = workflow.index("python -m venv .venv") + export_runtime = workflow.index( + "AKASHIC_PLUGIN_FIXTURE_PYTHON: ${{ github.workspace }}/.venv/bin/python" + ) + run_pytest = workflow.index("run: .venv/bin/python -m pytest -q tests/") + + assert create_runtime < export_runtime < run_pytest + @pytest.mark.asyncio async def test_manager_rebuilds_fitbit_runtime_on_exact_formal_root( From be6e109d56934f50f4a4537db7d41c8a9a3ae829 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Mon, 24 Aug 2026 20:46:58 +0800 Subject: [PATCH 9/9] fix: load Fitbit runtime as a package --- plugin.py | 6 +++--- src/content_adapter.py | 2 +- tests/conftest.py | 6 ++++++ tests/test_plugin.py | 4 ++-- 4 files changed, 12 insertions(+), 6 deletions(-) diff --git a/plugin.py b/plugin.py index bd304b5..c422dd4 100644 --- a/plugin.py +++ b/plugin.py @@ -21,13 +21,13 @@ MobileUiNavigation, ServiceKey, ) -from src.content_adapter import ( +from .src.content_adapter import ( BoundContentSource, FitbitContentRuntime, FitbitMonitorClient, ) -from src.mobile_reader import mobile_ui_query -from src.sleep_context import FitbitAdapterStore, SleepContextAppender +from .src.mobile_reader import mobile_ui_query +from .src.sleep_context import FitbitAdapterStore, SleepContextAppender class ContentSourceServices(Protocol): diff --git a/src/content_adapter.py b/src/content_adapter.py index e5828b1..fdff540 100644 --- a/src/content_adapter.py +++ b/src/content_adapter.py @@ -12,7 +12,7 @@ from agent.control.timer import TimerHandle, TimerStatus from agent.plugin_composition import Context, HealthHandle, PluginTimers -from src.sleep_context import FitbitAdapterStore +from .sleep_context import FitbitAdapterStore class BoundContentSource(Protocol): diff --git a/tests/conftest.py b/tests/conftest.py index 3eef713..d13f25f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,7 @@ import os import sys from pathlib import Path +from types import ModuleType repo_root = Path(__file__).resolve().parents[1] @@ -13,3 +14,8 @@ for path in (repo_root, agent_root): if str(path) not in sys.path: sys.path.insert(0, str(path)) + +package = ModuleType("fitbit_test_plugin") +package.__path__ = [str(repo_root)] +package.__package__ = "fitbit_test_plugin" +sys.modules["fitbit_test_plugin"] = package diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 724842e..11459aa 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -27,8 +27,8 @@ from agent.plugins.static_manifest import load_static_plugin_manifest from plugins.content import plugin as content_plugin -import plugin as plugin_module -from plugin import FitbitConfig +from fitbit_test_plugin import plugin as plugin_module # pyright: ignore[reportMissingImports] +from fitbit_test_plugin.plugin import FitbitConfig # pyright: ignore[reportMissingImports] from src import mobile_reader from src.mobile_reader import mobile_ui_query