feat(sdk): add host integration extension points

docs/python-sdk.md

@@ -490,6 +490,7 @@ Run the agent once and return a `RunResult`.
 | `sender_id` | `str` | `"user"` | Logical sender identifier used in runtime context. |
 | `media` | `list[str] \| None` | `None` | Optional local media paths attached to the message. |
 | `ephemeral` | `bool` | `False` | Run without persisting the turn or compacting session history. |
+| `attributes` | `Mapping[str, Any] \| None` | `None` | Caller-owned request data for host integrations. It is available to context providers and turn-hook factories, but is not added to trusted message metadata or persisted in session messages. |
 | `hooks` | `list[AgentHook] \| None` | `None` | Lifecycle hooks for this run only. |
 | `model` | `str \| None` | `None` | Override the model for this run only. |
 | `model_preset` | `str \| None` | `None` | Override the model preset for this run only. |
@@ -631,9 +632,67 @@ 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 sync or async handler to one runtime event type 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 applications can attach external context without copying or modifying the
+nanobot agent loop. A context provider receives a `RequestContext` before each
+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.
+Its handler may read the completed transcript through `bot.sessions`. Runtime
+event handlers run in registration order, and async handlers are awaited before
+the run continues.
+
+```python
+from nanobot import (
+    Nanobot,
+    RequestContext,
+    RuntimeContextBlock,
+    SessionTurnPersisted,
+)
+
+
+async def run_with_external_memory(openviking) -> None:
+    async with Nanobot.from_config() as bot:
+        async def load_context(request: RequestContext):
+            resource = request.attributes.get("resource")
+            if not resource:
+                return None
+            text = await openviking.search(resource, request.original_user_text or "")
+            return RuntimeContextBlock(source="openviking", content=text)
+
+        async def sync_saved_turn(event: SessionTurnPersisted):
+            snapshot = bot.sessions.export(event.context.session_key)
+            if snapshot is not None:
+                await openviking.sync(
+                    resource=event.context.attributes.get("resource"),
+                    messages=snapshot.messages,
+                )
+
+        remove_context = bot.runtime.add_context_provider(load_context)
+        remove_sync = bot.runtime.subscribe(SessionTurnPersisted, sync_saved_turn)
+        try:
+            await bot.run(
+                "Continue the architecture discussion",
+                session_key="project:openviking",
+                attributes={"resource": "viking://projects/openviking"},
+            )
+        finally:
+            remove_sync()
+            remove_context()
+```
+
+Context providers are trusted host extensions: their returned text becomes
+model-visible context. Validate and delimit untrusted external content before
+returning it. `SessionTurnPersisted` is not emitted for `ephemeral=True` runs.
+
 ## Hooks
 
 Hooks let you observe or customize the agent loop. Subclass `AgentHook` and override the methods you need.

nanobot/__init__.py

