[codex] Publish Steam 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 = "steam"
-version = "3.1.0"
+version = "3.2.1"
 api_version = 3
 entrypoint = "plugin.py"
 

context_source.py

@@ -1,36 +1,41 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import logging
 from collections.abc import Callable
 from datetime import UTC, datetime
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
 
-from agent.control.timer import TimerHandle, TimerStatus
-from agent.lifecycle.types import BeforeTurnCtx
-from agent.plugin_composition import HealthHandle, PluginTimers
+from agent.plugin_composition import (
+    HealthHandle,
+    PluginTimers,
+    TimerHandle,
+    TimerStatus,
+)
 
+from .eventmail import BoundContextSource
 from .steam_runtime import backend
 
 
 class SteamContextRuntime:
-    """用 Timer 刷新 Steam current state，并只为 Wake 追加 fresh hint。"""
+    """用 Timer 刷新 Steam current state，并上报可过期 Context。"""
 
     def __init__(
         self,
         data_root: Path,
         timers: PluginTimers,
         health: HealthHandle,
         report_incident: Callable[[str, str], object],
+        context: BoundContextSource,
         *,
         now: Callable[[], datetime] | None = None,
     ) -> None:
         self._data_root = data_root
         self._timers = timers
         self._health = health
         self._report_incident = report_incident
+        self._context = context
         self._now = now or (lambda: datetime.now(UTC))
         self._handle: TimerHandle | None = None
         self._task: asyncio.Task[None] | None = None
@@ -50,6 +55,7 @@ async def start(self) -> None:
         now = self._aware_now()
         await asyncio.to_thread(backend.initialize, self._data_root, now)
         deadline = await asyncio.to_thread(backend.next_deadline, self._data_root, now)
+        await asyncio.to_thread(self._report_current, now)
         self._arm(deadline)
 
     async def close(self) -> None:
@@ -70,19 +76,6 @@ async def close(self) -> None:
             await handle.cleanup()
         self._stop_diagnostics()
 
-    def prepare(self, ctx: BeforeTurnCtx) -> None:
-        """只在 Wake channel 读取 fresh state 并追加一个普通 hint。"""
-
-        if ctx.channel != "wake":
-            return
-        current = backend.wake_context(self._data_root, ctx.timestamp)
-        if current is None:
-            return
-        ctx.extra_hints.append(
-            "Steam current context:\n"
-            + json.dumps(current, sort_keys=True, separators=(",", ":"))
-        )
-
     def _arm(self, deadline: datetime) -> None:
         if self._closed or self._handle is not None:
             return
@@ -131,6 +124,7 @@ async def _wait_refresh_rearm(self, handle: TimerHandle) -> None:
             else:
                 self._health.recover()
                 next_due = result.next_due
