fix(sdk): narrow persisted turn callback API

docs/python-sdk.md

@@ -633,11 +633,11 @@ Do not expose exported snapshots directly to chat users.
 | `model` | Current runtime model name. |
 | `workspace` | Current runtime workspace path. |
 | `add_context_provider(provider)` | Register an async per-turn context provider and return an unsubscribe callback. |
-| `subscribe(event_type, handler)` | Subscribe a best-effort sync or async handler to one runtime event type and return an unsubscribe callback. |
+| `on_session_turn_persisted(handler)` | Register a best-effort sync or async callback for locally persisted turns and return an unsubscribe callback. |
 | `await compact_session(session_key)` | Run token/replay-window consolidation for a session. |
 | `await compact_idle_session(session_key, max_suffix=8)` | Run idle-session compaction and return its summary. |
 
-### Host integration context and persistence events
+### Host integration context and persisted-turn callbacks
 
 Host applications can attach external context without copying or modifying the
 nanobot agent loop. A context provider receives a `RequestContext` before each
@@ -645,14 +645,15 @@ model turn and may return one or more `RuntimeContextBlock` values. Use
 `attributes` for caller-owned routing data; nanobot keeps it separate from
 trusted channel metadata and does not persist it in session messages.
 
-`SessionTurnPersisted` is published after a non-ephemeral turn has been saved.
+`on_session_turn_persisted()` invokes its callback after a non-ephemeral turn
-Its handler may read the completed transcript through `bot.sessions`. Runtime
+has been saved. The callback receives `SessionTurnPersisted` and may read the
-event handlers run in registration order, and async handlers are awaited before
+completed transcript through `bot.sessions`. Callbacks run in registration
-the run continues. Subscriptions are observational: handler exceptions are
+order, and async callbacks are awaited before the run continues. They are
-logged and suppressed so the completed local turn remains successful. Durable
+observational: callback exceptions are logged and suppressed so the completed
-external synchronization must catch failures and persist retry work before the
+local turn remains successful. Durable external synchronization must catch
-handler returns. During SDK runs, handlers execute while the session is still
+failures and persist retry work before the callback returns. During SDK runs,
-serialized and must not re-enter `bot.run()` for the same session.
+callbacks execute while the session is still serialized and must not re-enter
+`bot.run()` for the same session.
 
 ```python
 import json
@@ -704,7 +705,7 @@ async def run_with_external_memory(external_memory, enqueue_retry) -> None:
                     await enqueue_retry(event, snapshot, exc)
 
         remove_context = bot.runtime.add_context_provider(load_context)
-        remove_sync = bot.runtime.subscribe(SessionTurnPersisted, sync_saved_turn)
+        remove_sync = bot.runtime.on_session_turn_persisted(sync_saved_turn)
         try:
             await bot.run(
                 "Continue the architecture discussion",
@@ -719,7 +720,7 @@ async def run_with_external_memory(external_memory, enqueue_retry) -> None:
 Context providers are trusted host extensions, and `RuntimeContextBlock.content`
 is appended verbatim to model-visible context. Apply equivalent bounding,
 encoding, and delimiter escaping to untrusted external content.
-`SessionTurnPersisted` is not emitted for `ephemeral=True` runs.
+Persisted-turn callbacks are not invoked for `ephemeral=True` runs.
 
 ## Hooks
 

nanobot/sdk/clients.py

@@ -2,12 +2,13 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable, Iterable, Mapping
+from collections.abc import Awaitable, Callable, Iterable, Mapping
 from copy import deepcopy
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
-from nanobot.runtime_context import RUNTIME_CONTEXT_HISTORY_META
+from nanobot.bus.runtime_events import SessionTurnPersisted
+from nanobot.runtime_context import RUNTIME_CONTEXT_HISTORY_META, RuntimeContextProvider
 from nanobot.sdk.types import (
     SessionInfo,
     SessionSnapshot,
@@ -18,8 +19,6 @@ from nanobot.session.manager import replay_max_messages_for_context
 
 if TYPE_CHECKING:
     from nanobot.agent.loop import AgentLoop
-    from nanobot.bus.runtime_events import RuntimeEventHandler, RuntimeEventType
-    from nanobot.runtime_context import RuntimeContextProvider
 
 
 class SessionClient:
@@ -202,13 +201,12 @@ class RuntimeClient:
         """Register per-turn model context and return an unsubscribe callback."""
         return self._loop.register_runtime_context_provider(provider)
 
-    def subscribe(
+    def on_session_turn_persisted(
         self,
-        event_type: RuntimeEventType,
+        handler: Callable[[SessionTurnPersisted], Awaitable[None] | None],
-        handler: RuntimeEventHandler,
     ) -> Callable[[], None]:
-        """Subscribe to one runtime event type and return an unsubscribe callback."""
+        """Register a persisted-turn callback and return an unsubscribe callback."""
-        return self._loop.runtime_events.subscribe(handler, event_type)
+        return self._loop.runtime_events.subscribe(handler, SessionTurnPersisted)
 
     async def compact_session(self, session_key: str) -> SessionSnapshot:
         """Run token/replay-window consolidation for one session."""

tests/test_nanobot_facade.py

@@ -335,7 +335,7 @@ async def test_run_exposes_attributes_to_context_provider_without_persisting_the
 
 
 @pytest.mark.asyncio
-async def test_runtime_subscription_is_best_effort_and_reads_display_safe_session(tmp_path):
+async def test_persisted_turn_callback_is_best_effort_and_reads_display_safe_session(tmp_path):
     from nanobot import SessionTurnPersisted
     from nanobot.agent.loop import AgentLoop
     from nanobot.bus.queue import MessageBus
@@ -374,8 +374,8 @@ async def test_runtime_subscription_is_best_effort_and_reads_display_safe_sessio
         seen.append((event, bot.sessions.get(event.context.session_key)))
 
     remove_context = bot.runtime.add_context_provider(provide_context)
-    remove_failure = bot.runtime.subscribe(SessionTurnPersisted, fail_sync)
+    remove_failure = bot.runtime.on_session_turn_persisted(fail_sync)
-    unsubscribe = bot.runtime.subscribe(SessionTurnPersisted, on_persisted)
+    unsubscribe = bot.runtime.on_session_turn_persisted(on_persisted)
     result = await bot.run(
         "hi",
         session_key="sdk:persisted",
@@ -405,7 +405,7 @@ async def test_runtime_subscription_is_best_effort_and_reads_display_safe_sessio
 
 
 @pytest.mark.asyncio
-async def test_runtime_subscription_observes_saved_command_turn(tmp_path):
+async def test_persisted_turn_callback_observes_saved_command_turn(tmp_path):
     from nanobot import SessionTurnPersisted
     from nanobot.agent.loop import AgentLoop
     from nanobot.bus.queue import MessageBus
@@ -417,7 +417,7 @@ async def test_runtime_subscription_observes_saved_command_turn(tmp_path):
         model="test-model",
     ))
     seen: list[SessionTurnPersisted] = []
-    bot.runtime.subscribe(SessionTurnPersisted, seen.append)
+    bot.runtime.on_session_turn_persisted(seen.append)
 
     await bot.run("/skill", session_key="sdk:command")
 
@@ -431,7 +431,7 @@ async def test_runtime_subscription_observes_saved_command_turn(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_ephemeral_run_does_not_publish_session_persisted_event(tmp_path):
+async def test_ephemeral_run_does_not_invoke_persisted_turn_callback(tmp_path):
     from nanobot import SessionTurnPersisted
     from nanobot.agent.loop import AgentLoop
     from nanobot.bus.queue import MessageBus
@@ -449,13 +449,20 @@ async def test_ephemeral_run_does_not_publish_session_persisted_event(tmp_path):
         model="test-model",
     ))
     seen: list[SessionTurnPersisted] = []
-    bot.runtime.subscribe(SessionTurnPersisted, seen.append)
+    bot.runtime.on_session_turn_persisted(seen.append)
 
     await bot.run("hi", session_key="sdk:ephemeral", ephemeral=True)
 
     assert seen == []
 
 
+def test_runtime_client_does_not_expose_generic_event_subscription():
+    from nanobot.sdk.clients import RuntimeClient
+
+    assert hasattr(RuntimeClient, "on_session_turn_persisted")
+    assert not hasattr(RuntimeClient, "subscribe")
+
+
 def test_import_from_top_level():
     import nanobot