[codex] Publish Fitbit alerts and context through EventMail

.github/workflows/plugin-api-v3.yml

@@ -36,7 +36,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           repository: kachofugetsu09/akashic-agent
-          ref: 9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5
+          ref: 39cbdcefc155aaf6c41deafd7754a37e6126c23c
           path: .akashic-core
       - uses: actions/setup-python@v5
         with:

akashic.plugin.toml

@@ -1,6 +1,6 @@
 schema_version = 1
 name = "fitbit"
-version = "3.1.0"
+version = "3.2.1"
 api_version = 3
 entrypoint = "plugin.py"
 

plugin.py

@@ -1,11 +1,8 @@
 from __future__ import annotations
 
 from datetime import UTC, datetime, timedelta
-from typing import Protocol
-
 from pydantic import BaseModel, ConfigDict, Field
 
-from agent.lifecycle.composition import CONTEXT_PREPARED_EVENT
 from agent.plugin_composition import (
     MANAGED_PROCESSES,
     MCP_SERVERS,
@@ -19,22 +16,14 @@
     McpServerDefinition,
     MobileUiDefinition,
     MobileUiNavigation,
-    ServiceKey,
 )
 from .src.content_adapter import (
-    BoundContentSource,
-    FitbitContentRuntime,
+    FitbitWakeRuntime,
     FitbitMonitorClient,
 )
+from .src.eventmail import EVENTMAIL_ALERT_SOURCE, EVENTMAIL_CONTEXT_SOURCE
 from .src.mobile_reader import mobile_ui_query
-from .src.sleep_context import FitbitAdapterStore, SleepContextAppender
-
-
-class ContentSourceServices(Protocol):
-    def bind(self, source_id: str) -> BoundContentSource: ...
-
-
-CONTENT_SOURCE = ServiceKey[ContentSourceServices]("content.source.v1")
+from .src.sleep_context import FitbitAdapterStore
 
 
 class FitbitContentConfig(BaseModel):
@@ -52,15 +41,20 @@ class FitbitConfig(BaseModel):
 
 api_version = 3
 name = "fitbit"
-version = "3.1.0"
-desc = "Fitbit health monitor, Content source, and sleep context"
+version = "3.2.1"
+desc = "Fitbit health Alert and sleep Context source"
 Config = FitbitConfig
-inject = (MANAGED_PROCESSES, MCP_SERVERS, TIMERS, CONTENT_SOURCE, UI_SLOTS)
+inject = (
+    MANAGED_PROCESSES,
+    MCP_SERVERS,
+    TIMERS,
+    UI_SLOTS,
+)
 dashboard_module = "dashboard.py"
 
 
 async def apply(ctx: Context, config: FitbitConfig) -> None:
-    """装配 monitor、工具、Content 采集、睡眠上下文与移动界面。"""
+    """装配 monitor、工具、Wake 来源和移动界面。"""
 
     # 1. 登记现有 monitor 与用户显式调用的普通 MCP 工具
     await ctx.require(MANAGED_PROCESSES).register(
@@ -90,34 +84,41 @@ async def apply(ctx: Context, config: FitbitConfig) -> None:
         ),
     )
 
-    # 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),
+    # 2. EventMail 存在时，独立子 Fiber 才启动健康来源。
+    async def apply_eventmail(source_ctx: Context) -> None:
+        store = FitbitAdapterStore(source_ctx.data_root / "adapter.sqlite3")
+        store.initialize(datetime.now(UTC))
+        runtime = FitbitWakeRuntime(
+            store,
+            source_ctx.require(TIMERS),
+            source_ctx.require(EVENTMAIL_ALERT_SOURCE).bind("fitbit-health-alerts"),
+            source_ctx.require(EVENTMAIL_CONTEXT_SOURCE).bind("fitbit-sleep"),
+            FitbitMonitorClient(),
+            poll_interval=timedelta(seconds=config.content.poll_interval_seconds),
+            sleep_ttl=timedelta(seconds=config.content.sleep_ttl_seconds),
+        )
+
+        def setup() -> object:
+            return runtime.close
+
+        _ = await source_ctx.effect(setup, label="fitbit-eventmail-runtime")
+        poll_health = await source_ctx.health("fitbit-eventmail-poll")
+
+        async def start(_event: object) -> None:
+            await runtime.start(source_ctx, poll_health)
+
+        async def stop(_event: object) -> None:
+            await runtime.close()
+
+        _ = await source_ctx.on(RUNTIME_STARTED, start)
+        _ = await source_ctx.on(RUNTIME_STOPPING, stop)
+
+    _ = await ctx.inject(
+        (TIMERS, EVENTMAIL_ALERT_SOURCE, EVENTMAIL_CONTEXT_SOURCE),
+        apply_eventmail,
+        name="fitbit-eventmail-source",
     )
 
-    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(ctx, poll_health)
-
-    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,

src/content_adapter.py

@@ -3,28 +3,24 @@
 import asyncio
 import hashlib
 import json
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Mapping
 from datetime import UTC, datetime, timedelta
-from typing import Protocol, cast
+from typing import cast
 from urllib.parse import quote
 
 import requests
 