+                await asyncio.to_thread(self._report_current, now)
                 self._log.info(
                     "refresh committed presence=%s history_appended=%s next_due=%s",
                     result.presence,
@@ -144,6 +138,19 @@ async def _wait_refresh_rearm(self, handle: TimerHandle) -> None:
         if not self._closed and next_due is not None:
             self._arm(next_due)
 
+    def _report_current(self, now: datetime) -> None:
+        current = backend.wake_context(self._data_root, now)
+        if current is None:
+            return
+        observed_at = datetime.fromisoformat(str(current["observed_at"]))
+        expires_at = datetime.fromisoformat(str(current["expires_at"]))
+        _ = self._context.report(
+            event_id="current",
+            payload=current,
+            observed_at=observed_at,
+            expires_at=expires_at,
+        )
+
     def _aware_now(self) -> datetime:
         value = self._now()
         if value.tzinfo is None:

eventmail.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Protocol
+
+from agent.plugin_composition import ServiceKey
+
+
+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_CONTEXT_SOURCE = ServiceKey[ContextSourceServices](
+    "eventmail.context_source.v1"
+)

plugin.py

@@ -2,7 +2,6 @@
 
 from pydantic import BaseModel
 
-from agent.lifecycle.composition import CONTEXT_PREPARED_EVENT
 from agent.plugin_composition import (
     MCP_SERVERS,
     RUNTIME_STARTED,
@@ -11,8 +10,8 @@
     Context,
     McpServerDefinition,
 )
-
 from .context_source import SteamContextRuntime
+from .eventmail import EVENTMAIL_CONTEXT_SOURCE
 
 
 class SteamConfig(BaseModel):
@@ -21,15 +20,15 @@ class SteamConfig(BaseModel):
 
 api_version = 3
 name = "steam"
-version = "3.1.0"
-desc = "Timer 刷新的 Steam current context 与用户 MCP"
+version = "3.2.1"
+desc = "Timer 上报的 Steam current Context 与用户 MCP"
 Config = SteamConfig
 inject = (MCP_SERVERS, TIMERS)
 skill_roots = ("skills",)
 
 
 async def apply(ctx: Context, config: object) -> None:
-    """组合用户 MCP、Timer current state 和 Wake context listener。"""
+    """组合用户 MCP、Timer current state 和 Wake Context 上报。"""
 
     if not isinstance(config, SteamConfig):
         raise TypeError("steam config 必须是 SteamConfig")
@@ -46,19 +45,26 @@ async def apply(ctx: Context, config: object) -> None:
         ),
     )
 
-    # 2. 正式 Root 独占 Timer 刷新；listener 只读 current state。
-    health = await ctx.health("context-refresh", required=True)
-    runtime = SteamContextRuntime(
-        ctx.data_root,
-        ctx.require(TIMERS),
-        health,
-        ctx.report_incident,
-    )
+    # 2. EventMail 存在时，独立子 Fiber 才刷新 current state。
+    async def apply_eventmail(source_ctx: Context) -> None:
+        health = await source_ctx.health("context-refresh", required=True)
+        runtime = SteamContextRuntime(
+            source_ctx.data_root,
+            source_ctx.require(TIMERS),
+            health,
+            source_ctx.report_incident,
+            source_ctx.require(EVENTMAIL_CONTEXT_SOURCE).bind("steam-presence"),
+        )
 
-    def setup() -> object:
-        return runtime.close
+        def setup() -> object:
+            return runtime.close
 
-    _ = await ctx.effect(setup, label="steam-context-runtime")
-    _ = await ctx.on(CONTEXT_PREPARED_EVENT, runtime.prepare)
-    _ = await ctx.on(RUNTIME_STARTED, lambda _: runtime.start())
-    _ = await ctx.on(RUNTIME_STOPPING, lambda _: runtime.close())
+        _ = await source_ctx.effect(setup, label="steam-context-runtime")
+        _ = await source_ctx.on(RUNTIME_STARTED, lambda _: runtime.start())
+        _ = await source_ctx.on(RUNTIME_STOPPING, lambda _: runtime.close())
+
+    _ = await ctx.inject(
+        (TIMERS, EVENTMAIL_CONTEXT_SOURCE),
+        apply_eventmail,
+        name="steam-eventmail-source",
+    )

tests/test_context_source.py

@@ -1,15 +1,14 @@
 from __future__ import annotations
 
 import asyncio
-import hashlib
 import json
+from collections.abc import Mapping
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 import pytest
 
 from agent.control.timer import TimerReceipt, TimerStatus
-from agent.lifecycle.types import BeforeTurnCtx
 from agent.plugin_composition import PluginTimers
 from steam_test_plugin.context_source import SteamContextRuntime  # pyright: ignore[reportMissingImports]
 from steam_test_plugin.steam_runtime import backend  # pyright: ignore[reportMissingImports]
@@ -86,6 +85,15 @@ def _config(data_root: Path) -> None:
     )
 
 