@@ -32,6 +32,9 @@ _LAZY_EXPORTS = {
     "Nanobot": ".nanobot",
     "RunStream": ".nanobot",
     "RunResult": ".nanobot",
+    "RequestContext": ".agent.tools.context",
+    "RuntimeContextBlock": ".runtime_context",
+    "RuntimeContextProvider": ".runtime_context",
     "SessionInfo": ".nanobot",
     "SessionSnapshot": ".nanobot",
     "STREAM_EVENT_REASONING_COMPLETED": ".nanobot",
@@ -47,6 +50,7 @@ _LAZY_EXPORTS = {
     "STREAM_EVENT_TYPES": ".nanobot",
     "StreamEvent": ".nanobot",
     "StreamEventType": ".nanobot",
+    "SessionTurnPersisted": ".bus.runtime_events",
 }
 
 
@@ -64,6 +68,9 @@ def __getattr__(name: str):
 __all__ = [
     "Nanobot",
     "RunResult",
+    "RequestContext",
+    "RuntimeContextBlock",
+    "RuntimeContextProvider",
     "RunStream",
     "SessionInfo",
     "SessionSnapshot",
@@ -80,4 +87,5 @@ __all__ = [
     "STREAM_EVENT_TYPES",
     "StreamEvent",
     "StreamEventType",
+    "SessionTurnPersisted",
 ]

nanobot/agent/hook.py

@@ -58,6 +58,7 @@ class AgentTurnHookContext:
     message_id: str | None = None
     session_key: str | None = None
     metadata: dict[str, Any] = field(default_factory=dict)
+    attributes: dict[str, Any] = field(default_factory=dict)
     ephemeral: bool = False
 
 

nanobot/agent/loop.py

@@ -122,6 +122,7 @@ class TurnContext:
     initial_messages: list[dict[str, Any]] = field(default_factory=list)
     request_context: RequestContext | None = None
     runtime_context_blocks: list[RuntimeContextBlock] = field(default_factory=list)
+    attributes: dict[str, Any] = field(default_factory=dict)
 
     final_content: str | None = None
     all_messages: list[dict[str, Any]] = field(default_factory=list)
@@ -612,11 +613,18 @@ class AgentLoop:
     def register_runtime_context_provider(
         self,
         provider: RuntimeContextProvider,
-    ) -> None:
+    ) -> Callable[[], None]:
-        """Register a provider resolved once before each inbound model turn."""
+        """Register a per-turn context provider and return an unsubscribe callback."""
-        if provider not in self._runtime_context_providers:
+        if provider in self._runtime_context_providers:
+            return lambda: None
         self._runtime_context_providers.append(provider)
 
+        def _unsubscribe() -> None:
+            with suppress(ValueError):
+                self._runtime_context_providers.remove(provider)
+
+        return _unsubscribe
+
     async def submit_cron_turn(self, msg: InboundMessage) -> OutboundMessage | None:
         return await self._cron_turns.submit(msg)
 
@@ -703,6 +711,7 @@ class AgentLoop:
             original_user_text=ctx.original_user_text,
             runtime=ctx.runtime,
             metadata=dict(ctx.msg.metadata or {}),
+            attributes=dict(ctx.attributes),
             sender_id=ctx.msg.sender_id,
             turn_id=ctx.turn_id,
             workspace=scope.project_path,
@@ -881,6 +890,7 @@ class AgentLoop:
                         original_user_text=pending_msg.content,
                         runtime=runtime,
                         metadata=dict(metadata),
+                        attributes=dict(request_ctx.attributes),
                         sender_id=pending_msg.sender_id,
                         turn_id=request_ctx.turn_id,
                         workspace=scope.project_path,
@@ -980,6 +990,7 @@ class AgentLoop:
                 chat_id=chat_id,
                 message_id=message_id,
                 metadata=metadata,
+                attributes=dict(request_ctx.attributes),
                 session_key=active_session_key,
                 workspace=effective_scope.project_path,
                 tool_hint_max_length=self.tool_hint_max_length,
@@ -1320,6 +1331,7 @@ class AgentLoop:
         runtime: LLMRuntime | None = None,
         delivery: TurnDelivery | None = None,
         on_runtime_admitted: Callable[[LLMRuntime], Awaitable[None]] | None = None,
+        attributes: Mapping[str, Any] | None = None,
     ) -> OutboundMessage | None:
         """Process a single inbound message and return the response."""
         kind = TurnKind.SYSTEM if msg.channel == "system" else TurnKind.USER
@@ -1367,6 +1379,7 @@ class AgentLoop:
             hooks=list(hooks or []),
             hook_factories=list(hook_factories or []),
             tools=tools,
+            attributes=dict(attributes or {}),
         )
         # A streaming callback may be present even when the final text comes from a
         # non-streaming recovery. Only the last completed segment can suppress the
@@ -1559,8 +1572,15 @@ class AgentLoop:
                 ctx.session.add_message(
                     "assistant", result.content, _command=True
                 )
-                self.sessions.save(ctx.session)
                 self._clear_pending_user_turn(ctx.session)
+                self.sessions.save(ctx.session)
+                if not ctx.ephemeral:
+                    await self._runtime_events().session_turn_persisted(
+                        ctx.msg,
+                        ctx.session_key,
+                        turn_id=ctx.turn_id,
+                        attributes=ctx.attributes,
+                    )
             return True
         return False
 
@@ -1702,6 +1722,13 @@ class AgentLoop:
         self._clear_pending_user_turn(ctx.session)
         self._clear_runtime_checkpoint(ctx.session)
         self.sessions.save(ctx.session)
