From dbf3c4b245aba51049e26255d535a54c31b8537c Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Sun, 21 Jun 2026 16:46:04 +0800 Subject: [PATCH] feat(sdk): expand Python runtime controls --- docs/README.md | 4 +- docs/python-sdk.md | 560 ++++++++++- nanobot/__init__.py | 37 +- nanobot/agent/hook.py | 14 + nanobot/agent/loop.py | 33 +- nanobot/nanobot.py | 234 ++++- nanobot/sdk/__init__.py | 1 + nanobot/sdk/clients.py | 165 ++++ nanobot/sdk/runtime.py | 192 ++++ nanobot/sdk/streaming.py | 222 +++++ nanobot/sdk/types.py | 153 +++ .../test_loop_direct_websocket_status.py | 25 + tests/test_nanobot_facade.py | 888 +++++++++++++++++- 13 files changed, 2448 insertions(+), 80 deletions(-) create mode 100644 nanobot/sdk/__init__.py create mode 100644 nanobot/sdk/clients.py create mode 100644 nanobot/sdk/runtime.py create mode 100644 nanobot/sdk/streaming.py create mode 100644 nanobot/sdk/types.py diff --git a/docs/README.md b/docs/README.md index 9b4ef7b9..9c996e2b 100644 --- a/docs/README.md +++ b/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) | diff --git a/docs/python-sdk.md b/docs/python-sdk.md index 7c475c2f..967872fe 100644 --- a/docs/python-sdk.md +++ b/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( - config_path="~/.nanobot/config.json", - workspace="/my/project", -) +async with Nanobot.from_config(workspace="/my/project") as bot: + result = await bot.run("Explain the project structure") ``` +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:`, `project:`, or +`eval:`. + +### 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. | -| `messages` | `list[dict]` | Reserved for richer SDK introspection; may be empty in current versions. | +| `tools_used` | `list[str]` | Tool names used during the run. | +| `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") - result = await bot.run( - "Explain the main function", - session_key="sdk:demo", - hooks=[TimingHook()], - ) + async with Nanobot.from_config(workspace="/my/project") as bot: + result = await bot.run( + "Explain the main function", + session_key="sdk:demo", + hooks=[TimingHook()], + ) print(result.content) diff --git a/nanobot/__init__.py b/nanobot/__init__.py index ac6484d3..84b09532 100644 --- a/nanobot/__init__.py +++ b/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", +] diff --git a/nanobot/agent/hook.py b/nanobot/agent/hook.py index b0a72246..f9ff37f3 100644 --- a/nanobot/agent/hook.py +++ b/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 diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 776cb849..9b744e04 100644 --- a/nanobot/agent/loop.py +++ b/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 not ephemeral and turn_hooks: - hook = CompositeHook([loop_hook] + turn_hooks) + if run_hooks and (not ephemeral or run_extra_hooks_for_ephemeral): + hook = CompositeHook([loop_hook, *run_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, diff --git a/nanobot/nanobot.py b/nanobot/nanobot.py index a31338c2..27f61eee 100644 --- a/nanobot/nanobot.py +++ b/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, +) - -@dataclass(slots=True) -class RunResult: - """Result of a single agent run.""" - - content: str - tools_used: list[str] - messages: list[dict[str, Any]] +__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", +] 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 []) - response = await self._loop.process_direct( + per_run_hooks = [capture, *(hooks or [])] + 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], - ) - - content = (response.content if response else None) or "" - return RunResult( - content=content, - tools_used=capture.tools_used, - messages=capture.messages, + channel=channel, + chat_id=chat_id, + sender_id=sender_id, + media=media, + ephemeral=ephemeral, + hooks=hooks, + model=model, + 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.).""" diff --git a/nanobot/sdk/__init__.py b/nanobot/sdk/__init__.py new file mode 100644 index 00000000..7c6fa32c --- /dev/null +++ b/nanobot/sdk/__init__.py @@ -0,0 +1 @@ +"""Internal helpers for the high-level nanobot Python SDK.""" diff --git a/nanobot/sdk/clients.py b/nanobot/sdk/clients.py new file mode 100644 index 00000000..49a68222 --- /dev/null +++ b/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, + ) diff --git a/nanobot/sdk/runtime.py b/nanobot/sdk/runtime.py new file mode 100644 index 00000000..38aae479 --- /dev/null +++ b/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 diff --git a/nanobot/sdk/streaming.py b/nanobot/sdk/streaming.py new file mode 100644 index 00000000..b53bf53d --- /dev/null +++ b/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, + )) diff --git a/nanobot/sdk/types.py b/nanobot/sdk/types.py new file mode 100644 index 00000000..5e4963c3 --- /dev/null +++ b/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, + ) diff --git a/tests/agent/test_loop_direct_websocket_status.py b/tests/agent/test_loop_direct_websocket_status.py index 879fa23a..ef3d34c4 100644 --- a/tests/agent/test_loop_direct_websocket_status.py +++ b/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")] diff --git a/tests/test_nanobot_facade.py b/tests/test_nanobot_facade.py index 9e44afc3..bc656b9f 100644 --- a/tests/test_nanobot_facade.py +++ b/tests/test_nanobot_facade.py @@ -5,11 +5,31 @@ from __future__ import annotations import asyncio import json from pathlib import Path +from types import SimpleNamespace from unittest.mock import ANY, AsyncMock, MagicMock, patch import pytest -from nanobot.nanobot import Nanobot, RunResult +from nanobot.nanobot 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, + Nanobot, + RunResult, + RunStream, + SessionInfo, + SessionSnapshot, + StreamEvent, + StreamEventType, +) def _write_config(tmp_path: Path, overrides: dict | None = None) -> Path: @@ -24,6 +44,17 @@ def _write_config(tmp_path: Path, overrides: dict | None = None) -> Path: return config_path +def _fake_provider(name: str, *, max_tokens: int = 8192) -> MagicMock: + provider = MagicMock(name=name) + provider.get_default_model.return_value = name + provider.generation = SimpleNamespace( + max_tokens=max_tokens, + temperature=0.1, + reasoning_effort=None, + ) + return provider + + def test_from_config_missing_file(): with pytest.raises(FileNotFoundError): Nanobot.from_config("/nonexistent/config.json") @@ -36,6 +67,50 @@ def test_from_config_creates_instance(tmp_path): assert bot._loop.workspace == tmp_path +def test_from_config_accepts_default_model_override(tmp_path): + config_path = _write_config(tmp_path) + + bot = Nanobot.from_config( + config_path, + workspace=tmp_path, + model="openai/gpt-4.1-mini", + ) + + assert bot.runtime.model == "openai/gpt-4.1-mini" + assert bot._loop.model_preset is None + + +def test_from_config_accepts_default_model_preset(tmp_path): + config_path = _write_config( + tmp_path, + { + "modelPresets": { + "fast": { + "model": "openai/gpt-4.1-mini", + "provider": "openrouter", + } + } + }, + ) + + bot = Nanobot.from_config(config_path, workspace=tmp_path, model_preset="fast") + + assert bot.runtime.model == "openai/gpt-4.1-mini" + assert bot._loop.model_preset == "fast" + + +def test_from_config_rejects_multiple_model_selectors(tmp_path): + config_path = _write_config(tmp_path) + + with pytest.raises(ValueError, match="mutually exclusive"): + Nanobot.from_config( + config_path, + workspace=tmp_path, + model="openai/gpt-4.1", + model_preset="fast", + ) + + def test_from_config_default_path(): from nanobot.config.schema import Config @@ -66,7 +141,9 @@ async def test_run_returns_result(tmp_path): assert isinstance(result, RunResult) assert result.content == "Hello back!" bot._loop.process_direct.assert_awaited_once_with( - "hi", session_key="sdk:default", extra_hooks=ANY + "hi", + session_key="sdk:default", + hooks=ANY, ) @@ -91,10 +168,10 @@ async def test_run_with_hooks(tmp_path): assert result.content == "done" assert bot._loop._extra_hooks == [] - extra_hooks = bot._loop.process_direct.await_args.kwargs["extra_hooks"] - assert len(extra_hooks) == 2 - assert isinstance(extra_hooks[0], SDKCaptureHook) - assert isinstance(extra_hooks[1], TestHook) + hooks = bot._loop.process_direct.await_args.kwargs["hooks"] + assert len(hooks) == 2 + assert isinstance(hooks[0], SDKCaptureHook) + assert isinstance(hooks[1], TestHook) @pytest.mark.asyncio @@ -167,7 +244,9 @@ async def test_run_custom_session_key(tmp_path): await bot.run("hi", session_key="user-alice") bot._loop.process_direct.assert_awaited_once_with( - "hi", session_key="user-alice", extra_hooks=ANY + "hi", + session_key="user-alice", + hooks=ANY, ) @@ -176,6 +255,42 @@ def test_import_from_top_level(): assert nanobot.Nanobot is Nanobot assert nanobot.RunResult is RunResult + assert nanobot.RunStream is RunStream + assert nanobot.SessionInfo is SessionInfo + assert nanobot.SessionSnapshot is SessionSnapshot + assert nanobot.StreamEvent is StreamEvent + assert nanobot.StreamEventType is StreamEventType + assert nanobot.STREAM_EVENT_TEXT_DELTA == STREAM_EVENT_TEXT_DELTA + assert nanobot.STREAM_EVENT_RUN_COMPLETED == STREAM_EVENT_RUN_COMPLETED + assert nanobot.STREAM_EVENT_TYPES == STREAM_EVENT_TYPES + + +def test_stream_event_constants_are_stable(): + assert STREAM_EVENT_TYPES == ( + 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, + ) + assert STREAM_EVENT_TYPES == ( + "run.started", + "text.delta", + "text.completed", + "reasoning.delta", + "reasoning.completed", + "tool.started", + "tool.completed", + "tool.failed", + "run.completed", + "run.failed", + ) + assert len(set(STREAM_EVENT_TYPES)) == len(STREAM_EVENT_TYPES) # --------------------------------------------------------------------------- @@ -192,20 +307,19 @@ async def test_run_populates_tools_used_across_iterations(tmp_path): config_path = _write_config(tmp_path) bot = Nanobot.from_config(config_path, workspace=tmp_path) - async def fake_process_direct(message, *, session_key, extra_hooks=None): - extras = extra_hooks or [] + async def fake_process_direct(message, *, session_key, hooks): messages = [{"role": "user", "content": message}] ctx1 = AgentHookContext(iteration=0, messages=messages) ctx1.tool_calls = [ ToolCallRequest(id="c1", name="read_file", arguments={}), ToolCallRequest(id="c2", name="grep", arguments={}), ] - for h in extras: + for h in hooks: await h.after_iteration(ctx1) messages.append({"role": "assistant", "content": "ok"}) ctx2 = AgentHookContext(iteration=1, messages=messages) ctx2.tool_calls = [ToolCallRequest(id="c3", name="web_fetch", arguments={})] - for h in extras: + for h in hooks: await h.after_iteration(ctx2) return OutboundMessage(channel="cli", chat_id="direct", content="final") @@ -224,14 +338,13 @@ async def test_run_populates_final_messages(tmp_path): config_path = _write_config(tmp_path) bot = Nanobot.from_config(config_path, workspace=tmp_path) - async def fake_process_direct(message, *, session_key, extra_hooks=None): - extras = extra_hooks or [] + async def fake_process_direct(message, *, session_key, hooks): messages = [ {"role": "user", "content": message}, {"role": "assistant", "content": "hi there"}, ] ctx = AgentHookContext(iteration=0, messages=messages) - for h in extras: + for h in hooks: await h.after_iteration(ctx) return OutboundMessage(channel="cli", chat_id="direct", content="hi there") @@ -256,6 +369,271 @@ async def test_run_no_iterations_leaves_defaults_empty(tmp_path): result = await bot.run("hi") assert result.tools_used == [] assert result.messages == [] + assert result.usage == {} + assert result.stop_reason is None + assert result.error is None + + +@pytest.mark.asyncio +async def test_run_populates_observability_fields(tmp_path): + from nanobot.agent.hook import AgentRunHookContext + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + async def fake_process_direct(message, *, session_key, hooks): + ctx = AgentRunHookContext( + messages=[ + {"role": "user", "content": message}, + {"role": "assistant", "content": "done"}, + ], + final_content="done", + tools_used=["read_file"], + usage={"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}, + stop_reason="completed", + error=None, + tool_events=[{"tool": "read_file", "status": "ok"}], + ) + for h in hooks: + await h.after_run(ctx) + return OutboundMessage( + channel="cli", + chat_id="direct", + content="done", + metadata={"latency_ms": 42}, + ) + + bot._loop.process_direct = fake_process_direct + result = await bot.run("work") + + assert result.content == "done" + assert result.tools_used == ["read_file"] + assert result.usage == {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12} + assert result.stop_reason == "completed" + assert result.error is None + assert result.metadata == {"latency_ms": 42} + + +@pytest.mark.asyncio +async def test_run_ephemeral_still_captures_runner_observability(tmp_path): + from nanobot.agent.loop import AgentLoop + from nanobot.bus.queue import MessageBus + from nanobot.providers.base import LLMResponse + + provider = MagicMock() + provider.get_default_model.return_value = "test-model" + provider.chat_with_retry = AsyncMock(return_value=LLMResponse( + content="done", + tool_calls=[], + usage={"total_tokens": 3}, + )) + bot = Nanobot(AgentLoop( + bus=MessageBus(), + provider=provider, + workspace=tmp_path, + model="test-model", + )) + + result = await bot.run("hi", ephemeral=True) + + assert result.content == "done" + assert result.usage["total_tokens"] == 3 + assert result.usage["provider_tokens"] == 3 + + +@pytest.mark.asyncio +async def test_run_forwards_non_default_runtime_options(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + bot._loop.process_direct = AsyncMock( + return_value=OutboundMessage(channel="sdk", chat_id="chat-a", content="ok"), + ) + + await bot.run( + "hi", + session_key="sdk:chat-a", + channel="sdk", + chat_id="chat-a", + sender_id="alice", + media=["/tmp/image.png"], + ephemeral=True, + ) + + bot._loop.process_direct.assert_awaited_once_with( + "hi", + session_key="sdk:chat-a", + channel="sdk", + chat_id="chat-a", + sender_id="alice", + media=["/tmp/image.png"], + ephemeral=True, + _run_extra_hooks_for_ephemeral=True, + hooks=ANY, + ) + + +@pytest.mark.asyncio +async def test_run_allows_parallel_sessions_without_model_override(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + entered: list[str] = [] + both_entered = asyncio.Event() + + async def fake_process_direct(message, *, session_key, hooks): + entered.append(session_key) + if len(entered) == 2: + both_entered.set() + await asyncio.wait_for(both_entered.wait(), timeout=1) + return OutboundMessage(channel="cli", chat_id="direct", content=message) + + bot._loop.process_direct = fake_process_direct + + left, right = await asyncio.gather( + bot.run("left", session_key="sdk:left"), + bot.run("right", session_key="sdk:right"), + ) + + assert left.content == "left" + assert right.content == "right" + assert set(entered) == {"sdk:left", "sdk:right"} + + +@pytest.mark.asyncio +async def test_run_model_overrides_are_serialized_before_snapshot_build(tmp_path): + from nanobot.bus.events import OutboundMessage + from nanobot.providers.factory import ProviderSnapshot + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + original_model = bot._loop.model + active_models: list[str] = [] + snapshot_base_models: list[str] = [] + first_entered = asyncio.Event() + release_first = asyncio.Event() + + def fake_snapshot(*, model, model_preset): + assert model is not None + assert model_preset is None + snapshot_base_models.append(bot._loop.model) + return ProviderSnapshot( + provider=_fake_provider(model, max_tokens=2048), + model=model, + context_window_tokens=4096, + signature=("sdk", model), + ) + + bot._runtime_overrides.model_override_snapshot = MagicMock(side_effect=fake_snapshot) + + async def fake_process_direct(message, *, session_key, hooks): + active_models.append(bot._loop.model) + if message == "first": + first_entered.set() + await asyncio.wait_for(release_first.wait(), timeout=1) + return OutboundMessage(channel="cli", chat_id="direct", content=message) + + bot._loop.process_direct = fake_process_direct + + first = asyncio.create_task(bot.run("first", model="model:first")) + await asyncio.wait_for(first_entered.wait(), timeout=1) + + second = asyncio.create_task(bot.run("second", model="model:second")) + await asyncio.sleep(0) + assert not second.done() + + release_first.set() + first_result, second_result = await asyncio.gather(first, second) + + assert first_result.content == "first" + assert second_result.content == "second" + assert active_models == ["model:first", "model:second"] + assert snapshot_base_models == [original_model, original_model] + assert bot._loop.model == original_model + + +@pytest.mark.asyncio +async def test_run_model_override_is_per_run_and_restores_default(tmp_path): + from nanobot.bus.events import OutboundMessage + from nanobot.providers.factory import ProviderSnapshot + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + original_provider = bot._loop.provider + original_model = bot._loop.model + original_signature = bot._loop._provider_signature + override_provider = _fake_provider("override-provider", max_tokens=2048) + override = ProviderSnapshot( + provider=override_provider, + model="openai/gpt-4.1-mini", + context_window_tokens=4096, + signature=("sdk", "override"), + ) + bot._runtime_overrides.model_override_snapshot = MagicMock(return_value=override) + + async def fake_process_direct(message, *, session_key, hooks): + assert bot._loop.provider is override_provider + assert bot._loop.runner.provider is override_provider + assert bot._loop.model == "openai/gpt-4.1-mini" + assert bot._loop.context_window_tokens == 4096 + return OutboundMessage(channel="cli", chat_id="direct", content="ok") + + bot._loop.process_direct = fake_process_direct + + result = await bot.run("hi", model="openai/gpt-4.1-mini") + + assert result.content == "ok" + bot._runtime_overrides.model_override_snapshot.assert_called_once_with( + model="openai/gpt-4.1-mini", + model_preset=None, + ) + assert bot._loop.provider is original_provider + assert bot._loop.runner.provider is original_provider + assert bot._loop.model == original_model + assert bot._loop._provider_signature == original_signature + + +@pytest.mark.asyncio +async def test_run_model_preset_override_is_per_run(tmp_path): + from nanobot.bus.events import OutboundMessage + from nanobot.providers.factory import ProviderSnapshot + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + original_model = bot._loop.model + override_provider = _fake_provider("preset-provider", max_tokens=1024) + override = ProviderSnapshot( + provider=override_provider, + model="openai/gpt-4.1-mini", + context_window_tokens=2048, + signature=("preset", "fast"), + ) + bot._loop._build_model_preset_snapshot = MagicMock(return_value=override) + + async def fake_process_direct(message, *, session_key, hooks): + assert bot._loop.provider is override_provider + assert bot._loop.model == "openai/gpt-4.1-mini" + return OutboundMessage(channel="cli", chat_id="direct", content="ok") + + bot._loop.process_direct = fake_process_direct + + await bot.run("hi", model_preset="fast") + + bot._loop._build_model_preset_snapshot.assert_called_once_with("fast") + assert bot._loop.model == original_model + assert bot._loop.model_preset is None + + +@pytest.mark.asyncio +async def test_run_rejects_multiple_model_selectors(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + with pytest.raises(ValueError, match="mutually exclusive"): + await bot.run("hi", model="openai/gpt-4.1", model_preset="fast") @pytest.mark.asyncio @@ -273,11 +651,10 @@ async def test_run_user_hooks_still_fire_alongside_capture(tmp_path): async def after_iteration(self, context: AgentHookContext) -> None: seen_iterations.append(context.iteration) - async def fake_process_direct(message, *, session_key, extra_hooks=None): - extras = extra_hooks or [] - assert len(extras) == 2, f"expected capture + user hook, got {len(extras)}" + async def fake_process_direct(message, *, session_key, hooks): + assert len(hooks) == 2, f"expected capture + user hook, got {len(hooks)}" ctx = AgentHookContext(iteration=7, messages=[]) - for h in extras: + for h in hooks: await h.after_iteration(ctx) return OutboundMessage(channel="cli", chat_id="direct", content="ok") @@ -307,20 +684,20 @@ async def test_concurrent_run_hooks_are_isolated_per_call(tmp_path): started = 0 both_started = asyncio.Event() - async def fake_process_direct(message, *, session_key, extra_hooks=None): + async def fake_process_direct(message, *, session_key, hooks=None): nonlocal started started += 1 if started == 2: both_started.set() await both_started.wait() - extras = extra_hooks or [] + active_hooks = hooks or [] messages = [{"role": "user", "content": message}] ctx = AgentHookContext(iteration=0, messages=messages) ctx.tool_calls = [ ToolCallRequest(id=f"call-{message}", name=f"tool_{message}", arguments={}) ] - for h in extras: + for h in active_hooks: await h.after_iteration(ctx) return OutboundMessage(channel="cli", chat_id="direct", content=f"done {message}") @@ -350,9 +727,9 @@ async def test_run_restores_extra_hooks_even_on_populated_iterations(tmp_path): sentinel_hook = AgentHook() bot._loop._extra_hooks = [sentinel_hook] - async def fake_process_direct(message, *, session_key, extra_hooks=None): + async def fake_process_direct(message, *, session_key, hooks): ctx = AgentHookContext(iteration=0, messages=[]) - for h in extra_hooks or []: + for h in [*bot._loop._extra_hooks, *hooks]: await h.after_iteration(ctx) return OutboundMessage(channel="cli", chat_id="direct", content="done") @@ -361,6 +738,348 @@ async def test_run_restores_extra_hooks_even_on_populated_iterations(tmp_path): assert bot._loop._extra_hooks == [sentinel_hook] +@pytest.mark.asyncio +async def test_stream_yields_text_events_in_order(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + assert message == "hi" + assert session_key == "sdk:default" + await on_stream("Hel") + await on_stream("lo") + await on_stream_end(resuming=False) + return OutboundMessage(channel="cli", chat_id="direct", content="Hello") + + bot._loop.process_direct = fake_process_direct + + events = [event async for event in bot.stream("hi")] + + assert all(event.type in STREAM_EVENT_TYPES for event in events) + assert [event.type for event in events] == [ + "run.started", + "text.delta", + "text.delta", + "text.completed", + "run.completed", + ] + assert events[1].delta == "Hel" + assert events[2].delta == "lo" + assert events[3].content == "Hello" + assert events[4].result is not None + assert events[4].result.content == "Hello" + + +@pytest.mark.asyncio +async def test_run_streamed_wait_returns_full_result_without_consuming_events(tmp_path): + from nanobot.agent.hook import AgentRunHookContext + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + await on_stream("done") + await on_stream_end(resuming=False) + ctx = AgentRunHookContext( + messages=[ + {"role": "user", "content": message}, + {"role": "assistant", "content": "done"}, + ], + final_content="done", + tools_used=["read_file"], + usage={"total_tokens": 9}, + stop_reason="completed", + ) + for hook in hooks: + await hook.after_run(ctx) + return OutboundMessage( + channel="cli", + chat_id="direct", + content="done", + metadata={"latency_ms": 5}, + ) + + bot._loop.process_direct = fake_process_direct + + run = await bot.run_streamed("work") + assert isinstance(run, RunStream) + result = await run.wait() + + assert result.content == "done" + assert result.tools_used == ["read_file"] + assert result.usage == {"total_tokens": 9} + assert result.stop_reason == "completed" + assert result.metadata == {"latency_ms": 5} + + +@pytest.mark.asyncio +async def test_run_streamed_cancel_releases_full_queue_without_consuming(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + for i in range(400): + await on_stream(str(i)) + await on_stream_end(resuming=False) + return OutboundMessage(channel="cli", chat_id="direct", content="done") + + bot._loop.process_direct = fake_process_direct + + run = await bot.run_streamed("many") + await asyncio.sleep(0.05) + assert not run.done + + await asyncio.wait_for(run.cancel(), timeout=1) + assert run.done + + +@pytest.mark.asyncio +async def test_run_streamed_text_returns_final_content(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + bot._loop.process_direct = AsyncMock( + return_value=OutboundMessage(channel="cli", chat_id="direct", content="plain text"), + ) + + run = await bot.run_streamed("hi") + + assert await run.text() == "plain text" + + +@pytest.mark.asyncio +async def test_run_streamed_forwards_runtime_options(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + bot._loop.process_direct = AsyncMock( + return_value=OutboundMessage(channel="sdk", chat_id="chat-a", content="ok"), + ) + + run = await bot.run_streamed( + "hi", + session_key="sdk:chat-a", + channel="sdk", + chat_id="chat-a", + sender_id="alice", + media=["/tmp/image.png"], + ephemeral=True, + ) + await run.wait() + + bot._loop.process_direct.assert_awaited_once() + args, kwargs = bot._loop.process_direct.call_args + assert args == ("hi",) + assert kwargs["session_key"] == "sdk:chat-a" + assert kwargs["channel"] == "sdk" + assert kwargs["chat_id"] == "chat-a" + assert kwargs["sender_id"] == "alice" + assert kwargs["media"] == ["/tmp/image.png"] + assert kwargs["ephemeral"] is True + assert callable(kwargs["on_stream"]) + assert callable(kwargs["on_stream_end"]) + assert kwargs["hooks"] + + +@pytest.mark.asyncio +async def test_run_streamed_model_override_reports_model_and_restores(tmp_path): + from nanobot.bus.events import OutboundMessage + from nanobot.providers.factory import ProviderSnapshot + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + original_model = bot._loop.model + override_provider = _fake_provider("stream-provider", max_tokens=2048) + override = ProviderSnapshot( + provider=override_provider, + model="openai/gpt-4.1-mini", + context_window_tokens=4096, + signature=("sdk", "stream"), + ) + bot._runtime_overrides.model_override_snapshot = MagicMock(return_value=override) + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + assert bot._loop.provider is override_provider + assert bot._loop.model == "openai/gpt-4.1-mini" + await on_stream("ok") + await on_stream_end(resuming=False) + return OutboundMessage(channel="cli", chat_id="direct", content="ok") + + bot._loop.process_direct = fake_process_direct + + run = await bot.run_streamed("hi", model="openai/gpt-4.1-mini") + events = [event async for event in run.stream_events()] + result = await run.wait() + + assert result.content == "ok" + assert events[0].type == "run.started" + assert events[0].metadata["model"] == "openai/gpt-4.1-mini" + assert events[0].metadata["model_preset"] is None + assert bot._loop.model == original_model + + +@pytest.mark.asyncio +async def test_stream_rejects_multiple_model_selectors(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + with pytest.raises(ValueError, match="mutually exclusive"): + _ = [event async for event in bot.stream( + "hi", + model="openai/gpt-4.1", + model_preset="fast", + )] + + +@pytest.mark.asyncio +async def test_run_streamed_emits_tool_events(tmp_path): + from nanobot.agent.hook import AgentHookContext + from nanobot.bus.events import OutboundMessage + from nanobot.providers.base import ToolCallRequest + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + calls = [ + ToolCallRequest(id="call_ok", name="read_file", arguments={"path": "README.md"}), + ToolCallRequest(id="call_bad", name="exec", arguments={"cmd": "false"}), + ] + ctx = AgentHookContext(iteration=2, messages=[{"role": "user", "content": message}]) + ctx.tool_calls = calls + for hook in hooks: + await hook.before_execute_tools(ctx) + ctx.tool_events = [ + {"name": "read_file", "status": "ok", "detail": "README.md"}, + {"name": "exec", "status": "error", "detail": "exit 1"}, + ] + for hook in hooks: + await hook.after_iteration(ctx) + return OutboundMessage(channel="cli", chat_id="direct", content="done") + + bot._loop.process_direct = fake_process_direct + + run = await bot.run_streamed("inspect") + events = [event async for event in run.stream_events()] + await run.wait() + + assert [event.type for event in events] == [ + "run.started", + "tool.started", + "tool.started", + "tool.completed", + "tool.failed", + "run.completed", + ] + assert events[1].name == "read_file" + assert events[1].tool_call_id == "call_ok" + assert events[1].arguments == {"path": "README.md"} + assert events[3].metadata["status"] == "ok" + assert events[4].name == "exec" + assert events[4].error == "exit 1" + + +@pytest.mark.asyncio +async def test_run_streamed_emits_reasoning_events(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + for hook in hooks: + await hook.emit_reasoning("thinking") + await hook.emit_reasoning_end() + return OutboundMessage(channel="cli", chat_id="direct", content="done") + + bot._loop.process_direct = fake_process_direct + + events = [event async for event in bot.stream("think")] + + assert [event.type for event in events] == [ + "run.started", + "reasoning.delta", + "reasoning.completed", + "run.completed", + ] + assert events[1].delta == "thinking" + + +@pytest.mark.asyncio +async def test_stream_generator_break_cancels_underlying_run(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + cancelled = asyncio.Event() + + async def fake_process_direct(message, *, session_key, on_stream, on_stream_end, hooks): + try: + await on_stream("first") + await asyncio.sleep(10) + finally: + cancelled.set() + return OutboundMessage(channel="cli", chat_id="direct", content="done") + + bot._loop.process_direct = fake_process_direct + + async for event in bot.stream("stop early"): + if event.type == STREAM_EVENT_TEXT_DELTA: + break + + await asyncio.wait_for(cancelled.wait(), timeout=1) + + +@pytest.mark.asyncio +async def test_run_streamed_restores_hooks_and_reports_failure(tmp_path): + from nanobot.agent.hook import AgentHook + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + sentinel_hook = AgentHook() + bot._loop._extra_hooks = [sentinel_hook] + + async def fake_process_direct(message, **kwargs): + raise RuntimeError("boom") + + bot._loop.process_direct = fake_process_direct + + run = await bot.run_streamed("fail") + events = [event async for event in run.stream_events()] + + assert [event.type for event in events] == ["run.started", "run.failed"] + assert events[1].error == "boom" + with pytest.raises(RuntimeError, match="boom"): + await run.wait() + assert bot._loop._extra_hooks == [sentinel_hook] + + +@pytest.mark.asyncio +async def test_run_streamed_stream_events_is_single_consumer(tmp_path): + from nanobot.bus.events import OutboundMessage + + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + bot._loop.process_direct = AsyncMock( + return_value=OutboundMessage(channel="cli", chat_id="direct", content="done"), + ) + + run = await bot.run_streamed("hi") + events = [event async for event in run.stream_events()] + assert [event.type for event in events] == ["run.started", "run.completed"] + await run.wait() + + with pytest.raises(RuntimeError, match="only be consumed once"): + _ = [event async for event in run.stream_events()] + + @pytest.mark.asyncio async def test_sdk_capture_prefers_run_level_snapshot(): from nanobot.agent.hook import AgentHookContext, AgentRunHookContext, SDKCaptureHook @@ -382,10 +1101,133 @@ async def test_sdk_capture_prefers_run_level_snapshot(): await hook.after_run(AgentRunHookContext( messages=final_messages, tools_used=["read_file"], + usage={"total_tokens": 3}, + stop_reason="completed", )) assert hook.tools_used == ["read_file"] assert hook.messages == final_messages + assert hook.usage == {"total_tokens": 3} + assert hook.stop_reason == "completed" + + +@pytest.mark.asyncio +async def test_sessions_ingest_imports_transcript_without_running_model(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + bot._loop.process_direct = AsyncMock() + bot._loop.consolidator.maybe_consolidate_by_tokens = AsyncMock() + + snapshot = await bot.sessions.ingest( + "sdk:history", + [ + { + "role": "user", + "content": "I graduated with a Business Administration degree.", + "timestamp": "2023/05/30 (Tue) 17:27", + "source_session_id": "answer_1", + }, + { + "role": "assistant", + "content": "Congratulations on your degree.", + "timestamp": "2023/05/30 (Tue) 17:27", + }, + ], + metadata={"title": "LongMemEval case"}, + source="longmemeval", + ) + + assert isinstance(snapshot, SessionSnapshot) + assert snapshot.key == "sdk:history" + assert snapshot.metadata["title"] == "LongMemEval case" + assert snapshot.messages[0]["role"] == "user" + assert snapshot.messages[0]["timestamp"] == "2023/05/30 (Tue) 17:27" + assert snapshot.messages[0]["source_session_id"] == "answer_1" + assert snapshot.messages[0]["source"] == "longmemeval" + assert snapshot.messages[1]["source"] == "longmemeval" + bot._loop.process_direct.assert_not_called() + bot._loop.consolidator.maybe_consolidate_by_tokens.assert_not_called() + + reloaded = bot.sessions.get("sdk:history") + assert reloaded is not None + assert reloaded.messages == snapshot.messages + + +@pytest.mark.asyncio +async def test_sessions_ingest_validates_message_shape(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + with pytest.raises(ValueError, match="role"): + await bot.sessions.ingest("sdk:bad", [{"content": "missing role"}]) + + with pytest.raises(ValueError, match="unsupported message role"): + await bot.sessions.ingest("sdk:bad", [{"role": "developer", "content": "nope"}]) + + +@pytest.mark.asyncio +async def test_session_helpers_get_list_export_clear_delete_flush(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + await bot.sessions.ingest("sdk:first", [{"role": "user", "content": "hello"}]) + + listed = bot.sessions.list() + assert listed + assert isinstance(listed[0], SessionInfo) + assert {row.key for row in listed} == {"sdk:first"} + + exported = bot.sessions.export("sdk:first") + assert exported is not None + exported.messages[0]["content"] = "mutated copy" + assert bot.sessions.get("sdk:first").messages[0]["content"] == "hello" + + cleared = bot.sessions.clear("sdk:first") + assert cleared.messages == [] + assert bot.sessions.flush() >= 1 + assert bot.sessions.delete("sdk:first") is True + assert bot.sessions.get("sdk:first") is None + + +def test_memory_helpers_read_write_append_and_filter_history(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + + assert bot.memory.read() == "" + bot.memory.write("# Memory\n- User likes concise APIs.") + assert "concise APIs" in bot.memory.read() + + c1 = bot.memory.append_history("general event") + c2 = bot.memory.append_history("session event", session_key="sdk:history") + + all_entries = bot.memory.read_history() + assert [entry["cursor"] for entry in all_entries] == [c1, c2] + + session_entries = bot.memory.read_history(session_key="sdk:history") + assert len(session_entries) == 1 + assert session_entries[0]["content"] == "session event" + + +@pytest.mark.asyncio +async def test_runtime_helpers_expose_model_workspace_and_compact(tmp_path): + config_path = _write_config(tmp_path) + bot = Nanobot.from_config(config_path, workspace=tmp_path) + await bot.sessions.ingest("sdk:history", [{"role": "user", "content": "hello"}]) + + bot._loop.consolidator.maybe_consolidate_by_tokens = AsyncMock() + snapshot = await bot.runtime.compact_session("sdk:history") + assert snapshot.key == "sdk:history" + bot._loop.consolidator.maybe_consolidate_by_tokens.assert_awaited_once() + assert bot.runtime.model == bot._loop.model + assert bot.runtime.workspace == tmp_path + + bot._loop.consolidator.compact_idle_session = AsyncMock(return_value="Summary.") + summary = await bot.runtime.compact_idle_session("sdk:history", max_suffix=4) + assert summary == "Summary." + bot._loop.consolidator.compact_idle_session.assert_awaited_once_with( + "sdk:history", + max_suffix=4, + ) @pytest.mark.asyncio