+class _Context:
+    def __init__(self) -> None:
+        self.reports: list[dict[str, object]] = []
+
+    def report(self, **kwargs: object) -> Mapping[str, object]:
+        self.reports.append(dict(kwargs))
+        return {"changed": True}
+
+
 async def _eventually(predicate) -> None:
     for _ in range(200):
         if predicate():
@@ -94,20 +102,6 @@ async def _eventually(predicate) -> None:
     raise AssertionError("condition did not settle")
 
 
-def _ctx(now: datetime, channel: str) -> BeforeTurnCtx:
-    return BeforeTurnCtx(
-        session_key="session",
-        channel=channel,
-        chat_id="chat",
-        content="hello",
-        timestamp=now,
-        retrieved_memory_block="",
-        retrieval_trace_raw=None,
-        history_messages=(),
-        turn_id="turn:1",
-    )
-
-
 @pytest.mark.asyncio
 async def test_network_incident_retries_and_recovers(
     tmp_path: Path,
@@ -133,6 +127,7 @@ def refresh(_data_root: Path, attempt: datetime) -> backend.RefreshResult:
         PluginTimers(timer),
         health,  # type: ignore[arg-type]
         lambda kind, message: incidents.append((kind, message)),
+        _Context(),  # type: ignore[arg-type]
         now=lambda: now,
     )
     await runtime.start()
@@ -153,7 +148,7 @@ def refresh(_data_root: Path, attempt: datetime) -> backend.RefreshResult:
     await runtime.close()
 
 
-def test_context_listener_is_wake_only_fresh_only_and_read_only(
+def test_current_presence_is_reported_as_expiring_context(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -166,29 +161,21 @@ def test_context_listener_is_wake_only_fresh_only_and_read_only(
     )
     monkeypatch.setattr(backend, "_fetch_recently_played", lambda _: [])
     _ = backend.refresh(tmp_path, now)
-    database = tmp_path / "steam_proactive.sqlite3"
-    before = hashlib.sha256(database.read_bytes()).hexdigest()
+    context = _Context()
     runtime = SteamContextRuntime(
         tmp_path,
         PluginTimers(None),
         _Health(),  # type: ignore[arg-type]
         lambda _kind, _message: None,
+        context,  # type: ignore[arg-type]
         now=lambda: now,
     )
 
-    passive = _ctx(now, "passive")
-    runtime.prepare(passive)
-    wake = _ctx(now + timedelta(minutes=1), "wake")
-    runtime.prepare(wake)
-    stale = _ctx(now + timedelta(minutes=6), "wake")
-    runtime.prepare(stale)
+    runtime._report_current(now)  # pyright: ignore[reportPrivateUsage]
 
-    assert passive.extra_hints == []
-    assert len(wake.extra_hints) == 1
-    assert wake.extra_hints[0].startswith("Steam current context:\n")
-    assert wake.abort is False
-    assert stale.extra_hints == []
-    assert hashlib.sha256(database.read_bytes()).hexdigest() == before
+    assert len(context.reports) == 1
+    assert context.reports[0]["event_id"] == "current"
+    assert context.reports[0]["expires_at"] == now + timedelta(minutes=5)
 
 
 @pytest.mark.asyncio
@@ -206,6 +193,7 @@ async def test_contract_failure_degrades_and_does_not_retry(
         PluginTimers(timer),
         health,  # type: ignore[arg-type]
         lambda kind, message: incidents.append((kind, message)),
+        _Context(),  # type: ignore[arg-type]
         now=lambda: now,
     )
     monkeypatch.setattr(
@@ -223,7 +211,5 @@ async def test_contract_failure_degrades_and_does_not_retry(
 
     assert len(timer.handles) == 1
     assert health.reason == "RuntimeError: schema mismatch"
-    assert incidents == [
-        ("steam_refresh_contract", "RuntimeError: schema mismatch")
-    ]
+    assert incidents == [("steam_refresh_contract", "RuntimeError: schema mismatch")]
     await runtime.close()