+        if not ctx.ephemeral:
+            await self._runtime_events().session_turn_persisted(
+                ctx.msg,
+                ctx.session_key,
+                turn_id=ctx.turn_id,
+                attributes=ctx.attributes,
+            )
 
     async def _prepare_outbound(self, ctx: TurnContext) -> None:
         if ctx.suppress_response:
@@ -1983,6 +2010,7 @@ class AgentLoop:
         persist_user_message: bool = True,
         runtime: LLMRuntime | None = None,
         on_runtime_admitted: Callable[[LLMRuntime], Awaitable[None]] | None = None,
+        attributes: Mapping[str, Any] | None = None,
     ) -> OutboundMessage | None:
         """Process an external message directly and return the outbound payload."""
         if channel == "system":
@@ -2018,6 +2046,8 @@ class AgentLoop:
                     kwargs["runtime"] = runtime
                 if on_runtime_admitted is not None:
                     kwargs["on_runtime_admitted"] = on_runtime_admitted
+                if attributes is not None:
+                    kwargs["attributes"] = dict(attributes)
                 return await self._process_message(
                     msg,
                     **kwargs,

nanobot/agent/tools/context.py

@@ -26,6 +26,7 @@ class RequestContext:
     original_user_text: str | None = None
     runtime: LLMRuntime | None = None
     metadata: dict[str, Any] = field(default_factory=dict)
+    attributes: dict[str, Any] = field(default_factory=dict)
     sender_id: str | None = None
     turn_id: str | None = None
     workspace: Path | None = None

nanobot/agent/turn_hooks.py

@@ -29,6 +29,7 @@ class AgentTurnHookSpec:
     chat_id: str = "direct"
     message_id: str | None = None
     metadata: dict[str, Any] | None = None
+    attributes: dict[str, Any] | None = None
     session_key: str | None = None
     workspace: Path | None = None
     tool_hint_max_length: int = 40
@@ -62,6 +63,7 @@ def build_agent_turn_hook(spec: AgentTurnHookSpec) -> AgentHook:
         message_id=spec.message_id,
         session_key=spec.session_key,
         metadata=dict(spec.metadata or {}),
+        attributes=dict(spec.attributes or {}),
         ephemeral=spec.ephemeral,
     )
     hook_chain: list[AgentHook] = [progress_hook]

nanobot/bus/runtime_events.py

@@ -27,6 +27,7 @@ class RuntimeEventContext:
     chat_id: str
     session_key: str
     metadata: dict[str, Any] = field(default_factory=dict)
+    attributes: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
@@ -54,6 +55,15 @@ class TurnCompleted:
     runtime: Any | None = None
 
 
+@dataclass(frozen=True)
+class SessionTurnPersisted:
+    """A completed turn has been written to local session storage."""
+
+    context: RuntimeEventContext
+    turn_id: str
+    sender_id: str
+
+
 @dataclass(frozen=True)
 class GoalStateChanged:
     """A session's sustained-goal state changed."""
@@ -72,6 +82,7 @@ class RuntimeModelChanged:
 
 RuntimeEvent = (
     SessionTurnStarted
+    | SessionTurnPersisted
     | TurnRunStatusChanged
     | TurnCompleted
     | GoalStateChanged
@@ -79,6 +90,7 @@ RuntimeEvent = (
 )
 RuntimeEventType = (
     type[SessionTurnStarted]
+    | type[SessionTurnPersisted]
     | type[TurnRunStatusChanged]
     | type[TurnCompleted]
     | type[GoalStateChanged]
@@ -152,12 +164,14 @@ class RuntimeEventPublisher:
         chat_id: str,
         session_key: str,
         metadata: dict[str, Any] | None,
+        attributes: dict[str, Any] | None = None,
     ) -> RuntimeEventContext:
         return RuntimeEventContext(
             channel=channel,
             chat_id=chat_id,
             session_key=session_key,
             metadata=dict(metadata or {}),
+            attributes=dict(attributes or {}),
         )
 
     def record_turn_runtime(self, session_key: str, runtime: Any) -> None:
@@ -208,6 +222,28 @@ class RuntimeEventPublisher:
             )
         )
 
