feat(sdk): expand Python runtime controls

docs/README.md

@@ -41,6 +41,7 @@ If a local `nanobot agent` session can already answer normally, you can also ask
 | Use nanobot in a browser | [`webui.md`](./webui.md) | Enable WebSocket, run `nanobot gateway`, open `http://127.0.0.1:8765` |
 | Talk through a chat app | [`chat-apps.md`](./chat-apps.md) | Merge one channel snippet, run `nanobot channels status`, keep `nanobot gateway` running |
 | Change provider or add fallbacks | [`provider-cookbook.md`](./provider-cookbook.md) | Keep `modelPresets` named and set `agents.defaults.modelPreset` |
+| Call nanobot from Python | [`python-sdk.md`](./python-sdk.md) | Reuse the same config/workspace from code, then run or stream one agent turn |
 | Understand before operating long-term | [`concepts.md`](./concepts.md) | Know what config, workspace, gateway, sessions, memory, and tools mean |
 | Diagnose a new failure | [`troubleshooting.md`](./troubleshooting.md) | Start with `nanobot status`, then `nanobot agent -m "Hello!"` |
 
@@ -68,7 +69,7 @@ If a local `nanobot agent` session can already answer normally, you can also ask
 | Observability | [`configuration.md#langfuse-observability`](./configuration.md#langfuse-observability) | Langfuse tracing setup and required environment variables |
 | WebSocket protocol | [`websocket.md`](./websocket.md) | Custom clients, token issuance, multiplexed chats, media, and protocol events |
 | OpenAI-compatible API | [`openai-api.md`](./openai-api.md) | `/v1/chat/completions`, `/v1/models`, file uploads, and SDK-compatible usage |
-| Python SDK | [`python-sdk.md`](./python-sdk.md) | Running nanobot from Python and attaching hooks |
+| Python SDK | [`python-sdk.md`](./python-sdk.md) | SDK 101, sessions, streaming, model overrides, runtime helpers, and hooks |
 | Runtime self-inspection | [`my-tool.md`](./my-tool.md) | Inspecting and tuning the current agent run |
 
 ## Fast Lookup
@@ -80,6 +81,7 @@ If a local `nanobot agent` session can already answer normally, you can also ask
 | Langfuse environment variables | [`configuration.md#langfuse-observability`](./configuration.md#langfuse-observability) |
 | WebSocket/WebUI protocol details | [`websocket.md`](./websocket.md) |
 | OpenAI-compatible API usage | [`openai-api.md`](./openai-api.md) |
+| Python SDK usage | [`python-sdk.md`](./python-sdk.md) |
 | Multiple configs, workspaces, and ports | [`multiple-instances.md`](./multiple-instances.md) |
 | Security, sandboxing, and SSRF controls | [`configuration.md#security`](./configuration.md#security) |
 | Channel plugin development | [`channel-plugin-guide.md`](./channel-plugin-guide.md) |

docs/python-sdk.md

@@ -1,16 +1,64 @@
 # Python SDK
 
-Use nanobot as a library — no CLI, no gateway, just Python.
+Use nanobot as a Python library. The SDK gives you the same agent runtime used
+by the CLI, but from code: model routing, tools, workspace access, conversation
+history, memory, streaming events, and runtime helpers.
 
-Before debugging SDK code, prove the same config works from the CLI:
+If you have used the OpenAI SDK before, the most important difference is this:
+
+- OpenAI SDK calls a model.
+- nanobot SDK runs an agent around a model.
+
+That means one SDK call can read files, call tools, keep session history, use
+memory, stream progress, and return structured runtime information.
+
+```text
+your Python code
+  -> Nanobot SDK
+    -> agent runtime
+      -> configured model provider
+      -> tools
+      -> workspace
+      -> session history
+      -> memory
+```
+
+## Before You Start
+
+Install and configure nanobot first. If you have not done that yet, follow the
+[Quick Start](quick-start.md) and complete the setup wizard. For SDK-only Python
+environments, install the package with:
+
+```bash
+python -m pip install nanobot-ai
+```
+
+`Nanobot.from_config()` reuses your normal `~/.nanobot/config.json` and
+`~/.nanobot/workspace/`. Provider, model, tools, memory, and session behavior
+match the CLI unless you override them. For the difference between config and
+workspace, see [Concepts: Config vs Workspace](concepts.md#config-vs-workspace).
+
+Before writing SDK code, run the same first-run checks from the main
+[Install and Quick Start](quick-start.md):
+
+```bash
+nanobot status
+```
+
+`nanobot status` should show the config path, workspace path, active model or
+preset, and provider summary. Then send one real message:
 
 ```bash
 nanobot agent -m "Hello!"
 ```
 
-`Nanobot.from_config()` reuses your normal `~/.nanobot/config.json`, so provider, model, tools, and workspace behavior match the CLI unless you override them.
+A normal assistant reply means install, config, provider/model selection, and
+workspace access are all usable. Once that works, the SDK should see the same
+runtime.
 
-## Quick Start
+## 5-Minute Quick Start
+
+### Ask One Question
 
 ```python
 import asyncio
@@ -27,21 +75,228 @@ async def main() -> None:
 asyncio.run(main())
 ```
 
-Use `async with` when possible so MCP connections and background cleanup work are closed before the event loop exits. If you manage the instance manually, call `await bot.aclose()` in a `finally` block.
+Use `async with` when possible so tool connections and background cleanup are
+closed before the event loop exits. If you manage the instance manually, call
+`await bot.aclose()` in a `finally` block.
+
+The SDK is async-first because agent runs may stream tokens, execute tools, and
+wait on external services. In a normal Python script, wrap your async function
+with `asyncio.run(...)` as shown above. In a notebook or another async app, call
+`await bot.run(...)` directly from your existing event loop.
+
+### Inspect What Happened
+
+`bot.run(...)` returns a `RunResult`, not just a string:
+
+```python
+result = await bot.run("Review this repository")
+
+print(result.content)     # final answer
+print(result.tools_used)  # tools the agent used
+print(result.usage)       # token usage when available
+print(result.stop_reason) # why the run stopped
+```
+
+### Continue A Conversation
+
+Use a `session_key` when you want history to carry across turns. Different
+session keys are isolated from each other:
+
+```python
+await bot.run("My name is Alice.", session_key="user:alice")
+result = await bot.run("What is my name?", session_key="user:alice")
+
+print(result.content)
+```
+
+This is the SDK equivalent of giving each user, task, eval case, or workflow
+its own conversation thread.
+
+### Stream A Long Answer
+
+For live output, use `bot.stream(...)`:
+
+```python
+from nanobot import STREAM_EVENT_TEXT_DELTA
+
+async for event in bot.stream("Write a migration plan"):
+    if event.type == STREAM_EVENT_TEXT_DELTA:
+        print(event.delta, end="", flush=True)
+```
+
+Streaming returns structured events, so you can also observe tool calls,
+reasoning chunks, completion, and failures.
+
+## Complete Starter Script
+
+Save this as `sdk_demo.py` after `nanobot agent -m "Hello!"` works:
+
+```python
+import asyncio
+import sys
+
+from nanobot import (
+    STREAM_EVENT_RUN_COMPLETED,
+    STREAM_EVENT_RUN_FAILED,
+    STREAM_EVENT_TEXT_DELTA,
+    STREAM_EVENT_TOOL_STARTED,
+    Nanobot,
+)
+
+
+async def main() -> None:
+    prompt = " ".join(sys.argv[1:]) or "Explain what nanobot is in one paragraph."
+    session_key = "sdk:demo"
+
+    async with Nanobot.from_config() as bot:
+        print(f"model: {bot.runtime.model}")
+        print(f"workspace: {bot.runtime.workspace}")
+        print()
+
+        final_result = None
+        async for event in bot.stream(prompt, session_key=session_key):
+            if event.type == STREAM_EVENT_TEXT_DELTA:
+                print(event.delta, end="", flush=True)
+            elif event.type == STREAM_EVENT_TOOL_STARTED:
+                print(f"\n[tool] {event.name}", flush=True)
+            elif event.type == STREAM_EVENT_RUN_COMPLETED:
+                final_result = event.result
+            elif event.type == STREAM_EVENT_RUN_FAILED:
+                raise RuntimeError(event.error or "nanobot run failed")
+
+        print()
+        if final_result is not None:
+            print(f"\nstop_reason: {final_result.stop_reason}")
+            print(f"tools_used: {final_result.tools_used}")
+            print(f"usage: {final_result.usage}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+Run it:
+
+```bash
+python sdk_demo.py "List the top-level files in the current workspace."
+```
+
+You should see the configured model, workspace path, streamed assistant text,
+and final run metadata. The exact answer depends on your config and workspace,
+but a file-listing prompt may look like this:
+
+```text
+model: openai/gpt-4.1-mini
+workspace: /Users/alice/.nanobot/workspace
+
+[tool] list_dir
+Here are the top-level files I found...
+
+stop_reason: completed
+tools_used: ['list_dir']
+usage: {'prompt_tokens': ..., 'completion_tokens': ..., 'total_tokens': ...}
+```
+
+This script shows the usual production shape: create one `Nanobot`, choose a
+stable `session_key`, stream events, keep the final `RunResult`, and let
+`async with` close runtime resources.
+
+## Core Concepts
+
+| Concept | Meaning |
+|---------|---------|
+| `Nanobot` | The SDK object that owns one configured agent runtime. |
+| Run | One call to `bot.run(...)`, `bot.run_streamed(...)`, or `bot.stream(...)`. |
+| `session_key` | The conversation history key. Reuse it to continue a thread; change it to isolate a thread. |
+| Workspace | The local directory where file tools and shell tools operate. |
+| Tools | Capabilities the agent may call, such as file access, shell, web, or custom tools from your config. |
+| Memory | Long-term memory files managed by nanobot. |
+| Stream event | A typed event such as `text.delta`, `tool.started`, or `run.completed`. |
+| Model override | A temporary model or model preset used for one SDK instance or one run. |
+
+For most users, the mental model is:
+
+1. Create a `Nanobot` from config.
+2. Pick a `session_key`.
+3. Call `run` or `stream`.
+4. Read `RunResult` or stream events.
+5. Use session/memory/runtime helpers only when you need more control.
+
+## SDK Or OpenAI-Compatible API?
+
+nanobot has two programming surfaces:
+
+| Use | Choose | Why |
+|-----|--------|-----|
+| Python code running in the same process as nanobot | Python SDK | Direct access to `RunResult`, sessions, memory, runtime helpers, hooks, and stream events. |
+| Existing OpenAI-compatible clients, another language, or a separate process | [OpenAI-Compatible API](openai-api.md) | HTTP `/v1/chat/completions` compatibility with familiar client libraries. |
+
+The Python SDK is best when you are writing evals, notebooks, benchmark
+runners, product backends, local scripts, or integrations that should control
+nanobot directly.
+
+The OpenAI-compatible API is best when you already have an HTTP client, want
+process isolation, or need to call nanobot from a non-Python service.
 
 ## Common Patterns
 
 ### Use a specific config or workspace
 
+Set the workspace when your agent should work inside a specific project:
+
 ```python
 from nanobot import Nanobot
 
-bot = Nanobot.from_config(
+async with Nanobot.from_config(workspace="/my/project") as bot:
-    config_path="~/.nanobot/config.json",
+    result = await bot.run("Explain the project structure")
-    workspace="/my/project",
-)
 ```
 
+Use a custom config when you run multiple nanobot instances or test an isolated
+setup:
+
+```python
+async with Nanobot.from_config(
+    config_path="./bot-a/config.json",
+    workspace="./bot-a/workspace",
+) as bot:
+    result = await bot.run("Hello from bot A")
+```
+
+The config controls what nanobot may use. The workspace is where nanobot keeps
+state for that instance. See [multiple-instances.md](multiple-instances.md) for
+multi-instance CLI and gateway examples.
+
+### Choose a default or per-run model
+
+Set the SDK instance default model when you create the bot:
+
+```python
+bot = Nanobot.from_config(model="openai/gpt-4.1")
+```
+
+Override the model for one run without changing the instance default:
+
+```python
+result = await bot.run("Summarize this file", model="openai/gpt-4.1-mini")
+```
+
+Model presets from `config.json` work the same way:
+
+```python
+bot = Nanobot.from_config(model_preset="fast")
+
+result = await bot.run("Think deeply about this bug", model_preset="reasoning")
+```
+
+`model` and `model_preset` are mutually exclusive.
+
+For first setup, prefer named presets in `config.json`. Mixing an API key from
+one provider with a model ID from another is the most common first-run failure.
+For the exact difference between `provider`, `model`, `apiKey`, and `apiBase`,
+see [Providers: Provider, Model, API Key, and Base URL](providers.md#provider-model-api-key-and-base-url).
+If a run fails before the SDK does anything interesting, confirm the same
+provider and model work with `nanobot agent -m "Hello!"` first.
+
 ### Isolate conversations with `session_key`
 
 Different session keys keep independent conversation history:
@@ -51,9 +306,131 @@ await bot.run("hi", session_key="user-alice")
 await bot.run("hi", session_key="task-42")
 ```
 
+Use stable keys in product code:
+
+```python
+session_key = f"user:{user_id}"
+result = await bot.run(user_message, session_key=session_key)
+```
+
+Avoid using the default `"sdk:default"` for multiple users or unrelated
+workflows. It is convenient for local experiments, but stable product code
+should choose explicit keys such as `user:<id>`, `project:<id>`, or
+`eval:<case-id>`.
+
+### Handle failures
+
+For a normal non-streamed run, catch exceptions around `bot.run(...)` and inspect
+`RunResult.error` when the runtime returns a structured failure:
+
+```python
+try:
+    result = await bot.run("Review this repo", session_key="project:demo")
+except Exception as exc:
+    print(f"SDK call failed before a result was returned: {exc}")
+else:
+    if result.error:
+        print(f"Agent run failed: {result.error}")
+    else:
+        print(result.content)
+```
+
+For streamed runs, either consume the stream to completion or close it:
+
+```python
+run = await bot.run_streamed("Write a long answer", session_key="task:123")
+try:
+    async for event in run.stream_events():
+        ...
+finally:
+    if not run.done:
+        await run.aclose()
+```
+
+Use `await run.cancel()` when the user presses a stop button or leaves the page
+before the stream finishes.
+
+### Stream long-running output
+
+Use `bot.stream()` when you want Cursor/OpenAI-style live events instead of
+waiting for the final `RunResult`:
+
+```python
+from nanobot import (
+    STREAM_EVENT_RUN_COMPLETED,
+    STREAM_EVENT_TEXT_DELTA,
+    STREAM_EVENT_TOOL_STARTED,
+)
+
+async for event in bot.stream("Review this repository"):
+    if event.type == STREAM_EVENT_TEXT_DELTA:
+        print(event.delta, end="", flush=True)
+    elif event.type == STREAM_EVENT_TOOL_STARTED:
+        print(f"\nusing {event.name}")
+    elif event.type == STREAM_EVENT_RUN_COMPLETED:
+        print("\nfinal:", event.result.content)
+```
+
+Use `run_streamed()` when you also want a handle you can wait on:
+
+```python
+from nanobot import STREAM_EVENT_TEXT_DELTA
+
+run = await bot.run_streamed("Write a detailed migration plan")
+
+async for event in run.stream_events():
+    if event.type == STREAM_EVENT_TEXT_DELTA:
+        print(event.delta, end="", flush=True)
+
+result = await run.wait()
+```
+
+Always either consume the stream, call `await run.wait()` / `await run.text()`,
+or close it with `await run.cancel()` / `await run.aclose()`. Exiting
+`stream_events()` or `bot.stream()` early cancels the underlying run so a
+half-consumed stream cannot leave a background task stuck behind backpressure.
+
+### Import an existing transcript
+
+This is useful for evals, benchmark runners, migrations, and tests.
+
+Use `bot.sessions.ingest()` when you already have a transcript and want it to
+become nanobot session history. Ingesting a transcript does not call the model,
+execute tools, update memory, or compact automatically.
+
+```python
+await bot.sessions.ingest(
+    "eval:case-1",
+    [
+        {
+            "role": "user",
+            "content": "I graduated with a degree in Business Administration.",
+            "timestamp": "2023/05/30 (Tue) 17:27",
+            "source_session_id": "answer_280352e9",
+        },
+        {
+            "role": "assistant",
+            "content": "Congratulations on your degree.",
+            "timestamp": "2023/05/30 (Tue) 17:27",
+        },
+    ],
+    source="longmemeval",
+)
+
+await bot.runtime.compact_session("eval:case-1")
+
+result = await bot.run(
+    "Current Date: 2023/05/30 (Tue) 23:40\n"
+    "Question: What degree did I graduate with?",
+    session_key="eval:case-1",
+)
+print(result.content)
+```
+
 ### Attach hooks for observability
 
-Hooks let you inspect tool calls, streaming, and iteration state without modifying nanobot internals:
+Hooks are an advanced escape hatch. Use them when you want custom logging,
+metrics, tracing, or output post-processing without modifying nanobot internals:
 
 ```python
 from nanobot.agent import AgentHook, AgentHookContext
@@ -68,9 +445,25 @@ class AuditHook(AgentHook):
 result = await bot.run("Review this change", hooks=[AuditHook()])
 ```
 
+## Where To Go Next
+
+The SDK page is the programming entry point. The fuller conceptual and
+configuration docs remain the source of truth for the runtime around it:
+
+| Need | Read |
+|------|------|
+| First working install and config | [Install and Quick Start](quick-start.md) |
+| Mental model for config, workspace, sessions, tools, and memory | [Concepts](concepts.md) |
+| Provider/model/API key/base URL matching | [Providers and Models](providers.md) |
+| Pasteable provider recipes | [Provider Cookbook](provider-cookbook.md) |
+| Complete configuration reference | [Configuration](configuration.md) |
+| Long-term memory design | [Memory](memory.md) |
+| HTTP API instead of Python SDK | [OpenAI-Compatible API](openai-api.md) |
+| Debugging install, config, provider, or runtime failures | [Troubleshooting](troubleshooting.md) |
+
 ## API Reference
 
-### `Nanobot.from_config(config_path=None, *, workspace=None)`
+### `Nanobot.from_config(config_path=None, *, workspace=None, model=None, model_preset=None)`
 
 Create a `Nanobot` instance from a config file.
 
@@ -78,10 +471,13 @@ Create a `Nanobot` instance from a config file.
 |-------|------|---------|-------------|
 | `config_path` | `str \| Path \| None` | `None` | Path to `config.json`. Defaults to `~/.nanobot/config.json`. |
 | `workspace` | `str \| Path \| None` | `None` | Override the workspace directory from config. |
+| `model` | `str \| None` | `None` | Override the instance default model. |
+| `model_preset` | `str \| None` | `None` | Override the instance default model preset from `config.json`. |
 
 Raises `FileNotFoundError` if an explicit config path does not exist.
+Raises `ValueError` if both `model` and `model_preset` are provided.
 
-### `await bot.run(message, *, session_key="sdk:default", hooks=None)`
+### `await bot.run(...)`
 
 Run the agent once and return a `RunResult`.
 
@@ -89,11 +485,93 @@ Run the agent once and return a `RunResult`.
 |-------|------|---------|-------------|
 | `message` | `str` | *(required)* | The user message to process. |
 | `session_key` | `str` | `"sdk:default"` | Session identifier for conversation isolation. Different keys get independent history. |
+| `channel` | `str` | `"cli"` | Logical channel label used in runtime context. |
+| `chat_id` | `str` | `"direct"` | Logical chat identifier used in runtime context. |
+| `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. |
 | `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. |
+
+`model` and `model_preset` are per-run overrides and do not change
+`bot.runtime.model` after the run completes. They are mutually exclusive.
+
+### `await bot.run_streamed(...)`
+
+Start a streamed agent turn and return a `RunStream`. It accepts the same
+parameters as `bot.run(...)`.
+
+```python
+run = await bot.run_streamed("Generate a long answer")
+
+async for event in run.stream_events():
+    ...
+
+result = await run.wait()
+```
+
+### `bot.stream(...)`
+
+Convenience wrapper around `run_streamed()` for direct event iteration. It
+accepts the same parameters as `bot.run(...)`.
+
+```python
+async for event in bot.stream("Generate a long answer"):
+    ...
+```
+
+### `RunStream`
+
+| Method | Description |
+|--------|-------------|
+| `stream_events()` | Single-consumer async iterator of `StreamEvent` objects. |
+| `await wait()` | Wait for the run to finish and return `RunResult`. |
+| `await text()` | Wait for the run to finish and return `RunResult.content`. |
+| `await cancel()` | Cancel the run and release stream resources. |
+| `await aclose()` | Close the stream; equivalent cleanup primitive for `async with` / manual lifecycle code. |
+
+Normal SDK runs with different session keys may overlap. Runs that use per-run
+`model` or `model_preset` overrides are exclusive while the override is active,
+because the current `AgentLoop` provider/model state is mutable.
+
+### `StreamEvent`
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `type` | `StreamEventType` | Event type, such as `text.delta` or `run.completed`. |
+| `delta` | `str` | Incremental text or reasoning chunk. |
+| `content` | `str` | Completed text segment or final content. |
+| `result` | `RunResult \| None` | Present on `run.completed`. |
+| `name` | `str \| None` | Tool name for tool events. |
+| `tool_call_id` | `str \| None` | Provider tool call id when available. |
+| `arguments` | `dict \| None` | Tool arguments when available. |
+| `iteration` | `int \| None` | Agent loop iteration when available. |
+| `resuming` | `bool \| None` | Whether a text segment ended before more tool work. |
+| `usage` | `dict[str, int]` | Token usage on completion events. |
+| `error` | `str \| None` | Error text on failed events. |
+| `metadata` | `dict` | Additional event metadata. |
+
+Use the exported constants instead of hard-coded strings when possible:
+
+| Constant | Value |
+|----------|-------|
+| `STREAM_EVENT_RUN_STARTED` | `run.started` |
+| `STREAM_EVENT_TEXT_DELTA` | `text.delta` |
+| `STREAM_EVENT_TEXT_COMPLETED` | `text.completed` |
+| `STREAM_EVENT_REASONING_DELTA` | `reasoning.delta` |
+| `STREAM_EVENT_REASONING_COMPLETED` | `reasoning.completed` |
+| `STREAM_EVENT_TOOL_STARTED` | `tool.started` |
+| `STREAM_EVENT_TOOL_COMPLETED` | `tool.completed` |
+| `STREAM_EVENT_TOOL_FAILED` | `tool.failed` |
+| `STREAM_EVENT_RUN_COMPLETED` | `run.completed` |
+| `STREAM_EVENT_RUN_FAILED` | `run.failed` |
+
+`STREAM_EVENT_TYPES` contains all stable v1 event values.
 
 ### `await bot.aclose()`
 
-Release resources held by the SDK instance, including MCP connections. The async context manager calls this automatically:
+Release resources held by the SDK instance, including tool connections. The async context manager calls this automatically:
 
 ```python
 async with Nanobot.from_config() as bot:
@@ -105,8 +583,48 @@ async with Nanobot.from_config() as bot:
 | Field | Type | Description |
 |-------|------|-------------|
 | `content` | `str` | The agent's final text response. |
-| `tools_used` | `list[str]` | Reserved for richer SDK introspection; may be empty in current versions. |
+| `tools_used` | `list[str]` | Tool names used during the run. |
-| `messages` | `list[dict]` | Reserved for richer SDK introspection; may be empty in current versions. |
+| `messages` | `list[dict]` | Final message list from the run. |
+| `usage` | `dict[str, int]` | Token usage reported or estimated by the runtime. |
+| `stop_reason` | `str \| None` | Why the run stopped, such as `"completed"` or `"max_iterations"`. |
+| `error` | `str \| None` | Error text when the run failed inside the agent runtime. |
+| `metadata` | `dict` | Outbound metadata such as latency. |
+
+## Session, Memory, And Runtime Helpers
+
+### `bot.sessions`
+
+| Method | Description |
+|--------|-------------|
+| `await ingest(session_key, messages, metadata=None, source=None, save=True)` | Import existing transcript messages without running the model. |
+| `get(session_key)` | Return a `SessionSnapshot`, or `None` if missing. |
+| `list()` | Return compact `SessionInfo` rows. |
+| `export(session_key)` | Return a full `SessionSnapshot` suitable for JSON serialization. |
+| `clear(session_key)` | Clear and persist one session. |
+| `delete(session_key)` | Delete one session from disk and cache. |
+| `flush()` | Flush cached sessions to durable storage. |
+
+Ingested messages must include `role` and `content`. Roles may be `user`,
+`assistant`, `tool`, or `system`. Other fields, such as `timestamp`,
+`source_session_id`, or `source_date`, are persisted as message metadata.
+
+### `bot.memory`
+
+| Method | Description |
+|--------|-------------|
+| `read()` | Read `memory/MEMORY.md`. |
+| `write(text)` | Overwrite `memory/MEMORY.md`. |
+| `append_history(text, session_key=None)` | Append one `memory/history.jsonl` entry and return its cursor. |
+| `read_history(session_key=None)` | Read memory history entries, optionally filtered by session key. |
+
+### `bot.runtime`
+
+| Method / Property | Description |
+|-------------------|-------------|
+| `model` | Current runtime model name. |
+| `workspace` | Current runtime workspace path. |
+| `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. |
 
 ## Hooks
 
@@ -223,12 +741,12 @@ class TimingHook(AgentHook):
 
 
 async def main() -> None:
-    bot = Nanobot.from_config(workspace="/my/project")
+    async with Nanobot.from_config(workspace="/my/project") as bot:
-    result = await bot.run(
+        result = await bot.run(
-        "Explain the main function",
+            "Explain the main function",
-        session_key="sdk:demo",
+            session_key="sdk:demo",
-        hooks=[TimingHook()],
+            hooks=[TimingHook()],
-    )
+        )
     print(result.content)
 
 

nanobot/__init__.py

@@ -30,7 +30,23 @@ __logo__ = "🐈"
 
 _LAZY_EXPORTS = {
     "Nanobot": ".nanobot",
+    "RunStream": ".nanobot",
     "RunResult": ".nanobot",
+    "SessionInfo": ".nanobot",
+    "SessionSnapshot": ".nanobot",
+    "STREAM_EVENT_REASONING_COMPLETED": ".nanobot",
+    "STREAM_EVENT_REASONING_DELTA": ".nanobot",
+    "STREAM_EVENT_RUN_COMPLETED": ".nanobot",
+    "STREAM_EVENT_RUN_FAILED": ".nanobot",
+    "STREAM_EVENT_RUN_STARTED": ".nanobot",
+    "STREAM_EVENT_TEXT_COMPLETED": ".nanobot",
+    "STREAM_EVENT_TEXT_DELTA": ".nanobot",
+    "STREAM_EVENT_TOOL_COMPLETED": ".nanobot",
+    "STREAM_EVENT_TOOL_FAILED": ".nanobot",
+    "STREAM_EVENT_TOOL_STARTED": ".nanobot",
+    "STREAM_EVENT_TYPES": ".nanobot",
+    "StreamEvent": ".nanobot",
+    "StreamEventType": ".nanobot",
 }
 
 
@@ -45,4 +61,23 @@ def __getattr__(name: str):
     return val
 
 
-__all__ = ["Nanobot", "RunResult"]
+__all__ = [
+    "Nanobot",
+    "RunResult",
+    "RunStream",
+    "SessionInfo",
+    "SessionSnapshot",
+    "STREAM_EVENT_REASONING_COMPLETED",
+    "STREAM_EVENT_REASONING_DELTA",
+    "STREAM_EVENT_RUN_COMPLETED",
+    "STREAM_EVENT_RUN_FAILED",
+    "STREAM_EVENT_RUN_STARTED",
+    "STREAM_EVENT_TEXT_COMPLETED",
+    "STREAM_EVENT_TEXT_DELTA",
+    "STREAM_EVENT_TOOL_COMPLETED",
+    "STREAM_EVENT_TOOL_FAILED",
+    "STREAM_EVENT_TOOL_STARTED",
+    "STREAM_EVENT_TYPES",
+    "StreamEvent",
+    "StreamEventType",
+]

nanobot/agent/hook.py

@@ -176,12 +176,26 @@ class SDKCaptureHook(AgentHook):
         super().__init__()
         self.tools_used: list[str] = []
         self.messages: list[dict[str, Any]] = []
+        self.usage: dict[str, int] = {}
+        self.stop_reason: str | None = None
+        self.error: str | None = None
+        self.tool_events: list[dict[str, str]] = []
+        self.had_injections: bool = False
 
     async def after_iteration(self, context: AgentHookContext) -> None:
         for call in context.tool_calls:
             self.tools_used.append(call.name)
         self.messages = list(context.messages)
+        self.usage = dict(context.usage)
+        self.stop_reason = context.stop_reason
+        self.error = context.error
+        self.tool_events = list(context.tool_events)
 
     async def after_run(self, context: AgentRunHookContext) -> None:
         self.tools_used = list(context.tools_used)
         self.messages = list(context.messages)
+        self.usage = dict(context.usage)
+        self.stop_reason = context.stop_reason
+        self.error = context.error
+        self.tool_events = list(context.tool_events)
+        self.had_injections = context.had_injections

nanobot/agent/loop.py

@@ -127,8 +127,9 @@ class TurnContext:
     pending_summary: str | None = None
 
     ephemeral: bool = False
+    run_extra_hooks_for_ephemeral: bool = False
+    hooks: list[AgentHook] = field(default_factory=list)
     tools: ToolRegistry | None = None
-    extra_hooks: list[AgentHook] | None = None
 
     turn_wall_started_at: float = field(default_factory=time.time)
     visible_run_started_at: float | None = None
@@ -693,8 +694,9 @@ class AgentLoop:
         session_key: str | None = None,
         pending_queue: asyncio.Queue | None = None,
         ephemeral: bool = False,
+        run_extra_hooks_for_ephemeral: bool = False,
+        hooks: list[AgentHook] | None = None,
         tools: ToolRegistry | None = None,
-        extra_hooks: list[AgentHook] | None = None,
     ) -> tuple[str | None, list[str], list[dict], str, bool]:
         """Run the agent iteration loop.
 
@@ -720,10 +722,10 @@ class AgentLoop:
             set_tool_context=self._set_tool_context,
             on_iteration=lambda iteration: setattr(self, "_current_iteration", iteration),
         )
+        run_hooks = [*self._extra_hooks, *(hooks or [])]
         hook: AgentHook = loop_hook
-        turn_hooks = extra_hooks if extra_hooks is not None else self._extra_hooks
+        if run_hooks and (not ephemeral or run_extra_hooks_for_ephemeral):
-        if not ephemeral and turn_hooks:
+            hook = CompositeHook([loop_hook, *run_hooks])
-            hook = CompositeHook([loop_hook] + turn_hooks)
 
         async def _checkpoint(payload: dict[str, Any]) -> None:
             if session is None:
@@ -1241,8 +1243,9 @@ class AgentLoop:
         on_stream_end: Callable[..., Awaitable[None]] | None = None,
         pending_queue: asyncio.Queue | None = None,
         ephemeral: bool = False,
+        run_extra_hooks_for_ephemeral: bool = False,
+        hooks: list[AgentHook] | None = None,
         tools: ToolRegistry | None = None,
-        extra_hooks: list[AgentHook] | None = None,
     ) -> OutboundMessage | None:
         """Process a single inbound message and return the response."""
         self._refresh_provider_snapshot()
@@ -1274,8 +1277,9 @@ class AgentLoop:
             on_stream_end=on_stream_end,
             pending_queue=pending_queue,
             ephemeral=ephemeral,
+            run_extra_hooks_for_ephemeral=run_extra_hooks_for_ephemeral,
+            hooks=list(hooks or []),
             tools=tools,
-            extra_hooks=extra_hooks,
         )
 
         while ctx.state is not TurnState.DONE:
@@ -1501,8 +1505,9 @@ class AgentLoop:
             session_key=ctx.session_key,
             pending_queue=ctx.pending_queue,
             ephemeral=ctx.ephemeral,
+            run_extra_hooks_for_ephemeral=ctx.run_extra_hooks_for_ephemeral,
+            hooks=ctx.hooks,
             tools=ctx.tools,
-            extra_hooks=ctx.extra_hooks,
         )
         final_content, tools_used, all_msgs, stop_reason, had_injections = result
         ctx.final_content = final_content
@@ -1811,14 +1816,16 @@ class AgentLoop:
         session_key: str = "cli:direct",
         channel: str = "cli",
         chat_id: str = "direct",
+        sender_id: str = "user",
         media: list[str] | None = None,
         on_progress: Callable[..., Awaitable[None]] | None = None,
         on_stream: Callable[[str], Awaitable[None]] | None = None,
         on_stream_end: Callable[..., Awaitable[None]] | None = None,
         ephemeral: bool = False,
+        _run_extra_hooks_for_ephemeral: bool = False,
+        hooks: list[AgentHook] | None = None,
         tools: ToolRegistry | None = None,
         persist_user_message: bool = True,
-        extra_hooks: list[AgentHook] | None = None,
     ) -> OutboundMessage | None:
         """Process a message directly and return the outbound payload."""
         await self._connect_mcp()
@@ -1826,7 +1833,7 @@ class AgentLoop:
         if not persist_user_message:
             metadata[turn_continuation.SKIP_USER_PERSIST_META] = True
         msg = InboundMessage(
-            channel=channel, sender_id="user", chat_id=chat_id,
+            channel=channel, sender_id=sender_id, chat_id=chat_id,
             content=content, media=media or [], metadata=metadata,
         )
         # Share the dispatch lock so direct calls serialize with bus turns.
@@ -1840,10 +1847,12 @@ class AgentLoop:
                     "on_stream_end": on_stream_end,
                     "ephemeral": ephemeral,
                 }
+                if _run_extra_hooks_for_ephemeral:
+                    kwargs["run_extra_hooks_for_ephemeral"] = True
+                if hooks is not None:
+                    kwargs["hooks"] = hooks
                 if tools is not None:
                     kwargs["tools"] = tools
-                if extra_hooks is not None:
-                    kwargs["extra_hooks"] = extra_hooks
                 return await self._process_message(
                     msg,
                     **kwargs,

nanobot/nanobot.py

@@ -2,22 +2,62 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+import asyncio
+from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Any
 
 from nanobot.agent.hook import AgentHook, SDKCaptureHook
 from nanobot.agent.loop import AgentLoop
+from nanobot.config.schema import Config
 from nanobot.providers.image_generation import image_gen_provider_configs
+from nanobot.sdk.clients import MemoryClient, RuntimeClient, SessionClient
+from nanobot.sdk.runtime import (
+    SDKRuntimeController,
+    build_process_direct_kwargs,
+    ensure_single_model_selector,
+)
+from nanobot.sdk.streaming import RunStream, SDKStreamEmitter, SDKStreamingHook
+from nanobot.sdk.types import (
+    STREAM_EVENT_REASONING_COMPLETED,
+    STREAM_EVENT_REASONING_DELTA,
+    STREAM_EVENT_RUN_COMPLETED,
+    STREAM_EVENT_RUN_FAILED,
+    STREAM_EVENT_RUN_STARTED,
+    STREAM_EVENT_TEXT_COMPLETED,
+    STREAM_EVENT_TEXT_DELTA,
+    STREAM_EVENT_TOOL_COMPLETED,
+    STREAM_EVENT_TOOL_FAILED,
+    STREAM_EVENT_TOOL_STARTED,
+    STREAM_EVENT_TYPES,
+    RunResult,
+    SessionInfo,
+    SessionSnapshot,
+    StreamEvent,
+    StreamEventType,
+    result_from_response,
+)
 
-
+__all__ = [
-@dataclass(slots=True)
+    "Nanobot",
-class RunResult:
+    "RunResult",
-    """Result of a single agent run."""
+    "RunStream",
-
+    "SessionInfo",
-    content: str
+    "SessionSnapshot",
-    tools_used: list[str]
+    "STREAM_EVENT_REASONING_COMPLETED",
-    messages: list[dict[str, Any]]
+    "STREAM_EVENT_REASONING_DELTA",
+    "STREAM_EVENT_RUN_COMPLETED",
+    "STREAM_EVENT_RUN_FAILED",
+    "STREAM_EVENT_RUN_STARTED",
+    "STREAM_EVENT_TEXT_COMPLETED",
+    "STREAM_EVENT_TEXT_DELTA",
+    "STREAM_EVENT_TOOL_COMPLETED",
+    "STREAM_EVENT_TOOL_FAILED",
+    "STREAM_EVENT_TOOL_STARTED",
+    "STREAM_EVENT_TYPES",
+    "StreamEvent",
+    "StreamEventType",
+]
 
 
 class Nanobot:
@@ -30,8 +70,13 @@ class Nanobot:
         print(result.content)
     """
 
-    def __init__(self, loop: AgentLoop) -> None:
+    def __init__(self, loop: AgentLoop, *, config: Config | None = None) -> None:
         self._loop = loop
+        self._config = config
+        self._runtime_overrides = SDKRuntimeController(loop, config=config)
+        self.sessions = SessionClient(loop)
+        self.memory = MemoryClient(loop)
+        self.runtime = RuntimeClient(loop)
 
     @classmethod
     def from_config(
@@ -39,6 +84,8 @@ class Nanobot:
         config_path: str | Path | None = None,
         *,
         workspace: str | Path | None = None,
+        model: str | None = None,
+        model_preset: str | None = None,
     ) -> Nanobot:
         """Create a Nanobot instance from a config file.
 
@@ -46,10 +93,12 @@ class Nanobot:
             config_path: Path to ``config.json``.  Defaults to
                 ``~/.nanobot/config.json``.
             workspace: Override the workspace directory from config.
+            model: Override the instance default model.
+            model_preset: Override the instance default model preset.
         """
         from nanobot.config.loader import load_config, resolve_config_env_vars
-        from nanobot.config.schema import Config
 
+        ensure_single_model_selector(model=model, model_preset=model_preset)
         resolved: Path | None = None
         if config_path is not None:
             resolved = Path(config_path).expanduser().resolve()
@@ -61,19 +110,32 @@ class Nanobot:
             config.agents.defaults.workspace = str(
                 Path(workspace).expanduser().resolve()
             )
+        if model is not None:
+            config.agents.defaults.model_preset = None
+            config.agents.defaults.model = model
+            config.agents.defaults.provider = "auto"
+        elif model_preset is not None:
+            config.agents.defaults.model_preset = model_preset
 
         loop = AgentLoop.from_config(
             config,
             image_generation_provider_configs=image_gen_provider_configs(config),
         )
-        return cls(loop)
+        return cls(loop, config=config)
 
     async def run(
         self,
         message: str,
         *,
         session_key: str = "sdk:default",
+        channel: str = "cli",
+        chat_id: str = "direct",
+        sender_id: str = "user",
+        media: list[str] | None = None,
+        ephemeral: bool = False,
         hooks: list[AgentHook] | None = None,
+        model: str | None = None,
+        model_preset: str | None = None,
     ) -> RunResult:
         """Run the agent once and return the result.
 
@@ -81,22 +143,150 @@ class Nanobot:
             message: The user message to process.
             session_key: Session identifier for conversation isolation.
                 Different keys get independent history.
+            channel: Logical channel label for runtime context.
+            chat_id: Logical chat identifier for runtime context.
+            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.
             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.
         """
         capture = SDKCaptureHook()
-        base_hooks = list(hooks) if hooks is not None else list(self._loop._extra_hooks or [])
+        per_run_hooks = [capture, *(hooks or [])]
-        response = await self._loop.process_direct(
+        async with self._runtime_overrides.override(model=model, model_preset=model_preset):
+            kwargs = build_process_direct_kwargs(
+                session_key=session_key,
+                channel=channel,
+                chat_id=chat_id,
+                sender_id=sender_id,
+                media=media,
+                ephemeral=ephemeral,
+            )
+            response = await self._loop.process_direct(
+                message,
+                **kwargs,
+                hooks=per_run_hooks,
+            )
+
+        return result_from_response(response, capture)
+
+    async def run_streamed(
+        self,
+        message: str,
+        *,
+        session_key: str = "sdk:default",
+        channel: str = "cli",
+        chat_id: str = "direct",
+        sender_id: str = "user",
+        media: list[str] | None = None,
+        ephemeral: bool = False,
+        hooks: list[AgentHook] | None = None,
+        model: str | None = None,
+        model_preset: str | None = None,
+    ) -> RunStream:
+        """Start a streamed run and return a handle for events and final result."""
+        ensure_single_model_selector(model=model, model_preset=model_preset)
+        queue: asyncio.Queue[StreamEvent | object] = asyncio.Queue(maxsize=256)
+        emitter = SDKStreamEmitter(queue)
+        stream_hook = SDKStreamingHook(emitter)
+        capture = SDKCaptureHook()
+        per_run_hooks = [capture, stream_hook, *(hooks or [])]
+
+        async def _on_stream(delta: str) -> None:
+            await emitter.text_delta(delta)
+
+        async def _on_stream_end(*_args: Any, resuming: bool = False, **_kwargs: Any) -> None:
+            await emitter.text_completed(resuming=resuming)
+
+        async def _run() -> RunResult:
+            async with self._runtime_overrides.override(model=model, model_preset=model_preset):
+                kwargs = build_process_direct_kwargs(
+                    session_key=session_key,
+                    channel=channel,
+                    chat_id=chat_id,
+                    sender_id=sender_id,
+                    media=media,
+                    ephemeral=ephemeral,
+                    on_stream=_on_stream,
+                    on_stream_end=_on_stream_end,
+                )
+                await emitter.emit(StreamEvent(
+                    type=STREAM_EVENT_RUN_STARTED,
+                    metadata={
+                        "session_key": session_key,
+                        "channel": channel,
+                        "chat_id": chat_id,
+                        "sender_id": sender_id,
+                        "model": self._loop.model,
+                        "model_preset": (
+                            model_preset if model_preset is not None else self._loop.model_preset
+                        ),
+                    },
+                ))
+                try:
+                    response = await self._loop.process_direct(
+                        message,
+                        **kwargs,
+                        hooks=per_run_hooks,
+                    )
+                    await emitter.text_completed(resuming=False, force=False)
+                    result = result_from_response(response, capture)
+                    await emitter.emit(StreamEvent(
+                        type=STREAM_EVENT_RUN_COMPLETED,
+                        content=result.content,
+                        result=result,
+                        usage=dict(result.usage),
+                        metadata=dict(result.metadata),
+                    ))
+                    return result
+                except Exception as exc:
+                    await emitter.emit(StreamEvent(
+                        type=STREAM_EVENT_RUN_FAILED,
+                        error=str(exc),
+                        metadata={"exception_type": type(exc).__name__},
+                    ))
+                    raise
+                finally:
+                    emitter.close()
+
+        task = asyncio.create_task(_run())
+        return RunStream(task, queue)
+
+    async def stream(
+        self,
+        message: str,
+        *,
+        session_key: str = "sdk:default",
+        channel: str = "cli",
+        chat_id: str = "direct",
+        sender_id: str = "user",
+        media: list[str] | None = None,
+        ephemeral: bool = False,
+        hooks: list[AgentHook] | None = None,
+        model: str | None = None,
+        model_preset: str | None = None,
+    ) -> AsyncIterator[StreamEvent]:
+        """Stream events for one agent turn."""
+        run = await self.run_streamed(
             message,
             session_key=session_key,
-            extra_hooks=[capture, *base_hooks],
+            channel=channel,
-        )
+            chat_id=chat_id,
-
+            sender_id=sender_id,
-        content = (response.content if response else None) or ""
+            media=media,
-        return RunResult(
+            ephemeral=ephemeral,
-            content=content,
+            hooks=hooks,
-            tools_used=capture.tools_used,
+            model=model,
-            messages=capture.messages,
+            model_preset=model_preset,
         )
+        try:
+            async for event in run.stream_events():
+                yield event
+            await run.wait()
+        finally:
+            if not run.done:
+                await run.aclose()
 
     async def aclose(self) -> None:
         """Release resources held by this instance (MCP connections, etc.)."""

nanobot/sdk/__init__.py

@@ -0,0 +1 @@
+"""Internal helpers for the high-level nanobot Python SDK."""

nanobot/sdk/clients.py

@@ -0,0 +1,165 @@
+"""Small convenience clients exposed by the high-level Python SDK."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from copy import deepcopy
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from nanobot.sdk.types import (
+    SessionInfo,
+    SessionSnapshot,
+    snapshot_from_payload,
+    snapshot_from_session,
+)
+
+if TYPE_CHECKING:
+    from nanobot.agent.loop import AgentLoop
+
+
+class SessionClient:
+    """Session management helpers exposed through ``bot.sessions``."""
+
+    _RESERVED_MESSAGE_KEYS = {"role", "content"}
+    _VALID_ROLES = {"user", "assistant", "tool", "system"}
+
+    def __init__(self, loop: AgentLoop) -> None:
+        self._loop = loop
+
+    async def ingest(
+        self,
+        session_key: str,
+        messages: Iterable[Mapping[str, Any]],
+        *,
+        metadata: Mapping[str, Any] | None = None,
+        source: str | None = None,
+        save: bool = True,
+    ) -> SessionSnapshot:
+        """Import an existing transcript without running the model."""
+        session = self._loop.sessions.get_or_create(session_key)
+        if metadata:
+            session.metadata.update(deepcopy(dict(metadata)))
+
+        for raw in messages:
+            if "role" not in raw:
+                raise ValueError("ingested messages must include a role")
+            if "content" not in raw:
+                raise ValueError("ingested messages must include content")
+            role = str(raw["role"]).strip()
+            if role not in self._VALID_ROLES:
+                raise ValueError(f"unsupported message role: {role!r}")
+            extra = {
+                key: deepcopy(value)
+                for key, value in raw.items()
+                if key not in self._RESERVED_MESSAGE_KEYS
+            }
+            if source is not None and "source" not in extra:
+                extra["source"] = source
+            session.add_message(role, deepcopy(raw["content"]), **extra)
+
+        if save:
+            self._loop.sessions.save(session)
+        return snapshot_from_session(session)
+
+    def get(self, session_key: str) -> SessionSnapshot | None:
+        """Return a session snapshot without creating a new session on disk."""
+        cached = self._loop.sessions._cache.get(session_key)
+        if cached is not None:
+            return snapshot_from_session(cached)
+        payload = self._loop.sessions.read_session_file(session_key)
+        if payload is None:
+            return None
+        return snapshot_from_payload(payload)
+
+    def list(self) -> list[SessionInfo]:
+        """List persisted sessions."""
+        return [
+            SessionInfo(
+                key=str(row.get("key") or ""),
+                created_at=row.get("created_at"),
+                updated_at=row.get("updated_at"),
+                title=str(row.get("title") or ""),
+                preview=str(row.get("preview") or ""),
+                path=row.get("path"),
+            )
+            for row in self._loop.sessions.list_sessions()
+        ]
+
+    def export(self, session_key: str) -> SessionSnapshot | None:
+        """Return a full session snapshot suitable for JSON serialization."""
+        return self.get(session_key)
+
+    def clear(self, session_key: str) -> SessionSnapshot:
+        """Clear one session and persist the empty session."""
+        session = self._loop.sessions.get_or_create(session_key)
+        session.clear()
+        self._loop.sessions.save(session)
+        return snapshot_from_session(session)
+
+    def delete(self, session_key: str) -> bool:
+        """Delete one session from disk and cache."""
+        return self._loop.sessions.delete_session(session_key)
+
+    def flush(self) -> int:
+        """Flush cached sessions to durable storage."""
+        return self._loop.sessions.flush_all()
+
+
+class MemoryClient:
+    """Long-term memory helpers exposed through ``bot.memory``."""
+
+    def __init__(self, loop: AgentLoop) -> None:
+        self._loop = loop
+
+    def read(self) -> str:
+        """Read ``memory/MEMORY.md``."""
+        return self._loop.context.memory.read_memory()
+
+    def write(self, text: str) -> None:
+        """Overwrite ``memory/MEMORY.md``."""
+        self._loop.context.memory.write_memory(text)
+
+    def append_history(self, text: str, *, session_key: str | None = None) -> int:
+        """Append one entry to ``memory/history.jsonl`` and return its cursor."""
+        return self._loop.context.memory.append_history(text, session_key=session_key)
+
+    def read_history(self, *, session_key: str | None = None) -> list[dict[str, Any]]:
+        """Read memory history entries, optionally filtered by session."""
+        entries = self._loop.context.memory.read_unprocessed_history(since_cursor=0)
+        if session_key is not None:
+            entries = [entry for entry in entries if entry.get("session_key") == session_key]
+        return deepcopy(entries)
+
+
+class RuntimeClient:
+    """Runtime control helpers exposed through ``bot.runtime``."""
+
+    def __init__(self, loop: AgentLoop) -> None:
+        self._loop = loop
+
+    @property
+    def model(self) -> str:
+        """Current runtime model name."""
+        return self._loop.model
+
+    @property
+    def workspace(self) -> Path:
+        """Current runtime workspace."""
+        return self._loop.workspace
+
+    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)
+        await self._loop.consolidator.maybe_consolidate_by_tokens(
+            session,
+            replay_max_messages=self._loop._max_messages,
+        )
+        return snapshot_from_session(self._loop.sessions.get_or_create(session_key))
+
+    async def compact_idle_session(self, session_key: str, *, max_suffix: int = 8) -> str | None:
+        """Run idle-session compaction for one session and return the summary."""
+        return await self._loop.consolidator.compact_idle_session(
+            session_key,
+            max_suffix=max_suffix,
+        )

nanobot/sdk/runtime.py

@@ -0,0 +1,192 @@
+"""Runtime helpers for SDK calls."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING, Any
+
+from nanobot.config.schema import Config, ModelPresetConfig
+from nanobot.providers.factory import ProviderSnapshot, build_provider_snapshot
+
+if TYPE_CHECKING:
+    from nanobot.agent.loop import AgentLoop
+
+
+def ensure_single_model_selector(
+    *,
+    model: str | None,
+    model_preset: str | None,
+) -> None:
+    if model is not None and model_preset is not None:
+        raise ValueError("model and model_preset are mutually exclusive")
+
+
+def build_process_direct_kwargs(
+    *,
+    session_key: str,
+    channel: str,
+    chat_id: str,
+    sender_id: str,
+    media: list[str] | None,
+    ephemeral: bool,
+    on_stream: Any | None = None,
+    on_stream_end: Any | None = None,
+) -> dict[str, Any]:
+    kwargs: dict[str, Any] = {"session_key": session_key}
+    if channel != "cli":
+        kwargs["channel"] = channel
+    if chat_id != "direct":
+        kwargs["chat_id"] = chat_id
+    if sender_id != "user":
+        kwargs["sender_id"] = sender_id
+    if media is not None:
+        kwargs["media"] = media
+    if ephemeral:
+        kwargs["ephemeral"] = True
+        kwargs["_run_extra_hooks_for_ephemeral"] = True
+    if on_stream is not None:
+        kwargs["on_stream"] = on_stream
+    if on_stream_end is not None:
+        kwargs["on_stream_end"] = on_stream_end
+    return kwargs
+
+
+class SDKRuntimeGate:
+    """Allow normal SDK runs to overlap while model overrides stay exclusive."""
+
+    def __init__(self) -> None:
+        self._condition = asyncio.Condition()
+        self._readers = 0
+        self._writer_active = False
+        self._writers_waiting = 0
+
+    def slot(self, *, exclusive: bool) -> SDKRuntimeGateSlot:
+        return SDKRuntimeGateSlot(self, exclusive=exclusive)
+
+    async def _acquire(self, *, exclusive: bool) -> None:
+        async with self._condition:
+            if exclusive:
+                self._writers_waiting += 1
+                try:
+                    await self._condition.wait_for(
+                        lambda: not self._writer_active and self._readers == 0
+                    )
+                    self._writer_active = True
+                finally:
+                    self._writers_waiting -= 1
+                    self._condition.notify_all()
+                return
+
+            await self._condition.wait_for(
+                lambda: not self._writer_active and self._writers_waiting == 0
+            )
+            self._readers += 1
+
+    async def _release(self, *, exclusive: bool) -> None:
+        async with self._condition:
+            if exclusive:
+                self._writer_active = False
+            else:
+                self._readers = max(0, self._readers - 1)
+            self._condition.notify_all()
+
+
+class SDKRuntimeGateSlot:
+    def __init__(self, gate: SDKRuntimeGate, *, exclusive: bool) -> None:
+        self._gate = gate
+        self._exclusive = exclusive
+
+    async def __aenter__(self) -> None:
+        await self._gate._acquire(exclusive=self._exclusive)
+
+    async def __aexit__(self, *exc: object) -> None:
+        await self._gate._release(exclusive=self._exclusive)
+
+
+class SDKRuntimeController:
+    """Apply per-run SDK model overrides without leaking global runtime state."""
+
+    def __init__(self, loop: AgentLoop, *, config: Config | None = None) -> None:
+        self._loop = loop
+        self._config = config
+        self._gate = SDKRuntimeGate()
+
+    @asynccontextmanager
+    async def override(
+        self,
+        *,
+        model: str | None,
+        model_preset: str | None,
+    ) -> AsyncIterator[None]:
+        ensure_single_model_selector(model=model, model_preset=model_preset)
+        exclusive = model is not None or model_preset is not None
+        async with self._gate.slot(exclusive=exclusive):
+            override = self.model_override_snapshot(model=model, model_preset=model_preset)
+            restore = self._current_snapshot() if override is not None else None
+            restore_signature = self._loop._provider_signature
+            if override is not None:
+                self._loop._apply_provider_snapshot(
+                    override,
+                    publish_update=False,
+                    model_preset=model_preset,
+                )
+            try:
+                yield
+            finally:
+                if restore is not None:
+                    self._restore_snapshot(
+                        restore,
+                        provider_signature=restore_signature,
+                    )
+
+    def model_override_snapshot(
+        self,
+        *,
+        model: str | None,
+        model_preset: str | None,
+    ) -> ProviderSnapshot | None:
+        ensure_single_model_selector(model=model, model_preset=model_preset)
+        if model_preset is not None:
+            return self._loop._build_model_preset_snapshot(model_preset)
+        if model is None:
+            return None
+
+        if self._config is not None:
+            base = self._config.resolve_preset(self._loop.model_preset)
+            preset = base.model_copy(update={"model": model, "provider": "auto"})
+            return build_provider_snapshot(self._config, preset=preset)
+
+        generation = getattr(self._loop.provider, "generation", None)
+        preset = ModelPresetConfig(
+            model=model,
+            provider="auto",
+            max_tokens=getattr(generation, "max_tokens", 8192),
+            context_window_tokens=self._loop.context_window_tokens,
+            temperature=getattr(generation, "temperature", 0.1),
+            reasoning_effort=getattr(generation, "reasoning_effort", None),
+        )
+        from nanobot.agent.model_presets import build_static_preset_snapshot
+
+        return build_static_preset_snapshot(self._loop.provider, "sdk:override", preset)
+
+    def _current_snapshot(self) -> ProviderSnapshot:
+        signature = self._loop._provider_signature
+        if signature is None:
+            signature = ("sdk:runtime", id(self._loop.provider), self._loop.model)
+        return ProviderSnapshot(
+            provider=self._loop.provider,
+            model=self._loop.model,
+            context_window_tokens=self._loop.context_window_tokens,
+            signature=signature,
+        )
+
+    def _restore_snapshot(
+        self,
+        snapshot: ProviderSnapshot,
+        *,
+        provider_signature: tuple[object, ...] | None,
+    ) -> None:
+        self._loop._apply_provider_snapshot(snapshot, publish_update=False)
+        self._loop._provider_signature = provider_signature

nanobot/sdk/streaming.py

@@ -0,0 +1,222 @@
+"""Streaming support for the high-level Python SDK."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import suppress
+from copy import deepcopy
+
+from nanobot.agent.hook import AgentHook, AgentHookContext
+from nanobot.sdk.types import (
+    STREAM_EVENT_REASONING_COMPLETED,
+    STREAM_EVENT_REASONING_DELTA,
+    STREAM_EVENT_TEXT_COMPLETED,
+    STREAM_EVENT_TEXT_DELTA,
+    STREAM_EVENT_TOOL_COMPLETED,
+    STREAM_EVENT_TOOL_FAILED,
+    STREAM_EVENT_TOOL_STARTED,
+    RunResult,
+    StreamEvent,
+)
+
+_STREAM_SENTINEL = object()
+
+
+class RunStream:
+    """A running SDK turn with Cursor/OpenAI-style event streaming."""
+
+    def __init__(
+        self,
+        task: asyncio.Task[RunResult],
+        queue: asyncio.Queue[StreamEvent | object],
+    ) -> None:
+        self._task = task
+        self._queue = queue
+        self._events_started = False
+        self._events_done = False
+        self._stream_active = False
+        self._closed = False
+
+    @property
+    def done(self) -> bool:
+        """Whether the underlying run task has finished."""
+        return self._task.done()
+
+    async def stream_events(self) -> AsyncIterator[StreamEvent]:
+        """Yield streaming events for this run.
+
+        The event stream is single-consumer: call this method only once. Closing
+        the iterator before completion cancels the underlying run.
+        """
+        if self._events_started:
+            raise RuntimeError("RunStream.stream_events() can only be consumed once")
+        self._events_started = True
+        self._stream_active = True
+        try:
+            while True:
+                item = await self._queue.get()
+                if item is _STREAM_SENTINEL:
+                    self._events_done = True
+                    break
+                yield item
+        finally:
+            self._stream_active = False
+            if not self._events_done:
+                await self.aclose()
+
+    async def wait(self) -> RunResult:
+        """Wait for the run to finish and return its final result."""
+        if not self._events_done and not self._stream_active:
+            if not self._events_started:
+                self._events_started = True
+            await self._drain_events()
+        return await self._task
+
+    async def text(self) -> str:
+        """Wait for the run to finish and return the final text."""
+        return (await self.wait()).content
+
+    async def cancel(self) -> None:
+        """Cancel the running turn and release stream resources."""
+        await self.aclose()
+
+    async def aclose(self) -> None:
+        """Close the stream, cancelling the run if it is still active."""
+        if self._closed:
+            return
+        self._closed = True
+        if not self._task.done():
+            self._task.cancel()
+        self._finish_events()
+        try:
+            await self._task
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            # Closing is cleanup; wait() remains the API that surfaces run errors.
+            pass
+
+    async def _drain_events(self) -> None:
+        while not self._events_done:
+            item = await self._queue.get()
+            if item is _STREAM_SENTINEL:
+                self._events_done = True
+                break
+
+    def _finish_events(self) -> None:
+        self._events_done = True
+        while True:
+            with suppress(asyncio.QueueEmpty):
+                self._queue.get_nowait()
+                continue
+            break
+        with suppress(asyncio.QueueFull):
+            self._queue.put_nowait(_STREAM_SENTINEL)
+
+
+class SDKStreamEmitter:
+    """Serialize SDK streaming events onto a bounded async queue."""
+
+    def __init__(self, queue: asyncio.Queue[StreamEvent | object]) -> None:
+        self._queue = queue
+        self._text_parts: list[str] = []
+        self._closed = False
+
+    async def emit(self, event: StreamEvent) -> None:
+        if self._closed:
+            return
+        await self._queue.put(event)
+
+    async def text_delta(self, delta: str, *, iteration: int | None = None) -> None:
+        if not delta:
+            return
+        self._text_parts.append(delta)
+        await self.emit(StreamEvent(
+            type=STREAM_EVENT_TEXT_DELTA,
+            delta=delta,
+            iteration=iteration,
+        ))
+
+    async def text_completed(
+        self,
+        *,
+        resuming: bool = False,
+        iteration: int | None = None,
+        force: bool = True,
+    ) -> None:
+        content = "".join(self._text_parts)
+        if not content and (resuming or not force):
+            return
+        self._text_parts = []
+        await self.emit(StreamEvent(
+            type=STREAM_EVENT_TEXT_COMPLETED,
+            content=content,
+            iteration=iteration,
+            resuming=resuming,
+        ))
+
+    def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if self._queue.full():
+            with suppress(asyncio.QueueEmpty):
+                self._queue.get_nowait()
+        with suppress(asyncio.QueueFull):
+            self._queue.put_nowait(_STREAM_SENTINEL)
+
+
+class SDKStreamingHook(AgentHook):
+    """Convert agent lifecycle hooks into public SDK stream events."""
+
+    def __init__(self, emitter: SDKStreamEmitter) -> None:
+        super().__init__()
+        self._emitter = emitter
+        self._reasoning_open = False
+
+    async def before_execute_tools(self, context: AgentHookContext) -> None:
+        for call in context.tool_calls:
+            await self._emitter.emit(StreamEvent(
+                type=STREAM_EVENT_TOOL_STARTED,
+                name=call.name,
+                tool_call_id=call.id,
+                arguments=deepcopy(call.arguments),
+                iteration=context.iteration,
+            ))
+
+    async def emit_reasoning(self, reasoning_content: str | None) -> None:
+        if not reasoning_content:
+            return
+        self._reasoning_open = True
+        await self._emitter.emit(StreamEvent(
+            type=STREAM_EVENT_REASONING_DELTA,
+            delta=reasoning_content,
+        ))
+
+    async def emit_reasoning_end(self) -> None:
+        if not self._reasoning_open:
+            return
+        self._reasoning_open = False
+        await self._emitter.emit(StreamEvent(type=STREAM_EVENT_REASONING_COMPLETED))
+
+    async def after_iteration(self, context: AgentHookContext) -> None:
+        if not context.tool_events:
+            return
+        for index, raw_event in enumerate(context.tool_events):
+            call = context.tool_calls[index] if index < len(context.tool_calls) else None
+            event = dict(raw_event)
+            status = event.get("status")
+            name = str(event.get("name") or (call.name if call else ""))
+            event_type = (
+                STREAM_EVENT_TOOL_COMPLETED if status == "ok" else STREAM_EVENT_TOOL_FAILED
+            )
+            await self._emitter.emit(StreamEvent(
+                type=event_type,
+                name=name or None,
+                tool_call_id=call.id if call else None,
+                arguments=deepcopy(call.arguments) if call else None,
+                iteration=context.iteration,
+                error=None if status == "ok" else str(event.get("detail") or ""),
+                metadata=event,
+            ))

nanobot/sdk/types.py

@@ -0,0 +1,153 @@
+"""Public SDK value objects and event constants."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import dataclass, field
+from typing import Any, Literal, Mapping, TypeAlias
+
+StreamEventType: TypeAlias = Literal[
+    "run.started",
+    "text.delta",
+    "text.completed",
+    "reasoning.delta",
+    "reasoning.completed",
+    "tool.started",
+    "tool.completed",
+    "tool.failed",
+    "run.completed",
+    "run.failed",
+]
+
+STREAM_EVENT_RUN_STARTED: StreamEventType = "run.started"
+STREAM_EVENT_TEXT_DELTA: StreamEventType = "text.delta"
+STREAM_EVENT_TEXT_COMPLETED: StreamEventType = "text.completed"
+STREAM_EVENT_REASONING_DELTA: StreamEventType = "reasoning.delta"
+STREAM_EVENT_REASONING_COMPLETED: StreamEventType = "reasoning.completed"
+STREAM_EVENT_TOOL_STARTED: StreamEventType = "tool.started"
+STREAM_EVENT_TOOL_COMPLETED: StreamEventType = "tool.completed"
+STREAM_EVENT_TOOL_FAILED: StreamEventType = "tool.failed"
+STREAM_EVENT_RUN_COMPLETED: StreamEventType = "run.completed"
+STREAM_EVENT_RUN_FAILED: StreamEventType = "run.failed"
+
+STREAM_EVENT_TYPES: tuple[StreamEventType, ...] = (
+    STREAM_EVENT_RUN_STARTED,
+    STREAM_EVENT_TEXT_DELTA,
+    STREAM_EVENT_TEXT_COMPLETED,
+    STREAM_EVENT_REASONING_DELTA,
+    STREAM_EVENT_REASONING_COMPLETED,
+    STREAM_EVENT_TOOL_STARTED,
+    STREAM_EVENT_TOOL_COMPLETED,
+    STREAM_EVENT_TOOL_FAILED,
+    STREAM_EVENT_RUN_COMPLETED,
+    STREAM_EVENT_RUN_FAILED,
+)
+
+
+@dataclass(slots=True)
+class RunResult:
+    """Result of a single agent run."""
+
+    content: str
+    tools_used: list[str] = field(default_factory=list)
+    messages: list[dict[str, Any]] = field(default_factory=list)
+    usage: dict[str, int] = field(default_factory=dict)
+    stop_reason: str | None = None
+    error: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class StreamEvent:
+    """A typed event emitted by ``Nanobot.stream()`` and ``RunStream``."""
+
+    type: StreamEventType
+    delta: str = ""
+    content: str = ""
+    result: RunResult | None = None
+    name: str | None = None
+    tool_call_id: str | None = None
+    arguments: dict[str, Any] | None = None
+    iteration: int | None = None
+    resuming: bool | None = None
+    usage: dict[str, int] = field(default_factory=dict)
+    error: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class SessionSnapshot:
+    """A durable snapshot of one nanobot session."""
+
+    key: str
+    messages: list[dict[str, Any]]
+    metadata: dict[str, Any] = field(default_factory=dict)
+    created_at: str | None = None
+    updated_at: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-serializable copy of the snapshot."""
+        return {
+            "key": self.key,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "metadata": deepcopy(self.metadata),
+            "messages": deepcopy(self.messages),
+        }
+
+
+@dataclass(slots=True)
+class SessionInfo:
+    """Compact session metadata for listings."""
+
+    key: str
+    created_at: str | None = None
+    updated_at: str | None = None
+    title: str = ""
+    preview: str = ""
+    path: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-serializable copy of the listing row."""
+        return {
+            "key": self.key,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "title": self.title,
+            "preview": self.preview,
+            "path": self.path,
+        }
+
+
+def snapshot_from_session(session: Any) -> SessionSnapshot:
+    return SessionSnapshot(
+        key=session.key,
+        created_at=session.created_at.isoformat(),
+        updated_at=session.updated_at.isoformat(),
+        metadata=deepcopy(session.metadata),
+        messages=deepcopy(session.messages),
+    )
+
+
+def snapshot_from_payload(payload: Mapping[str, Any]) -> SessionSnapshot:
+    return SessionSnapshot(
+        key=str(payload.get("key") or ""),
+        created_at=payload.get("created_at"),
+        updated_at=payload.get("updated_at"),
+        metadata=deepcopy(dict(payload.get("metadata") or {})),
+        messages=deepcopy(list(payload.get("messages") or [])),
+    )
+
+
+def result_from_response(response: Any, capture: Any) -> RunResult:
+    content = (response.content if response else None) or ""
+    metadata = dict(response.metadata) if response and response.metadata else {}
+    return RunResult(
+        content=content,
+        tools_used=capture.tools_used,
+        messages=capture.messages,
+        usage=capture.usage,
+        stop_reason=capture.stop_reason,
+        error=capture.error,
+        metadata=metadata,
+    )

tests/agent/test_loop_direct_websocket_status.py

@@ -95,3 +95,28 @@ async def test_process_direct_reuses_existing_session_lock(tmp_path) -> None:
             task.cancel()
             with pytest.raises(asyncio.CancelledError):
                 await task
+
+
+@pytest.mark.asyncio
+async def test_process_direct_applies_per_run_hooks(tmp_path) -> None:
+    from nanobot.agent.hook import AgentHook, AgentRunHookContext
+
+    loop = _make_loop(tmp_path)
+    events: list[tuple[str, str | None]] = []
+
+    class RecordingHook(AgentHook):
+        async def before_run(self, context: AgentRunHookContext) -> None:
+            events.append(("before", None))
+
+        async def after_run(self, context: AgentRunHookContext) -> None:
+            events.append(("after", context.final_content))
+
+    response = await loop.process_direct(
+        "hello",
+        session_key="api:per-run-hook",
+        hooks=[RecordingHook()],
+    )
+
+    assert response is not None
+    assert response.content == "done"
+    assert events == [("before", None), ("after", "done")]