-from agent.control.timer import TimerHandle, TimerStatus
-from agent.plugin_composition import Context, HealthHandle, PluginTimers
+from agent.plugin_composition import (
+    Context,
+    HealthHandle,
+    PluginTimers,
+    TimerHandle,
+    TimerStatus,
+)
+from .eventmail import BoundAlertSource, BoundContextSource
 from .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 FitbitMonitorTransientError(RuntimeError):
     """表示 monitor HTTP/IO 边界可在下一次 Timer 重试。"""
 
@@ -63,14 +59,15 @@ def ensure_not_pending(self, event_id: str) -> None:
             raise RuntimeError(f"Fitbit event ACK 后仍在 pending 队列: {event_id}")
 
 
-class FitbitContentRuntime:
-    """结算已投递 ACK，提交一次 monitor 快照，再登记一个 Timer。"""
+class FitbitWakeRuntime:
+    """上报 Fitbit Alert 与 Context，再登记一个 Timer。"""
 
     def __init__(
         self,
         store: FitbitAdapterStore,
         timers: PluginTimers,
-        content: BoundContentSource,
+        alerts: BoundAlertSource,
+        context: BoundContextSource,
         monitor: FitbitMonitorClient,
         *,
         poll_interval: timedelta,
@@ -80,7 +77,8 @@ def __init__(
     ) -> None:
         self._store = store
         self._timers = timers
-        self._content = content
+        self._alerts = alerts
+        self._context = context
         self._monitor = monitor
         self._poll_interval = poll_interval
         self._sleep_ttl = sleep_ttl
@@ -94,10 +92,10 @@ async def start(self, ctx: Context, health: HealthHandle) -> None:
         """恢复来源 deadline，并启动唯一 Fiber-owned 采集循环。"""
 
         if self._closed:
-            raise RuntimeError("Fitbit Content runtime 已关闭")
+            raise RuntimeError("Fitbit Wake runtime 已关闭")
         if self._task is None:
             self._task = await ctx.spawn(
-                self._run(ctx, health), name="fitbit-content-poll"
+                self._run(ctx, health), name="fitbit-wake-poll"
             )
 
     async def close(self) -> None:
@@ -143,43 +141,42 @@ async def _run(self, ctx: Context, health: HealthHandle) -> None:
     def tick(self) -> None:
         """先结算历史投递，再发布当前 monitor 快照。"""
 
-        # 1. 先完成外部 ACK，再结算 Content
-        self._drain_unsettled()
-
-        # 2. 只拉取一次，独立归一化健康事件，并优先提交 Content
+        # 1. 只拉取一次；终态 Alert 先 ACK，其余按稳定身份上报。
         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())
+        for item in items:
+            event_id = str(item["item_id"])
+            status = self._alerts.status(
+                event_id=event_id,
+            )
+            if status in {"delivered", "skipped", "expired"}:
+                self._ensure_not_pending(event_id)
+                if self._after_provider_ack is not None:
+                    self._after_provider_ack()
+                continue
+            _ = self._alerts.report(
+                event_id=event_id,
+                payload=_mapping(item, "payload"),
+                observed_at=now,
+            )
+
+        # 2. 睡眠状态是可覆盖、会过期的 Context，不参与 Content 初筛。
         sleep = normalize_sleep(snapshot)
+        expires_at = now + self._sleep_ttl
+        _ = self._context.report(
+            event_id="current",
+            payload=sleep,
+            observed_at=now,
+            expires_at=expires_at,
+        )
         self._store.commit_snapshot(
             sleep,
             observed_at=now,
-            expires_at=now + self._sleep_ttl,
+            expires_at=expires_at,
             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._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 _monitor_snapshot(self) -> Mapping[str, object]:
         try:
             return self._monitor.snapshot()
@@ -249,19 +246,6 @@ def normalize_sleep(snapshot: Mapping[str, object]) -> Mapping[str, object]:
     }
 
 
-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 必须是对象")
@@ -283,10 +267,12 @@ def _string(payload: Mapping[str, object], name: str) -> str:
 
 
 def _canonical(payload: object) -> str:
-    return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    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 必须带时区")
+        raise ValueError("Fitbit Wake clock 必须带时区")
     return value.astimezone(UTC)

src/eventmail.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Protocol
+
+from agent.plugin_composition import ServiceKey
+
+
+class BoundAlertSource(Protocol):
+    def report(
+        self,
+        *,
+        event_id: str,
+        payload: Mapping[str, object],
+        observed_at: datetime,
+        expires_at: datetime | None = None,
+    ) -> Mapping[str, object]: ...
+
+    def status(self, *, event_id: str) -> str | None: ...
+
+
+class AlertSourceServices(Protocol):
+    def bind(self, source_id: str) -> BoundAlertSource: ...
+
+
+class BoundContextSource(Protocol):
+    def report(
+        self,
+        *,
+        event_id: str,
+        payload: Mapping[str, object],
+        observed_at: datetime,
+        expires_at: datetime | None = None,
+    ) -> Mapping[str, object]: ...
+
+
+class ContextSourceServices(Protocol):
+    def bind(self, source_id: str) -> BoundContextSource: ...
+
+
+EVENTMAIL_ALERT_SOURCE = ServiceKey[AlertSourceServices](
+    "eventmail.alert_source.v1"
+)
+EVENTMAIL_CONTEXT_SOURCE = ServiceKey[ContextSourceServices](
+    "eventmail.context_source.v1"
+)