+    async def session_turn_persisted(
+        self,
+        msg: InboundMessage,
+        session_key: str,
+        *,
+        turn_id: str,
+        attributes: dict[str, Any] | None = None,
+    ) -> None:
+        await self.bus.publish(
+            SessionTurnPersisted(
+                context=self._context(
+                    channel=msg.channel,
+                    chat_id=msg.chat_id,
+                    session_key=session_key,
+                    metadata=msg.metadata,
+                    attributes=attributes,
+                ),
+                turn_id=turn_id,
+                sender_id=msg.sender_id,
+            )
+        )
+
     async def turn_completed(
         self,
         *,

nanobot/nanobot.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping
 from pathlib import Path
 from typing import Any
 
@@ -134,6 +134,7 @@ class Nanobot:
         sender_id: str = "user",
         media: list[str] | None = None,
         ephemeral: bool = False,
+        attributes: Mapping[str, Any] | None = None,
         hooks: list[AgentHook] | None = None,
         model: str | None = None,
         model_preset: str | None = None,
@@ -149,6 +150,9 @@ class Nanobot:
             sender_id: Logical sender identifier for runtime context.
             media: Optional local media paths attached to the message.
             ephemeral: If true, do not persist the turn or compact session history.
+            attributes: Optional caller-owned request data exposed to context
+                providers and turn-hook factories. Attributes are kept separate
+                from nanobot's trusted internal message metadata.
             hooks: Optional lifecycle hooks for this run.
             model: Override the model for this run only.
             model_preset: Override the model preset for this run only.
@@ -167,6 +171,7 @@ class Nanobot:
             sender_id=sender_id,
             media=media,
             ephemeral=ephemeral,
+            attributes=attributes,
         )
         if runtime is not None:
             kwargs["runtime"] = runtime
@@ -188,6 +193,7 @@ class Nanobot:
         sender_id: str = "user",
         media: list[str] | None = None,
         ephemeral: bool = False,
+        attributes: Mapping[str, Any] | None = None,
         hooks: list[AgentHook] | None = None,
         model: str | None = None,
         model_preset: str | None = None,
@@ -242,6 +248,7 @@ class Nanobot:
                 sender_id=sender_id,
                 media=media,
                 ephemeral=ephemeral,
+                attributes=attributes,
                 on_stream=_on_stream,
                 on_stream_end=_on_stream_end,
             )
@@ -289,6 +296,7 @@ class Nanobot:
         sender_id: str = "user",
         media: list[str] | None = None,
         ephemeral: bool = False,
+        attributes: Mapping[str, Any] | None = None,
         hooks: list[AgentHook] | None = None,
         model: str | None = None,
         model_preset: str | None = None,
@@ -302,6 +310,7 @@ class Nanobot:
             sender_id=sender_id,
             media=media,
             ephemeral=ephemeral,
+            attributes=attributes,
             hooks=hooks,
             model=model,
             model_preset=model_preset,

nanobot/sdk/clients.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from copy import deepcopy
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -18,6 +18,8 @@ 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:
@@ -193,6 +195,21 @@ class RuntimeClient:
         """Current runtime workspace."""
         return self._loop.workspace
 
+    def add_context_provider(
+        self,
+        provider: RuntimeContextProvider,
+    ) -> Callable[[], None]:
+        """Register per-turn model context and return an unsubscribe callback."""
+        return self._loop.register_runtime_context_provider(provider)
+
+    def subscribe(
+        self,
+        event_type: RuntimeEventType,
+        handler: RuntimeEventHandler,
+    ) -> Callable[[], None]:
+        """Subscribe to one runtime event type and return an unsubscribe callback."""
+        return self._loop.runtime_events.subscribe(handler, event_type)
+
     async def compact_session(self, session_key: str) -> SessionSnapshot:
         """Run token/replay-window consolidation for one session."""
         session = self._loop.sessions.get_or_create(session_key)

nanobot/sdk/runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 
@@ -22,6 +23,7 @@ def build_process_direct_kwargs(
     sender_id: str,
     media: list[str] | None,
     ephemeral: bool,
+    attributes: Mapping[str, Any] | None = None,
     on_stream: Any | None = None,
     on_stream_end: Any | None = None,
 ) -> dict[str, Any]:
@@ -37,6 +39,8 @@ def build_process_direct_kwargs(
     if ephemeral:
         kwargs["ephemeral"] = True
         kwargs["_run_extra_hooks_for_ephemeral"] = True
+    if attributes is not None:
+        kwargs["attributes"] = dict(attributes)
     if on_stream is not None:
         kwargs["on_stream"] = on_stream
     if on_stream_end is not None:

tests/agent/test_turn_hooks.py

@@ -65,6 +65,7 @@ async def test_turn_hook_builder_runs_factories_with_matching_registration_order
         session_key="websocket:chat-1",
         workspace=tmp_path,
         metadata={"source": "test"},
+        attributes={"tenant": "acme"},
         registered_hook_factories=[factory("registered_factory")],
         registered_hooks=[RecordingHook(events, "registered")],
         turn_hook_factories=[factory("turn_factory")],
@@ -92,6 +93,10 @@ async def test_turn_hook_builder_runs_factories_with_matching_registration_order
         {"source": "test"},
         {"source": "test"},
     ]
+    assert [context.attributes for context in captured] == [
+        {"tenant": "acme"},
+        {"tenant": "acme"},
+    ]
 
 
 @pytest.mark.asyncio

tests/bus/test_runtime_events.py

@@ -6,6 +6,7 @@ from nanobot.bus.runtime_events import (
     RuntimeEventContext,
     RuntimeEventPublisher,
     RuntimeModelChanged,
+    SessionTurnPersisted,
     SessionTurnStarted,
     TurnCompleted,
     TurnRunStatusChanged,
@@ -120,3 +121,33 @@ async def test_runtime_event_publisher_consumes_turn_metadata_on_complete() -> N
     assert isinstance(second, TurnCompleted)
     assert second.latency_ms is None
     assert second.runtime is None
+
+
+@pytest.mark.asyncio
+async def test_runtime_event_publisher_emits_persisted_turn_attributes() -> None:
+    bus = RuntimeEventBus()
+    seen: list[object] = []
+    publisher = RuntimeEventPublisher(bus)
+    msg = InboundMessage(
+        channel="sdk",
+        sender_id="alice",
+        chat_id="chat-a",
+        content="hello",
+        metadata={"internal": "routing"},
+    )
+
+    bus.subscribe(seen.append, SessionTurnPersisted)
+    await publisher.session_turn_persisted(
+        msg,
+        "sdk:chat-a",
+        turn_id="turn-1",
+        attributes={"tenant": "acme"},
+    )
+
+    event = seen[0]
+    assert isinstance(event, SessionTurnPersisted)
+    assert event.context.session_key == "sdk:chat-a"
+    assert event.context.metadata == {"internal": "routing"}
+    assert event.context.attributes == {"tenant": "acme"}
+    assert event.turn_id == "turn-1"
+    assert event.sender_id == "alice"

tests/test_nanobot_facade.py

@@ -264,10 +264,157 @@ async def test_run_custom_session_key(tmp_path):
     )
 
 
+@pytest.mark.asyncio
+async def test_run_exposes_attributes_to_context_provider_without_persisting_them(tmp_path):
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.agent.tools.context import RequestContext
+    from nanobot.bus.queue import MessageBus
+    from nanobot.providers.base import LLMResponse
+
+    provider = _fake_provider("test-model")
+    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
+        content="done",
+        tool_calls=[],
+    ))
+    bot = Nanobot(AgentLoop(
+        bus=MessageBus(),
+        provider=provider,
+        workspace=tmp_path,
+        model="test-model",
+    ))
+    seen: list[RequestContext] = []
+
+    async def provide_context(context: RequestContext):
+        seen.append(context)
+        return None
+
+    unsubscribe = bot.runtime.add_context_provider(provide_context)
+    result = await bot.run(
+        "hi",
+        session_key="sdk:attributes",
+        attributes={"tenant": "acme"},
+    )
+
+    assert result.content == "done"
+    assert seen[0].attributes == {"tenant": "acme"}
+    assert seen[0].metadata == {}
+    snapshot = bot.sessions.export("sdk:attributes")
+    assert snapshot is not None
+    assert all("attributes" not in message for message in snapshot.messages)
+
+    unsubscribe()
+    await bot.run(
+        "again",
+        session_key="sdk:attributes",
+        attributes={"tenant": "other"},
+    )
+    assert len(seen) == 1
+
+
+@pytest.mark.asyncio
+async def test_runtime_subscription_observes_saved_session_and_can_unsubscribe(tmp_path):
+    from nanobot import SessionTurnPersisted
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.queue import MessageBus
+    from nanobot.providers.base import LLMResponse
+
+    provider = _fake_provider("test-model")
+    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
+        content="saved reply",
+        tool_calls=[],
+    ))
+    bot = Nanobot(AgentLoop(
+        bus=MessageBus(),
+        provider=provider,
+        workspace=tmp_path,
+        model="test-model",
+    ))
+    seen: list[tuple[SessionTurnPersisted, SessionSnapshot | None]] = []
+
+    def on_persisted(event: SessionTurnPersisted) -> None:
+        seen.append((event, bot.sessions.export(event.context.session_key)))
+
+    unsubscribe = bot.runtime.subscribe(SessionTurnPersisted, on_persisted)
+    await bot.run(
+        "hi",
+        session_key="sdk:persisted",
+        sender_id="alice",
+        attributes={"tenant": "acme"},
+    )
+
+    assert len(seen) == 1
+    event, snapshot = seen[0]
+    assert event.sender_id == "alice"
+    assert event.context.attributes == {"tenant": "acme"}
+    assert snapshot is not None
+    assert snapshot.messages[-1]["role"] == "assistant"
+    assert snapshot.messages[-1]["content"] == "saved reply"
+
+    unsubscribe()
+    await bot.run("again", session_key="sdk:persisted")
+    assert len(seen) == 1
+
+
+@pytest.mark.asyncio
+async def test_runtime_subscription_observes_saved_command_turn(tmp_path):
+    from nanobot import SessionTurnPersisted
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.queue import MessageBus
+
+    bot = Nanobot(AgentLoop(
+        bus=MessageBus(),
+        provider=_fake_provider("test-model"),
+        workspace=tmp_path,
+        model="test-model",
+    ))
+    seen: list[SessionTurnPersisted] = []
+    bot.runtime.subscribe(SessionTurnPersisted, seen.append)
+
+    await bot.run("/skill", session_key="sdk:command")
+
+    assert len(seen) == 1
+    snapshot = bot.sessions.export("sdk:command")
+    assert snapshot is not None
+    assert [message["role"] for message in snapshot.messages[-2:]] == [
+        "user",
+        "assistant",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_ephemeral_run_does_not_publish_session_persisted_event(tmp_path):
+    from nanobot import SessionTurnPersisted
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.queue import MessageBus
+    from nanobot.providers.base import LLMResponse
+
+    provider = _fake_provider("test-model")
+    provider.chat_with_retry = AsyncMock(return_value=LLMResponse(
+        content="temporary",
+        tool_calls=[],
+    ))
+    bot = Nanobot(AgentLoop(
+        bus=MessageBus(),
+        provider=provider,
+        workspace=tmp_path,
+        model="test-model",
+    ))
+    seen: list[SessionTurnPersisted] = []
+    bot.runtime.subscribe(SessionTurnPersisted, seen.append)
+
+    await bot.run("hi", session_key="sdk:ephemeral", ephemeral=True)
+
+    assert seen == []
+
+
 def test_import_from_top_level():
     import nanobot
 
     assert nanobot.Nanobot is Nanobot
+    assert nanobot.RequestContext.__name__ == "RequestContext"
+    assert nanobot.RuntimeContextBlock.__name__ == "RuntimeContextBlock"
+    assert nanobot.RuntimeContextProvider is not None
+    assert nanobot.SessionTurnPersisted.__name__ == "SessionTurnPersisted"
     assert nanobot.RunResult is RunResult
     assert nanobot.RunStream is RunStream
     assert nanobot.SessionInfo is SessionInfo
@@ -920,6 +1067,7 @@ async def test_run_streamed_forwards_runtime_options(tmp_path):
         sender_id="alice",
         media=["/tmp/image.png"],
         ephemeral=True,
+        attributes={"tenant": "acme"},
     )
     await run.wait()
 
@@ -932,6 +1080,7 @@ async def test_run_streamed_forwards_runtime_options(tmp_path):
     assert kwargs["sender_id"] == "alice"
     assert kwargs["media"] == ["/tmp/image.png"]
     assert kwargs["ephemeral"] is True
+    assert kwargs["attributes"] == {"tenant": "acme"}
     assert callable(kwargs["on_stream"])
     assert callable(kwargs["on_stream_end"])
     assert kwargs["hooks"]