Compare commits

..
Author SHA1 Message Date
Xubin Ren 3f29b10f0d Merge remote-tracking branch 'origin/main' into codex/unified-extension-platform 2026-07-27 14:37:18 +08:00
Xubin Ren bed0db4922 refactor(extensions): simplify the native extension platform 2026-07-27 14:37:05 +08:00
chengyongruandGitHub 281b4b7f0b chore: remove expired v0.3.1 compatibility shims (#5106) 2026-07-27 13:53:04 +08:00
chengyongruandchengyongru 39348dfafe refactor(agent): remove dead lifecycle scaffolding 2026-07-27 12:00:06 +08:00
Xubin Ren d68857bb2d Merge remote-tracking branch 'origin/main' into codex/unified-extension-platform
# Conflicts:
#	docs/configuration.md
2026-07-27 10:36:20 +08:00
chengyongruandXubin Ren b3d3a3e6c3 fix(image): delegate DNS to explicit proxy 2026-07-27 10:06:19 +08:00
chengyongruandXubin Ren d73794bc68 fix(image): honor provider proxy for URL downloads 2026-07-27 10:06:19 +08:00
Xubin Ren cc3dbbe804 fix(security): block IPv6 unspecified SSRF targets 2026-07-27 10:06:19 +08:00
Xubin Ren 4408cde019 fix(security): harden generated image downloads 2026-07-27 10:06:19 +08:00
Xubin Ren cf1e801a29 fix(image): align Gemini hints with model capabilities 2026-07-27 03:07:41 +08:00
Xubin Ren a8604a3172 fix(image): scope Gemini image sizes by model 2026-07-27 03:07:41 +08:00
ef445cc246 fix(image): narrow Gemini Flash aspect-ratio and image-size scoping
Address review feedback that the capability checks were broader than the
documented per-model matrix:

- Drop the extreme aspect ratios (1:4, 4:1, 1:8, 8:1) from the Flash
  allow-list. They are only documented for 3.1 Flash / Flash Lite, so the
  global set could send an unsupported ratio to 2.5 Flash Image or 3.1 Pro
  Image. Keep the ratios common to every Flash image model.
- Identify imageSize support positively via "gemini-3" instead of excluding
  "2.5". The old predicate also matched gemini-2.0-flash-preview-image-
  generation, which (with the default 1K size) altered that model's request
  shape even though only Gemini 3+ image models accept a configurable size.

Add tests for the gemini-2.0 image-size drop and the extreme-ratio drop.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-27 03:07:41 +08:00
4986590bd7 fix(image): pass aspect ratio and size to Gemini Flash image models
The Gemini Flash image path (`generateContent`) dropped both `aspect_ratio`
and `image_size`: `generate()` never forwarded them and
`_generate_gemini_flash` did not accept them, so every request fell back to
1:1 / input-matched output. The Imagen path was unaffected.

Forward the hints and emit them under
`generationConfig.responseFormat.image` per the current Gemini API. Aspect
ratio is validated against the accepted set; `imageSize` is validated against
{512,1K,2K,4K} and only sent to Gemini 3+ image models, since
`gemini-2.5-flash-image` supports only `aspectRatio`.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-27 03:07:41 +08:00
Xubin Ren b695a7e875 fix(cli): harden quick start OAuth handling 2026-07-27 02:51:04 +08:00
Xubin Ren a4ec83fb0d fix(cli): scope Codex proxy env resolution 2026-07-27 02:51:04 +08:00
chengyongruandXubin Ren 2a1f840ce2 fix(cli): support Codex OAuth in quick start 2026-07-27 02:51:04 +08:00
Xubin Ren addaf2d3fc fix(dingtalk): harden group reply sender labels 2026-07-27 02:33:41 +08:00
9f3dee0192 docs(dingtalk): clarify disable_private_chat intent in comments
Addresses automated review: document that the guard is an intentional hard group-only switch (allowlisted DMs blocked by design) and that str() guards a None sender_id. Comment-only.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-27 02:33:41 +08:00
205889f9e0 feat(dingtalk): prefix group replies with sender mention
In group chats, prefix the outbound markdown reply with an H1 naming the sender (# @<nick>) so the addressed user can spot it in a busy group. Private replies are sent verbatim.

Visual only: DingTalk markdown robot messages do not push real @ notifications (that would require staffId plumbing and a different message type). sender_name is read from OutboundMessage.metadata, which the agent loop already propagates from inbound metadata.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-27 02:33:41 +08:00
14e692e40d feat(dingtalk): add disable_private_chat to reject 1:1 DMs
Add a `disable_private_chat` config flag (JSON alias `disablePrivateChat`,
default False) to the DingTalk channel. When enabled, any non-group (1:1)
message is rejected with a Chinese notice directing the user to group chat
("该机器人未开启私聊,请在群聊中与我对话。") before permission/pairing logic
runs, so even allowlisted senders are redirected. Group messages are
unaffected.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-27 02:33:41 +08:00
Xubin Ren 68717937e8 fix(agent): throttle idle scans by default 2026-07-27 02:15:48 +08:00
Andrew KhmylovandXubin Ren 7aab7e8830 feat(agent): make idle compaction scan interval configurable
Before this change, idle compaction is triggered every 1 second
if the incoming message stream is idle.
When triggered, it enumerates all session files, loads and parses them,
and then checks their expiration.

This becomes too CPU-intensive, especially on low-power devices like Raspberry Pi.
It's unlikely that you actually need to compact every second over the long time.

This change adds a configurable throttling for idle-compaction.

Default behavior is unchanged.
2026-07-27 02:15:48 +08:00
Xubin Ren 4e2640f2d2 fix(memory): keep failed Dream batches retryable 2026-07-27 02:00:41 +08:00
shixi-liandXubin Ren 15e42059bd fix(memory): progress past completed no-op batches 2026-07-27 02:00:41 +08:00
Xubin Ren b55b76d755 fix(streaming): preserve recovered segments across channels 2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren e6baecafcd fix(agent): close length recovery lifecycle gaps 2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren 27a00c7a4f fix(webui): merge length recovery stream segments 2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren 3cc5a98d9f refactor(agent): derive recovery count from segments 2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren 1d2ed6e4d2 fix(agent): reset recovery chains across injections
Reset both the recovered segments and retry budget whenever injected input starts a new logical answer. Cover fatal tool-error boundaries and rename the prompt test module so pytest can collect the full suite.
2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren 154cbc1974 refactor(agent): trim recovery tail anchor 2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren df2e5b7225 fix(agent): anchor truncated response continuations 2026-07-27 01:39:46 +08:00
chengyongruandXubin Ren b19039f9d0 fix(agent): preserve length-recovered output 2026-07-27 01:39:46 +08:00
Xubin Ren c1899e2cb4 fix(mcp): decode URI-encoded schema refs 2026-07-27 01:14:41 +08:00
amplifierplusandXubin Ren 9aae7485d6 fix(mcp): normalize local schema refs 2026-07-27 01:14:41 +08:00
chengyongruandXubin Ren 2e2f15dd0c fix(channels): serialize Feishu connect completion 2026-07-27 01:00:12 +08:00
KDBandXubin Ren 4835814746 fix(channels): ignore confirmations after connect cancellation 2026-07-27 01:00:12 +08:00
Xubin Ren d236883e2d fix(pairing): reject malformed store entries 2026-07-27 00:46:40 +08:00
santhrealandXubin Ren f7bf4c972e fix(pairing): treat null approved/pending maps as empty 2026-07-27 00:46:40 +08:00
Xubin Ren cf6ca13b6d fix(exec): preserve bwrap workspace masking 2026-07-27 00:31:00 +08:00
yu-xin-candXubin Ren 22e61003f9 test(exec): make bwrap bind tests portable 2026-07-27 00:31:00 +08:00
yu-xin-candXubin Ren 01a11b3980 feat(exec): allow extra bwrap bind roots 2026-07-27 00:31:00 +08:00
Xubin Ren 5d8046deef test(heartbeat): cover ignored unified routes 2026-07-27 00:12:44 +08:00
yu-xin-candXubin Ren a7a6c26eab fix(heartbeat): route unified sessions to last channel 2026-07-27 00:12:44 +08:00
chengyongruandchengyongru be43a54570 fix(webui): prevent mobile thread overflow 2026-07-26 23:59:28 +08:00
Xubin Ren ff379b91cf fix(agent): preserve merged runtime context markers 2026-07-26 23:46:54 +08:00
yu-xin-candXubin Ren eb93060f95 fix(agent): preserve pending runtime context 2026-07-26 23:46:54 +08:00
santhrealandXubin Ren 07c3e02d5c fix(triggers): treat null runHistory as empty when loading triggers 2026-07-26 23:33:28 +08:00
santhrealandXubin Ren aaf2eef568 fix(feishu): tolerate null multi_url and list fields in card extract 2026-07-26 23:19:35 +08:00
santhrealandXubin Ren 1e505ff405 fix(triggers): coerce string lastRunAtMs when loading local triggers 2026-07-26 23:05:19 +08:00
Xubin Ren 30750060ce test(feishu): cover null post metadata fields 2026-07-26 22:51:49 +08:00
santhrealandXubin Ren a7cac65c76 fix(feishu): move post extract test import to module top 2026-07-26 22:51:49 +08:00
santhrealandXubin Ren fb88154377 fix(feishu): tolerate null text fields when extracting post content 2026-07-26 22:51:49 +08:00
Xubin Ren a521cf31d9 test(extensions): make marker coverage platform-neutral 2026-07-26 21:35:53 +08:00
chengyongruandchengyongru d576804f23 feat(channels): enable tool hints by default 2026-07-26 21:22:12 +08:00
Xubin Ren 0b81858378 fix(extensions): harden compatibility and lifecycle 2026-07-26 21:19:29 +08:00
Xubin Ren bbaafd0f4f docs(extensions): document management and compatibility 2026-07-26 21:19:29 +08:00
Xubin Ren 41bebdcdb5 feat(extensions): add first-class WebUI management 2026-07-26 21:19:29 +08:00
Xubin Ren 55e497be14 feat(extensions): add management CLI 2026-07-26 21:19:29 +08:00
Xubin Ren 83b54212ae feat(extensions): wire runtime management and marketplace 2026-07-26 21:19:29 +08:00
Xubin Ren 6e0950833a feat(extensions): enforce dependencies and permission grants 2026-07-26 21:19:29 +08:00
Xubin Ren 012c7ce034 feat(extensions): load published OpenClaw runtimes 2026-07-26 21:19:29 +08:00
Xubin Ren 76d7a33b3b feat(extensions): install compatible packages atomically 2026-07-26 21:19:29 +08:00
Xubin Ren 0cd091ba93 feat(extensions): activate packages transactionally 2026-07-26 21:19:29 +08:00
Xubin Ren 029f9bc53b feat(extensions): add Pi and OpenClaw sidecar protocol 2026-07-26 21:19:29 +08:00
Xubin Ren 1e573c75ae feat(extensions): project native capabilities into catalog 2026-07-26 21:19:29 +08:00
Xubin Ren 863b02d215 feat(extensions): add manifest discovery and policy config 2026-07-26 21:19:29 +08:00
Xubin Ren 336b2876d4 feat(extensions): add unified manifest and registry 2026-07-26 21:19:29 +08:00
chengyongruandGitHub ee93725e83 fix(webui): restore file edit diff display (#5096) 2026-07-26 19:05:53 +08:00
santhrealandchengyongru 7c94ba9643 fix(session): coerce null session metadata to empty dict 2026-07-26 17:47:16 +08:00
santhrealandchengyongru 745757cc37 fix(memory): skip non-dict history.jsonl lines when reading 2026-07-26 17:45:51 +08:00
santhrealandchengyongru 259d8a018c fix(skills): tolerate null requires/bins/env in skill metadata 2026-07-26 17:44:41 +08:00
175 changed files with 9721 additions and 908 deletions
+2 -2
View File
@@ -14,9 +14,9 @@ Shell execution (`ExecTool`, `agent/tools/shell.py`) also respects `restrict_to_
## SSRF Protection
All outbound HTTP requests from agent tools must pass through `validate_url_target` (`security/network.py`). By default it blocks loopback, RFC1918 private addresses, CGNAT ranges, link-local ranges, and cloud metadata endpoints (including `169.254.169.254`).
All outbound HTTP requests from agent tools must pass through the shared URL guards in `security/network.py` (`validate_url_target` or `resolve_url_target`). By default they block loopback, RFC1918 private addresses, CGNAT ranges, link-local ranges, and cloud metadata endpoints (including `169.254.169.254`).
The only escape hatch is `configure_ssrf_whitelist(cidrs)`, which reads from `config.tools.ssrf_whitelist` at load time.
For direct requests, the only escape hatch is `configure_ssrf_whitelist(cidrs)`, which reads from `config.tools.ssrf_whitelist` at load time. An explicitly configured `providers.<name>.proxy` is a separate user-authorized trust boundary for provider requests and provider-returned image URL downloads. Those downloads still reject malformed URLs and locally identifiable private/internal targets on every redirect, but hostnames unavailable to local DNS are delegated to the trusted proxy. The user-selected proxy owns final DNS resolution and network egress policy.
HTTP/SSE MCP transports are part of this boundary: validate configured MCP URLs before probing or constructing clients, and validate each outgoing HTTP request before redirects are followed. Local/private HTTP MCP endpoints are allowed only through the explicit SSRF whitelist. Stdio MCP servers are not part of the HTTP SSRF path.
+2
View File
@@ -33,6 +33,7 @@ Pick the row that matches what you want to accomplish next:
| Add model fallbacks | [Configure Model Fallback](./guides/configure-model-fallback.md) |
| Enable web search | [Configure Web Search](./guides/configure-web-search.md) |
| Add an MCP tool server | [Configure MCP Tools](./guides/configure-mcp-tools.md) |
| Install and govern an extension | [Extensions](./extensions.md) |
| Generate images | [Image Generation](./image-generation.md) |
| Schedule work or create a local trigger | [Automations](./automations.md) |
| Understand and manage long-term memory | [Memory](./memory.md) |
@@ -79,6 +80,7 @@ These pages explain implementation and extension points. You do not need them to
| Understand source ownership and runtime flow | [Architecture](./architecture.md) |
| Set up a development environment | [Development](./development.md) and [CONTRIBUTING.md](../CONTRIBUTING.md) |
| Add a channel package | [Channel Package Guide](./channel-package-guide.md) |
| Publish an extension package | [Extension Authoring](./extension-authoring.md) |
| Build the WebUI source | [WebUI Development](../webui/README.md) |
If a command or screen no longer matches these docs, please [open an issue](https://github.com/HKUDS/nanobot/issues) with your nanobot version, operating system, and the page that needs correction.
+2 -2
View File
@@ -618,7 +618,7 @@ async def send(self, msg: OutboundMessage) -> None:
await self._send_message(msg.chat_id, msg.content, media=msg.media)
```
Tool hints are off by default for most channels. Users can enable them globally or per channel:
Tool hints are on by default. Users can disable them globally or per channel:
```json
{
@@ -626,7 +626,7 @@ Tool hints are off by default for most channels. Users can enable them globally
"sendToolHints": true,
"webhook": {
"enabled": true,
"sendToolHints": true
"sendToolHints": false
}
}
}
+34
View File
@@ -18,6 +18,7 @@ Use this page when you know what you want to run and need the command shape. For
| Deliver a local trigger | `nanobot trigger <id> "message"` | Created first with `/trigger <name>` in the target chat/session |
| Serve an OpenAI-compatible API | `nanobot serve` | Starts `/v1/chat/completions`, `/v1/models`, and `/health` |
| Check chat channel setup | `nanobot channels status` | Useful before starting `nanobot gateway` |
| Manage extension packages | `nanobot extensions list` | Install, inspect, trust, enable, and remove native nanobot packages |
| Manage optional features | `nanobot plugins list` | Shows channels and optional capabilities you can turn on |
| Log in to QR/OAuth-style channels | `nanobot channels login <channel>` | Used by channels such as WhatsApp and WeChat |
| Log in to OAuth model providers | `nanobot provider login <provider>` | Used by OpenAI Codex, xAI subscription, and GitHub Copilot providers |
@@ -248,6 +249,39 @@ nanobot channels status
See [`chat-apps.md`](./chat-apps.md) for channel-specific setup.
## Extensions
Extension installation, trust, permission grants, and enablement are separate
operations:
| Command | Description |
|---|---|
| `nanobot extensions list` | Show installed packages and activation policy |
| `nanobot extensions inspect <id>` | Show identity, dependencies, requested permissions, and diagnostics |
| `nanobot extensions install <url> --kind git [--ref <ref>]` | Install from a Git branch, tag, or commit |
| `nanobot extensions install <path> --kind local` | Install from a local package directory |
| `nanobot extensions permissions <id> [permissions...]` | Replace the exact granted permission set; omit values to revoke all |
| `nanobot extensions trust <id>` | Approve executing the installed package |
| `nanobot extensions untrust <id>` | Revoke trust and stop activation |
| `nanobot extensions enable <id>` | Allow activation when every other gate passes |
| `nanobot extensions disable <id>` | Stop activation without uninstalling |
| `nanobot extensions uninstall <id>` | Remove the user-scope package after confirmation |
| `nanobot extensions uninstall <id> --yes` | Remove without an interactive confirmation |
Example:
```bash
nanobot extensions install https://github.com/acme/nanobot-review.git
nanobot extensions inspect acme.review
nanobot extensions permissions acme.review workspace.read
nanobot extensions trust acme.review
nanobot extensions enable acme.review
```
Installed packages live under `~/.nanobot/extensions/`. They do not execute
until trusted. See [Extensions](./extensions.md) for the safety model and
[Extension Authoring](./extension-authoring.md) for the native package contract.
## Optional Features
Use these commands when you want nanobot to add or remove a built-in capability
+39 -7
View File
@@ -27,6 +27,7 @@ the focused guides first and come back here for exact fields and defaults.
| Configure model fallback | [`guides/configure-model-fallback.md`](./guides/configure-model-fallback.md) |
| Add an OpenAI-compatible provider | [`guides/configure-openai-compatible-provider.md`](./guides/configure-openai-compatible-provider.md) |
| Add Langfuse observability | [`guides/configure-langfuse-observability.md`](./guides/configure-langfuse-observability.md) |
| Install and govern extensions | [`extensions.md`](./extensions.md) |
| Secure a local AI agent | [`guides/secure-local-ai-agent.md`](./guides/secure-local-ai-agent.md) |
| Deploy the gateway | [`guides/deploy-nanobot-gateway.md`](./guides/deploy-nanobot-gateway.md) |
@@ -45,6 +46,7 @@ the focused guides first and come back here for exact fields and defaults.
| Configure web search and fetch | [Web Tools](#web-tools) |
| Enable image generation | [Image Generation](#image-generation) |
| Add MCP servers | [MCP](#mcp-model-context-protocol) |
| Enable or disable external extensions | [Extensions](#extensions) |
| Review shell, workspace, and SSRF controls | [Security](#security) |
| Control access and pairing | [Pairing](#pairing) |
| Tune gateway jobs, sessions, and tools | [Gateway Heartbeat](#gateway-heartbeat), [Auto Compact](#auto-compact), [Unified Session](#unified-session), [Tool Hint Max Length](#tool-hint-max-length) |
@@ -64,6 +66,7 @@ If the WebUI does not expose the option you need, start from the task below. Mos
| Enable web search or fetch | `tools.web.search.*`, `tools.web.fetch.*`, optional `tools.ssrfWhitelist` | Ask a question that requires current web information, then inspect logs if needed | [Web Tools](#web-tools), [Security](#security) |
| Enable image generation | `tools.imageGeneration.enabled`, `tools.imageGeneration.provider`, `tools.imageGeneration.model`, matching provider credentials | Enable Image Generation in the WebUI and send one image request | [Image Generation](#image-generation) |
| Add external tools through MCP | `tools.mcpServers.<name>` | Start `nanobot gateway --verbose` and check startup/tool logs | [MCP](#mcp-model-context-protocol) |
| Enable external extension packages | `extensions.enabled` | `nanobot extensions list`, then inspect the package | [Extensions](#extensions), [Extension guide](./extensions.md) |
| Tighten tool and network safety | `tools.restrictToWorkspace`, `tools.exec.sandbox`, `tools.ssrfWhitelist`, `channels.*.allowFrom` | Run the same workflow through the channel or CLI you plan to expose | [Security](#security), [Pairing](#pairing) |
| Tune request timeouts or process concurrency | `NANOBOT_LLM_TIMEOUT_S`, `NANOBOT_STREAM_IDLE_TIMEOUT_S`, `NANOBOT_MAX_CONCURRENT_REQUESTS` | Start nanobot from the same environment and inspect startup/runtime logs | [Runtime Environment Variables](#runtime-environment-variables) |
| Run multiple isolated bots | separate `--config` and `--workspace` paths, plus distinct `gateway.port` or channel ports when processes run together | Use the same explicit paths with `nanobot status`, `agent`, `webui`, `gateway`, and `serve` | [Multiple Instances](./multiple-instances.md), [CLI Reference](./cli-reference.md) |
@@ -1555,7 +1558,7 @@ Global settings that apply to all channels. Configure under the `channels` secti
{
"channels": {
"sendProgress": true,
"sendToolHints": false,
"sendToolHints": true,
"extractDocumentText": true,
"sendMaxRetries": 3,
"telegram": {
@@ -1568,7 +1571,7 @@ Global settings that apply to all channels. Configure under the `channels` secti
| Setting | Default | Description |
|---------|---------|-------------|
| `sendProgress` | `true` | Stream agent's text progress to the channel |
| `sendToolHints` | `false` | Stream tool-call hints (e.g. `read_file("…")`) |
| `sendToolHints` | `true` | Stream tool-call hints (e.g. `read_file("…")`) |
| `showReasoning` | `true` | Allow channels to surface model reasoning/thinking content (DeepSeek-R1 `reasoning_content`, Anthropic `thinking_blocks`, inline `<think>` tags). Reasoning flows as a dedicated stream with `_reasoning_delta` / `_reasoning_end` markers — channels override `send_reasoning_delta` / `send_reasoning_end` to render in-place updates. Even with `true`, channels without those overrides stay no-op silently. Currently surfaced on CLI and WebSocket/WebUI (italic shimmer header, auto-collapses after the stream ends); Telegram / Slack / Discord / Feishu / WeChat / Matrix / Mattermost keep the base no-op until their bubble UI is adapted. Independent of `sendProgress`. |
| `extractDocumentText` | `true` | Extract supported document/text attachments into the model prompt. PDF, DOCX, XLSX, and PPTX readers are included in the standard installation. Set to `false` to keep document content out of the prompt and include attachment path references instead. |
| `sendMaxRetries` | `3` | Max delivery attempts per outbound message, including the initial send (0-10 configured, minimum 1 actual attempt) |
@@ -1581,10 +1584,11 @@ Global settings that apply to all channels. Configure under the `channels` secti
{
"channels": {
"sendProgress": true,
"sendToolHints": false,
"sendToolHints": true,
"telegram": {
"enabled": true,
"sendProgress": false
"sendProgress": false,
"sendToolHints": false
},
"websocket": {
"enabled": true,
@@ -1994,7 +1998,9 @@ For API keys, tokens, and other secrets, see [Environment Variables for Secrets]
| `tools.exec.timeout` | `60` | Default hard timeout in seconds for shell commands. Config values may exceed the per-call tool cap; set `0` to disable the hard timeout for trusted long-running commands. |
| `tools.exec.pathPrepend` | `""` | Extra directories to prepend to `PATH` when running shell commands. Use this when configured tools should win executable lookup precedence, such as a Python virtual environment's `bin` or `Scripts` directory. |
| `tools.exec.pathAppend` | `""` | Extra directories to append to `PATH` when running shell commands (e.g. `/usr/sbin` for `ufw`). |
| `tools.webuiAllowRemotePackageInstall` | `false` | When `false`, the WebUI can install missing optional packages only from a browser opened on the same machine as nanobot. Set to `true` only when a trusted remote admin is allowed to install Python packages into this nanobot environment. |
| `tools.exec.sandboxRoBinds` | `[]` | Extra absolute paths to read-only bind into the `"bwrap"` sandbox with `--ro-bind-try`, such as `/home/user/.local/bin` or `/home/user/.cargo/bin` when those paths are also in `pathPrepend`/`pathAppend`. These roots are also accepted by the shell absolute-path guard only while bwrap is active. Bind only directories whose contents are safe for agent commands to read; paths equal to or containing the active workspace are ignored so they cannot uncover its masked parent directory. |
| `tools.exec.sandboxRwBinds` | `[]` | Extra absolute paths to read-write bind into the `"bwrap"` sandbox with `--bind-try`, for trusted tool caches or scratch directories. Use sparingly: paths listed here are intentionally writable by shell commands inside the sandbox. Paths equal to or containing the active workspace are ignored. |
| `tools.webuiAllowRemotePackageInstall` | `false` | When `false`, the WebUI can install optional support or extension packages only from a browser opened on the same machine as nanobot. Set to `true` only when a trusted remote admin may install packages into this environment. |
| `tools.ssrfWhitelist` | `[]` | CIDR ranges exempted from the shared SSRF guard used by web fetches and HTTP/SSE MCP connections. Prefer exact host CIDRs such as `192.168.1.50/32`; broad ranges increase SSRF exposure. |
| `channels.*.allowFrom` | omitted | Access control per channel. Omit to use pairing-only mode; set `["*"]` to allow everyone; or list specific user IDs. See [Pairing](#pairing) for details. |
@@ -2155,7 +2161,8 @@ When a user is idle for longer than a configured threshold, nanobot **proactivel
{
"agents": {
"defaults": {
"idleCompactAfterMinutes": 15
"idleCompactAfterMinutes": 15,
"idleCompactCheckIntervalSeconds": 60
}
}
}
@@ -2164,11 +2171,12 @@ When a user is idle for longer than a configured threshold, nanobot **proactivel
| Option | Default | Description |
|--------|---------|-------------|
| `agents.defaults.idleCompactAfterMinutes` | `15` | Minutes of idle time before auto-compaction starts. Set to `0` to disable. The default is close to a typical LLM KV cache expiry window, so stale sessions get compacted before the user returns. |
| `agents.defaults.idleCompactCheckIntervalSeconds` | `60` | Minimum number of seconds between scans for idle sessions. Set to `0` to scan on every idle tick (~1 s). |
`sessionTtlMinutes` remains accepted as a legacy alias for backward compatibility, but `idleCompactAfterMinutes` is the preferred config key going forward.
How it works:
1. **Idle detection**: On each idle tick (~1 s), checks all sessions for expiration.
1. **Idle detection**: On each idle tick (~1 s), checks whether an idle-session scan is due. By default, the full scan runs at most once per minute.
2. **Background compaction**: Idle sessions summarize the older live prefix via LLM and keep the most recent legal suffix (currently 8 messages).
3. **Summary injection**: When the user returns, the summary is injected as runtime context (one-shot, not persisted) alongside the retained recent suffix.
4. **Restart-safe resume**: The summary is also mirrored into session metadata so it can still be recovered after a process restart.
@@ -2228,6 +2236,30 @@ When enabled, all incoming messages — regardless of which channel they arrive
> This is designed for single-user, multi-device setups. It is **off by default** — existing users see zero behavior change.
## Extensions
Use the WebUI **Extensions** page or `nanobot extensions` commands for normal
installation and trust decisions. Extension support can be disabled globally:
```json
{
"extensions": {
"enabled": true
}
}
```
| Option | Default | Description |
|---|---|---|
| `extensions.enabled` | `true` | Enable external extension discovery and activation |
Installed packages and their trust, permission, and activation state live
under `~/.nanobot/extensions/`. Managing an extension does not rewrite
`config.json`.
See [Extensions](./extensions.md) for the safe install flow and
[Extension Authoring](./extension-authoring.md) for the package contract.
## Disabled Skills
nanobot ships with built-in skills, and your workspace can also define custom skills under `skills/`. If you want to hide specific skills from the agent, set `agents.defaults.disabledSkills` to a list of skill directory names:
+147
View File
@@ -0,0 +1,147 @@
# Extension Authoring
A native nanobot extension is a directory containing:
```text
nanobot-review/
├── nanobot.extension.json
└── extension.py
```
The manifest describes identity, activation prerequisites, and requested
permissions. The Python entry point performs the real registration. This keeps
one authoritative source for tool, command, and hook ownership.
## Manifest
```json
{
"id": "acme.review",
"name": "Acme Review",
"version": "1.0.0",
"entry": "extension:register",
"description": "Adds repository review tools.",
"apiVersion": 1,
"license": "MIT",
"homepage": "https://github.com/acme/nanobot-review",
"dependencies": [
{
"kind": "executable",
"name": "git"
}
],
"permissions": [
{
"name": "workspace.read",
"reason": "Read files selected for review."
}
]
}
```
Required fields are `id`, `name`, and `version`. `entry` defaults to
`"extension:register"` and `apiVersion` defaults to `1`.
IDs use lowercase letters, digits, dots, underscores, and hyphens. Entry points
use `module:function` syntax and must resolve inside the package.
### Dependencies
| Kind | Meaning |
|---|---|
| `python` | Installed Python distribution; `specifier` accepts a version constraint |
| `executable` | Command available on `PATH` |
| `environment` | Non-empty environment variable |
Set `"optional": true` when a missing dependency should not block activation.
### Permissions
Permissions are lowercase namespaced identifiers chosen by the package, such
as `workspace.read` or `network`. Give each permission a concrete reason.
Activation waits until every requested permission is granted.
The host currently uses permissions as explicit user consent. They do not
sandbox Python code, so do not describe a permission as stronger isolation
than it provides.
## Registration API
The entry point receives `PythonExtensionApi` and must return `None`:
```python
from typing import Any
from nanobot.agent.tools.base import Tool
class ReviewTool(Tool):
@property
def name(self) -> str:
return "review_repository"
@property
def description(self) -> str:
return "Review the current repository."
@property
def parameters(self) -> dict[str, Any]:
return {"type": "object", "properties": {}}
async def execute(self, **kwargs: Any) -> str:
return "No findings."
def register(api) -> None:
api.register_tool(ReviewTool())
```
The API has three stable methods:
```python
api.register_tool(tool)
api.register_command("review", handler)
api.register_hook_factory(factory)
```
Command handlers use nanobot's `CommandContext` and return an
`OutboundMessage` or `None`. Hook factories receive `AgentTurnHookContext` and
return an `AgentHook` or `None`.
Do not modify `AgentLoop` or global registries directly. The API tags every
registration with the extension ID so reload, failure rollback, and uninstall
can remove exactly what the package owns.
## Collision and failure behavior
Tool and command names are unique across core and active extensions. If an
extension registers a duplicate name, activation fails for that extension and
all of its partial registrations are rolled back.
Missing dependencies are reported as diagnostics instead of crashing the
gateway.
## Develop locally
1. Create the manifest and entry module.
2. Install the directory with `--kind local`.
3. Inspect and grant its permissions.
4. Trust it.
5. Reinstall after editing so nanobot records a new integrity digest.
```bash
nanobot extensions install "$PWD" --kind local
nanobot extensions inspect acme.review
nanobot extensions permissions acme.review workspace.read
nanobot extensions trust acme.review
```
Keep tests in the extension repository. At minimum, test registration,
duplicate-name failure, and behavior when each required dependency is missing.
## Distribution
Publish the directory in a Git repository. Users can pin a release tag or
commit with `--ref`. The repository root must contain
`nanobot.extension.json`; install scripts and generated compatibility manifests
are not part of the native contract.
+89
View File
@@ -0,0 +1,89 @@
# Extensions
Extensions add native tools, slash commands, or lifecycle hooks without
changing nanobot core. An extension is a Python package with one manifest and
one registration entry point.
Use an extension when a capability needs executable integration with nanobot.
Use a [skill](./skills.md) when instructions alone are enough, an App when the
agent should call an external CLI, and MCP when a service already exposes an
MCP server.
## Install
Install from a Git repository:
```bash
nanobot extensions install https://github.com/acme/nanobot-review.git
```
Install a local package while developing it:
```bash
nanobot extensions install /absolute/path/to/nanobot-review --kind local
```
Git installs may select a branch, tag, or commit:
```bash
nanobot extensions install https://github.com/acme/nanobot-review.git \
--ref v1.2.0
```
The WebUI **Extensions** page exposes the same Git and local installation
flows. Local paths are accepted only from a browser running on the nanobot
host.
## Review before activation
New packages are installed enabled but untrusted. They cannot execute until
you review the manifest, grant every requested permission, and trust them:
```bash
nanobot extensions inspect acme.review
nanobot extensions permissions acme.review workspace.read
nanobot extensions trust acme.review
```
Use `list` to check the result:
```bash
nanobot extensions list
```
Disable, untrust, or remove a package at any time:
```bash
nanobot extensions disable acme.review
nanobot extensions untrust acme.review
nanobot extensions uninstall acme.review
```
Changes made in the WebUI reload its gateway extension host immediately.
Changes made by the standalone CLI take effect the next time the gateway or
agent process starts. Failed registrations are rolled back and reported as
diagnostics.
## Safety model
Extensions are executable Python code. nanobot provides these controls:
- packages are copied into `~/.nanobot/extensions/` with an integrity digest;
- package symlinks and special files are rejected;
- installation, permission grants, trust, and activation are separate steps;
- changed package contents invalidate trust;
- registration is transactional, so a failed extension does not leave tools,
commands, or hooks behind;
- remote WebUI clients cannot grant trust or permissions.
Permission declarations are consent gates, not an operating-system sandbox.
Only install code you are willing to run with the same account as nanobot.
## Package compatibility
The core runtime intentionally executes only the native nanobot Python
contract. Pi and OpenClaw packages are not loaded directly. Compatibility
adapters can be distributed as separate nanobot extensions later without
adding JavaScript runtimes or package-market policy to the agent core.
See [Extension Authoring](./extension-authoring.md) to build a package.
+3
View File
@@ -70,6 +70,9 @@ Provider settings reuse normal provider config fields:
| `providers.<name>.apiBase` | Optional custom base URL |
| `providers.<name>.extraHeaders` | Headers merged into provider requests |
| `providers.<name>.extraBody` | Extra JSON fields merged into provider request bodies |
| `providers.<name>.proxy` | Explicit trusted HTTP proxy for provider requests and returned image URL downloads |
For providers that return image URLs, direct downloads use DNS pinning. When an explicit provider `proxy` is configured, nanobot rejects malformed URLs and locally identifiable private/internal targets on the initial URL and every redirect. Hostnames unavailable to local DNS are delegated to that trusted proxy, which owns final DNS resolution and network egress. Process-wide proxy environment variables are not used for these downloads.
Both camelCase and snake_case config keys are accepted, but docs use camelCase to match `config.json`.
+1 -6
View File
@@ -186,9 +186,7 @@ Dream is configured under `agents.defaults.dream`:
"defaults": {
"dream": {
"intervalH": 2,
"modelOverride": null,
"maxBatchSize": 20,
"maxIterations": 10
"modelOverride": null
}
}
}
@@ -200,15 +198,12 @@ Dream is configured under `agents.defaults.dream`:
| `intervalH` | How often Dream runs, in hours |
| `cron` | Cron expression override (takes precedence over `intervalH`) |
| `modelOverride` | Optional Dream-specific model override *(pending implementation)* |
| `maxBatchSize` | *(Deprecated — not used)* |
| `maxIterations` | *(Deprecated — not used)* |
In practical terms:
- `intervalH` is the normal way to configure Dream frequency. Internally it runs as an `every` schedule.
- `cron` overrides `intervalH` when set, allowing precise cron expressions (e.g. `0 */4 * * *`).
- `modelOverride` is reserved for a future release. Currently Dream uses the same model as the main agent.
- `maxBatchSize` and `maxIterations` are preserved for config compatibility but no longer affect behavior.
## In Practice
+10 -7
View File
@@ -285,9 +285,9 @@ The gateway refuses to start with `host` set to `"0.0.0.0"` unless `token` or
form.
Remote WebUI clients with a valid token can view and use Apps. Actions that
install missing nanobot support packages, such as adding a channel dependency,
are blocked by default. To let trusted remote administrators change the Python
environment through the WebUI, opt in explicitly:
install missing nanobot support packages or first-class extension packages are
blocked by default. To let trusted remote administrators place packages into
this nanobot installation through the WebUI, opt in explicitly:
```json
{
@@ -298,12 +298,15 @@ environment through the WebUI, opt in explicitly:
```
Use this only for a private deployment where every authenticated WebUI user is
trusted to change the Python environment that nanobot runs in. If you publish
the WebUI through Nginx, Caddy, Cloudflare Tunnel, or a similar service, treat it
as remote access and leave package installs disabled unless that is intentional.
trusted to change the nanobot installation. A remotely installed extension
remains untrusted and inactive: trust, permission grants, activation, disabling,
and removal stay restricted to a browser on the nanobot host. If you publish the
WebUI through Nginx, Caddy, Cloudflare Tunnel, or a similar service, treat it as
remote access and leave package installs disabled unless that is intentional.
Optional feature installs use pip's configured package index, including
`PIP_INDEX_URL`.
`PIP_INDEX_URL`. Extension packages install from an explicit Git repository or
local directory.
Leave remote package installs disabled when the WebUI is exposed beyond a
private, trusted network.
+2 -6
View File
@@ -69,7 +69,7 @@ class ContextBuilder:
def build_system_prompt(
self,
skill_names: list[str] | None = None,
*,
channel: str | None = None,
session_summary: str | None = None,
workspace: Path | None = None,
@@ -196,14 +196,11 @@ class ContextBuilder:
self,
history: list[dict[str, Any]],
current_message: str,
skill_names: list[str] | None = None,
*,
media: list[str] | None = None,
channel: str | None = None,
chat_id: str | None = None,
current_role: str = "user",
sender_id: str | None = None,
session_summary: str | None = None,
session_metadata: Mapping[str, Any] | None = None,
runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None,
workspace: Path | None = None,
include_memory_recent_history: bool = True,
@@ -219,7 +216,6 @@ class ContextBuilder:
{
"role": "system",
"content": self.build_system_prompt(
skill_names,
channel=channel,
session_summary=session_summary,
workspace=root,
+1
View File
@@ -25,6 +25,7 @@ class AgentHookContext:
tool_events: list[dict[str, str]] = field(default_factory=list)
streamed_content: bool = False
streamed_reasoning: bool = False
stream_continues_current_message: bool = False
final_content: str | None = None
stop_reason: str | None = None
error: str | None = None
+176 -138
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
import dataclasses
import inspect
import os
import time
from collections.abc import Mapping
@@ -12,7 +13,7 @@ from dataclasses import dataclass, field
from enum import Enum, auto
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING, Any, Awaitable, Callable
from typing import TYPE_CHECKING, Any, Awaitable, Callable, TypeVar
from loguru import logger
@@ -73,7 +74,7 @@ from nanobot.session.goal_state import (
sustained_goal_active,
)
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
from nanobot.session.keys import UNIFIED_SESSION_KEY
from nanobot.session.keys import UNIFIED_SESSION_KEY, remember_last_channel
from nanobot.session.manager import (
Session,
SessionManager,
@@ -102,15 +103,7 @@ if TYPE_CHECKING:
)
from nanobot.cron.service import CronService
class TurnState(Enum):
RESTORE = auto()
COMPACT = auto()
COMMAND = auto()
BUILD = auto()
RUN = auto()
SAVE = auto()
RESPOND = auto()
DONE = auto()
_T = TypeVar("_T")
class TurnKind(Enum):
@@ -118,20 +111,10 @@ class TurnKind(Enum):
SYSTEM = auto()
@dataclass
class StateTraceEntry:
state: TurnState
started_at: float
duration_ms: float
event: str
error: str | None = None
@dataclass
class TurnContext:
msg: InboundMessage
session_key: str
state: TurnState
turn_id: str
runtime: LLMRuntime | None
kind: TurnKind
@@ -145,7 +128,6 @@ class TurnContext:
runtime_context_blocks: list[RuntimeContextBlock] = field(default_factory=list)
final_content: str | None = None
tools_used: list[str] = field(default_factory=list)
all_messages: list[dict[str, Any]] = field(default_factory=list)
stop_reason: str = ""
had_injections: bool = False
@@ -177,8 +159,6 @@ class TurnContext:
visible_run_started_at: float | None = None
turn_latency_ms: int | None = None
trace: list[StateTraceEntry] = field(default_factory=list)
class AgentLoop:
"""
@@ -243,19 +223,6 @@ class AgentLoop:
_RUNTIME_CHECKPOINT_KEY = "runtime_checkpoint"
_PENDING_USER_TURN_KEY = "pending_user_turn"
# Event-driven state transition table.
# Handlers return an event string; the driver looks up the next state here.
_TRANSITIONS: dict[tuple[TurnState, str], TurnState] = {
(TurnState.RESTORE, "ok"): TurnState.COMPACT,
(TurnState.COMPACT, "ok"): TurnState.COMMAND,
(TurnState.COMMAND, "dispatch"): TurnState.BUILD,
(TurnState.COMMAND, "shortcut"): TurnState.DONE,
(TurnState.BUILD, "ok"): TurnState.RUN,
(TurnState.RUN, "ok"): TurnState.SAVE,
(TurnState.SAVE, "ok"): TurnState.RESPOND,
(TurnState.RESPOND, "ok"): TurnState.DONE,
}
def __init__(
self,
bus: MessageBus,
@@ -296,6 +263,7 @@ class AgentLoop:
runtime_model_publisher: Callable[[str, str | None], None] | None = None,
restart_mode: str = "auto",
local_trigger_store: Any | None = None,
idle_compact_check_interval_seconds: int = 0,
):
from nanobot.config.schema import ToolsConfig
@@ -444,6 +412,8 @@ class AgentLoop:
consolidator=self.consolidator,
session_ttl_minutes=session_ttl_minutes,
)
self._idle_compact_check_interval_s = idle_compact_check_interval_seconds
self._next_idle_compact_check_at = time.monotonic()
if model_preset:
self.set_model_preset(model_preset, publish_update=False)
self._register_default_tools(provider_snapshot_loader=provider_snapshot_loader)
@@ -499,6 +469,7 @@ class AgentLoop:
unified_session=defaults.unified_session,
disabled_skills=defaults.disabled_skills,
session_ttl_minutes=defaults.session_ttl_minutes,
idle_compact_check_interval_seconds=defaults.idle_compact_check_interval_seconds,
consolidation_ratio=defaults.consolidation_ratio,
tools_config=config.tools,
model_presets=preset_helpers.configured_model_presets(config),
@@ -707,13 +678,8 @@ class AgentLoop:
current_message=ctx.msg.content,
media=ctx.msg.media if ctx.kind is TurnKind.USER and ctx.msg.media else None,
channel=ctx.delivery.route.channel,
chat_id=str(
ctx.msg.metadata.get("context_chat_id") or ctx.delivery.route.chat_id
),
current_role="user",
sender_id=ctx.msg.sender_id,
session_summary=ctx.pending_summary,
session_metadata=ctx.session.metadata,
workspace=scope.project_path,
runtime_context_blocks=ctx.runtime_context_blocks,
include_memory_recent_history=not ctx.ephemeral,
@@ -745,14 +711,23 @@ class AgentLoop:
self,
ctx: TurnContext,
) -> list[RuntimeContextBlock]:
tools = ctx.tools or self.tools
assert ctx.request_context is not None
return await self._resolve_runtime_context_for_request(
ctx.request_context,
ctx.tools or self.tools,
)
async def _resolve_runtime_context_for_request(
self,
request: RequestContext,
tools: ToolRegistry,
) -> list[RuntimeContextBlock]:
providers = [
*tools.get_runtime_context_providers(),
*self._runtime_context_providers,
]
assert ctx.request_context is not None
blocks = runtime_context_blocks_from_metadata(ctx.request_context.metadata)
blocks.extend(await resolve_runtime_context(providers, ctx.request_context))
blocks = runtime_context_blocks_from_metadata(request.metadata)
blocks.extend(await resolve_runtime_context(providers, request))
return blocks
async def _dispatch_command_inline(
@@ -789,6 +764,27 @@ class AgentLoop:
return UNIFIED_SESSION_KEY
return msg.session_key
def _remember_unified_session_route(
self,
session: Session,
msg: InboundMessage,
*,
is_user_turn: bool,
) -> None:
"""Remember the latest user-facing route for unified-session delivery."""
if (
not self._unified_session
or session.key != UNIFIED_SESSION_KEY
or not is_user_turn
or msg.channel in {"cli", "system"}
or msg.sender_id == "subagent"
):
return
_, automation_metadata = automation_history_overrides(msg.metadata)
if automation_metadata:
return
remember_last_channel(session.metadata, msg.channel, msg.chat_id)
@staticmethod
def _replay_token_budget(runtime: LLMRuntime) -> int:
"""Derive a token budget for session history replay from the context window."""
@@ -830,9 +826,9 @@ class AgentLoop:
"""Run the agent iteration loop.
*on_stream*: called with each content delta during streaming.
*on_stream_end(resuming)*: called when a streaming session finishes.
``resuming=True`` means tool calls follow (spinner should restart);
``resuming=False`` means this is the final response.
*on_stream_end(resuming, merge_next)*: called when a streaming session finishes.
``resuming=True`` means the active turn continues. ``merge_next=True`` means
the next text segment belongs to the same user-visible assistant message.
Returns (final_content, tools_used, messages, stop_reason, had_injections).
"""
@@ -855,7 +851,7 @@ class AgentLoop:
if pending_queue is None:
return []
def _to_user_message(pending_msg: InboundMessage) -> dict[str, Any]:
async def _to_user_message(pending_msg: InboundMessage) -> dict[str, Any]:
content = pending_msg.content
media = pending_msg.media if pending_msg.media else None
if media:
@@ -864,6 +860,31 @@ class AgentLoop:
user_content = self.context._build_user_content(content, media)
row: dict[str, Any] = {"role": "user", "content": user_content}
metadata = pending_msg.metadata if isinstance(pending_msg.metadata, dict) else {}
if pending_msg.channel != "system":
scope = self.workspace_scopes.for_turn(
channel=pending_msg.channel,
message_metadata=metadata,
session_metadata=session.metadata if session is not None else None,
)
pending_request = RequestContext(
channel=pending_msg.channel,
chat_id=pending_msg.chat_id,
message_id=metadata.get("message_id"),
session_key=active_session_key,
original_user_text=pending_msg.content,
runtime=runtime,
metadata=dict(metadata),
sender_id=pending_msg.sender_id,
turn_id=request_ctx.turn_id,
workspace=scope.project_path,
)
blocks = await self._resolve_runtime_context_for_request(
pending_request,
effective_tools,
)
row["content"], marker = append_runtime_context(user_content, blocks)
if marker is not None:
row["_meta"] = {RUNTIME_CONTEXT_MESSAGE_META: marker}
if (
pending_msg.sender_id == "subagent"
and metadata.get("injected_event") == "subagent_result"
@@ -880,7 +901,7 @@ class AgentLoop:
items: list[dict[str, Any]] = []
while len(items) < limit:
try:
items.append(_to_user_message(pending_queue.get_nowait()))
items.append(await _to_user_message(pending_queue.get_nowait()))
except asyncio.QueueEmpty:
break
@@ -898,10 +919,10 @@ class AgentLoop:
session.key,
)
return items
items.append(_to_user_message(msg))
items.append(await _to_user_message(msg))
while len(items) < limit:
try:
items.append(_to_user_message(pending_queue.get_nowait()))
items.append(await _to_user_message(pending_queue.get_nowait()))
except asyncio.QueueEmpty:
break
@@ -1014,12 +1035,29 @@ class AgentLoop:
# Push final content through stream so streaming channels (e.g. Feishu)
# update the card instead of leaving it empty.
if on_stream and on_stream_end and should_stream:
await on_stream(result.final_content or "")
stream_content = (
result.pending_stream_content
if result.pending_stream_content is not None
else result.final_content or ""
)
await on_stream(stream_content)
await on_stream_end(resuming=False)
elif result.stop_reason == "error":
logger.error("LLM returned error: {}", (result.final_content or "")[:200])
return result.final_content, result.tools_used, result.messages, result.stop_reason, result.had_injections
def _check_expired_sessions_if_due(self) -> None:
"""Scan idle sessions no more often than the configured interval."""
now = time.monotonic()
if now < self._next_idle_compact_check_at:
return
self._next_idle_compact_check_at = now + self._idle_compact_check_interval_s
self.auto_compact.check_expired(
self._schedule_background,
self.runtime_for_session,
active_session_keys=self._pending_queues.keys(),
)
async def run(self) -> None:
"""Run the agent loop, dispatching messages as tasks to stay responsive to /stop."""
self._running = True
@@ -1031,11 +1069,7 @@ class AgentLoop:
try:
msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
except asyncio.TimeoutError:
self.auto_compact.check_expired(
self._schedule_background,
self.runtime_for_session,
active_session_keys=self._pending_queues.keys(),
)
self._check_expired_sessions_if_due()
continue
except asyncio.CancelledError:
# Preserve real task cancellation so shutdown can complete cleanly.
@@ -1161,6 +1195,14 @@ class AgentLoop:
for _, coordinator in self._automation_turn_coordinators:
coordinator.complete(msg, error=asyncio.CancelledError())
logger.info("Task cancelled for session {}", session_key)
try:
await delivery.abort_stream()
except Exception:
logger.debug(
"Could not close stream for cancelled session {}",
session_key,
exc_info=True,
)
# Preserve partial context from the interrupted turn so
# the user does not lose tool results and assistant
# messages accumulated before /stop. The checkpoint was
@@ -1298,7 +1340,6 @@ class AgentLoop:
msg=msg,
session=None,
session_key=key,
state=TurnState.RESTORE,
turn_id=f"{key}:{time.time_ns()}",
runtime=runtime,
kind=kind,
@@ -1330,6 +1371,19 @@ class AgentLoop:
if ctx.on_stream is not None:
stream_callback = ctx.on_stream
stream_end_callback = ctx.on_stream_end
stream_end_accepts_merge_next = False
if stream_end_callback is not None:
try:
stream_end_signature = inspect.signature(stream_end_callback)
stream_end_accepts_merge_next = (
"merge_next" in stream_end_signature.parameters
or any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
for parameter in stream_end_signature.parameters.values()
)
)
except (TypeError, ValueError):
pass
segment_streamed_content = False
async def _tracked_stream(delta: str) -> None:
@@ -1338,75 +1392,64 @@ class AgentLoop:
segment_streamed_content = True
await stream_callback(delta)
async def _tracked_stream_end(*, resuming: bool = False) -> None:
async def _tracked_stream_end(
*,
resuming: bool = False,
merge_next: bool = False,
) -> None:
nonlocal segment_streamed_content
ctx.streamed_content = segment_streamed_content
segment_streamed_content = False
if stream_end_callback is not None:
await stream_end_callback(resuming=resuming)
if merge_next and stream_end_accepts_merge_next:
await stream_end_callback(resuming=resuming, merge_next=True)
else:
await stream_end_callback(resuming=resuming)
ctx.on_stream = _tracked_stream
ctx.on_stream_end = _tracked_stream_end
while ctx.state is not TurnState.DONE:
handler_name = f"_state_{ctx.state.name.lower()}"
handler = getattr(self, handler_name, None)
if handler is None:
raise RuntimeError(f"Missing state handler for {ctx.state}")
t0 = time.perf_counter()
try:
event = await handler(ctx)
except Exception:
duration = (time.perf_counter() - t0) * 1000
ctx.trace.append(
StateTraceEntry(
state=ctx.state,
started_at=t0,
duration_ms=duration,
event="",
error="exception",
)
)
raise
duration = (time.perf_counter() - t0) * 1000
ctx.trace.append(
StateTraceEntry(
state=ctx.state,
started_at=t0,
duration_ms=duration,
event=event,
)
)
logger.debug(
"[turn {}] State {} took {:.1f}ms -> event {}",
ctx.turn_id,
ctx.state.name,
duration,
event,
)
next_state = self._TRANSITIONS.get((ctx.state, event))
if next_state is None:
raise RuntimeError(
f"[turn {ctx.turn_id}] No transition from {ctx.state} "
f"on event {event!r}"
)
ctx.state = next_state
logger.debug(
"[turn {}] Turn completed after {} states",
ctx.turn_id,
len(ctx.trace),
)
await self._run_turn_stage(ctx, "restore", self._restore_turn)
await self._run_turn_stage(ctx, "compact", self._compact_session)
if await self._run_turn_stage(ctx, "command", self._dispatch_command):
return ctx.outbound
await self._run_turn_stage(ctx, "build", self._build_turn)
await self._run_turn_stage(ctx, "run", self._run_turn)
await self._run_turn_stage(ctx, "save", self._persist_turn)
await self._run_turn_stage(ctx, "respond", self._prepare_outbound)
return ctx.outbound
async def _run_turn_stage(
self,
ctx: TurnContext,
name: str,
handler: Callable[[TurnContext], Awaitable[_T]],
) -> _T:
started_at = time.perf_counter()
try:
result = await handler(ctx)
except Exception:
duration_ms = (time.perf_counter() - started_at) * 1000
logger.debug(
"[turn {}] Stage {} failed after {:.1f}ms",
ctx.turn_id,
name,
duration_ms,
)
raise
duration_ms = (time.perf_counter() - started_at) * 1000
logger.debug(
"[turn {}] Stage {} completed in {:.1f}ms",
ctx.turn_id,
name,
duration_ms,
)
return result
def _assemble_outbound(
self,
msg: InboundMessage,
final_content: str,
all_msgs: list[dict[str, Any]],
stop_reason: str,
had_injections: bool,
streamed_content: bool,
@@ -1437,7 +1480,7 @@ class AgentLoop:
metadata=meta,
)
async def _state_restore(self, ctx: TurnContext) -> TurnState:
async def _restore_turn(self, ctx: TurnContext) -> None:
"""Restore checkpoint / pending user turn; extract documents."""
msg = ctx.msg
@@ -1456,6 +1499,11 @@ class AgentLoop:
# ensure it exists in case this handler is invoked independently.
if ctx.session is None:
ctx.session = self.sessions.get_or_create(ctx.session_key)
self._remember_unified_session_route(
ctx.session,
msg,
is_user_turn=ctx.original_user_text is not None,
)
await ctx.delivery.started()
if ctx.kind is TurnKind.USER:
self.workspace_scopes.persist_message_scope(ctx.session, msg)
@@ -1465,8 +1513,6 @@ class AgentLoop:
if self._restore_pending_user_turn(ctx.session):
self.sessions.save(ctx.session)
return "ok"
def _prepare_message_media(self, content: str, media: list[str]) -> tuple[str, list[str]]:
if self._should_extract_document_text():
return extract_documents(content, media)
@@ -1477,14 +1523,13 @@ class AgentLoop:
return True
return self.channels_config.extract_document_text
async def _state_compact(self, ctx: TurnContext) -> str:
async def _compact_session(self, ctx: TurnContext) -> None:
ctx.session, pending = self.auto_compact.prepare_session(ctx.session, ctx.session_key)
ctx.pending_summary = pending
return "ok"
async def _state_command(self, ctx: TurnContext) -> str:
async def _dispatch_command(self, ctx: TurnContext) -> bool:
if ctx.kind is TurnKind.SYSTEM:
return "dispatch"
return False
raw = ctx.msg.content.strip()
_, automation_metadata = automation_history_overrides(ctx.msg.metadata)
is_user_turn = (
@@ -1520,10 +1565,10 @@ class AgentLoop:
)
self.sessions.save(ctx.session)
self._clear_pending_user_turn(ctx.session)
return "shortcut"
return "dispatch"
return True
return False
async def _state_build(self, ctx: TurnContext) -> str:
async def _build_turn(self, ctx: TurnContext) -> None:
runtime = ctx.runtime
if runtime is None:
runtime = self.runtime_for_session(ctx.session)
@@ -1579,9 +1624,7 @@ class AgentLoop:
if ctx.on_retry_wait is None:
ctx.on_retry_wait = ctx.delivery.retry_wait_callback()
return "ok"
async def _state_run(self, ctx: TurnContext) -> str:
async def _run_turn(self, ctx: TurnContext) -> None:
if ctx.visible_run_started_at is None:
ctx.visible_run_started_at = time.time()
await ctx.delivery.running(started_at=ctx.visible_run_started_at)
@@ -1608,17 +1651,15 @@ class AgentLoop:
tools=ctx.tools,
request_context=ctx.request_context,
)
final_content, tools_used, all_msgs, stop_reason, had_injections = result
final_content, _, all_msgs, stop_reason, had_injections = result
ctx.final_content = final_content
ctx.tools_used = tools_used
ctx.all_messages = all_msgs
ctx.stop_reason = stop_reason
ctx.had_injections = had_injections
if ctx.kind is TurnKind.USER:
await turn_continuation.maybe_continue_turn(ctx)
return "ok"
async def _state_save(self, ctx: TurnContext) -> str:
async def _persist_turn(self, ctx: TurnContext) -> None:
turn_continuation.prepare_save_boundary(ctx)
if (
@@ -1659,12 +1700,11 @@ class AgentLoop:
self._clear_pending_user_turn(ctx.session)
self._clear_runtime_checkpoint(ctx.session)
self.sessions.save(ctx.session)
return "ok"
async def _state_respond(self, ctx: TurnContext) -> str:
async def _prepare_outbound(self, ctx: TurnContext) -> None:
if ctx.suppress_response:
ctx.outbound = None
return "ok"
return
if ctx.kind is TurnKind.SYSTEM:
ctx.outbound = ctx.delivery.background_response(
ctx.final_content,
@@ -1672,11 +1712,10 @@ class AgentLoop:
streamed=ctx.streamed_content,
latency_ms=ctx.turn_latency_ms,
)
return "ok"
return
ctx.outbound = self._assemble_outbound(
ctx.msg,
ctx.final_content,
ctx.all_messages,
ctx.stop_reason,
ctx.had_injections,
ctx.streamed_content,
@@ -1684,7 +1723,6 @@ class AgentLoop:
)
if ctx.ephemeral and ctx.outbound is not None:
ctx.outbound.metadata["_stop_reason"] = ctx.stop_reason
return "ok"
def _sanitize_persisted_blocks(
self,
+41 -14
View File
@@ -43,13 +43,33 @@ if TYPE_CHECKING:
# MemoryStore — pure file I/O layer
# ---------------------------------------------------------------------------
class DreamRunProgress:
"""Track tool failures that make a nominally completed Dream run unsafe to advance."""
def __init__(self) -> None:
self.had_tool_errors = False
async def __call__(
self,
*_args: Any,
tool_events: list[dict[str, Any]] | None = None,
**_kwargs: Any,
) -> None:
if any(
isinstance(event, dict) and event.get("phase") == "error"
for event in tool_events or ()
):
self.had_tool_errors = True
class MemoryStore:
"""Pure file I/O for memory files: MEMORY.md, history.jsonl, SOUL.md, USER.md."""
_DEFAULT_MAX_HISTORY = 1000
# Durable files whose real working-tree delta grounds Dream commit messages
# and the cursor-advance gate. Deliberately excludes memory/.dream_cursor so
# that advancing the cursor itself is never mistaken for a productive edit.
# Durable files whose real working-tree delta grounds Dream commit messages.
# Deliberately excludes memory/.dream_cursor so progress bookkeeping never
# appears as a durable-memory edit in the audit record.
_DREAM_CONTENT_PATHS = ("SOUL.md", "USER.md", "memory/MEMORY.md")
# Per-file cap when embedding current contents into the Dream prompt. The
# durable files are tiny in practice (~5 KB total), but a runaway file must
@@ -433,9 +453,11 @@ class MemoryStore:
line = line.strip()
if line:
try:
entries.append(json.loads(line))
parsed = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(parsed, dict):
entries.append(parsed)
return entries
@@ -453,7 +475,8 @@ class MemoryStore:
lines = [line for line in data.split("\n") if line.strip()]
if not lines:
return None
return json.loads(lines[-1])
parsed = json.loads(lines[-1])
return parsed if isinstance(parsed, dict) else None
except (FileNotFoundError, json.JSONDecodeError, UnicodeDecodeError):
return None
@@ -583,8 +606,7 @@ class MemoryStore:
"""Structured summary of uncommitted changes to the durable memory files.
Returns "" when git is unavailable or no content file changed. This is
the ground-truth input for diff-grounded Dream commit messages and for
gating cursor advance on real edits (never on LLM self-report).
the ground-truth input for diff-grounded Dream commit messages.
"""
if not self._git.is_initialized():
return ""
@@ -633,10 +655,18 @@ class MemoryStore:
return tools
@staticmethod
def dream_run_completed(resp: object | None) -> bool:
"""Return True only when an ephemeral Dream agent turn completed cleanly."""
def dream_run_completed(
resp: object | None,
*,
had_tool_errors: bool = False,
) -> bool:
"""Return True only when a Dream turn completed without tool failures."""
metadata = getattr(resp, "metadata", None)
return isinstance(metadata, dict) and metadata.get("_stop_reason") == "completed"
return (
not had_tool_errors
and isinstance(metadata, dict)
and metadata.get("_stop_reason") == "completed"
)
# -- message formatting utility ------------------------------------------
@@ -882,7 +912,7 @@ class Consolidator:
) -> tuple[int, str]:
"""Estimate prompt size from the full unconsolidated session tail."""
history = self._full_unconsolidated_history(session)
channel, chat_id = (session.key.split(":", 1) if ":" in session.key else (None, None))
channel = session.key.split(":", 1)[0] if ":" in session.key else None
# Include archived summary in estimation so the budget accounts for it.
meta = session.metadata.get("_last_summary")
summary = meta.get("text") if isinstance(meta, dict) else (meta if isinstance(meta, str) else None)
@@ -890,10 +920,7 @@ class Consolidator:
history=history,
current_message="[token-probe]",
channel=channel,
chat_id=chat_id,
sender_id=None,
session_summary=summary,
session_metadata=session.metadata,
session_key=session.key,
unified_session=self.unified_session,
)
+7 -1
View File
@@ -85,7 +85,13 @@ class AgentProgressHook(AgentHook):
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None:
await self.emit_reasoning_end()
if self._on_stream_end:
await self._on_stream_end(resuming=resuming)
kwargs: dict[str, bool] = {"resuming": resuming}
if (
context.stream_continues_current_message
and self._on_progress_accepts(self._on_stream_end, "merge_next")
):
kwargs["merge_next"] = True
await self._on_stream_end(**kwargs)
self._stream_buf = ""
self._think_extractor.reset()
+120 -16
View File
@@ -19,6 +19,11 @@ from nanobot.agent.context_governance import (
from nanobot.agent.hook import AgentHook, AgentHookContext, AgentRunHookContext
from nanobot.agent.tools.registry import ToolRegistry, is_tool_error_result
from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
from nanobot.runtime_context import (
RUNTIME_CONTEXT_MESSAGE_META,
detach_runtime_context,
reattach_runtime_context,
)
from nanobot.session.history_visibility import is_hidden_history_message
from nanobot.utils.helpers import (
IncrementalThinkExtractor,
@@ -55,6 +60,18 @@ _MAX_LENGTH_RECOVERIES = 3
_MAX_INJECTIONS_PER_TURN = 3
_MAX_INJECTION_CYCLES = 5
def _restore_outer_whitespace(content: str, original: str | None) -> str:
"""Restore boundary whitespace stripped while cleaning one recovered segment."""
if not original:
return content
leading_size = len(original) - len(original.lstrip())
trailing_size = len(original) - len(original.rstrip())
leading = original[:leading_size]
trailing = original[-trailing_size:] if trailing_size else ""
return f"{leading}{content}{trailing}"
@dataclass(slots=True)
class AgentRunSpec:
"""Configuration for a single agent execution."""
@@ -96,6 +113,8 @@ class AgentRunResult:
error: str | None = None
tool_events: list[dict[str, str]] = field(default_factory=list)
had_injections: bool = False
# Terminal tail to emit when the preceding final-content prefix was already streamed.
pending_stream_content: str | None = None
class AgentRunner:
@@ -137,10 +156,51 @@ class AgentRunner:
and not is_hidden_history_message(messages[-1])
):
merged = dict(messages[-1])
merged["content"] = cls._merge_message_content(
merged.get("content"),
injection.get("content"),
left_meta = merged.get("_meta")
right_meta = injection.get("_meta")
left_marker = (
left_meta.get(RUNTIME_CONTEXT_MESSAGE_META)
if isinstance(left_meta, dict)
else None
)
right_marker = (
right_meta.get(RUNTIME_CONTEXT_MESSAGE_META)
if isinstance(right_meta, dict)
else None
)
detached_left = (
detach_runtime_context(merged.get("content"), left_marker)
if isinstance(left_marker, dict)
else (merged.get("content"), [], [])
)
detached_right = (
detach_runtime_context(injection.get("content"), right_marker)
if isinstance(right_marker, dict)
else (injection.get("content"), [], [])
)
if detached_left is not None and detached_right is not None:
left_content, left_sources, left_blocks = detached_left
right_content, right_sources, right_blocks = detached_right
merged_content = cls._merge_message_content(left_content, right_content)
context_blocks = [*left_blocks, *right_blocks]
if context_blocks:
merged_content, marker = reattach_runtime_context(
merged_content,
[*left_sources, *right_sources],
context_blocks,
)
internal_meta = dict(left_meta) if isinstance(left_meta, dict) else {}
if isinstance(right_meta, dict):
for key, value in right_meta.items():
internal_meta.setdefault(key, value)
internal_meta[RUNTIME_CONTEXT_MESSAGE_META] = marker
merged["_meta"] = internal_meta
merged["content"] = merged_content
else:
merged["content"] = cls._merge_message_content(
merged.get("content"),
injection.get("content"),
)
messages[-1] = merged
continue
messages.append(injection)
@@ -334,10 +394,13 @@ class AgentRunner:
# Per-turn throttle for repeated attempts against the same outside target.
workspace_violation_counts: dict[str, int] = {}
empty_content_retries = 0
length_recovery_count = 0
# Segments from one uninterrupted length-recovery chain. Tool work or
# injected user input starts a new logical answer and clears the chain.
length_recovery_parts: list[str] = []
had_injections = False
injection_cycles = 0
compacted_tool_call_ids: set[str] = set()
pending_stream_content: str | None = None
governance_config = ContextGovernanceConfig(
provider=spec.runtime.provider,
model=spec.runtime.model,
@@ -372,6 +435,7 @@ class AgentRunner:
context.response = response
context.tool_calls = list(response.tool_calls)
original_content = response.content
reasoning_text, cleaned_content = extract_reasoning(
response.reasoning_content,
response.thinking_blocks,
@@ -458,6 +522,7 @@ class AgentRunner:
)
if should_continue:
had_injections = True
length_recovery_parts.clear()
continue
break
await self._emit_checkpoint(
@@ -472,7 +537,7 @@ class AgentRunner:
},
)
empty_content_retries = 0
length_recovery_count = 0
length_recovery_parts.clear()
# Checkpoint 1: drain injections after tools, before next LLM call
_drained, injection_cycles = await self._try_drain_injections(
spec, messages, None, injection_cycles,
@@ -521,29 +586,50 @@ class AgentRunner:
context.response = response
context.usage = dict(raw_usage)
context.tool_calls = list(response.tool_calls)
original_content = response.content
clean = hook.finalize_content(context, response.content)
if response.finish_reason == "length" and not is_blank_text(clean):
length_recovery_count += 1
if length_recovery_count <= _MAX_LENGTH_RECOVERIES:
if len(length_recovery_parts) < _MAX_LENGTH_RECOVERIES:
length_recovery_parts.append(
_restore_outer_whitespace(clean, original_content)
)
logger.info(
"Output truncated on turn {} for {} ({}/{}); continuing",
iteration,
spec.session_key or "default",
length_recovery_count,
len(length_recovery_parts),
_MAX_LENGTH_RECOVERIES,
)
if hook.wants_streaming():
context.stream_continues_current_message = True
await hook.on_stream_end(context, resuming=True)
messages.append(build_assistant_message(
clean,
reasoning_content=response.reasoning_content,
thinking_blocks=response.thinking_blocks,
))
messages.append(build_length_recovery_message())
messages.append(build_length_recovery_message(clean))
await hook.after_iteration(context)
continue
# Some streaming providers recover with a complete response but no
# content deltas. When an earlier length segment is already visible,
# emit this terminal segment into the same stream; otherwise the
# regular full response would duplicate the visible prefix.
if (
length_recovery_parts
and hook.wants_streaming()
and not context.streamed_content
and response.finish_reason != "error"
and not is_blank_text(clean)
):
await hook.on_stream(
context,
_restore_outer_whitespace(clean, original_content),
)
context.streamed_content = True
assistant_message: dict[str, Any] | None = None
if response.finish_reason != "error" and not is_blank_text(clean):
assistant_message = build_assistant_message(
@@ -568,6 +654,7 @@ class AgentRunner:
await hook.on_stream_end(context, resuming=should_continue)
if should_continue:
length_recovery_parts.clear()
await hook.after_iteration(context)
continue
@@ -589,6 +676,7 @@ class AgentRunner:
)
if should_continue:
had_injections = True
length_recovery_parts.clear()
continue
break
if is_blank_text(clean):
@@ -606,6 +694,7 @@ class AgentRunner:
)
if should_continue:
had_injections = True
length_recovery_parts.clear()
continue
break
@@ -625,7 +714,13 @@ class AgentRunner:
"pending_tool_calls": [],
},
)
final_content = clean
if length_recovery_parts:
final_content = (
"".join(length_recovery_parts)
+ _restore_outer_whitespace(clean, original_content)
).strip()
else:
final_content = clean
context.final_content = final_content
context.stop_reason = stop_reason
await hook.after_iteration(context)
@@ -643,17 +738,25 @@ class AgentRunner:
)
if drained_after_max_iterations:
had_injections = True
final_content = None
terminal_content = None
if spec.finalize_on_max_iterations:
final_content = await self._try_finalize_after_max_iterations(
terminal_content = await self._try_finalize_after_max_iterations(
spec,
hook,
messages,
usage,
)
if final_content is None:
final_content = self._max_iterations_fallback(spec)
self._append_final_message(messages, final_content)
if terminal_content is None:
terminal_content = self._max_iterations_fallback(spec)
if length_recovery_parts:
terminal_tail = f"\n\n{terminal_content.lstrip()}"
final_content = (
"".join(length_recovery_parts).rstrip() + terminal_tail
).strip()
pending_stream_content = terminal_tail
else:
final_content = terminal_content
self._append_final_message(messages, terminal_content)
return AgentRunResult(
final_content=final_content,
@@ -664,6 +767,7 @@ class AgentRunner:
error=error,
tool_events=tool_events,
had_injections=had_injections,
pending_stream_content=pending_stream_content,
)
def _build_request_kwargs(
@@ -1223,7 +1327,7 @@ class AgentRunner:
return payload, event, exc
return payload, event, None
if is_tool_error_result(tool_call.name, result):
if is_tool_error_result(result):
await hook.on_execute_tool_error(context, tool_call, tool, params, result)
event = {
"name": tool_call.name,
+15 -9
View File
@@ -154,11 +154,21 @@ class SkillsLoader:
sections.append("\n".join(lines))
return "\n\n".join(sections)
@staticmethod
def _requirement_lists(skill_meta: dict) -> tuple[list[str], list[str]]:
"""Return (bins, env) lists from skill metadata, tolerating null/wrong shapes."""
requires = skill_meta.get("requires") or {}
if not isinstance(requires, dict):
return [], []
bins_raw = requires.get("bins") or []
env_raw = requires.get("env") or []
bins = [str(v) for v in bins_raw if isinstance(v, str) and v.strip()] if isinstance(bins_raw, list) else []
env = [str(v) for v in env_raw if isinstance(v, str) and v.strip()] if isinstance(env_raw, list) else []
return bins, env
def _get_missing_requirements(self, skill_meta: dict) -> str:
"""Get a description of missing requirements."""
requires = skill_meta.get("requires", {})
required_bins = requires.get("bins", [])
required_env_vars = requires.get("env", [])
required_bins, required_env_vars = self._requirement_lists(skill_meta)
return ", ".join(
[f"CLI: {command_name}" for command_name in required_bins if not shutil.which(command_name)]
+ [f"ENV: {env_name}" for env_name in required_env_vars if not os.environ.get(env_name)]
@@ -172,9 +182,7 @@ class SkillsLoader:
def get_skill_requirements(self, name: str) -> dict[str, list[str]]:
"""Return explicit command/env requirements and currently missing entries."""
requires = self._get_skill_meta(name).get("requires", {})
bins = [str(value) for value in requires.get("bins", [])]
env = [str(value) for value in requires.get("env", [])]
bins, env = self._requirement_lists(self._get_skill_meta(name))
return {
"bins": bins,
"env": env,
@@ -219,9 +227,7 @@ class SkillsLoader:
def _check_requirements(self, skill_meta: dict) -> bool:
"""Check if skill requirements are met (bins, env vars)."""
requires = skill_meta.get("requires", {})
required_bins = requires.get("bins", [])
required_env_vars = requires.get("env", [])
required_bins, required_env_vars = self._requirement_lists(skill_meta)
return all(shutil.which(cmd) for cmd in required_bins) and all(
os.environ.get(var) for var in required_env_vars
)
-6
View File
@@ -39,12 +39,6 @@ def _validate_patch_path(path: str) -> str:
return normalized
def _lines_to_text(lines: list[str]) -> str:
if not lines:
return ""
return "\n".join(lines) + "\n"
def _text_line_count(text: str) -> int:
if not text:
return 0
+1 -3
View File
@@ -28,7 +28,7 @@ _CRON_PARAMETERS = tool_parameters_schema(
"(e.g., 'Send a reminder to WeChat: xxx' or 'Check system status and report'). "
"Not used for action='list' or action='remove'."
),
every_seconds=IntegerSchema(0, description="Interval in seconds (for recurring tasks)"),
every_seconds=IntegerSchema(description="Interval in seconds (for recurring tasks)"),
cron_expr=StringSchema("Cron expression like '0 9 * * *' (for scheduled tasks)"),
tz=StringSchema(
"Optional IANA timezone for cron expressions (e.g. 'America/Vancouver'). "
@@ -138,8 +138,6 @@ class CronTool(Tool):
tz: str | None = None,
at: str | None = None,
job_id: str | None = None,
deliver: bool = True,
**kwargs: Any,
) -> str:
if action == "add":
if self._in_cron_context.get():
-4
View File
@@ -447,7 +447,6 @@ def format_session_poll(session_id: str, poll: _SessionPoll) -> str:
default=False,
),
yield_time_ms=IntegerSchema(
DEFAULT_YIELD_MS,
description="Milliseconds to wait before returning recent output (default 1000, max 30000).",
minimum=0,
maximum=MAX_YIELD_MS,
@@ -458,20 +457,17 @@ def format_session_poll(session_id: str, poll: _SessionPoll) -> str:
nullable=True,
),
wait_timeout_ms=IntegerSchema(
DEFAULT_WAIT_FOR_MS,
description="Maximum milliseconds to wait for wait_for text (default 10000, max 120000).",
minimum=0,
maximum=MAX_WAIT_FOR_MS,
nullable=True,
),
max_output_chars=IntegerSchema(
DEFAULT_MAX_OUTPUT_CHARS,
description="Maximum output characters to return from this poll (default 10000, max 50000).",
minimum=1000,
maximum=MAX_OUTPUT_CHARS,
),
max_output_tokens=IntegerSchema(
DEFAULT_MAX_OUTPUT_CHARS,
description="Compatibility alias for max_output_chars. The current runtime uses a character budget.",
minimum=1000,
maximum=MAX_OUTPUT_CHARS,
-6
View File
@@ -226,12 +226,10 @@ def _builtin_skill_read_path(path: str) -> Path | None:
tool_parameters_schema(
path=StringSchema("The file path to read"),
offset=IntegerSchema(
1,
description="Line number to start reading from (1-indexed, default 1)",
minimum=1,
),
limit=IntegerSchema(
2000,
description="Maximum number of lines to read (default 2000)",
minimum=1,
),
@@ -790,13 +788,11 @@ def _find_match(content: str, old_text: str) -> tuple[str | None, int]:
new_text=StringSchema("The text to replace with"),
replace_all=BooleanSchema(description="Replace all occurrences (default false)"),
occurrence=IntegerSchema(
1,
description="Optional 1-based occurrence to replace when old_text appears multiple times.",
minimum=1,
nullable=True,
),
line_hint=IntegerSchema(
1,
description=(
"Optional exact 1-based target line copied from read_file. "
"The selected old_text match must cover this line."
@@ -805,7 +801,6 @@ def _find_match(content: str, old_text: str) -> tuple[str | None, int]:
nullable=True,
),
expected_replacements=IntegerSchema(
1,
description="Optional guard for the number of replacements that must be made.",
minimum=1,
nullable=True,
@@ -1036,7 +1031,6 @@ class EditFileTool(_FsTool):
path=StringSchema("The directory path to list"),
recursive=BooleanSchema(description="Recursively list all files (default false)"),
max_entries=IntegerSchema(
200,
description="Maximum entries to return (default 200)",
minimum=1,
),
+126 -21
View File
@@ -315,13 +315,87 @@ def _extract_nullable_branch(options: Any) -> tuple[dict[str, Any], bool] | None
return None
def _normalize_schema_for_openai(schema: Any) -> dict[str, Any]:
"""Normalize only nullable JSON Schema patterns for tool definitions."""
if not isinstance(schema, dict):
return {"type": "object", "properties": {}}
def _resolve_local_schema_ref(root: dict[str, Any], ref: str) -> Any:
"""Resolve a local JSON Pointer without accepting remote references."""
if not ref.startswith("#"):
raise ValueError("not a local JSON Pointer")
pointer = urllib.parse.unquote(ref[1:], errors="strict")
if not pointer:
return root
if not pointer.startswith("/"):
raise ValueError("not a local JSON Pointer")
current: Any = root
for raw_part in pointer[1:].split("/"):
part = raw_part.replace("~1", "/").replace("~0", "~")
if isinstance(current, dict):
current = current[part]
elif isinstance(current, list):
current = current[int(part)]
else:
raise KeyError(part)
return current
def _rewrite_local_schema_refs(schema: dict[str, Any]) -> dict[str, Any]:
"""Hoist arbitrary local JSON-Pointer refs into provider-compatible ``$defs``."""
rewritten_refs: dict[str, str] = {}
generated_defs: dict[str, Any] = {}
def rewrite(value: Any) -> Any:
if isinstance(value, list):
return [rewrite(item) for item in value]
if not isinstance(value, dict):
return value
rewritten = dict(value)
ref = rewritten.get("$ref")
is_rewritable_ref = False
if isinstance(ref, str) and not ref.startswith("#/$defs/"):
try:
pointer = urllib.parse.unquote(ref[1:], errors="strict")
except (UnicodeDecodeError, ValueError):
pass
else:
is_rewritable_ref = ref.startswith("#") and (
not pointer or pointer.startswith("/")
)
if is_rewritable_ref:
name = rewritten_refs.get(ref)
if name is None:
try:
target = _resolve_local_schema_ref(schema, ref)
except (KeyError, IndexError, TypeError, UnicodeDecodeError, ValueError):
logger.warning("MCP tool schema contains an unresolved local $ref: {}", ref)
else:
assert isinstance(ref, str)
name = f"ref_{hashlib.sha256(ref.encode()).hexdigest()[:12]}"
existing_defs = schema.get("$defs")
while isinstance(existing_defs, dict) and name in existing_defs:
name += "_"
rewritten_refs[ref] = name
# Reserve the name before descending so recursive refs terminate.
generated_defs[name] = {}
generated_defs[name] = rewrite(target)
if name is not None:
rewritten["$ref"] = f"#/$defs/{name}"
return {key: rewrite(item) for key, item in rewritten.items()}
result = rewrite(schema)
if generated_defs:
existing_defs = result.get("$defs")
result["$defs"] = {
**(existing_defs if isinstance(existing_defs, dict) else {}),
**generated_defs,
}
return result
def _normalize_nullable_schema(schema: dict[str, Any]) -> dict[str, Any]:
"""Normalize nullable forms in structural subschemas only."""
normalized = dict(schema)
raw_type = normalized.get("type")
if isinstance(raw_type, list):
non_null = [item for item in raw_type if item != "null"]
@@ -339,23 +413,34 @@ def _normalize_schema_for_openai(schema: Any) -> dict[str, Any]:
normalized["nullable"] = True
break
if "properties" in normalized and isinstance(normalized["properties"], dict):
if isinstance(normalized.get("properties"), dict):
normalized["properties"] = {
name: _normalize_schema_for_openai(prop) if isinstance(prop, dict) else prop
name: _normalize_nullable_schema(prop) if isinstance(prop, dict) else prop
for name, prop in normalized["properties"].items()
}
if isinstance(normalized.get("items"), dict):
normalized["items"] = _normalize_nullable_schema(normalized["items"])
if isinstance(normalized.get("$defs"), dict):
normalized["$defs"] = {
name: _normalize_nullable_schema(definition)
if isinstance(definition, dict)
else definition
for name, definition in normalized["$defs"].items()
}
if "items" in normalized and isinstance(normalized["items"], dict):
normalized["items"] = _normalize_schema_for_openai(normalized["items"])
if normalized.get("type") != "object":
return normalized
normalized.setdefault("properties", {})
normalized.setdefault("required", [])
if normalized.get("type") == "object":
normalized.setdefault("properties", {})
normalized.setdefault("required", [])
return normalized
def _normalize_schema_for_openai(schema: Any) -> dict[str, Any]:
"""Normalize MCP JSON Schema patterns for tool definitions."""
if not isinstance(schema, dict):
return {"type": "object", "properties": {}}
return _normalize_nullable_schema(_rewrite_local_schema_refs(schema))
class _MCPWrapperBase(Tool):
"""Common reconnect handling for wrappers bound to one MCP server session."""
@@ -830,6 +915,23 @@ class MCPPromptWrapper(_MCPWrapperBase):
return "\n".join(parts) or "(no output)"
def _register_mcp_capability(
registry: ToolRegistry,
capability: Tool,
server_name: str,
) -> bool:
owner = f"nanobot.mcp.{server_name}"
if registry.register_if_absent(capability, owner=owner):
return True
logger.warning(
"MCP: skipping capability '{}' from server '{}' because it is already registered by '{}'",
capability.name,
server_name,
registry.owner(capability.name),
)
return False
async def connect_mcp_servers(
mcp_servers: dict, registry: ToolRegistry
) -> dict[str, MCPConnection]:
@@ -963,7 +1065,8 @@ async def connect_mcp_servers(
)
continue
wrapper = MCPToolWrapper(session, name, tool_def, tool_timeout=cfg.tool_timeout)
registry.register(wrapper)
if not _register_mcp_capability(registry, wrapper, name):
continue
logger.debug("MCP: registered tool '{}' from server '{}'", wrapper.name, name)
registered_count += 1
if enabled_tools:
@@ -1000,7 +1103,8 @@ async def connect_mcp_servers(
wrapper = MCPResourceWrapper(
session, name, resource, resource_timeout=cfg.tool_timeout
)
registry.register(wrapper)
if not _register_mcp_capability(registry, wrapper, name):
continue
registered_count += 1
logger.debug(
"MCP: registered resource '{}' from server '{}'",
@@ -1018,7 +1122,8 @@ async def connect_mcp_servers(
wrapper = MCPPromptWrapper(
session, name, prompt, prompt_timeout=cfg.tool_timeout
)
registry.register(wrapper)
if not _register_mcp_capability(registry, wrapper, name):
continue
registered_count += 1
logger.debug(
"MCP: registered prompt '{}' from server '{}'",
@@ -1188,7 +1293,7 @@ async def reload_servers(state: Any, registry: ToolRegistry) -> dict[str, Any]:
tools_removed = 0
for name in [*removed, *changed]:
tools_removed += _unregister_server_tools(state, registry, name)
tools_removed += _unregister_server_tools(registry, name)
await _close_server(state, name)
state._mcp_servers = next_servers
@@ -1362,7 +1467,7 @@ async def _refresh_terminated_server(
return current_tool
logger.warning("MCP server '{}' session terminated; refreshing connection", server_name)
_unregister_server_tools(state, registry, server_name)
_unregister_server_tools(registry, server_name)
await _close_server(state, server_name)
connected = await connect_mcp_servers({server_name: cfg}, registry)
@@ -1394,7 +1499,7 @@ def _tool_belongs_to_server(tool: Tool | None, tool_name: str, server_name: str)
return tool_name.startswith(_tool_prefix(server_name))
def _unregister_server_tools(state: Any, registry: ToolRegistry, server_name: str) -> int:
def _unregister_server_tools(registry: ToolRegistry, server_name: str) -> int:
removed = 0
for tool_name in list(registry.tool_names):
tool = registry.get(tool_name)
+25 -3
View File
@@ -12,7 +12,7 @@ if TYPE_CHECKING:
from nanobot.runtime_context import RuntimeContextProvider
def is_tool_error_result(name: str, result: Any) -> bool:
def is_tool_error_result(result: Any) -> bool:
return isinstance(result, ToolResult) and result.is_error
@@ -25,22 +25,44 @@ class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
self._owners: dict[str, str] = {}
self._cached_definitions: list[dict[str, Any]] | None = None
def register(self, tool: Tool) -> None:
def register(self, tool: Tool, *, owner: str = "nanobot.core") -> None:
"""Register a tool."""
self._tools[tool.name] = tool
self._owners[tool.name] = owner
self._cached_definitions = None
def register_if_absent(self, tool: Tool, *, owner: str = "nanobot.core") -> bool:
"""Register a tool without replacing an existing capability."""
if tool.name in self._tools:
return False
self.register(tool, owner=owner)
return True
def unregister(self, name: str) -> None:
"""Unregister a tool by name."""
self._tools.pop(name, None)
self._owners.pop(name, None)
self._cached_definitions = None
def unregister_owner(self, owner: str) -> None:
"""Remove all tools registered by one extension."""
for name in [
name for name, registered_owner in self._owners.items()
if registered_owner == owner
]:
self.unregister(name)
def get(self, name: str) -> Tool | None:
"""Get a tool by name."""
return self._tools.get(name)
def owner(self, name: str) -> str | None:
"""Return the extension ID that registered a tool."""
return self._owners.get(name)
def get_runtime_context_providers(self) -> list[RuntimeContextProvider]:
"""Return tool-owned providers in stable tool-name order."""
providers: list[RuntimeContextProvider] = []
@@ -193,7 +215,7 @@ class ToolRegistry:
try:
assert tool is not None # guarded by prepare_call()
result = await tool.execute(**params)
if is_tool_error_result(name, result):
if is_tool_error_result(result):
return ToolResult.error(str(result) + hint)
return result
except Exception as e:
+63 -5
View File
@@ -5,13 +5,54 @@ To add a new backend, implement a function with the signature:
and register it in _BACKENDS below.
"""
import os
import shlex
from pathlib import Path
from typing import Iterable
from nanobot.config.paths import get_media_dir
def _bwrap(command: str, workspace: str, cwd: str) -> str:
def _normalize_bind_paths(
paths: Iterable[str] | None,
*,
workspace: Path | None = None,
) -> list[str]:
out: list[str] = []
seen: set[str] = set()
for raw in paths or []:
value = str(raw).strip()
if not value:
continue
path = Path(os.path.expandvars(value)).expanduser()
if not path.is_absolute():
continue
resolved_path = path.resolve(strict=False)
if workspace is not None:
try:
workspace.relative_to(resolved_path)
except ValueError:
pass
else:
# A later bind of the workspace or one of its parents could
# cover the tmpfs that hides the config directory.
continue
resolved = str(resolved_path)
if resolved in seen:
continue
seen.add(resolved)
out.append(resolved)
return out
def _bwrap(
command: str,
workspace: str,
cwd: str,
*,
sandbox_ro_binds: Iterable[str] | None = None,
sandbox_rw_binds: Iterable[str] | None = None,
) -> str:
"""Wrap command in a bubblewrap sandbox (requires bwrap in container).
Only the workspace is bind-mounted read-write; its parent dir (which holds
@@ -51,17 +92,34 @@ def _bwrap(command: str, workspace: str, cwd: str) -> str:
"--dir", str(ws), # recreate workspace mount point
"--bind", str(ws), str(ws),
"--ro-bind-try", str(media), str(media), # read-only access to media
"--chdir", sandbox_cwd,
"--", "sh", "-c", command,
]
for p in _normalize_bind_paths(sandbox_ro_binds, workspace=ws):
args += ["--ro-bind-try", p, p]
for p in _normalize_bind_paths(sandbox_rw_binds, workspace=ws):
args += ["--bind-try", p, p]
args += ["--chdir", sandbox_cwd, "--", "sh", "-c", command]
return shlex.join(args)
_BACKENDS = {"bwrap": _bwrap}
def wrap_command(sandbox: str, command: str, workspace: str, cwd: str) -> str:
def wrap_command(
sandbox: str,
command: str,
workspace: str,
cwd: str,
*,
sandbox_ro_binds: Iterable[str] | None = None,
sandbox_rw_binds: Iterable[str] | None = None,
) -> str:
"""Wrap *command* using the named sandbox backend."""
if backend := _BACKENDS.get(sandbox):
return backend(command, workspace, cwd)
return backend(
command,
workspace,
cwd,
sandbox_ro_binds=sandbox_ro_binds,
sandbox_rw_binds=sandbox_rw_binds,
)
raise ValueError(f"Unknown sandbox backend {sandbox!r}. Available: {list(_BACKENDS)}")
+1 -5
View File
@@ -52,11 +52,10 @@ class StringSchema(Schema):
class IntegerSchema(Schema):
"""Integer parameter: optional placeholder int (legacy ctor signature), description, and bounds."""
"""Integer parameter with a description and optional bounds."""
def __init__(
self,
value: int = 0,
*,
description: str = "",
minimum: int | None = None,
@@ -64,7 +63,6 @@ class IntegerSchema(Schema):
enum: tuple[int, ...] | list[int] | None = None,
nullable: bool = False,
) -> None:
self._value = value
self._description = description
self._minimum = minimum
self._maximum = maximum
@@ -92,7 +90,6 @@ class NumberSchema(Schema):
def __init__(
self,
value: float = 0.0,
*,
description: str = "",
minimum: float | None = None,
@@ -100,7 +97,6 @@ class NumberSchema(Schema):
enum: tuple[float, ...] | list[float] | None = None,
nullable: bool = False,
) -> None:
self._value = value
self._description = description
self._minimum = minimum
self._maximum = maximum
+56 -2
View File
@@ -84,6 +84,8 @@ class ExecToolConfig(Base):
path_prepend: str = ""
path_append: str = ""
sandbox: str = ""
sandbox_ro_binds: list[str] = Field(default_factory=list)
sandbox_rw_binds: list[str] = Field(default_factory=list)
allowed_env_keys: list[str] = Field(default_factory=list)
allow_patterns: list[str] = Field(default_factory=list)
deny_patterns: list[str] = Field(default_factory=list)
@@ -106,7 +108,6 @@ class _PreparedCommand:
working_dir=StringSchema("Optional working directory for the command"),
workdir=StringSchema("Compatibility alias for working_dir"),
timeout=IntegerSchema(
60,
description=(
"Timeout in seconds. Increase for long-running commands "
"like compilation or installation (default 60, max 600)."
@@ -187,6 +188,8 @@ class ExecTool(Tool):
sandbox=cfg.sandbox,
path_prepend=cfg.path_prepend,
path_append=cfg.path_append,
sandbox_ro_binds=cfg.sandbox_ro_binds,
sandbox_rw_binds=cfg.sandbox_rw_binds,
allowed_env_keys=cfg.allowed_env_keys,
allow_patterns=cfg.allow_patterns,
deny_patterns=cfg.deny_patterns,
@@ -205,6 +208,8 @@ class ExecTool(Tool):
sandbox: str = "",
path_prepend: str = "",
path_append: str = "",
sandbox_ro_binds: list[str] | None = None,
sandbox_rw_binds: list[str] | None = None,
allowed_env_keys: list[str] | None = None,
session_manager: Any | None = None,
):
@@ -237,6 +242,8 @@ class ExecTool(Tool):
self.webui_allow_local_service_access = webui_allow_local_service_access
self.path_prepend = path_prepend
self.path_append = path_append
self.sandbox_ro_binds = self._normalize_bind_roots(sandbox_ro_binds)
self.sandbox_rw_binds = self._normalize_bind_roots(sandbox_rw_binds)
self.allowed_env_keys = allowed_env_keys or []
self._session_manager = session_manager or DEFAULT_EXEC_SESSION_MANAGER
@@ -464,7 +471,14 @@ class ExecTool(Tool):
)
else:
workspace = workspace_root or cwd
command = wrap_command(self.sandbox, command, workspace, cwd)
command = wrap_command(
self.sandbox,
command,
workspace,
cwd,
sandbox_ro_binds=[str(p) for p in self.sandbox_ro_binds],
sandbox_rw_binds=[str(p) for p in self.sandbox_rw_binds],
)
cwd = str(Path(workspace).resolve())
effective_timeout = self._resolve_timeout(timeout)
@@ -794,6 +808,9 @@ class ExecTool(Tool):
if workspace_root
else None
)
sandbox_bind_roots = self._active_sandbox_bind_roots(
resolved_workspace or cwd_path
)
for raw in self._extract_absolute_paths(cmd):
try:
@@ -817,6 +834,8 @@ class ExecTool(Tool):
)
if not allowed and resolved_workspace is not None:
allowed = is_path_within(p, resolved_workspace)
if not allowed and sandbox_bind_roots:
allowed = any(is_path_within(p, root) for root in sandbox_bind_roots)
if p.is_absolute() and not allowed:
return ToolResult.error(
"Error: Command blocked by safety guard (path outside working dir)"
@@ -921,3 +940,38 @@ class ExecTool(Tool):
posix_paths = re.findall(r"(?:^|[\s|>='\"])(/[^\s\"'>;|<]+)", command) # POSIX: /absolute only
home_paths = re.findall(r"(?:^|[\s>='\"])(~[/+][^\s\"'>;|<]*)", command) # POSIX/Windows home shortcut: ~/ or ~+
return win_paths + posix_paths + home_paths
@staticmethod
def _normalize_bind_roots(paths: list[str] | None) -> list[Path]:
roots: list[Path] = []
seen: set[str] = set()
for raw in paths or []:
value = str(raw).strip()
if not value:
continue
path = Path(os.path.expandvars(value)).expanduser()
if not path.is_absolute():
continue
with suppress(OSError, RuntimeError, ValueError):
resolved = path.resolve(strict=False)
key = os.path.normcase(os.fspath(resolved))
if key in seen:
continue
seen.add(key)
roots.append(resolved)
return roots
def _active_sandbox_bind_roots(
self,
workspace_root: Path | None = None,
) -> list[Path]:
if self.sandbox != "bwrap" or _IS_WINDOWS:
return []
roots = [*self.sandbox_ro_binds, *self.sandbox_rw_binds]
if workspace_root is None:
return roots
return [
root
for root in roots
if not is_path_within(workspace_root, root)
]
+2 -3
View File
@@ -271,13 +271,12 @@ def _normalize_volcengine_auth_level(value: Any) -> int | None:
@tool_parameters(
tool_parameters_schema(
query=StringSchema("Search query"),
count=IntegerSchema(1, description="Results (1-10)", minimum=1, maximum=10),
count=IntegerSchema(description="Results (1-10)", minimum=1, maximum=10),
timeRange=StringSchema(
"Optional time filter for providers that support it: "
"OneDay, OneWeek, OneMonth, OneYear, or YYYY-MM-DD..YYYY-MM-DD",
),
authLevel=IntegerSchema(
0,
description="Optional authority filter for providers that support it: 0=all, 1=authoritative",
minimum=0,
maximum=1,
@@ -939,7 +938,7 @@ class WebSearchTool(Tool):
"enum": ["markdown", "text"],
"default": "markdown",
},
maxChars=IntegerSchema(0, minimum=100),
maxChars=IntegerSchema(minimum=100),
required=["url"],
)
)
+17 -2
View File
@@ -126,6 +126,7 @@ class TurnDelivery:
lifecycle_message: InboundMessage = field(init=False)
_stream_base_id: str | None = field(init=False, default=None)
_stream_segment: int = field(init=False, default=0)
_stream_open: bool = field(init=False, default=False)
def __post_init__(self) -> None:
self.delivery_message = dataclasses.replace(
@@ -284,8 +285,14 @@ class TurnDelivery:
metadata=self.delivery_message.metadata,
)
)
self._stream_open = True
async def _publish_stream_end(self, *, resuming: bool = False) -> None:
async def _publish_stream_end(
self,
*,
resuming: bool = False,
merge_next: bool = False,
) -> None:
await self.bus.publish_outbound(
outbound_message_for_event(
channel=self.delivery_message.channel,
@@ -293,8 +300,16 @@ class TurnDelivery:
event=StreamEndEvent(
stream_id=self._stream_id(),
resuming=resuming,
merge_next=merge_next,
),
metadata=self.delivery_message.metadata,
)
)
self._stream_segment += 1
self._stream_open = merge_next
if not merge_next:
self._stream_segment += 1
async def abort_stream(self) -> None:
"""Close an interrupted stream so stateful channels can release its buffer."""
if self._stream_open:
await self._publish_stream_end()
+2
View File
@@ -46,6 +46,7 @@ class StreamEndEvent(OutboundEvent):
content: str = ""
stream_id: str | None = None
resuming: bool = False
merge_next: bool = False
@dataclass(frozen=True)
@@ -176,6 +177,7 @@ def _legacy_event_from_metadata(msg: OutboundMessage) -> OutboundEvent | None:
content=msg.content,
stream_id=_metadata_str(meta, "_stream_id"),
resuming=bool(meta.get("_resuming")),
merge_next=bool(meta.get("_merge_next")),
)
if meta.get("_stream_delta"):
return StreamDeltaEvent(
+5 -1
View File
@@ -29,7 +29,7 @@ class BaseChannel(ABC):
name: str = "base"
display_name: str = "Base"
send_progress: bool = True
send_tool_hints: bool = False
send_tool_hints: bool = True
show_reasoning: bool = True
def __init__(self, config: Any, bus: MessageBus):
@@ -110,6 +110,7 @@ class BaseChannel(ABC):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
"""Deliver a streaming text chunk.
@@ -118,6 +119,9 @@ class BaseChannel(ABC):
Stateful implementations should key buffers by ``stream_id`` rather
than only by ``chat_id`` when it is provided.
``merge_next`` marks a resumable provider boundary whose next text
segment belongs to the same user-visible message.
"""
pass
+46 -3
View File
@@ -24,6 +24,17 @@ from nanobot.security.network import validate_resolved_url, validate_url_target
DINGTALK_MAX_REMOTE_MEDIA_BYTES = 20 * 1024 * 1024
DINGTALK_MAX_REMOTE_MEDIA_REDIRECTS = 3
_DINGTALK_MARKDOWN_INLINE_SPECIALS = frozenset(r"\`*_{}[]()<>#+-.!|~")
_DINGTALK_SENDER_NAME_MAX_CHARS = 80
def _escape_markdown_sender_name(value: str) -> str:
"""Render an untrusted display name as one bounded Markdown-safe line."""
normalized = " ".join(value.split())[:_DINGTALK_SENDER_NAME_MAX_CHARS]
return "".join(
f"\\{char}" if char in _DINGTALK_MARKDOWN_INLINE_SPECIALS else char
for char in normalized
)
try:
from dingtalk_stream import (
@@ -175,6 +186,7 @@ class DingTalkConfig(Base):
allow_remote_media_redirects: bool = False
remote_media_redirect_allowed_hosts: list[str] = Field(default_factory=list)
group_user_isolation: bool = False # If True, each user in group chat gets their own session
disable_private_chat: bool = False # If True, reject 1:1 DMs with a notice; group chats only
class DingTalkChannel(BaseChannel):
@@ -712,8 +724,20 @@ class DingTalkChannel(BaseChannel):
if not token:
raise RuntimeError("DingTalk access token unavailable")
if msg.content and msg.content.strip():
if not await self._send_markdown_text(token, msg.chat_id, msg.content.strip()):
content = msg.content.strip() if msg.content else ""
if content:
# In group chats, prefix the reply with a markdown header naming the
# sender so the addressed user can spot the reply. Visual only —
# DingTalk's markdown robot messages do not push real @ notifications.
sender_name = msg.metadata.get("sender_name") if msg.metadata else None
safe_sender_name = (
_escape_markdown_sender_name(sender_name)
if isinstance(sender_name, str)
else ""
)
if msg.chat_id.startswith("group:") and safe_sender_name:
content = f"# @{safe_sender_name}\n\n{content}"
if not await self._send_markdown_text(token, msg.chat_id, content):
raise RuntimeError("DingTalk text message was not delivered")
for media_ref in msg.media or []:
@@ -733,7 +757,7 @@ class DingTalkChannel(BaseChannel):
async def _on_message(
self,
content: str,
sender_id: str,
sender_id: str | None,
sender_name: str,
conversation_type: str | None = None,
conversation_id: str | None = None,
@@ -745,11 +769,30 @@ class DingTalkChannel(BaseChannel):
"""
try:
self.logger.info("inbound: {} from {}", content, sender_name)
if not sender_id:
self.logger.warning("dropping DingTalk message without a sender ID")
return
is_group = conversation_type == "2" and conversation_id
chat_id = f"group:{conversation_id}" if is_group else sender_id
session_key = None
if is_group and self.config.group_user_isolation:
session_key = f"{self.name}:group:{conversation_id}:{sender_id}"
if not is_group and self.config.disable_private_chat:
# Group-only kill switch: drop DMs with a notice *before* any
# allow_from / pairing check, so even allowlisted senders are
# redirected — intentional, this is a hard private-chat guard
# rather than an authorization decision. No session is created.
self.logger.info("private chat disabled; rejecting DM from {}", sender_name)
await self.send(
OutboundMessage(
channel=self.name,
chat_id=chat_id,
content="该机器人未开启私聊,请在群聊中与我对话。",
)
)
return
await self._handle_message(
sender_id=sender_id,
chat_id=chat_id,
@@ -1,4 +1,5 @@
import asyncio
import json
import zipfile
from io import BytesIO
from types import SimpleNamespace
@@ -9,15 +10,15 @@ import pytest
# Check optional dingtalk dependencies before running tests
try:
from nanobot.channels import dingtalk
DINGTALK_AVAILABLE = getattr(dingtalk, "DINGTALK_AVAILABLE", False)
import nanobot.channels.dingtalk.runtime as dingtalk_module
DINGTALK_AVAILABLE = dingtalk_module.DINGTALK_AVAILABLE
except ImportError:
DINGTALK_AVAILABLE = False
if not DINGTALK_AVAILABLE:
pytest.skip("DingTalk dependencies not installed (dingtalk-stream)", allow_module_level=True)
import nanobot.channels.dingtalk.runtime as dingtalk_module
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.dingtalk.runtime import (
@@ -153,6 +154,92 @@ async def test_group_user_isolation_true_separates_sessions() -> None:
assert msg1.chat_id == msg2.chat_id == "group:conv123"
def test_disable_private_chat_uses_camel_case_config_key() -> None:
config = DingTalkConfig.model_validate({"disablePrivateChat": True})
assert config.disable_private_chat is True
assert config.model_dump(mode="json", by_alias=True)["disablePrivateChat"] is True
@pytest.mark.asyncio
async def test_dm_rejected_when_private_chat_disabled(monkeypatch) -> None:
"""With disable_private_chat=True, a 1:1 DM is rejected: nothing reaches the
bus (no session is created) and the bot replies with a notice directing the
user to group chat. Even allowlisted senders are blocked in DMs."""
config = DingTalkConfig(
client_id="app",
client_secret="secret",
allow_from=["*"], # even allowlisted senders are blocked in DMs
disable_private_chat=True,
)
bus = MessageBus()
channel = DingTalkChannel(config, bus)
async def fake_get_token():
return "test-token"
monkeypatch.setattr(channel, "_get_access_token", fake_get_token)
channel._http = _FakeHttp()
await channel._on_message(
"hello",
sender_id="user1",
sender_name="Alice",
conversation_type="1",
)
# No inbound message was published -> no session created
assert bus.inbound.empty()
# A notice was sent back to the DM user via the private-chat API
assert len(channel._http.calls) == 1
call = channel._http.calls[0]
assert call["url"] == "https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend"
assert call["json"]["msgKey"] == "sampleMarkdown"
assert call["json"]["userIds"] == ["user1"]
assert "该机器人未开启私聊,请在群聊中与我对话。" in call["json"]["msgParam"]
@pytest.mark.asyncio
async def test_dm_allowed_when_private_chat_not_disabled() -> None:
"""By default (disable_private_chat=False), a 1:1 DM still reaches the bus."""
config = DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"])
bus = MessageBus()
channel = DingTalkChannel(config, bus)
await channel._on_message(
"hello",
sender_id="user1",
sender_name="Alice",
conversation_type="1",
)
msg = await bus.consume_inbound()
assert msg.chat_id == "user1"
assert msg.metadata["conversation_type"] == "1"
@pytest.mark.asyncio
async def test_group_message_allowed_when_private_chat_disabled() -> None:
"""Disabling private chat must not affect group messages."""
config = DingTalkConfig(
client_id="app", client_secret="secret", allow_from=["*"], disable_private_chat=True
)
bus = MessageBus()
channel = DingTalkChannel(config, bus)
await channel._on_message(
"hello",
sender_id="user1",
sender_name="Alice",
conversation_type="2",
conversation_id="conv123",
)
msg = await bus.consume_inbound()
assert msg.chat_id == "group:conv123"
@pytest.mark.asyncio
async def test_group_send_uses_group_messages_api() -> None:
config = DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"])
@@ -173,6 +260,105 @@ async def test_group_send_uses_group_messages_api() -> None:
assert call["json"]["msgKey"] == "sampleMarkdown"
@pytest.mark.asyncio
async def test_group_send_prepends_sender_mention(monkeypatch) -> None:
"""Group replies are prefixed with a markdown header naming the sender."""
config = DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"])
channel = DingTalkChannel(config, MessageBus())
channel._http = _FakeHttp()
async def _fake_token() -> str:
return "token"
monkeypatch.setattr(channel, "_get_access_token", _fake_token)
await channel.send(
OutboundMessage(
channel="dingtalk",
chat_id="group:conv123",
content="hello",
metadata={"sender_name": "Alice"},
)
)
sent_text = json.loads(channel._http.calls[0]["json"]["msgParam"])["text"]
assert sent_text == "# @Alice\n\nhello"
@pytest.mark.asyncio
async def test_group_send_escapes_untrusted_sender_name(monkeypatch) -> None:
"""A sender nickname cannot inject extra Markdown blocks into the reply."""
config = DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"])
channel = DingTalkChannel(config, MessageBus())
channel._http = _FakeHttp()
async def _fake_token() -> str:
return "token"
monkeypatch.setattr(channel, "_get_access_token", _fake_token)
await channel.send(
OutboundMessage(
channel="dingtalk",
chat_id="group:conv123",
content="hello",
metadata={"sender_name": "Alice\n# [click](https://evil) *admin*"},
)
)
sent_text = json.loads(channel._http.calls[0]["json"]["msgParam"])["text"]
assert sent_text == r"# @Alice \# \[click\]\(https://evil\) \*admin\*" + "\n\nhello"
@pytest.mark.asyncio
async def test_private_send_does_not_prepend_mention(monkeypatch) -> None:
"""Private replies are sent verbatim, without the sender header."""
config = DingTalkConfig(client_id="app", client_secret="secret", allow_from=["*"])
channel = DingTalkChannel(config, MessageBus())
channel._http = _FakeHttp()
async def _fake_token() -> str:
return "token"
monkeypatch.setattr(channel, "_get_access_token", _fake_token)
await channel.send(
OutboundMessage(
channel="dingtalk",
chat_id="user1", # private chat: no "group:" prefix
content="hello",
metadata={"sender_name": "Alice"},
)
)
sent_text = json.loads(channel._http.calls[0]["json"]["msgParam"])["text"]
assert sent_text == "hello"
@pytest.mark.asyncio
async def test_message_without_sender_id_is_dropped() -> None:
"""Malformed inbound events must not publish or attempt an invalid reply."""
config = DingTalkConfig(
client_id="app",
client_secret="secret",
allow_from=["*"],
disable_private_chat=True,
)
bus = MessageBus()
channel = DingTalkChannel(config, bus)
channel._http = _FakeHttp()
await channel._on_message(
"hello",
sender_id=None,
sender_name="Unknown",
conversation_type="1",
)
assert bus.inbound.empty()
assert channel._http.calls == []
@pytest.mark.asyncio
async def test_handler_uses_voice_recognition_text_when_text_is_empty(monkeypatch) -> None:
bus = MessageBus()
+5
View File
@@ -489,6 +489,7 @@ class DiscordChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
"""Progressive Discord delivery: send once, then edit until the stream ends."""
client = self._client
@@ -496,6 +497,10 @@ class DiscordChannel(BaseChannel):
self.logger.warning("client not ready; dropping stream delta")
return
if stream_end and merge_next:
if not delta:
return
stream_end = False
if stream_end:
buf = self._stream_bufs.get(chat_id)
if not buf or buf.message is None or not buf.text:
@@ -754,6 +754,36 @@ async def test_send_delta_streams_by_editing_message(monkeypatch) -> None:
assert owner._stream_bufs == {}
@pytest.mark.asyncio
async def test_send_delta_merge_next_keeps_one_message(monkeypatch) -> None:
owner = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus())
client = _FakeDiscordClient(owner, intents=None)
owner._client = client
owner._running = True
target = _FakeChannel(channel_id=123)
client.channels[123] = target
times = iter([1.0, 3.0, 5.0])
monkeypatch.setattr("nanobot.channels.discord.runtime.time.monotonic", lambda: next(times, 5.0))
await owner.send_delta(
"123",
"first-",
stream_id="s1",
stream_end=True,
merge_next=True,
)
await owner.send_delta("123", "second", stream_id="s1")
await owner.send_delta("123", "", stream_id="s1", stream_end=True)
assert target.sent_payloads == [{"content": "first-"}]
assert target.sent_messages[0].edits == [
{"content": "first-second"},
{"content": "first-second"},
]
assert owner._stream_bufs == {}
@pytest.mark.asyncio
async def test_send_delta_stream_end_splits_oversized_reply(monkeypatch) -> None:
owner = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus())
+29 -17
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
import json
import secrets
import threading
import time
from dataclasses import dataclass
from typing import Any
@@ -41,6 +42,7 @@ class FeishuConnectStore:
def __init__(self) -> None:
self._sessions: dict[str, FeishuConnectSession] = {}
self._completion_lock = threading.Lock()
async def handle(self, action: str, query: QueryParams) -> dict[str, Any]:
"""Handle one generic settings connection action."""
@@ -58,7 +60,7 @@ class FeishuConnectStore:
if action == "poll":
return await asyncio.to_thread(self.poll, session_id)
if action == "cancel":
return self.cancel(session_id)
return await asyncio.to_thread(self.cancel, session_id)
raise ChannelConnectError(f"unsupported Feishu connect action: {action}", status=404)
def start(
@@ -127,24 +129,33 @@ class FeishuConnectStore:
session.last_error = str(exc)
return _pending_payload(session)
session.domain = str(result.get("domain") or session.domain)
status = result.get("status")
if status == "succeeded":
session.instance_id = feishu.save_registration_result(
result,
instance_id=session.instance_id,
name=session.instance_name,
)
self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "succeeded",
"message": "Feishu is connected.",
"domain": session.domain,
"app_id": result.get("app_id"),
}
with self._completion_lock:
if self._sessions.get(session_id) is not session:
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "cancelled",
"message": "Feishu connection cancelled.",
}
session.domain = str(result.get("domain") or session.domain)
session.instance_id = feishu.save_registration_result(
result,
instance_id=session.instance_id,
name=session.instance_name,
)
self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "succeeded",
"message": "Feishu is connected.",
"domain": session.domain,
"app_id": result.get("app_id"),
}
session.domain = str(result.get("domain") or session.domain)
if status == "failed":
self._sessions.pop(session_id, None)
return {
@@ -158,7 +169,8 @@ class FeishuConnectStore:
return _pending_payload(session)
def cancel(self, session_id: str) -> dict[str, Any]:
session = self._sessions.pop(session_id, None)
with self._completion_lock:
session = self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id if session else DEFAULT_INSTANCE_ID,
+28 -10
View File
@@ -269,7 +269,7 @@ def _extract_element_content(element: dict) -> list[str]:
parts.append(text_content)
elif isinstance(text, str):
parts.append(text)
for field in element.get("fields", []):
for field in element.get("fields") or []:
if isinstance(field, dict):
field_text = field.get("text", {})
if isinstance(field_text, dict):
@@ -291,7 +291,10 @@ def _extract_element_content(element: dict) -> list[str]:
c = text.get("content", "")
if c:
parts.append(c)
url = element.get("url", "") or element.get("multi_url", {}).get("url", "")
multi_url = element.get("multi_url") or {}
url = element.get("url", "") or (
multi_url.get("url", "") if isinstance(multi_url, dict) else ""
)
if url:
parts.append(f"link: {url}")
@@ -300,12 +303,14 @@ def _extract_element_content(element: dict) -> list[str]:
parts.append(alt.get("content", "[image]") if isinstance(alt, dict) else "[image]")
elif tag == "note":
for ne in element.get("elements", []):
for ne in element.get("elements") or []:
parts.extend(_extract_element_content(ne))
elif tag == "column_set":
for col in element.get("columns", []):
for ce in col.get("elements", []):
for col in element.get("columns") or []:
if not isinstance(col, dict):
continue
for ce in col.get("elements") or []:
parts.extend(_extract_element_content(ce))
elif tag == "plain_text":
@@ -319,7 +324,7 @@ def _extract_element_content(element: dict) -> list[str]:
for column in (element.get("columns") or [])
if isinstance(column, dict) and column.get("name")
]
rows = element.get("rows", [])
rows = element.get("rows") or []
if columns:
parts.append(" | ".join(header for _, header in columns))
if isinstance(rows, list):
@@ -337,7 +342,7 @@ def _extract_element_content(element: dict) -> list[str]:
parts.append(row_text)
else:
for ne in element.get("elements", []):
for ne in element.get("elements") or []:
parts.extend(_extract_element_content(ne))
return parts
@@ -356,7 +361,8 @@ def _extract_post_content(content_json: dict) -> tuple[str, list[str]]:
if not isinstance(block, dict) or not isinstance(block.get("content"), list):
return None, []
texts, images = [], []
if title := block.get("title"):
title = block.get("title")
if isinstance(title, str) and title:
texts.append(title)
for row in block["content"]:
if not isinstance(row, list):
@@ -366,12 +372,19 @@ def _extract_post_content(content_json: dict) -> tuple[str, list[str]]:
continue
tag = el.get("tag")
if tag in ("text", "a"):
texts.append(el.get("text", ""))
text = el.get("text", "")
if isinstance(text, str):
texts.append(text)
elif tag == "at":
texts.append(f"@{el.get('user_name', 'user')}")
user = el.get("user_name", "user")
texts.append(f"@{user if isinstance(user, str) and user else 'user'}")
elif tag == "code_block":
lang = el.get("language", "")
code_text = el.get("text", "")
if not isinstance(lang, str):
lang = ""
if not isinstance(code_text, str):
code_text = ""
texts.append(f"\n```{lang}\n{code_text}\n```\n")
elif tag == "img" and (key := el.get("image_key")):
images.append(key)
@@ -2203,6 +2216,7 @@ class FeishuChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
"""Progressive streaming via CardKit: create card on first delta, stream-update on subsequent.
@@ -2218,6 +2232,10 @@ class FeishuChannel(BaseChannel):
rid_type = "chat_id" if chat_id.startswith("oc_") else "open_id"
# --- stream end: final update or fallback ---
if stream_end and merge_next:
if not delta:
return
stream_end = False
if stream_end:
message_id = meta.get("message_id")
# Only finalize the OnIt -> DONE reaction transition on the truly
@@ -0,0 +1,122 @@
from __future__ import annotations
import asyncio
import threading
from typing import Any
import pytest
from nanobot.channels.feishu import runtime as feishu
from nanobot.channels.feishu.connect import FeishuConnectStore
@pytest.mark.asyncio
async def test_feishu_cancel_wins_over_inflight_confirmation(
monkeypatch: pytest.MonkeyPatch,
) -> None:
poll_started = threading.Event()
release_poll = threading.Event()
saved_results: list[dict[str, Any]] = []
monkeypatch.setattr(feishu, "_init_registration", lambda _domain: None)
monkeypatch.setattr(
feishu,
"_begin_registration",
lambda _domain: {
"device_code": "device-cancel",
"qr_url": "https://qr.example/cancel",
"expire_in": 600,
"interval": 2,
},
)
def fake_poll_registration_once(**_kwargs: Any) -> dict[str, str]:
poll_started.set()
assert release_poll.wait(timeout=5)
return {
"status": "succeeded",
"domain": "feishu",
"app_id": "late-app",
"app_secret": "late-secret",
}
def fake_save_registration_result(
result: dict[str, Any],
**_kwargs: Any,
) -> str:
saved_results.append(result)
return "default"
monkeypatch.setattr(feishu, "poll_registration_once", fake_poll_registration_once)
monkeypatch.setattr(feishu, "save_registration_result", fake_save_registration_result)
store = FeishuConnectStore()
started = await store.handle("start", {})
query = {"session_id": [started["session_id"]]}
poll_task = asyncio.create_task(store.handle("poll", query))
assert await asyncio.to_thread(poll_started.wait, 5)
cancelled = await store.handle("cancel", query)
release_poll.set()
completed = await poll_task
assert cancelled["status"] == "cancelled"
assert completed["status"] == "cancelled"
assert saved_results == []
@pytest.mark.asyncio
async def test_feishu_cancel_does_not_interleave_with_registration_save(
monkeypatch: pytest.MonkeyPatch,
) -> None:
save_started = threading.Event()
release_save = threading.Event()
monkeypatch.setattr(feishu, "_init_registration", lambda _domain: None)
monkeypatch.setattr(
feishu,
"_begin_registration",
lambda _domain: {
"device_code": "device-lock",
"qr_url": "https://qr.example/lock",
"expire_in": 600,
"interval": 2,
},
)
monkeypatch.setattr(
feishu,
"poll_registration_once",
lambda **_kwargs: {
"status": "succeeded",
"domain": "feishu",
"app_id": "saved-app",
"app_secret": "saved-secret",
},
)
def fake_save_registration_result(
_result: dict[str, Any],
**_kwargs: Any,
) -> str:
save_started.set()
assert release_save.wait(timeout=5)
return "default"
monkeypatch.setattr(feishu, "save_registration_result", fake_save_registration_result)
store = FeishuConnectStore()
started = await store.handle("start", {})
query = {"session_id": [started["session_id"]]}
poll_task = asyncio.create_task(store.handle("poll", query))
assert await asyncio.to_thread(save_started.wait, 5)
cancel_task = asyncio.create_task(store.handle("cancel", query))
await asyncio.sleep(0)
assert not cancel_task.done()
release_save.set()
completed = await poll_task
cancelled = await cancel_task
assert completed["status"] == "succeeded"
assert cancelled["status"] == "cancelled"
@@ -1,6 +1,10 @@
import json
from nanobot.channels.feishu.runtime import _extract_share_card_content
from nanobot.channels.feishu.runtime import (
_extract_element_content,
_extract_post_content,
_extract_share_card_content,
)
def test_extract_interactive_card_reads_user_dsl_body_elements() -> None:
@@ -37,3 +41,48 @@ def test_extract_interactive_card_reads_table_rows() -> None:
}
assert _extract_share_card_content(content, "interactive") == "Name | Score\nAlice | 98"
def test_extract_post_content_tolerates_null_fields() -> None:
text, images = _extract_post_content(
{
"title": None,
"content": [
[
{"tag": "text", "text": None},
{"tag": "a", "text": None},
{"tag": "at", "user_name": None},
{"tag": "text", "text": "ok"},
{"tag": "code_block", "language": None, "text": None},
]
],
}
)
assert "@user" in text
assert "ok" in text
assert images == []
def test_extract_button_tolerates_null_multi_url() -> None:
element = {"tag": "button", "text": {"content": "Go"}, "multi_url": None}
assert _extract_element_content(element) == ["Go"]
def test_extract_column_set_tolerates_null_columns_and_elements() -> None:
assert _extract_element_content({"tag": "column_set", "columns": None}) == []
assert _extract_element_content(
{"tag": "column_set", "columns": [{"elements": None}]}
) == []
def test_extract_div_tolerates_null_fields() -> None:
assert _extract_element_content(
{"tag": "div", "text": {"content": "hi"}, "fields": None}
) == ["hi"]
def test_interactive_card_button_null_multi_url() -> None:
content = {
"elements": [{"tag": "button", "text": {"content": "Go"}, "multi_url": None}]
}
assert _extract_share_card_content(content, "interactive") == "Go"
@@ -285,6 +285,27 @@ class TestSendDelta:
settings_call = ch._client.cardkit.v1.card.settings.call_args[0][0]
assert settings_call.body.sequence == 5 # after final content seq 4
@pytest.mark.asyncio
async def test_stream_end_merge_next_preserves_buffer(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="first-",
card_id="card_1",
sequence=3,
last_edit=time.monotonic(),
)
await ch.send_delta(
"oc_chat1",
"boundary",
stream_end=True,
merge_next=True,
)
assert ch._stream_bufs["oc_chat1"].text == "first-boundary"
ch._client.cardkit.v1.card_element.content.assert_not_called()
ch._client.cardkit.v1.card.settings.assert_not_called()
@pytest.mark.asyncio
async def test_stream_end_fallback_when_no_card_id(self):
"""If card creation failed, stream_end falls back to a plain card message."""
+27 -3
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
import hashlib
import inspect
from collections.abc import Callable, Iterable
from contextlib import suppress
from pathlib import Path
@@ -97,6 +98,7 @@ class ChannelManager:
webui_static_dist: bool = True,
webui_runtime_surface: str = "browser",
webui_runtime_capabilities: dict[str, Any] | None = None,
webui_extension_service: Any | None = None,
):
self.config = config
self.bus = bus
@@ -109,6 +111,7 @@ class ChannelManager:
self._webui_static_dist = webui_static_dist
self._webui_runtime_surface = webui_runtime_surface
self._webui_runtime_capabilities = dict(webui_runtime_capabilities or {})
self._webui_extension_service = webui_extension_service
self.channels: dict[str, BaseChannel] = {}
self._channel_owners: dict[str, str] = {}
self._channel_runtime_specs: dict[str, tuple[str, str]] = {}
@@ -175,6 +178,10 @@ class ChannelManager:
local_trigger_pending_ids=self._webui_local_trigger_pending_ids,
channel_feature_action=self.apply_channel_feature_action,
channel_runtime_status=self.get_status,
extension_service=self._webui_extension_service,
allow_remote_package_install=(
self.config.tools.webui_allow_remote_package_install
),
logger=logger,
)
kwargs["gateway"] = gateway
@@ -763,13 +770,29 @@ class ChannelManager:
msg: OutboundMessage,
event: StreamDeltaEvent | StreamEndEvent,
) -> None:
kwargs: dict[str, Any] = {
"stream_id": event.stream_id,
"stream_end": isinstance(event, StreamEndEvent),
"resuming": event.resuming if isinstance(event, StreamEndEvent) else False,
}
if isinstance(event, StreamEndEvent) and event.merge_next:
try:
signature = inspect.signature(channel.send_delta)
if (
"merge_next" in signature.parameters
or any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
for parameter in signature.parameters.values()
)
):
kwargs["merge_next"] = True
except (TypeError, ValueError):
pass
await channel.send_delta(
msg.chat_id,
msg.content,
msg.metadata,
stream_id=event.stream_id,
stream_end=isinstance(event, StreamEndEvent),
resuming=event.resuming if isinstance(event, StreamEndEvent) else False,
**kwargs,
)
@staticmethod
@@ -850,6 +873,7 @@ class ChannelManager:
final_event = StreamEndEvent(
stream_id=next_stream_id,
resuming=next_event.resuming,
merge_next=next_event.merge_next,
)
# Stream ended - stop coalescing this stream
break
+5
View File
@@ -598,9 +598,14 @@ class MatrixChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
relates_to = self._build_thread_relates_to(metadata)
if stream_end and merge_next:
if not delta:
return
stream_end = False
if stream_end:
stream_key = _matrix_stream_key(chat_id, stream_id)
buf = self._stream_bufs.pop(stream_key, None)
@@ -1937,6 +1937,29 @@ async def test_send_delta_stream_end_replaces_existing_message() -> None:
}
@pytest.mark.asyncio
async def test_send_delta_merge_next_preserves_buffer() -> None:
channel = MatrixChannel(_make_config(), MessageBus())
client = _FakeAsyncClient("", "", "", None)
channel.client = client
channel._stream_bufs["!room:matrix.org"] = matrix_module._StreamBuf(
text="first-",
event_id="event-1",
last_edit=100.0,
)
channel.monotonic_time = lambda: 100.1
await channel.send_delta(
"!room:matrix.org",
"boundary",
stream_end=True,
merge_next=True,
)
assert channel._stream_bufs["!room:matrix.org"].text == "first-boundary"
assert client.room_send_calls == []
@pytest.mark.asyncio
async def test_send_delta_keeps_same_room_stream_ids_independent(monkeypatch) -> None:
channel = MatrixChannel(_make_config(), MessageBus())
+7 -2
View File
@@ -56,7 +56,7 @@ class MattermostConfig(Base):
react_emoji: str = "eyes"
done_emoji: str = "white_check_mark"
send_progress: bool = True
send_tool_hints: bool = False
send_tool_hints: bool = True
dm: MattermostDMConfig = Field(default_factory=MattermostDMConfig)
@@ -515,6 +515,7 @@ class MattermostChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
if not self._http_client:
return
@@ -532,7 +533,11 @@ class MattermostChannel(BaseChannel):
final += delta
if resuming:
self._clear_stream_state(stream_id)
if merge_next:
self._stream_buffers[stream_id] = final
self._stream_committed[stream_id] = final
else:
self._clear_stream_state(stream_id)
return
if final and not meta.get("_progress"):
@@ -119,6 +119,7 @@ def test_config_defaults():
assert config.token == ""
assert config.streaming is True
assert config.streaming_max_chars == 16000
assert config.send_tool_hints is True
assert config.dm.enabled is True
assert config.dm.policy == "open"
assert config.reply_in_thread is True
@@ -131,6 +132,7 @@ def test_config_camelcase_aliases():
"allowFromMatchMode": "username",
"streamingMaxChars": 8000,
"replyInThread": False,
"sendToolHints": False,
}
config = MattermostConfig.model_validate(raw)
assert config.server_url == "https://mm.example.com"
@@ -138,11 +140,13 @@ def test_config_camelcase_aliases():
assert config.allow_from_match_mode == "username"
assert config.streaming_max_chars == 8000
assert config.reply_in_thread is False
assert config.send_tool_hints is False
def test_config_default_config_classmethod():
d = MattermostChannel.default_config()
assert d["enabled"] is False
assert d["sendToolHints"] is True
assert d["serverUrl"] == ""
assert d["token"] == ""
@@ -578,6 +582,33 @@ async def test_stream_end_keyword_resuming_does_not_post_or_mark_done():
assert "s1" not in channel._stream_buffers
@pytest.mark.asyncio
async def test_stream_end_merge_next_preserves_buffer_until_final_end():
channel, fake = _make_channel()
channel._self_id = "bot_id"
fake.set_post_response("/api/v4/posts", {"id": "stream_post_1"})
await channel.send_delta("chan_1", "first ", stream_id="s1")
await channel.send_delta(
"chan_1",
"boundary ",
stream_id="s1",
stream_end=True,
resuming=True,
merge_next=True,
)
assert channel._stream_buffers["s1"] == "first boundary "
await channel.send_delta("chan_1", "second", stream_id="s1")
await channel.send_delta("chan_1", "", stream_id="s1", stream_end=True)
posts = [call for call in fake.post_calls if call["path"] == "/api/v4/posts"]
assert len(posts) == 1
assert posts[0]["json"]["message"] == "first boundary second"
assert "s1" not in channel._stream_buffers
@pytest.mark.asyncio
async def test_stream_end_failure_keeps_buffer_for_retry():
channel, fake = _make_channel()
-19
View File
@@ -3,8 +3,6 @@
from __future__ import annotations
import pkgutil
from functools import cache
from importlib.metadata import entry_points
from typing import TYPE_CHECKING
from loguru import logger
@@ -19,22 +17,6 @@ if TYPE_CHECKING:
from nanobot.channels.base import BaseChannel
@cache
def _warn_legacy_channel_entry_points() -> None:
# TODO(v0.3.1): Remove this detection and warning. v0.3.0 is the final
# migration window for installed legacy channel entry points.
names = sorted({entry_point.name for entry_point in entry_points(group="nanobot.channels")})
if not names:
return
logger.warning(
"Legacy channel entry points were detected but will not be loaded: {}. "
"The '{}' entry-point group is no longer supported; use a built-in channel or "
"migrate it into nanobot/channels/<channel>/.",
", ".join(names),
"nanobot.channels",
)
def _channel_package_names() -> list[str]:
import nanobot.channels as package
@@ -49,7 +31,6 @@ def discover_plugins(
enabled_names: set[str] | None = None,
) -> dict[str, ChannelPlugin]:
"""Load dependency-free descriptors from self-contained channel packages."""
_warn_legacy_channel_entry_points()
plugins: dict[str, ChannelPlugin] = {}
for name in _channel_package_names():
if enabled_names is not None and name not in enabled_names:
+5
View File
@@ -923,6 +923,7 @@ class TelegramChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
"""Progressive message editing: send on first delta, edit on subsequent ones."""
if not self._app:
@@ -930,6 +931,10 @@ class TelegramChannel(BaseChannel):
meta = metadata or {}
int_chat_id = int(chat_id)
if stream_end and merge_next:
if not delta:
return
stream_end = False
if stream_end:
buf = self._stream_bufs.get(chat_id)
if not buf or not buf.message_id or not buf.text:
@@ -675,6 +675,33 @@ async def test_send_delta_stream_end_raises_and_keeps_buffer_on_failure() -> Non
assert "123" in channel._stream_bufs
@pytest.mark.asyncio
async def test_send_delta_merge_next_preserves_buffer() -> None:
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
channel._app.bot.edit_message_text = AsyncMock()
channel._stream_bufs["123"] = _StreamBuf(
text="first-",
message_id=7,
last_edit=float("inf"),
stream_id="s:0",
)
await channel.send_delta(
"123",
"boundary",
stream_id="s:0",
stream_end=True,
merge_next=True,
)
assert channel._stream_bufs["123"].text == "first-boundary"
channel._app.bot.edit_message_text.assert_not_awaited()
@pytest.mark.asyncio
async def test_send_delta_stream_end_treats_not_modified_as_success() -> None:
from telegram.error import BadRequest
+8 -1
View File
@@ -995,13 +995,18 @@ class WebSocketChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
conns = list(self._subs.get(chat_id, ()))
meta = metadata or {}
stream_key = (chat_id, str(stream_id or ""))
if stream_end:
body: dict[str, Any] = {"event": "stream_end", "chat_id": chat_id}
buffered = self._stream_text_buffers.pop(stream_key, [])
buffered = (
self._stream_text_buffers.setdefault(stream_key, [])
if merge_next
else self._stream_text_buffers.pop(stream_key, [])
)
if delta:
buffered.append(delta)
full_text = "".join(buffered)
@@ -1019,6 +1024,8 @@ class WebSocketChannel(BaseChannel):
body["stream_id"] = stream_id
if stream_end and resuming:
body["resuming"] = True
if stream_end and merge_next:
body["merge_next"] = True
self._transcripts.prepare_and_append(
chat_id,
body,
@@ -1350,6 +1350,39 @@ async def test_send_delta_marks_resuming_stream_end() -> None:
assert payload["resuming"] is True
@pytest.mark.asyncio
async def test_send_delta_keeps_buffer_across_merged_stream_boundary() -> None:
bus = MagicMock()
channel = WebSocketChannel(
{"enabled": True, "allowFrom": ["*"], "streaming": True},
bus,
gateway=_basic_handler(bus),
)
mock_ws = AsyncMock()
channel._attach(mock_ws, "chat-1")
await channel.send_delta("chat-1", "first ", stream_id="sid")
await channel.send_delta(
"chat-1",
"",
stream_id="sid",
stream_end=True,
resuming=True,
merge_next=True,
)
await channel.send_delta("chat-1", "second", stream_id="sid")
await channel.send_delta("chat-1", "", stream_id="sid", stream_end=True)
payloads = [json.loads(call.args[0]) for call in mock_ws.send.await_args_list]
assert payloads[1]["merge_next"] is True
assert payloads[1]["resuming"] is True
assert [payload["text"] for payload in payloads if payload["event"] == "delta"] == [
"first ",
"second",
]
assert ("chat-1", "sid") not in channel._stream_text_buffers
@pytest.mark.asyncio
async def test_send_delta_stream_end_includes_inline_final_text() -> None:
bus = MagicMock()
+6
View File
@@ -130,6 +130,12 @@ class WeixinConnectStore:
status = status_data.get("status", "")
if status == "confirmed":
if self._sessions.get(session_id) is not session:
return {
"session_id": session_id,
"status": "cancelled",
"message": "WeChat login cancelled.",
}
token = str(status_data.get("bot_token", "") or "")
if not token:
self._sessions.pop(session_id, None)
+5
View File
@@ -1243,6 +1243,7 @@ class WeixinChannel(BaseChannel):
stream_id: str | None = None,
stream_end: bool = False,
resuming: bool = False,
merge_next: bool = False,
) -> None:
"""Deliver a streamed reply to WeChat.
@@ -1256,6 +1257,10 @@ class WeixinChannel(BaseChannel):
return
is_end = stream_end or bool(meta.get("_stream_end"))
buffer_key = stream_id or chat_id
if is_end and merge_next:
if delta:
self._stream_buffers.setdefault(buffer_key, []).append(delta)
return
# Accumulate intermediate deltas. The stream_end message's own content
# (present when the manager coalesces deltas into the end message) is
# folded into `full` below instead of appended here, so a send retry
@@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import json
from typing import Any
@@ -97,3 +98,52 @@ async def test_weixin_reconnect_keeps_existing_account_until_scan_succeeds(
cancelled = await store.cancel(started["session_id"])
assert cancelled["status"] == "cancelled"
assert json.loads(state_file.read_text(encoding="utf-8")) == existing
@pytest.mark.asyncio
async def test_weixin_cancel_wins_over_inflight_confirmation(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
state_dir = tmp_path / "weixin-state"
config_path = tmp_path / "config.json"
save_config(
Config.model_validate({"channels": {"weixin": {"stateDir": str(state_dir)}}}),
config_path,
)
monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
poll_started = asyncio.Event()
release_poll = asyncio.Event()
async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
return "qr-cancel", "https://qr.example/cancel"
async def fake_api_get_with_base(
self: WeixinChannel,
**_kwargs: Any,
) -> dict[str, str]:
poll_started.set()
await release_poll.wait()
return {
"status": "confirmed",
"bot_token": "late-token",
"ilink_user_id": "late-user",
}
monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code)
monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base)
store = WeixinConnectStore()
started = await store.handle("start", {})
query = {"session_id": [started["session_id"]]}
poll_task = asyncio.create_task(store.handle("poll", query))
await asyncio.wait_for(poll_started.wait(), timeout=5)
cancelled = await store.handle("cancel", query)
release_poll.set()
completed = await poll_task
assert cancelled["status"] == "cancelled"
assert completed["status"] == "cancelled"
assert not (state_dir / "account.json").exists()
@@ -1824,6 +1824,29 @@ async def test_stream_end_flushes_buffered_answer() -> None:
assert "wx-user" not in channel._stream_buffers
@pytest.mark.asyncio
async def test_stream_end_merge_next_preserves_buffer_until_final_end() -> None:
channel, _bus = _make_channel()
channel._client = object()
channel._token = "token"
channel._context_tokens["wx-user"] = "ctx-1"
channel._context_token_at["wx-user"] = time.time()
channel._send_text = AsyncMock()
await channel.send_delta(
"wx-user",
"first-",
stream_id="s1",
stream_end=True,
merge_next=True,
)
await channel.send_delta("wx-user", "second", stream_id="s1")
await channel.send_delta("wx-user", "", stream_id="s1", stream_end=True)
channel._send_text.assert_awaited_once_with("wx-user", "first-second", "ctx-1")
assert "s1" not in channel._stream_buffers
@pytest.mark.asyncio
async def test_stream_end_send_failure_keeps_buffer_for_retry() -> None:
channel, _bus = _make_channel()
+93 -57
View File
@@ -72,11 +72,16 @@ from nanobot.bus.outbound_events import ( # noqa: E402
StreamEndEvent,
outbound_event_from_message,
)
from nanobot.cli.extensions import create_extensions_app # noqa: E402
from nanobot.cli.gateway import create_gateway_app # noqa: E402
from nanobot.cli.stream import StreamRenderer, ThinkingSpinner # noqa: E402
from nanobot.config.paths import get_workspace_path, is_default_workspace # noqa: E402
from nanobot.config.schema import Config # noqa: E402
from nanobot.security.network import is_loopback_host # noqa: E402
from nanobot.session.keys import ( # noqa: E402
UNIFIED_SESSION_KEY,
last_channel_from_metadata,
)
from nanobot.utils.evaluator import evaluate_response, resolve_evaluator_prompt # noqa: E402
from nanobot.utils.helpers import ( # noqa: E402
sanitize_surrogates as _sanitize_surrogates,
@@ -264,6 +269,7 @@ def _pick_heartbeat_target_from_sessions(
enabled_channels: Iterable[str],
sessions: Iterable[dict[str, Any]],
archived_keys: Iterable[str],
unified_session_metadata: dict[str, Any] | None = None,
) -> tuple[str, str]:
enabled = set(enabled_channels)
archived = set(archived_keys)
@@ -271,6 +277,13 @@ def _pick_heartbeat_target_from_sessions(
key = item.get("key") or ""
if key in archived:
continue
if key == UNIFIED_SESSION_KEY:
route = last_channel_from_metadata(unified_session_metadata)
if route is not None:
channel, chat_id = route
if channel not in {"cli", "system"} and channel in enabled:
return channel, chat_id
continue
if ":" not in key:
continue
channel, chat_id = key.split(":", 1)
@@ -803,7 +816,6 @@ def _load_runtime_config(config: str | None = None, workspace: str | None = None
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
_warn_deprecated_config_keys(config_path)
if workspace:
loaded.agents.defaults.workspace = workspace
return loaded
@@ -824,24 +836,6 @@ def _read_trigger_cli_message(message: str | None) -> str:
raise typer.Exit(1)
def _warn_deprecated_config_keys(config_path: Path | None) -> None:
"""Hint users to remove obsolete keys from their config file."""
import json
from nanobot.config.loader import get_config_path
path = config_path or get_config_path()
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except Exception:
return
if "memoryWindow" in raw.get("agents", {}).get("defaults", {}):
console.print(
"[dim]Hint: `memoryWindow` in your config is no longer used "
"and can be safely removed.[/dim]"
)
def _load_inspection_config(
config: str | None = None,
workspace: str | None = None,
@@ -861,7 +855,6 @@ def _load_inspection_config(
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
_warn_deprecated_config_keys(display_path)
if workspace:
loaded.agents.defaults.workspace = workspace
return display_path, loaded
@@ -1336,6 +1329,7 @@ def serve(
from nanobot.api.server import create_app
from nanobot.bus.queue import MessageBus
from nanobot.extensions.host import ExtensionHost
from nanobot.providers.image_generation import image_gen_provider_configs
from nanobot.session.manager import SessionManager
@@ -1384,12 +1378,17 @@ def serve(
agent_loop, model_name=model_name, request_timeout=timeout,
api_key=api_key,
)
extension_host = ExtensionHost(agent_loop, lambda: runtime_config)
async def on_startup(_app):
await agent_loop._connect_mcp()
await extension_host.reload()
async def on_cleanup(_app):
await agent_loop.close_mcp()
try:
await extension_host.close()
finally:
await agent_loop.close_mcp()
api_app.on_startup.append(on_startup)
api_app.on_cleanup.append(on_cleanup)
@@ -1633,6 +1632,8 @@ def _run_gateway(
from nanobot.cron.service import CronJobSkippedError, CronService
from nanobot.cron.session_turns import is_bound_cron_job
from nanobot.cron.types import CronJob
from nanobot.extensions.host import ExtensionHost
from nanobot.extensions.service import ExtensionService
from nanobot.providers.factory import (
build_provider_snapshot,
build_unconfigured_provider_snapshot,
@@ -1752,6 +1753,8 @@ def _run_gateway(
local_trigger_store=trigger_store,
hook_factories=[create_file_edit_activity_hook],
)
extension_host = ExtensionHost(agent, lambda: config)
extension_service = ExtensionService(host=extension_host)
webui_turn_coordinator = WebuiTurnCoordinator(
bus=bus,
sessions=session_manager,
@@ -1812,12 +1815,13 @@ def _run_gateway(
# Dream is an internal job — run directly, not through the agent loop.
if job.name == "dream":
from nanobot.agent.memory import MemoryStore
from nanobot.agent.memory import DreamRunProgress, MemoryStore
dream_session_key = MemoryStore.dream_session_key
prune_dream_sessions = MemoryStore.prune_dream_sessions
store = agent.context.memory
progress = DreamRunProgress()
resp = None
diff_body = ""
try:
@@ -1832,22 +1836,28 @@ def _run_gateway(
session_key=key,
ephemeral=True,
tools=store.build_dream_tools(),
on_progress=_silent,
on_progress=progress,
)
# Ground truth: the real file delta, not the LLM's self-report.
# The real file delta grounds the audit record; clean completion
# decides whether this history batch has finished processing.
diff_body = store.dream_content_diff()
productive = bool(diff_body) or (
not store.git.is_initialized()
and MemoryStore.dream_run_completed(resp)
completed = MemoryStore.dream_run_completed(
resp,
had_tool_errors=progress.had_tool_errors,
)
if productive:
if completed:
store.set_last_dream_cursor(last_cursor)
logger.info("Dream cron job completed, cursor advanced to {}", last_cursor)
elif MemoryStore.dream_run_completed(resp):
logger.info(
"Dream cron job completed with no memory changes; "
"cursor not advanced",
)
if diff_body:
logger.info(
"Dream cron job completed, cursor advanced to {}",
last_cursor,
)
else:
logger.info(
"Dream cron job completed with no memory changes; "
"cursor advanced to {}",
last_cursor,
)
else:
logger.warning(
"Dream cron job did not complete; cursor remains at {}",
@@ -1979,15 +1989,22 @@ def _run_gateway(
webui_static_dist=webui_static_dist,
webui_runtime_surface=webui_runtime_surface,
webui_runtime_capabilities=webui_runtime_capabilities,
webui_extension_service=extension_service,
)
def _pick_heartbeat_target() -> tuple[str, str]:
"""Pick a routable channel/chat target for heartbeat-triggered messages."""
sidebar_state = read_webui_sidebar_state()
unified_metadata = None
if config.agents.defaults.unified_session:
record = session_manager.read_session_metadata(UNIFIED_SESSION_KEY)
if isinstance(record, dict) and isinstance(record.get("metadata"), dict):
unified_metadata = record["metadata"]
return _pick_heartbeat_target_from_sessions(
enabled_channels=channels.enabled_channels,
sessions=session_manager.list_sessions(),
archived_keys=sidebar_state.get("archived_keys", []),
unified_session_metadata=unified_metadata,
)
if channels.enabled_channels:
@@ -2128,6 +2145,7 @@ def _run_gateway(
console.print,
)
try:
await extension_host.reload()
await cron.start()
# Re-read once on first admission to close the watcher subscription window.
agent.runtime_resolver.invalidate()
@@ -2207,7 +2225,10 @@ def _run_gateway(
if flushed:
logger.info("Shutdown: flushed {} session(s) to disk", flushed)
finally:
restore_shutdown_handlers()
try:
await extension_host.close()
finally:
restore_shutdown_handlers()
asyncio.run(run())
@@ -2244,6 +2265,7 @@ def agent(
"""Interact with the agent directly."""
from nanobot.bus.queue import MessageBus
from nanobot.cron.service import CronService
from nanobot.extensions.host import ExtensionHost
from nanobot.providers.image_generation import image_gen_provider_configs
config = _load_runtime_config(config, workspace)
@@ -2271,6 +2293,7 @@ def agent(
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
extension_host = ExtensionHost(agent_loop, lambda: config)
restart_notice = consume_restart_notice_from_env()
if restart_notice and should_show_cli_restart_notice(restart_notice, session_id):
_print_agent_response(
@@ -2312,29 +2335,36 @@ def agent(
if message:
# Single message mode — direct call, no bus needed
async def run_once():
renderer = StreamRenderer(
render_markdown=markdown,
bot_name=config.agents.defaults.bot_name,
bot_icon=config.agents.defaults.bot_icon,
)
response = await agent_loop.process_direct(
message, session_id,
on_progress=_make_progress(renderer),
on_stream=renderer.on_delta,
on_stream_end=renderer.on_end,
)
if not renderer.streamed:
await renderer.close()
print_kwargs: dict[str, Any] = {}
if renderer.header_printed:
print_kwargs["show_header"] = False
_print_agent_response(
response.content if response else "",
try:
await extension_host.reload()
renderer = StreamRenderer(
render_markdown=markdown,
metadata=response.metadata if response else None,
**print_kwargs,
bot_name=config.agents.defaults.bot_name,
bot_icon=config.agents.defaults.bot_icon,
)
await agent_loop.close_mcp()
response = await agent_loop.process_direct(
message,
session_id,
on_progress=_make_progress(renderer),
on_stream=renderer.on_delta,
on_stream_end=renderer.on_end,
)
if not renderer.streamed:
await renderer.close()
print_kwargs: dict[str, Any] = {}
if renderer.header_printed:
print_kwargs["show_header"] = False
_print_agent_response(
response.content if response else "",
render_markdown=markdown,
metadata=response.metadata if response else None,
**print_kwargs,
)
finally:
try:
await agent_loop.close_mcp()
finally:
await extension_host.close()
asyncio.run(run_once())
else:
@@ -2367,6 +2397,7 @@ def agent(
signal.signal(signal.SIGPIPE, signal.SIG_IGN)
async def run_interactive():
await extension_host.reload()
bus_task = asyncio.create_task(agent_loop.run())
turn_done = asyncio.Event()
turn_done.set()
@@ -2498,7 +2529,10 @@ def agent(
agent_loop.stop()
outbound_task.cancel()
await asyncio.gather(bus_task, outbound_task, return_exceptions=True)
await agent_loop.close_mcp()
try:
await agent_loop.close_mcp()
finally:
await extension_host.close()
asyncio.run(run_interactive())
@@ -2508,6 +2542,8 @@ def agent(
# ============================================================================
app.add_typer(create_extensions_app(console=console), name="extensions")
channels_app = typer.Typer(help="Manage channels")
app.add_typer(channels_app, name="channels")
+180
View File
@@ -0,0 +1,180 @@
"""Typer commands for installing and governing extensions."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from typing import Any
import typer
from rich.console import Console
from rich.table import Table
from nanobot.extensions.service import ExtensionService
ServiceFactory = Callable[[], ExtensionService]
def create_extensions_app(
*,
console: Console,
service_factory: ServiceFactory = ExtensionService,
) -> typer.Typer:
"""Build the extension command group around the transport-neutral service."""
app = typer.Typer(help="Install, inspect, and govern native extensions.")
def service() -> ExtensionService:
return service_factory()
def run(awaitable: Any) -> dict[str, Any]:
try:
return asyncio.run(awaitable)
except (KeyError, RuntimeError, ValueError) as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
@app.command("list")
def list_extensions() -> None:
"""List installed extensions and their activation policy."""
payload = run(service().status())
table = Table(show_header=True, header_style="bold")
table.add_column("Extension")
table.add_column("State")
table.add_column("Trust")
table.add_column("Version")
for item in payload["extensions"]:
state = "active" if item["active"] else ("enabled" if item["enabled"] else "disabled")
table.add_row(
item["name"],
state,
"trusted" if item["trusted"] else "untrusted",
item["version"],
)
console.print(table)
if not payload["extensions"]:
console.print("[dim]No extensions installed.[/dim]")
if payload["diagnostics"]:
console.print(f"[yellow]{len(payload['diagnostics'])} diagnostic(s)[/yellow]")
@app.command("inspect")
def inspect_extension(extension_id: str = typer.Argument(..., help="Extension ID")) -> None:
"""Show manifest, dependencies, permissions, and diagnostics."""
payload = run(service().status())
item = next(
(candidate for candidate in payload["extensions"] if candidate["id"] == extension_id),
None,
)
if item is None:
console.print(f"[red]Extension not found: {extension_id}[/red]")
raise typer.Exit(1)
console.print(f"[bold]{item['name']}[/bold] [dim]{item['version']}[/dim]")
console.print(item["description"] or "[dim]No description.[/dim]")
console.print(
f"State: {'active' if item['active'] else 'inactive'} "
f"Trust: {'trusted' if item['trusted'] else 'untrusted'}"
)
_print_named_rows(console, "Dependencies", item["dependencies"], "kind", "name")
_print_permissions(console, item["permissions"], set(item["granted_permissions"]))
diagnostics = [
diagnostic
for diagnostic in payload["diagnostics"]
if diagnostic["extension_id"] == extension_id
]
if diagnostics:
console.print("\n[bold]Diagnostics[/bold]")
for diagnostic in diagnostics:
console.print(
f" [yellow]{diagnostic['code']}[/yellow] {diagnostic['message']}"
)
@app.command("install")
def install_extension(
source: str = typer.Argument(..., help="Git URL or local package path"),
kind: str = typer.Option("git", "--kind", help="git or local"),
ref: str = typer.Option("", "--ref", help="Git branch, tag, or commit"),
) -> None:
"""Install an extension without granting trust or permissions."""
payload = run(service().install(source, kind=kind, ref=ref, trusted=False))
record = payload["record"]
console.print(
f"[green]Installed {record['id']} {record['version']}[/green] "
"[yellow](untrusted)[/yellow]"
)
console.print(
f"Review with [bold]nanobot extensions inspect {record['id']}[/bold], "
"then grant permissions and trust it explicitly."
)
def policy_command(name: str, value: bool, label: str, help_text: str) -> None:
@app.command(name, help=help_text)
def update(extension_id: str = typer.Argument(..., help="Extension ID")) -> None:
payload = run(
service().set_enabled(extension_id, value)
if name in {"enable", "disable"}
else service().set_trusted(extension_id, value)
)
console.print(f"[green]{label}: {payload['record']['id']}[/green]")
policy_command("enable", True, "Enabled", "Allow an installed extension to activate.")
policy_command("disable", False, "Disabled", "Prevent an installed extension from activating.")
policy_command("trust", True, "Trusted", "Trust an installed extension's executable code.")
policy_command("untrust", False, "Trust revoked", "Revoke trust and stop extension activation.")
@app.command("permissions")
def set_permissions(
extension_id: str = typer.Argument(..., help="Extension ID"),
permissions: list[str] = typer.Argument(
None,
help="Exact permissions to grant; omit all to revoke every grant",
),
) -> None:
"""Replace the extension's granted host permissions."""
payload = run(service().set_permissions(extension_id, set(permissions or [])))
granted = payload["record"]["granted_permissions"]
console.print(
f"[green]Updated permissions for {extension_id}:[/green] "
+ (", ".join(granted) if granted else "none")
)
@app.command("uninstall")
def uninstall_extension(
extension_id: str = typer.Argument(..., help="Extension ID"),
yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation"),
) -> None:
"""Remove an installed extension."""
if not yes and not typer.confirm(f"Uninstall extension '{extension_id}'?"):
raise typer.Abort()
run(service().uninstall(extension_id))
console.print(f"[green]Uninstalled {extension_id}[/green]")
return app
def _print_named_rows(
console: Console,
title: str,
rows: list[dict[str, Any]],
category_key: str,
name_key: str,
) -> None:
console.print(f"\n[bold]{title}[/bold]")
if not rows:
console.print(" [dim]None[/dim]")
return
for row in rows:
console.print(f" {row[category_key]}: {row[name_key]}")
def _print_permissions(
console: Console,
permissions: list[dict[str, str]],
granted: set[str],
) -> None:
console.print("\n[bold]Permissions[/bold]")
if not permissions:
console.print(" [dim]None requested[/dim]")
return
for permission in permissions:
status = "[green]granted[/green]" if permission["name"] in granted else "[yellow]pending[/yellow]"
reason = f"{permission['reason']}" if permission["reason"] else ""
console.print(f" {permission['name']} ({status}){reason}")
+104 -11
View File
@@ -3,6 +3,7 @@
import asyncio
import json
import types
from contextlib import suppress
from dataclasses import dataclass
from functools import lru_cache
from typing import Any, Literal, NamedTuple, get_args, get_origin
@@ -14,6 +15,7 @@ except ModuleNotFoundError: # pragma: no cover - exercised in environments with
from loguru import logger
from pydantic import BaseModel
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from rich.table import Table
@@ -22,7 +24,7 @@ from nanobot.cli.models import (
get_model_context_limit,
get_model_suggestions,
)
from nanobot.config.loader import get_config_path, load_config
from nanobot.config.loader import get_config_path, load_config, resolve_config_env_vars
from nanobot.config.schema import Config, ModelPresetConfig
console = Console()
@@ -44,6 +46,8 @@ class _QuickStartProviderInfo(NamedTuple):
default_api_base: str
backend: str
is_direct: bool
is_oauth: bool
default_model: str
class _QuickStartEndpointChoice(NamedTuple):
@@ -73,6 +77,7 @@ _BACK_PRESSED = object() # Sentinel value for back navigation
_MODEL_PRESET_CACHE: set[str] = set()
_QUICK_START_CUSTOM_PROVIDER_CHOICE = "Other OpenAI-compatible"
_QUICK_START_OAUTH_PROVIDERS = {"openai_codex"}
_CLEAR_CHOICE = "Clear value"
_QUICK_START_MENU_CHOICE = "[Q] Quick Start"
@@ -1576,7 +1581,11 @@ def _get_quick_start_provider_info() -> dict[str, _QuickStartProviderInfo]:
result: dict[str, _QuickStartProviderInfo] = {}
for spec in PROVIDERS:
if spec.name == "custom" or spec.is_oauth or spec.is_transcription_only:
if (
spec.name == "custom"
or spec.is_transcription_only
or (spec.is_oauth and spec.name not in _QUICK_START_OAUTH_PROVIDERS)
):
continue
result[spec.name] = _QuickStartProviderInfo(
display_name=spec.display_name or spec.name,
@@ -1584,6 +1593,8 @@ def _get_quick_start_provider_info() -> dict[str, _QuickStartProviderInfo]:
default_api_base=spec.default_api_base,
backend=spec.backend,
is_direct=spec.is_direct,
is_oauth=spec.is_oauth,
default_model=spec.builtin_models[0].id if spec.builtin_models else "",
)
return result
@@ -1599,7 +1610,71 @@ def _get_quick_start_provider_choices() -> dict[str, str]:
def _quick_start_requires_api_key(provider_name: str, info: _QuickStartProviderInfo | None) -> bool:
"""Return whether Quick Start should ask for an API key."""
return provider_name == "custom" or not (info and info.is_local)
return provider_name == "custom" or not (info and (info.is_local or info.is_oauth))
def _quick_start_codex_proxy(config: Config) -> str | None:
"""Resolve only the Codex proxy without validating unrelated provider secrets."""
proxy_config = Config()
proxy_config.providers.openai_codex.proxy = config.providers.openai_codex.proxy
return resolve_config_env_vars(proxy_config).providers.openai_codex.proxy or None
def _quick_start_oauth_login(config: Config, provider_name: str) -> bool:
"""Authenticate an OAuth provider supported by Quick Start."""
if provider_name != "openai_codex":
console.print(f"[red]OAuth login is not supported for {provider_name}[/red]")
return False
try:
from oauth_cli_kit import get_token, login_oauth_interactive
except ImportError:
console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
return False
try:
proxy = _quick_start_codex_proxy(config)
except ValueError as exc:
console.print(f"[red]{escape(str(exc))}[/red]")
return False
token = None
with suppress(Exception):
token = get_token(proxy=proxy)
if not getattr(token, "access", None):
console.print("[cyan]Starting interactive OAuth login...[/cyan]\n")
try:
token = login_oauth_interactive(
print_fn=lambda message: console.print(message, markup=False),
prompt_fn=lambda prompt: _get_questionary().text(prompt).ask() or "",
proxy=proxy,
)
except Exception as exc:
console.print(f"[red]OAuth login failed: {escape(str(exc))}[/red]")
return False
if not getattr(token, "access", None):
console.print("[red]OAuth login failed[/red]")
return False
account = getattr(token, "account_id", None)
suffix = f" [dim]{escape(str(account))}[/dim]" if account else ""
console.print(f"[green]Authenticated with OpenAI Codex[/green]{suffix}")
return True
def _quick_start_oauth_is_authenticated(config: Config, provider_name: str) -> bool:
"""Return whether Quick Start can load a usable OAuth token."""
if provider_name != "openai_codex":
return False
try:
from oauth_cli_kit import get_token
proxy = _quick_start_codex_proxy(config)
token = get_token(proxy=proxy)
except Exception:
return False
return bool(getattr(token, "access", None))
def _quick_start_requires_base_url(provider_name: str, info: _QuickStartProviderInfo | None) -> bool:
@@ -1710,7 +1785,11 @@ def _configure_quick_start_provider(config: Config) -> bool | object:
console.print(f"[red]Unknown provider: {provider_name}[/red]")
return False
model = _input_model_with_autocomplete("Model ID", "", provider_name)
model = _input_model_with_autocomplete(
"Model ID",
provider_info.default_model if provider_info else "",
provider_name,
)
if model is _BACK_PRESSED:
continue
model = (model or "").strip()
@@ -1718,6 +1797,10 @@ def _configure_quick_start_provider(config: Config) -> bool | object:
console.print("[yellow]! Model ID is required for Quick Start[/yellow]")
return False
if provider_info and provider_info.is_oauth:
if not _quick_start_oauth_login(config, provider_name):
return False
if api_key is not None:
provider_config.api_key = api_key
if api_base:
@@ -1784,17 +1867,27 @@ def _show_quick_start_summary(config: Config) -> None:
_show_quick_start_progress(3)
preset = config.model_presets.get("primary")
provider_label = "AI provider"
has_api_key = True
credentials_ready = True
credential_name = "API key"
if preset:
provider_config = getattr(config.providers, preset.provider, None)
provider_label, _is_gateway, is_local, _api_base = _get_provider_info().get(
preset.provider, (preset.provider, False, False, "")
)
has_api_key = is_local or bool(provider_config and provider_config.api_key)
provider_info = _get_quick_start_provider_info().get(preset.provider)
if provider_info:
provider_label = provider_info.display_name
if provider_info.is_oauth:
credential_name = "OAuth login"
credentials_ready = _quick_start_oauth_is_authenticated(config, preset.provider)
else:
credentials_ready = provider_info.is_local or bool(
provider_config and provider_config.api_key
)
else:
provider_label = _get_provider_names().get(preset.provider, preset.provider)
credentials_ready = bool(provider_config and provider_config.api_key)
status = "Ready"
if not has_api_key:
status = f"{provider_label} API key missing"
if not credentials_ready:
status = f"{provider_label} {credential_name} missing"
rows = [
("Status", status),
+13 -13
View File
@@ -404,16 +404,14 @@ async def cmd_dream(ctx: CommandContext) -> OutboundMessage:
msg = ctx.msg
async def _run_dream():
async def _silent(*_args, **_kwargs):
pass
from nanobot.agent.memory import MemoryStore
from nanobot.agent.memory import DreamRunProgress, MemoryStore
dream_session_key = MemoryStore.dream_session_key
build_dream_commit_message = MemoryStore.build_dream_commit_message
prune_dream_sessions = MemoryStore.prune_dream_sessions
store = loop.context.memory
progress = DreamRunProgress()
content = ""
resp = None
diff_body = ""
@@ -434,20 +432,22 @@ async def cmd_dream(ctx: CommandContext) -> OutboundMessage:
session_key=key,
ephemeral=True,
tools=store.build_dream_tools(),
on_progress=_silent,
on_progress=progress,
)
elapsed = time.monotonic() - t0
# Ground truth: the real file delta, not the LLM's self-report.
# The real file delta grounds the audit record; clean completion
# decides whether this history batch has finished processing.
diff_body = store.dream_content_diff()
productive = bool(diff_body) or (
not store.git.is_initialized()
and MemoryStore.dream_run_completed(resp)
completed = MemoryStore.dream_run_completed(
resp,
had_tool_errors=progress.had_tool_errors,
)
if productive:
if completed:
store.set_last_dream_cursor(last_cursor)
content = f"Dream completed in {elapsed:.1f}s."
elif MemoryStore.dream_run_completed(resp):
content = f"Dream completed in {elapsed:.1f}s; no memory changes."
if diff_body:
content = f"Dream completed in {elapsed:.1f}s."
else:
content = f"Dream completed in {elapsed:.1f}s; no memory changes."
else:
content = (
f"Dream did not complete after {elapsed:.1f}s; "
+44 -3
View File
@@ -64,16 +64,57 @@ class CommandRouter:
self._priority: dict[str, Handler] = {}
self._exact: dict[str, Handler] = {}
self._prefix: list[tuple[str, Handler]] = []
self._owners: dict[tuple[str, str], str] = {}
def priority(self, cmd: str, handler: Handler) -> None:
def priority(
self,
cmd: str,
handler: Handler,
*,
owner: str = "nanobot.core",
) -> None:
self._priority[cmd] = handler
self._owners[("priority", cmd)] = owner
def exact(self, cmd: str, handler: Handler) -> None:
def exact(
self,
cmd: str,
handler: Handler,
*,
owner: str = "nanobot.core",
) -> None:
self._exact[cmd] = handler
self._owners[("exact", cmd)] = owner
def prefix(self, pfx: str, handler: Handler) -> None:
def prefix(
self,
pfx: str,
handler: Handler,
*,
owner: str = "nanobot.core",
) -> None:
self._prefix.append((pfx, handler))
self._prefix.sort(key=lambda p: len(p[0]), reverse=True)
self._owners[("prefix", pfx)] = owner
def owner(self, tier: str, command: str) -> str | None:
"""Return the extension that owns one command registration."""
return self._owners.get((tier, command))
def unregister_owner(self, owner: str) -> None:
"""Remove all command tiers registered by one extension."""
for (tier, command), registered_owner in list(self._owners.items()):
if registered_owner != owner:
continue
if tier == "priority":
self._priority.pop(command, None)
elif tier == "exact":
self._exact.pop(command, None)
else:
self._prefix = [
item for item in self._prefix if item[0] != command
]
self._owners.pop((tier, command), None)
def is_priority(self, text: str) -> bool:
return normalize_command_text(text).lower() in self._priority
-18
View File
@@ -7,7 +7,6 @@ from pathlib import Path
from typing import Any
import pydantic
from loguru import logger
from pydantic import BaseModel
from nanobot.config.schema import Config, _resolve_tool_config_refs
@@ -200,23 +199,6 @@ def _env_replace(match: re.Match[str]) -> str:
def _migrate_config(data: dict) -> dict:
"""Migrate old config formats to current."""
agents = data.get("agents", {})
defaults = agents.get("defaults", {}) if isinstance(agents, dict) else {}
if isinstance(defaults, dict):
had_legacy_max_messages = (
"maxMessages" in defaults or "max_messages" in defaults
)
defaults.pop("maxMessages", None)
defaults.pop("max_messages", None)
if had_legacy_max_messages:
# TODO(v0.3.1): Remove this legacy cleanup branch. v0.3.0 is the
# final release that warns before the schema silently ignores the field.
logger.warning(
"agents.defaults.maxMessages/max_messages is legacy and ignored; "
"replay max messages is now an internal safety cap. Remove it from "
"config. This compatibility warning will be removed in the next version."
)
# Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace
tools = data.get("tools", {})
exec_cfg = tools.get("exec", {})
+14 -6
View File
@@ -30,7 +30,7 @@ class ChannelsConfig(Base):
model_config = ConfigDict(extra="allow")
send_progress: bool = True # stream agent's text progress to the channel
send_tool_hints: bool = False # stream tool-call hints (e.g. read_file("…"))
send_tool_hints: bool = True # stream tool-call hints (e.g. read_file("…"))
show_reasoning: bool = True # surface model reasoning when channel implements it
extract_document_text: bool = True # extract text from document attachments before sending to the model
send_max_retries: int = Field(default=3, ge=0, le=10) # Max delivery attempts (initial send included)
@@ -64,9 +64,6 @@ class DreamConfig(Base):
default=None,
validation_alias=AliasChoices("modelOverride", "model", "model_override"),
) # Override model for Dream sessions (pending implementation)
max_batch_size: int = Field(default=20, ge=1) # Deprecated: no longer used
max_iterations: int = Field(default=15, ge=1) # Deprecated: no longer used
annotate_line_ages: bool = True # Deprecated: no longer used
def build_schedule(self, timezone: str) -> CronSchedule:
"""Build the runtime schedule, preferring the legacy cron override if present."""
@@ -154,6 +151,10 @@ class AgentDefaults(Base):
validation_alias=AliasChoices("idleCompactAfterMinutes", "sessionTtlMinutes"),
serialization_alias="idleCompactAfterMinutes",
) # Auto-compact idle threshold in minutes (0 = disabled)
idle_compact_check_interval_seconds: int = Field(
default=60,
ge=0,
) # Minimum interval in seconds between scans for idle sessions
consolidation_ratio: float = Field(
default=0.5,
ge=0.1,
@@ -195,7 +196,7 @@ class ProviderConfig(Base):
extra_headers: dict[str, str] | None = None # Custom headers (e.g. APP-Code for AiHubMix)
extra_body: dict[str, Any] | None = None # Extra provider request fields; shape depends on provider/API surface
extra_query: dict[str, str] | None = None # Extra query params (e.g. api-version for Azure-style gateways)
proxy: str | None = None # OpenAI-compatible/Codex HTTP proxy URL
proxy: str | None = None # Explicit HTTP proxy; image downloads trust its DNS and egress
thinking_style: str | None = None # Thinking/reasoning style for custom providers
# Valid values mirror the keys of _THINKING_STYLE_MAP in
@@ -402,11 +403,17 @@ class ToolsConfig(Base):
"webuiAllowRemotePackageInstall",
"webui_allow_remote_package_install",
),
) # allow non-local WebUI clients to install optional Python packages
) # allow non-local WebUI clients to install optional support and extension packages
mcp_servers: dict[str, MCPServerConfig] = Field(default_factory=dict)
ssrf_whitelist: list[str] = Field(default_factory=list) # CIDR ranges to exempt from SSRF blocking (e.g. ["100.64.0.0/10"] for Tailscale)
class ExtensionsConfig(Base):
"""Global switch for external extension activation."""
enabled: bool = True
class Config(BaseSettings):
"""Root configuration for nanobot."""
@@ -417,6 +424,7 @@ class Config(BaseSettings):
api: ApiConfig = Field(default_factory=ApiConfig)
gateway: GatewayConfig = Field(default_factory=GatewayConfig)
tools: ToolsConfig = Field(default_factory=ToolsConfig)
extensions: ExtensionsConfig = Field(default_factory=ExtensionsConfig)
model_presets: dict[str, ModelPresetConfig] = Field(
default_factory=dict,
validation_alias=AliasChoices("modelPresets", "model_presets"),
+19
View File
@@ -0,0 +1,19 @@
"""Stable author-facing API for native nanobot extensions."""
from nanobot.extensions.manifest import (
EXTENSION_API_VERSION,
DependencyKind,
ExtensionDependency,
ExtensionManifest,
ExtensionPermission,
)
from nanobot.extensions.runtime import PythonExtensionApi
__all__ = [
"EXTENSION_API_VERSION",
"DependencyKind",
"ExtensionDependency",
"ExtensionManifest",
"ExtensionPermission",
"PythonExtensionApi",
]
+64
View File
@@ -0,0 +1,64 @@
"""Discover installed extensions and resolve one activation snapshot."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
from nanobot.extensions.preflight import evaluate_dependencies
from nanobot.extensions.registry import (
ExtensionCandidate,
ExtensionDiagnostic,
ExtensionRegistry,
ExtensionSnapshot,
)
from nanobot.extensions.store import ExtensionStore
if TYPE_CHECKING:
from nanobot.config.schema import Config
@dataclass(frozen=True, slots=True)
class ExtensionCatalog:
"""Discovered candidates plus the active, policy-resolved snapshot."""
candidates: tuple[ExtensionCandidate, ...]
snapshot: ExtensionSnapshot
diagnostics: tuple[ExtensionDiagnostic, ...]
def build_extension_catalog(
config: Config,
*,
user_root: Path | None = None,
) -> ExtensionCatalog:
"""Build the authoritative extension view without executing package code."""
if not config.extensions.enabled:
return ExtensionCatalog((), ExtensionSnapshot((), ()), ())
discovery = ExtensionStore(user_root).discover()
candidates, dependency_diagnostics = evaluate_dependencies(
discovery.candidates
)
registry = ExtensionRegistry()
registry_diagnostics: list[ExtensionDiagnostic] = []
for candidate in candidates:
try:
registry.register(candidate)
except ValueError as exc:
registry_diagnostics.append(
ExtensionDiagnostic(
code="duplicate_installation",
extension_id=candidate.manifest.id,
message=str(exc),
)
)
snapshot = registry.snapshot()
diagnostics = (
discovery.diagnostics
+ dependency_diagnostics
+ tuple(registry_diagnostics)
+ snapshot.diagnostics
)
return ExtensionCatalog(candidates, snapshot, diagnostics)
+56
View File
@@ -0,0 +1,56 @@
"""JSON persistence for extension manifests."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from pydantic import ValidationError
from nanobot.extensions.manifest import ExtensionManifest
MANIFEST_FILENAME = "nanobot.extension.json"
class ManifestFormatError(ValueError):
"""Raised when a manifest cannot be decoded unambiguously."""
def load_manifest(path: Path) -> ExtensionManifest:
"""Read and validate one canonical JSON manifest."""
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise ManifestFormatError(f"cannot read extension manifest {path}: {exc}") from exc
return manifest_from_mapping(data)
def dump_manifest(manifest: ExtensionManifest, path: Path) -> None:
"""Write one canonical JSON manifest."""
path.write_text(
json.dumps(manifest_to_mapping(manifest), ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def manifest_from_mapping(data: object) -> ExtensionManifest:
"""Decode a manifest and reject unknown or invalid fields."""
try:
return ExtensionManifest.model_validate(data)
except ValidationError as exc:
unknown = sorted(
".".join(str(part) for part in error["loc"])
for error in exc.errors()
if error["type"] == "extra_forbidden"
)
if unknown:
raise ManifestFormatError(
f"extension manifest has unknown fields: {', '.join(unknown)}"
) from exc
raise ManifestFormatError(f"invalid extension manifest: {exc}") from exc
def manifest_to_mapping(manifest: ExtensionManifest) -> dict[str, Any]:
"""Return the canonical JSON representation."""
return manifest.model_dump(mode="json", by_alias=True)
+71
View File
@@ -0,0 +1,71 @@
"""Side-effect-free discovery of extension manifests on disk."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from nanobot.extensions.codec import MANIFEST_FILENAME, load_manifest
from nanobot.extensions.registry import (
ExtensionCandidate,
ExtensionDiagnostic,
)
@dataclass(frozen=True, slots=True)
class ExtensionDiscoveryResult:
candidates: tuple[ExtensionCandidate, ...] = ()
diagnostics: tuple[ExtensionDiagnostic, ...] = ()
def discover_manifest_root(
root: Path,
) -> ExtensionDiscoveryResult:
"""Discover direct children containing ``nanobot.extension.json``."""
if not root.exists():
return ExtensionDiscoveryResult()
if not root.is_dir():
return ExtensionDiscoveryResult(
diagnostics=(
ExtensionDiagnostic(
code="invalid_extension_root",
extension_id="",
message=f"extension root is not a directory: {root}",
),
)
)
manifests = []
direct_manifest = root / MANIFEST_FILENAME
if direct_manifest.is_file():
manifests.append(direct_manifest)
manifests.extend(
sorted(
path / MANIFEST_FILENAME
for path in root.iterdir()
if not path.name.startswith(".")
and path.is_dir()
and (path / MANIFEST_FILENAME).is_file()
)
)
candidates: list[ExtensionCandidate] = []
diagnostics: list[ExtensionDiagnostic] = []
for path in manifests:
try:
manifest = load_manifest(path)
candidates.append(
ExtensionCandidate(
manifest=manifest,
location=path.parent.resolve(),
)
)
except Exception as exc:
diagnostics.append(
ExtensionDiagnostic(
code="invalid_manifest",
extension_id=path.parent.name,
message=str(exc),
)
)
return ExtensionDiscoveryResult(tuple(candidates), tuple(diagnostics))
+89
View File
@@ -0,0 +1,89 @@
"""Agent-side lifecycle for first-class extensions."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
from loguru import logger
from nanobot.extensions.catalog import ExtensionCatalog, build_extension_catalog
from nanobot.extensions.registry import ExtensionDiagnostic
from nanobot.extensions.runtime import ActivationResult, ExtensionRuntimeManager
if TYPE_CHECKING:
from nanobot.agent.loop import AgentLoop
from nanobot.config.schema import Config
@dataclass(frozen=True, slots=True)
class ExtensionHostSnapshot:
"""Current discovery and activation result."""
catalog: ExtensionCatalog
activation: ActivationResult
@property
def diagnostics(self) -> tuple[ExtensionDiagnostic, ...]:
return self.catalog.diagnostics + self.activation.diagnostics
class ExtensionHost:
"""Reload external extensions without coupling their lifecycle to AgentLoop."""
def __init__(
self,
agent: AgentLoop,
config_loader: Callable[[], Config],
*,
user_root: Path | None = None,
) -> None:
self._agent = agent
self._config_loader = config_loader
self._user_root = user_root
self._manager: ExtensionRuntimeManager | None = None
self._snapshot: ExtensionHostSnapshot | None = None
self._lock = asyncio.Lock()
@property
def snapshot(self) -> ExtensionHostSnapshot | None:
return self._snapshot
async def reload(self) -> ExtensionHostSnapshot:
async with self._lock:
await self._close_manager()
self._snapshot = None
config = self._config_loader()
catalog = build_extension_catalog(
config,
user_root=self._user_root,
)
manager = ExtensionRuntimeManager(
tools=self._agent.tools,
commands=self._agent.commands,
hook_factories=self._agent._hook_factories,
)
activation = await manager.activate(catalog.snapshot)
self._manager = manager
self._snapshot = ExtensionHostSnapshot(catalog, activation)
for diagnostic in self._snapshot.diagnostics:
logger.warning(
"Extension {} [{}]: {}",
diagnostic.extension_id,
diagnostic.code,
diagnostic.message,
)
return self._snapshot
async def close(self) -> None:
async with self._lock:
await self._close_manager()
self._snapshot = None
async def _close_manager(self) -> None:
if self._manager is not None:
await self._manager.close()
self._manager = None
+126
View File
@@ -0,0 +1,126 @@
"""Strict schema for native nanobot extension packages."""
from __future__ import annotations
import re
from enum import Enum
from pathlib import Path
from typing import Literal, Self
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
EXTENSION_API_VERSION = 1
_IDENTIFIER = re.compile(r"[a-z0-9](?:[a-z0-9._-]*[a-z0-9])?")
_PERMISSION = re.compile(r"[a-z][a-z0-9]*(?:[._:-][a-z0-9]+)*")
class DependencyKind(str, Enum):
"""Kinds of prerequisites resolved before activation."""
PYTHON = "python"
EXECUTABLE = "executable"
ENVIRONMENT = "environment"
class _ManifestModel(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True)
class ExtensionDependency(_ManifestModel):
"""One activation prerequisite declared by an extension."""
kind: DependencyKind
name: str
specifier: str = ""
optional: bool = False
@field_validator("name")
@classmethod
def validate_name(cls, value: str) -> str:
return _require_text(value, "extension dependency name")
class ExtensionPermission(_ManifestModel):
"""A privileged host capability requested by an extension."""
name: str
reason: str = ""
@field_validator("name")
@classmethod
def validate_name(cls, value: str) -> str:
if _PERMISSION.fullmatch(value) is None:
raise ValueError(
"extension permission must be a lowercase namespaced identifier"
)
return value
class ExtensionManifest(_ManifestModel):
"""Identity, prerequisites, and consent declarations for one extension."""
id: str
name: str
version: str
entry: str = "extension:register"
description: str = ""
dependencies: tuple[ExtensionDependency, ...] = ()
permissions: tuple[ExtensionPermission, ...] = ()
api_version: Literal[EXTENSION_API_VERSION] = Field(
default=EXTENSION_API_VERSION,
alias="apiVersion",
)
homepage: str = ""
license: str = ""
@field_validator("id")
@classmethod
def validate_id(cls, value: str) -> str:
return _require_identifier(value, "extension id")
@field_validator("name", "version")
@classmethod
def validate_required_text(cls, value: str, info) -> str:
return _require_text(value, f"extension {info.field_name}")
@field_validator("entry")
@classmethod
def validate_entry(cls, value: str) -> str:
value = _require_text(value, "extension entry")
module_name = value.partition(":")[0]
if Path(module_name).is_absolute() or ".." in Path(module_name).parts:
raise ValueError("extension entry cannot escape the package root")
return value
@model_validator(mode="after")
def reject_duplicates(self) -> Self:
dependencies = [(item.kind, item.name) for item in self.dependencies]
if len(set(dependencies)) != len(dependencies):
raise ValueError("extension manifest contains duplicate dependencies")
permissions = [item.name for item in self.permissions]
if len(set(permissions)) != len(permissions):
raise ValueError("extension manifest contains duplicate permissions")
return self
def _require_text(value: str, label: str) -> str:
if not value.strip():
raise ValueError(f"{label} must be a non-empty string")
return value
def _require_identifier(value: str, label: str) -> str:
value = _require_text(value, label)
if _IDENTIFIER.fullmatch(value) is None:
raise ValueError(
f"{label} must use lowercase letters, digits, dots, underscores, or hyphens"
)
return value
def validate_extension_id(value: object) -> str:
"""Validate and return one portable extension identifier."""
if not isinstance(value, str):
raise ValueError("extension id must be a string")
return _require_identifier(value, "extension id")
+59
View File
@@ -0,0 +1,59 @@
"""Activation preflight for extension runtime prerequisites."""
from __future__ import annotations
import importlib.metadata
import os
import shutil
from dataclasses import replace
from nanobot.extensions.manifest import DependencyKind, ExtensionDependency
from nanobot.extensions.registry import ExtensionCandidate, ExtensionDiagnostic
from nanobot.extensions.versioning import dependency_version_failure
def evaluate_dependencies(
candidates: tuple[ExtensionCandidate, ...],
) -> tuple[tuple[ExtensionCandidate, ...], tuple[ExtensionDiagnostic, ...]]:
"""Disable candidates with missing required software and explain why."""
checked: list[ExtensionCandidate] = []
diagnostics: list[ExtensionDiagnostic] = []
for candidate in candidates:
failures = [
message
for dependency in candidate.manifest.dependencies
if not dependency.optional
if (message := _dependency_failure(dependency))
]
if failures:
candidate = replace(candidate, enabled=False)
diagnostics.extend(
ExtensionDiagnostic(
code="dependency_missing",
extension_id=candidate.manifest.id,
message=message,
)
for message in failures
)
checked.append(candidate)
return tuple(checked), tuple(diagnostics)
def _dependency_failure(
dependency: ExtensionDependency,
) -> str:
if dependency.kind is DependencyKind.EXECUTABLE:
if shutil.which(dependency.name) is None:
return f"Required executable is not installed: {dependency.name}"
return ""
if dependency.kind is DependencyKind.ENVIRONMENT:
if not os.getenv(dependency.name):
return f"Required environment variable is not set: {dependency.name}"
return ""
if dependency.kind is DependencyKind.PYTHON:
try:
version = importlib.metadata.version(dependency.name)
except importlib.metadata.PackageNotFoundError:
return f"Required Python package is not installed: {dependency.name}"
return dependency_version_failure(dependency, version, "Python package")
return f"Unsupported dependency kind: {dependency.kind.value}"
+82
View File
@@ -0,0 +1,82 @@
"""Deterministic extension activation planning."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from nanobot.extensions.manifest import ExtensionManifest
@dataclass(frozen=True, slots=True)
class ExtensionCandidate:
"""One discovered extension package and its activation state."""
manifest: ExtensionManifest
location: Path | None = None
enabled: bool = True
trusted: bool = False
integrity_valid: bool = True
granted_permissions: frozenset[str] = frozenset()
@dataclass(frozen=True, slots=True)
class ExtensionDiagnostic:
"""A non-fatal discovery or activation problem."""
code: str
extension_id: str
message: str
severity: str = "warning"
@dataclass(frozen=True, slots=True)
class ExtensionSnapshot:
"""Immutable activation plan consumed by the runtime."""
extensions: tuple[ExtensionCandidate, ...]
diagnostics: tuple[ExtensionDiagnostic, ...]
class ExtensionRegistry:
"""Select trusted candidates and report missing permission grants."""
def __init__(self) -> None:
self._candidates: dict[str, ExtensionCandidate] = {}
def register(self, candidate: ExtensionCandidate) -> None:
extension_id = candidate.manifest.id
if extension_id in self._candidates:
raise ValueError(f"extension '{extension_id}' is already installed")
self._candidates[extension_id] = candidate
def snapshot(self) -> ExtensionSnapshot:
active: list[ExtensionCandidate] = []
diagnostics: list[ExtensionDiagnostic] = []
for candidate in sorted(
self._candidates.values(),
key=lambda item: item.manifest.id,
):
requested = {
permission.name for permission in candidate.manifest.permissions
}
missing = sorted(requested - candidate.granted_permissions)
if (
candidate.enabled
and candidate.integrity_valid
and candidate.trusted
and not missing
):
active.append(candidate)
elif candidate.enabled and candidate.trusted and missing:
diagnostics.append(
ExtensionDiagnostic(
code="permission_required",
extension_id=candidate.manifest.id,
message=(
"Grant required extension permissions: "
+ ", ".join(missing)
),
)
)
return ExtensionSnapshot(tuple(active), tuple(diagnostics))
+218
View File
@@ -0,0 +1,218 @@
"""Transactional activation at nanobot's tool, command, and hook seams."""
from __future__ import annotations
import importlib
import shutil
import sys
from dataclasses import dataclass
from importlib.machinery import ModuleSpec
from pathlib import Path
from types import ModuleType
from typing import Any
from nanobot.agent.hook import AgentHook, AgentTurnHookFactory
from nanobot.agent.tools.base import Tool
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.command.router import CommandRouter, Handler
from nanobot.extensions.registry import (
ExtensionCandidate,
ExtensionDiagnostic,
ExtensionSnapshot,
)
@dataclass(frozen=True, slots=True)
class ActivationResult:
"""Immutable activation outcome consumed by the agent assembly layer."""
extensions: tuple[ExtensionCandidate, ...]
hook_factories: tuple[AgentTurnHookFactory, ...]
diagnostics: tuple[ExtensionDiagnostic, ...]
class PythonExtensionApi:
"""Small native API; extensions register into existing nanobot interfaces."""
def __init__(
self,
*,
owner: str,
tools: ToolRegistry,
commands: CommandRouter,
hook_factories: list[AgentTurnHookFactory],
) -> None:
self.owner = owner
self._tools = tools
self._commands = commands
self._hook_factories = hook_factories
def register_tool(self, tool: Tool) -> None:
if not self._tools.register_if_absent(tool, owner=self.owner):
existing = self._tools.owner(tool.name) or "unknown"
raise ValueError(
f"tool '{tool.name}' is already registered by '{existing}'"
)
def register_command(
self,
command: str,
handler: Handler,
*,
prefix: bool = False,
) -> None:
command = f"/{command.lstrip('/')}"
if prefix:
command = f"{command} "
register = self._commands.prefix if prefix else self._commands.exact
tier = "prefix" if prefix else "exact"
if existing := self._commands.owner(tier, command):
raise ValueError(
f"command '{command}' is already registered by '{existing}'"
)
register(command, handler, owner=self.owner)
def register_hook_factory(self, factory: AgentTurnHookFactory) -> None:
self._hook_factories.append(_owned_hook_factory(factory, self.owner))
class ExtensionRuntimeManager:
"""Activate a resolved snapshot and roll back failed registrations."""
def __init__(
self,
*,
tools: ToolRegistry,
commands: CommandRouter,
hook_factories: list[AgentTurnHookFactory] | None = None,
) -> None:
self._tools = tools
self._commands = commands
self._active: list[ExtensionCandidate] = []
self._hook_factories = hook_factories if hook_factories is not None else []
async def activate(self, snapshot: ExtensionSnapshot) -> ActivationResult:
diagnostics: list[ExtensionDiagnostic] = []
for candidate in snapshot.extensions:
try:
active = self._activate_candidate(candidate)
self._active.append(active)
except Exception as exc:
self._rollback_owner(candidate.manifest.id)
diagnostics.append(
ExtensionDiagnostic(
code="activation_failed",
extension_id=candidate.manifest.id,
message=str(exc),
)
)
return ActivationResult(
tuple(self._active),
tuple(self._hook_factories),
tuple(diagnostics),
)
async def close(self) -> None:
for active in reversed(self._active):
self._rollback_owner(active.manifest.id)
_unload_extension_modules(active)
self._active.clear()
def _activate_candidate(
self,
candidate: ExtensionCandidate,
) -> ExtensionCandidate:
self._activate_python(candidate)
return candidate
def _activate_python(self, candidate: ExtensionCandidate) -> None:
raw_entry = candidate.manifest.entry
module_name, separator, attribute = raw_entry.partition(":")
if not separator:
module_name = raw_entry
attribute = "register"
assert candidate.location is not None
importlib.invalidate_caches()
module_prefix = _module_prefix(candidate.manifest.id)
_unload_extension_modules(candidate)
package = ModuleType(module_prefix)
package.__package__ = module_prefix
package.__path__ = [str(candidate.location)]
package.__spec__ = ModuleSpec(module_prefix, loader=None, is_package=True)
sys.modules[module_prefix] = package
try:
module = importlib.import_module(f"{module_prefix}.{module_name}")
module_path = getattr(module, "__file__", None)
if not module_path or not Path(module_path).resolve().is_relative_to(
candidate.location.resolve()
):
raise ValueError(
f"Python extension entry resolves outside its package: {module_name}"
)
register = getattr(module, attribute)
api = PythonExtensionApi(
owner=candidate.manifest.id,
tools=self._tools,
commands=self._commands,
hook_factories=self._hook_factories,
)
result = register(api)
if result is not None:
raise TypeError("Python extension register function must return None")
except Exception:
_unload_extension_modules(candidate)
raise
def _rollback_owner(
self,
owner: str,
) -> None:
self._tools.unregister_owner(owner)
self._commands.unregister_owner(owner)
self._hook_factories[:] = [
factory
for factory in self._hook_factories
if getattr(factory, "__nanobot_extension_owner__", None) != owner
]
def _owned_hook_factory(
factory: AgentTurnHookFactory,
owner: str,
) -> AgentTurnHookFactory:
def owned(context: Any) -> AgentHook | None:
return factory(context)
setattr(owned, "__nanobot_extension_owner__", owner)
return owned
def _modules_under(root: Path) -> tuple[str, ...]:
package_root = root.resolve()
return tuple(
name
for name, module in tuple(sys.modules.items())
if (raw_path := getattr(module, "__file__", None))
and Path(raw_path).resolve().is_relative_to(package_root)
)
def _unload_modules_under(root: Path) -> None:
package_root = root.resolve()
for module_name in _modules_under(package_root):
sys.modules.pop(module_name, None)
for cache in package_root.rglob("__pycache__"):
shutil.rmtree(cache, ignore_errors=True)
def _module_prefix(extension_id: str) -> str:
return "_nanobot_extension_" + extension_id.encode().hex()
def _unload_extension_modules(candidate: ExtensionCandidate) -> None:
assert candidate.location is not None
prefix = _module_prefix(candidate.manifest.id)
for name in tuple(sys.modules):
if name == prefix or name.startswith(f"{prefix}."):
sys.modules.pop(name, None)
_unload_modules_under(candidate.location)
+167
View File
@@ -0,0 +1,167 @@
"""Transport-neutral extension management service."""
from __future__ import annotations
import asyncio
from dataclasses import asdict
from pathlib import Path
from typing import Any
from nanobot.extensions.host import ExtensionHost
from nanobot.extensions.manifest import ExtensionManifest
from nanobot.extensions.registry import ExtensionCandidate
from nanobot.extensions.store import ExtensionStore, InstalledExtension
class ExtensionService:
"""One management boundary shared by CLI and WebUI."""
def __init__(
self,
*,
host: ExtensionHost | None = None,
store: ExtensionStore | None = None,
) -> None:
self.host = host
self.store = store or ExtensionStore()
self._mutation_lock = asyncio.Lock()
async def status(self) -> dict[str, Any]:
snapshot = self.host.snapshot if self.host else None
catalog = snapshot.catalog if snapshot else None
if catalog is None:
discovery = self.store.discover()
candidates = discovery.candidates
diagnostics = discovery.diagnostics
active_ids: set[str] = set()
else:
candidates = catalog.candidates
active_ids = {
active.manifest.id
for active in snapshot.activation.extensions
}
diagnostics = catalog.diagnostics + snapshot.activation.diagnostics
records = self.store.records()
return {
"extensions": [
_candidate_payload(candidate, active_ids, records.get(candidate.manifest.id))
for candidate in sorted(
candidates,
key=lambda item: item.manifest.name.lower(),
)
],
"diagnostics": [asdict(item) for item in diagnostics],
}
async def install(
self,
source: str,
*,
kind: str = "git",
ref: str = "",
trusted: bool = False,
) -> dict[str, Any]:
async with self._mutation_lock:
if kind == "git":
result = await asyncio.to_thread(
self.store.install_git,
source,
ref=ref,
trusted=trusted,
)
elif kind == "local":
result = await asyncio.to_thread(
self.store.install_local,
Path(source),
trusted=trusted,
)
else:
raise ValueError(f"unknown extension source kind: {kind}")
await self._reload()
return {
"record": _record_payload(result.record),
"manifest": _manifest_payload(result.manifest),
}
async def set_enabled(self, extension_id: str, enabled: bool) -> dict[str, Any]:
return await self._update(extension_id, self.store.set_enabled, enabled)
async def set_trusted(self, extension_id: str, trusted: bool) -> dict[str, Any]:
return await self._update(extension_id, self.store.set_trusted, trusted)
async def set_permissions(
self,
extension_id: str,
permissions: set[str] | frozenset[str],
) -> dict[str, Any]:
return await self._update(
extension_id,
self.store.set_permissions,
permissions,
)
async def uninstall(self, extension_id: str) -> dict[str, Any]:
async with self._mutation_lock:
await asyncio.to_thread(self.store.uninstall, extension_id)
await self._reload()
return {"removed": extension_id}
async def _update(self, extension_id: str, action: Any, value: Any) -> dict[str, Any]:
async with self._mutation_lock:
record = await asyncio.to_thread(action, extension_id, value)
await self._reload()
return {"record": _record_payload(record)}
async def _reload(self) -> None:
if self.host is not None:
await self.host.reload()
def _candidate_payload(
candidate: ExtensionCandidate,
active_ids: set[str],
record: InstalledExtension | None,
) -> dict[str, Any]:
manifest = candidate.manifest
requested = [permission.name for permission in manifest.permissions]
return {
**_manifest_payload(manifest),
"location": str(candidate.location) if candidate.location else None,
"enabled": candidate.enabled,
"trusted": candidate.trusted,
"active": manifest.id in active_ids,
"requested_permissions": requested,
"granted_permissions": sorted(candidate.granted_permissions),
"source": record.source.value if record else "path",
"source_ref": record.source_ref if record else "",
"integrity": record.integrity if record else "",
"installed_at": record.installed_at if record else "",
}
def _manifest_payload(manifest: ExtensionManifest) -> dict[str, Any]:
return {
"id": manifest.id,
"name": manifest.name,
"version": manifest.version,
"description": manifest.description,
"homepage": manifest.homepage,
"license": manifest.license,
"dependencies": [
{
"kind": dependency.kind.value,
"name": dependency.name,
"specifier": dependency.specifier,
"optional": dependency.optional,
}
for dependency in manifest.dependencies
],
"permissions": [
{"name": permission.name, "reason": permission.reason}
for permission in manifest.permissions
],
}
def _record_payload(record: InstalledExtension) -> dict[str, Any]:
return record.model_dump(mode="json")
+511
View File
@@ -0,0 +1,511 @@
"""Atomic installation store and trust state for external extensions."""
from __future__ import annotations
import hashlib
import json
import os
import re
import shutil
import subprocess
import tempfile
from dataclasses import dataclass, replace
from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from uuid import uuid4
from filelock import FileLock
from pydantic import BaseModel, ConfigDict, field_validator
from nanobot.extensions.codec import MANIFEST_FILENAME, load_manifest
from nanobot.extensions.discovery import (
ExtensionDiscoveryResult,
discover_manifest_root,
)
from nanobot.extensions.manifest import ExtensionManifest, validate_extension_id
from nanobot.extensions.registry import ExtensionDiagnostic
_REGISTRY_FILENAME = ".registry.json"
_GIT_SCHEMES = frozenset({"git", "http", "https", "ssh"})
_SCP_GIT_URL = re.compile(
r"(?:[A-Za-z0-9._-]+@)?[A-Za-z0-9](?:[A-Za-z0-9.-]*[A-Za-z0-9])?:\S+"
)
_SHA256_INTEGRITY = re.compile(r"sha256:[0-9a-f]{64}")
class ExtensionSourceKind(str, Enum):
LOCAL = "local"
GIT = "git"
class InstalledExtension(BaseModel):
"""Persistent installation and policy record."""
model_config = ConfigDict(extra="forbid", frozen=True)
id: str
version: str
source: ExtensionSourceKind
source_ref: str
integrity: str
installed_at: str
enabled: bool = True
trusted: bool = False
granted_permissions: tuple[str, ...] = ()
@field_validator("id")
@classmethod
def validate_id(cls, value: str) -> str:
return validate_extension_id(value)
@field_validator("version", "source_ref", "installed_at")
@classmethod
def validate_metadata(cls, value: str) -> str:
if not value:
raise ValueError("extension registry metadata must use non-empty strings")
return value
@field_validator("integrity")
@classmethod
def validate_integrity(cls, value: str) -> str:
if _SHA256_INTEGRITY.fullmatch(value) is None:
raise ValueError("extension registry integrity must be a sha256 digest")
return value
@field_validator("granted_permissions")
@classmethod
def reject_duplicate_permissions(cls, value: tuple[str, ...]) -> tuple[str, ...]:
if len(set(value)) != len(value):
raise ValueError("extension granted permissions cannot contain duplicates")
return value
@dataclass(frozen=True, slots=True)
class InstallResult:
"""Installed package metadata."""
record: InstalledExtension
manifest: ExtensionManifest
class ExtensionStore:
"""Own the user extension directory and its atomic registry."""
def __init__(self, root: Path | None = None) -> None:
self.root = (root or Path.home() / ".nanobot" / "extensions").expanduser()
self.root.mkdir(parents=True, exist_ok=True)
self.registry_path = self.root / _REGISTRY_FILENAME
self._lock = FileLock(str(self.root / ".lock"))
def records(self, *, strict: bool = False) -> dict[str, InstalledExtension]:
if not self.registry_path.is_file():
return {}
try:
data = json.loads(self.registry_path.read_text(encoding="utf-8"))
if not isinstance(data, dict) or data.get("version") != 1:
raise ValueError("extension registry must be a version 1 object")
rows = data.get("extensions")
if not isinstance(rows, list):
raise ValueError("extension registry extensions must be an array")
records: dict[str, InstalledExtension] = {}
for item in rows:
record = InstalledExtension.model_validate(item)
if record.id in records:
raise ValueError(
f"extension registry contains duplicate id: {record.id}"
)
records[record.id] = record
return records
except (
OSError,
UnicodeError,
json.JSONDecodeError,
KeyError,
ValueError,
) as exc:
if strict:
raise ValueError(
f"invalid extension registry {self.registry_path}: {exc}"
) from exc
return {}
def discover(self) -> ExtensionDiscoveryResult:
"""Discover packages and apply persisted enable/trust state."""
result = discover_manifest_root(self.root)
diagnostics = list(result.diagnostics)
try:
records = self.records(strict=True)
except ValueError as exc:
records = {}
diagnostics.append(
ExtensionDiagnostic(
code="invalid_extension_registry",
extension_id="",
message=str(exc),
)
)
candidates = []
for candidate in result.candidates:
record = records.get(candidate.manifest.id)
trusted = record.trusted if record else False
integrity_valid = True
if candidate.location is not None and record is not None:
try:
_reject_unsafe_files(candidate.location)
actual_integrity = _tree_hash(candidate.location)
except (OSError, ValueError) as exc:
actual_integrity = ""
diagnostics.append(
ExtensionDiagnostic(
code="extension_integrity_error",
extension_id=candidate.manifest.id,
message=f"Could not verify installed package: {exc}",
)
)
if actual_integrity != record.integrity:
trusted = False
integrity_valid = False
diagnostics.append(
ExtensionDiagnostic(
code="extension_integrity_mismatch",
extension_id=candidate.manifest.id,
message=(
"Installed package contents changed after installation; "
"reinstall it before trusting it again"
),
)
)
candidates.append(
replace(
candidate,
enabled=record.enabled if record else True,
trusted=trusted,
integrity_valid=integrity_valid,
granted_permissions=frozenset(
record.granted_permissions if record else ()
),
)
)
return ExtensionDiscoveryResult(tuple(candidates), tuple(diagnostics))
def install_local(
self,
source: Path,
*,
trusted: bool = False,
) -> InstallResult:
return self._install_from_directory(
source.resolve(),
source_kind=ExtensionSourceKind.LOCAL,
source_ref=str(source.resolve()),
trusted=trusted,
)
def install_git(
self,
url: str,
*,
ref: str = "",
trusted: bool = False,
) -> InstallResult:
_validate_git_url(url)
with tempfile.TemporaryDirectory(prefix="nanobot-extension-git-") as raw:
checkout = Path(raw) / "checkout"
if ref:
_run(
[
"git",
"clone",
"--filter=blob:none",
"--no-checkout",
"--",
url,
str(checkout),
]
)
_run(
[
"git",
"-C",
str(checkout),
"fetch",
"--depth",
"1",
"--",
"origin",
ref,
]
)
_run(
[
"git",
"-C",
str(checkout),
"checkout",
"--detach",
"FETCH_HEAD",
]
)
else:
_run(["git", "clone", "--depth", "1", "--", url, str(checkout)])
return self._install_from_directory(
checkout,
source_kind=ExtensionSourceKind.GIT,
source_ref=f"{url}#{ref}" if ref else url,
trusted=trusted,
)
def set_enabled(self, extension_id: str, enabled: bool) -> InstalledExtension:
return self._update_record(extension_id, enabled=enabled)
def set_trusted(self, extension_id: str, trusted: bool) -> InstalledExtension:
return self._update_record(extension_id, trusted=trusted)
def set_permissions(
self,
extension_id: str,
permissions: set[str] | frozenset[str],
) -> InstalledExtension:
with self._lock:
records = self.records(strict=True)
if extension_id not in records:
raise KeyError(f"extension '{extension_id}' is not installed")
manifest = load_manifest(
self.root / extension_id / MANIFEST_FILENAME
)
requested = {
permission.name for permission in manifest.permissions
}
unknown = sorted(set(permissions) - requested)
if unknown:
raise ValueError(
"Cannot grant permissions not requested by the extension: "
+ ", ".join(unknown)
)
return self._update_record_locked(
records,
extension_id,
granted_permissions=tuple(sorted(permissions)),
)
def uninstall(self, extension_id: str) -> None:
with self._lock:
records = self.records(strict=True)
if extension_id not in records:
raise KeyError(f"extension '{extension_id}' is not installed")
target = self.root / extension_id
backup = self.root / f".uninstall-{uuid4().hex}"
if target.exists():
target.rename(backup)
try:
records.pop(extension_id)
self._write_records(records)
except Exception:
if backup.exists():
backup.rename(target)
raise
shutil.rmtree(backup, ignore_errors=True)
def _install_from_directory(
self,
source: Path,
*,
source_kind: ExtensionSourceKind,
source_ref: str,
trusted: bool,
) -> InstallResult:
with self._lock:
return self._install_from_directory_locked(
source,
source_kind=source_kind,
source_ref=source_ref,
trusted=trusted,
)
def _install_from_directory_locked(
self,
source: Path,
*,
source_kind: ExtensionSourceKind,
source_ref: str,
trusted: bool,
) -> InstallResult:
if not source.is_dir():
raise ValueError(f"extension source is not a directory: {source}")
if self.root.resolve().is_relative_to(source.resolve()):
raise ValueError("extension source cannot contain the extension store")
_reject_unsafe_files(source)
manifest = load_manifest(source / MANIFEST_FILENAME)
extension_id = manifest.id
self.root.mkdir(parents=True, exist_ok=True)
staging = self.root / f".install-{uuid4().hex}"
target = self.root / extension_id
backup = self.root / f".backup-{uuid4().hex}"
records = self.records(strict=True)
previous = records.get(extension_id)
backup_created = False
target_installed = False
try:
shutil.copytree(
source,
staging,
ignore=shutil.ignore_patterns(".git", "__pycache__", "*.pyc"),
)
_reject_unsafe_files(staging)
integrity = _tree_hash(staging)
if target.exists():
target.rename(backup)
backup_created = True
staging.rename(target)
target_installed = True
requested_permissions = {
permission.name for permission in manifest.permissions
}
unchanged = bool(previous and previous.integrity == integrity)
record = InstalledExtension(
id=extension_id,
version=manifest.version,
source=source_kind,
source_ref=source_ref,
integrity=integrity,
installed_at=datetime.now(UTC).isoformat(),
enabled=previous.enabled if previous else True,
trusted=trusted or bool(unchanged and previous and previous.trusted),
granted_permissions=(
tuple(
permission
for permission in previous.granted_permissions
if permission in requested_permissions
)
if previous
else ()
),
)
records[extension_id] = record
self._write_records(records)
shutil.rmtree(backup, ignore_errors=True)
return InstallResult(record, manifest)
except Exception:
shutil.rmtree(staging, ignore_errors=True)
if target_installed:
shutil.rmtree(target, ignore_errors=True)
if backup_created:
backup.rename(target)
raise
def _update_record(
self,
extension_id: str,
**changes: Any,
) -> InstalledExtension:
with self._lock:
records = self.records(strict=True)
return self._update_record_locked(records, extension_id, **changes)
def _update_record_locked(
self,
records: dict[str, InstalledExtension],
extension_id: str,
**changes: Any,
) -> InstalledExtension:
try:
record = records[extension_id].model_copy(update=changes)
except KeyError as exc:
raise KeyError(
f"extension '{extension_id}' is not installed"
) from exc
records[extension_id] = record
self._write_records(records)
return record
def _write_records(self, records: dict[str, InstalledExtension]) -> None:
self.root.mkdir(parents=True, exist_ok=True)
payload = {
"version": 1,
"extensions": [
record.model_dump(mode="json")
for record in sorted(records.values(), key=lambda item: item.id)
],
}
temp = self.registry_path.with_suffix(".tmp")
temp.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
os.replace(temp, self.registry_path)
def _run(command: list[str], *, cwd: Path | None = None) -> str:
try:
return subprocess.run(
command,
cwd=cwd,
check=True,
capture_output=True,
text=True,
).stdout
except FileNotFoundError as exc:
raise RuntimeError(f"required executable not found: {command[0]}") from exc
except subprocess.CalledProcessError as exc:
detail = (exc.stderr or exc.stdout or "").strip()
raise RuntimeError(f"{command[0]} failed: {detail}") from exc
def _validate_git_url(url: str) -> None:
if not isinstance(url, str) or not url.strip() or any(
character in url for character in ("\0", "\r", "\n")
):
raise ValueError("extension Git source must be a remote repository URL")
value = url.strip()
parsed = urlparse(value)
if parsed.scheme:
if parsed.scheme.lower() not in _GIT_SCHEMES or not parsed.hostname or not parsed.path:
raise ValueError(
"extension Git source must use git, http, https, or ssh"
)
if parsed.password or (
parsed.scheme.lower() in {"http", "https"} and parsed.username
):
raise ValueError(
"extension Git URLs cannot contain credentials; use a Git credential helper"
)
if parsed.query or parsed.fragment:
raise ValueError(
"extension Git URLs cannot contain query parameters or fragments; "
"pass the revision separately"
)
return
if _SCP_GIT_URL.fullmatch(value) is None or any(
character in value for character in ("?", "#")
):
raise ValueError("extension Git source must be a remote repository URL")
def _reject_unsafe_files(root: Path) -> None:
for path in root.rglob("*"):
if path.is_symlink():
raise ValueError(f"extension packages cannot contain symlinks: {path}")
if not path.is_file() and not path.is_dir():
raise ValueError(f"extension package contains a special file: {path}")
def _tree_hash(root: Path) -> str:
digest = hashlib.sha256()
for path in sorted(
item
for item in root.rglob("*")
if (item.is_file() or item.is_symlink())
and "__pycache__" not in item.parts
and item.suffix not in {".pyc", ".pyo"}
):
digest.update(path.relative_to(root).as_posix().encode())
digest.update(b"\0")
if path.is_symlink():
digest.update(b"link\0")
digest.update(os.fsencode(os.readlink(path)))
continue
digest.update(b"file\0")
with path.open("rb") as handle:
while chunk := handle.read(1024 * 1024):
digest.update(chunk)
return f"sha256:{digest.hexdigest()}"
+29
View File
@@ -0,0 +1,29 @@
"""Version constraint checks shared by extension activation gates."""
from packaging.specifiers import InvalidSpecifier, SpecifierSet
from packaging.version import InvalidVersion, Version
from nanobot.extensions.manifest import ExtensionDependency
def dependency_version_failure(
dependency: ExtensionDependency,
version: str,
label: str,
) -> str:
"""Return a user-facing constraint failure, or an empty string on success."""
if not dependency.specifier:
return ""
try:
matches = Version(version) in SpecifierSet(dependency.specifier)
except (InvalidSpecifier, InvalidVersion):
return (
f"{label} {dependency.name} has an unsupported version constraint: "
f"{dependency.specifier}"
)
if matches:
return ""
return (
f"{label} {dependency.name} {version} does not satisfy "
f"{dependency.specifier}"
)
+20 -1
View File
@@ -11,6 +11,7 @@ from nanobot.agent.hook import AgentHook, SDKCaptureHook
from nanobot.agent.hooks import create_file_edit_activity_hook
from nanobot.agent.loop import AgentLoop
from nanobot.config.schema import Config
from nanobot.extensions.host import ExtensionHost
from nanobot.providers.image_generation import image_gen_provider_configs
from nanobot.sdk.clients import MemoryClient, RuntimeClient, SessionClient
from nanobot.sdk.runtime import (
@@ -77,6 +78,9 @@ class Nanobot:
self.sessions = SessionClient(loop)
self.memory = MemoryClient(loop)
self.runtime = RuntimeClient(loop)
self._extensions = ExtensionHost(loop, lambda: config) if config else None
self._extensions_started = False
self._extensions_lock = asyncio.Lock()
@classmethod
def from_config(
@@ -153,6 +157,7 @@ class Nanobot:
model: Override the model for this run only.
model_preset: Override the model preset for this run only.
"""
await self._ensure_extensions()
capture = SDKCaptureHook()
per_run_hooks = [capture, *(hooks or [])]
runtime = self._loop.runtime_resolver.resolve_override(
@@ -193,6 +198,7 @@ class Nanobot:
model_preset: str | None = None,
) -> RunStream:
"""Start a streamed run and return a handle for events and final result."""
await self._ensure_extensions()
override_runtime = self._loop.runtime_resolver.resolve_override(
model=model,
model_preset=model_preset,
@@ -316,7 +322,20 @@ class Nanobot:
async def aclose(self) -> None:
"""Release resources held by this instance (MCP connections, etc.)."""
await self._loop.close_mcp()
try:
if self._extensions is not None:
await self._extensions.close()
self._extensions_started = False
finally:
await self._loop.close_mcp()
async def _ensure_extensions(self) -> None:
if self._extensions is None or self._extensions_started:
return
async with self._extensions_lock:
if not self._extensions_started:
await self._extensions.reload()
self._extensions_started = True
async def __aenter__(self) -> Nanobot:
return self
+42 -5
View File
@@ -43,9 +43,22 @@ def _load() -> dict[str, Any]:
except (json.JSONDecodeError, OSError):
logger.warning("Corrupted pairing store, resetting")
return {"approved": {}, "pending": {}}
if not isinstance(data, dict):
logger.warning("Corrupted pairing store, resetting")
return {"approved": {}, "pending": {}}
# JSON stores may contain null maps after partial edits; treat like {}.
approved = data.get("approved") or {}
if not isinstance(approved, dict):
approved = {}
data["approved"] = approved
pending = data.get("pending") or {}
if not isinstance(pending, dict):
pending = {}
data["pending"] = pending
# Convert approved lists to str sets for O(1) lookup.
for channel, users in data.get("approved", {}).items():
for channel, users in approved.items():
if not isinstance(users, list):
users = []
data["approved"][channel] = {str(u) for u in users}
@@ -56,9 +69,15 @@ def _save(data: dict[str, Any]) -> None:
path = _store_path()
path.parent.mkdir(parents=True, exist_ok=True)
# Convert sets back to lists for JSON serialization
approved = data.get("approved") or {}
pending = data.get("pending") or {}
if not isinstance(approved, dict):
approved = {}
if not isinstance(pending, dict):
pending = {}
payload = {
"approved": {ch: sorted(list(users)) for ch, users in data.get("approved", {}).items()},
"pending": dict(data.get("pending", {})),
"approved": {ch: sorted(list(users)) for ch, users in approved.items()},
"pending": dict(pending),
}
_write_text_atomic(path, json.dumps(payload, indent=2, ensure_ascii=False))
@@ -66,10 +85,26 @@ def _save(data: dict[str, Any]) -> None:
def _gc_pending(data: dict[str, Any]) -> None:
"""Remove expired pending entries in-place."""
now = time.time()
pending: dict[str, Any] = data.get("pending", {})
expired = [code for code, info in pending.items() if info.get("expires_at", 0) < now]
pending: dict[str, Any] = data.get("pending") or {}
if not isinstance(pending, dict):
data["pending"] = {}
return
expired = [
code
for code, info in pending.items()
if (
not isinstance(info, dict)
or not isinstance(info.get("channel"), str)
or not info.get("channel")
or info.get("sender_id") is None
or isinstance(info.get("expires_at"), bool)
or not isinstance(info.get("expires_at"), (int, float))
or info["expires_at"] < now
)
]
for code in expired:
del pending[code]
data["pending"] = pending
def generate_code(
@@ -152,6 +187,7 @@ def list_pending() -> list[dict[str, Any]]:
return [
{"code": code, **info}
for code, info in data.get("pending", {}).items()
if isinstance(info, dict)
]
@@ -195,6 +231,7 @@ def clear_channel(channel: str) -> dict[str, int]:
"""Remove approved senders and pending requests for *channel*."""
with _LOCK:
data = _load()
_gc_pending(data)
approved: dict[str, set[str]] = data.get("approved", {})
approved_users = approved.pop(channel, set())
+196 -50
View File
@@ -10,15 +10,27 @@ from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urljoin
import httpx
from loguru import logger
from nanobot.providers.openrouter_attribution import OPENROUTER_ATTRIBUTION_HEADERS
from nanobot.providers.registry import find_by_name
from nanobot.security.network import (
PinnedDNSAsyncTransport,
UnsafeURLRequestError,
resolve_url_target,
)
from nanobot.utils.helpers import detect_image_mime
_OPENROUTER_ATTRIBUTION_HEADERS = {
"HTTP-Referer": "https://github.com/HKUDS/nanobot",
"X-OpenRouter-Title": "nanobot",
"X-OpenRouter-Categories": "cli-agent,personal-agent",
}
_DEFAULT_TIMEOUT_S = 120.0
_IMAGE_DOWNLOAD_MAX_BYTES = 32 * 1024 * 1024
_IMAGE_DOWNLOAD_MAX_REDIRECTS = 5
_AIHUBMIX_TIMEOUT_S = 300.0
_AIHUBMIX_ASPECT_RATIO_SIZES = {
"1:1": "1024x1024",
@@ -29,6 +41,23 @@ _AIHUBMIX_ASPECT_RATIO_SIZES = {
}
_GEMINI_DEFAULT_TIMEOUT_S = 120.0
_GEMINI_IMAGEN_ASPECT_RATIOS = {"1:1", "9:16", "16:9", "3:4", "4:3"}
# Aspect ratios documented for every Gemini image model using generateContent.
_GEMINI_FLASH_COMMON_ASPECT_RATIOS = {
"1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9",
}
# Gemini 3.1 Flash and Flash Lite additionally accept extreme aspect ratios.
_GEMINI_31_FLASH_ASPECT_RATIOS = {
*_GEMINI_FLASH_COMMON_ASPECT_RATIOS,
"1:4",
"4:1",
"1:8",
"8:1",
}
# Gemini 3 Pro image models accept these sizes. Gemini 3.1 Flash adds 512,
# while Gemini 3.1 Flash Lite supports only 1K.
_GEMINI_3_IMAGE_SIZES = {"1K", "2K", "4K"}
_GEMINI_31_FLASH_IMAGE_SIZES = {"512", *_GEMINI_3_IMAGE_SIZES}
_GEMINI_31_FLASH_LITE_IMAGE_SIZES = {"1K"}
_OLLAMA_DEFAULT_SIDE = 1024
_OLLAMA_SIZE_PRESETS = {
"1K": 1024,
@@ -110,16 +139,81 @@ def _aihubmix_model_path(model: str) -> str:
async def _download_image_data_url(
client: httpx.AsyncClient,
url: str,
*,
proxy: str | None = None,
transport: httpx.AsyncBaseTransport | None = None,
) -> str:
response = await client.get(url)
try:
response.raise_for_status()
except httpx.HTTPStatusError as exc:
detail = response.text[:500]
raise ImageGenerationError(f"failed to download generated image: {detail}") from exc
raw = response.content
client_kwargs: dict[str, Any] = {
"follow_redirects": False,
"timeout": _DEFAULT_TIMEOUT_S,
"trust_env": False,
}
if proxy:
# An explicit provider proxy is a user-selected trusted egress boundary.
# Validate each URL locally, while the proxy owns final DNS resolution.
client_kwargs["proxy"] = proxy
else:
client_kwargs["transport"] = PinnedDNSAsyncTransport(inner=transport)
async with httpx.AsyncClient(**client_kwargs) as client:
current_url = url
for _ in range(_IMAGE_DOWNLOAD_MAX_REDIRECTS + 1):
if proxy:
ok, error, _ = resolve_url_target(
current_url,
trust_remote_dns=True,
)
if not ok:
raise ImageGenerationError(
f"blocked unsafe generated image URL: {error}"
)
async with client.stream("GET", current_url) as response:
if response.is_redirect:
location = response.headers.get("location")
if not location:
raise ImageGenerationError(
"generated image URL redirected without a location"
)
current_url = urljoin(str(response.url), location)
continue
try:
response.raise_for_status()
except httpx.HTTPStatusError as exc:
raise ImageGenerationError(
f"failed to download generated image (HTTP {response.status_code})"
) from exc
declared_size = response.headers.get("content-length")
if declared_size:
try:
if int(declared_size) > _IMAGE_DOWNLOAD_MAX_BYTES:
raise ImageGenerationError(
"generated image exceeded the 32 MiB download limit"
)
except ValueError:
pass
chunks: list[bytes] = []
total = 0
async for chunk in response.aiter_bytes():
total += len(chunk)
if total > _IMAGE_DOWNLOAD_MAX_BYTES:
raise ImageGenerationError(
"generated image exceeded the 32 MiB download limit"
)
chunks.append(chunk)
raw = b"".join(chunks)
break
else:
raise ImageGenerationError("generated image URL exceeded the redirect limit")
except UnsafeURLRequestError as exc:
raise ImageGenerationError(f"blocked unsafe generated image URL: {exc}") from exc
except httpx.RequestError as exc:
raise ImageGenerationError(f"failed to download generated image: {exc}") from exc
mime = detect_image_mime(raw)
if mime is None:
raise ImageGenerationError("generated image URL did not return a supported image")
@@ -227,6 +321,13 @@ class ImageGenerationProvider(ABC):
raise ImageGenerationError(f"{label} returned no images: {provider_error}")
raise ImageGenerationError(f"{label} returned no images for this request")
def _http_client_kwargs(self) -> dict[str, Any]:
kwargs: dict[str, Any] = {"timeout": self.timeout}
if self.proxy:
kwargs["proxy"] = self.proxy
kwargs["trust_env"] = False
return kwargs
async def _http_post(
self,
url: str,
@@ -239,11 +340,7 @@ class ImageGenerationProvider(ABC):
return await client.post(url, headers=headers, json=body)
if self._client is not None:
return await self._client.post(url, headers=headers, json=body)
client_kwargs: dict[str, Any] = {"timeout": self.timeout}
if self.proxy:
client_kwargs["proxy"] = self.proxy
client_kwargs["trust_env"] = False
async with httpx.AsyncClient(**client_kwargs) as c:
async with httpx.AsyncClient(**self._http_client_kwargs()) as c:
return await c.post(url, headers=headers, json=body)
@@ -301,7 +398,7 @@ class OpenRouterImageGenerationClient(ImageGenerationProvider):
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
**OPENROUTER_ATTRIBUTION_HEADERS,
**_OPENROUTER_ATTRIBUTION_HEADERS,
**self.extra_headers,
}
url = f"{self.api_base}/chat/completions"
@@ -371,7 +468,7 @@ class AIHubMixImageGenerationClient(ImageGenerationProvider):
}
size = _aihubmix_size(aspect_ratio, image_size)
client = self._client or httpx.AsyncClient(timeout=self.timeout)
client = self._client or httpx.AsyncClient(**self._http_client_kwargs())
try:
return await self._generate_with_client(
client,
@@ -431,7 +528,7 @@ class AIHubMixImageGenerationClient(ImageGenerationProvider):
raise ImageGenerationError(f"AIHubMix image generation failed: {detail}") from exc
payload = response.json()
images = await _aihubmix_images_from_payload(client, payload)
images = await _aihubmix_images_from_payload(payload, proxy=self.proxy)
self._require_images(images, payload)
@@ -631,7 +728,11 @@ class GeminiImageGenerationClient(ImageGenerationProvider):
prompt=prompt, model=model, aspect_ratio=aspect_ratio
)
return await self._generate_gemini_flash(
prompt=prompt, model=model, reference_images=reference_images or []
prompt=prompt,
model=model,
reference_images=reference_images or [],
aspect_ratio=aspect_ratio,
image_size=image_size,
)
async def _generate_imagen(
@@ -687,15 +788,22 @@ class GeminiImageGenerationClient(ImageGenerationProvider):
prompt: str,
model: str,
reference_images: list[str],
aspect_ratio: str | None = None,
image_size: str | None = None,
) -> GeneratedImageResponse:
parts: list[dict[str, Any]] = [
{"inlineData": image_path_to_inline_data(path)} for path in reference_images
]
parts.append({"text": prompt})
generation_config: dict[str, Any] = {"responseModalities": ["TEXT", "IMAGE"]}
image_config = _gemini_flash_image_config(model, aspect_ratio, image_size)
if image_config:
generation_config["responseFormat"] = {"image": image_config}
body: dict[str, Any] = {
"contents": [{"role": "user", "parts": parts}],
"generationConfig": {"responseModalities": ["TEXT", "IMAGE"]},
"generationConfig": generation_config,
}
body.update(self.extra_body)
@@ -744,9 +852,60 @@ class GeminiImageGenerationClient(ImageGenerationProvider):
)
def _gemini_flash_image_config(
model: str,
aspect_ratio: str | None,
image_size: str | None,
) -> dict[str, str]:
"""Build the ``responseFormat.image`` config for Gemini Flash image models.
Capabilities are model-specific: Gemini 3.1 Flash variants support four
additional extreme ratios, while configurable image sizes are limited to
the documented Gemini 3 image model families.
"""
config: dict[str, str] = {}
if aspect_ratio and aspect_ratio in _gemini_flash_supported_aspect_ratios(model):
config["aspectRatio"] = aspect_ratio
if image_size:
normalized = image_size.strip().upper()
if normalized in _gemini_flash_supported_image_sizes(model):
config["imageSize"] = normalized
return config
def _gemini_flash_supported_aspect_ratios(model: str) -> set[str]:
"""Return the documented aspect ratios for a generateContent image model."""
normalized = model.lower()
if (
"gemini-3.1-flash-lite-image" in normalized
or "gemini-3.1-flash-image" in normalized
):
return _GEMINI_31_FLASH_ASPECT_RATIOS
if "gemini-" in normalized and "image" in normalized:
return _GEMINI_FLASH_COMMON_ASPECT_RATIOS
return set()
def _gemini_flash_supported_image_sizes(model: str) -> set[str]:
"""Return the ``imageSize`` values documented for a Flash-path model.
Earlier Flash image models (2.0, 2.5) expose no configurable size. Gemini
3.1 Flash Lite is intentionally checked before the broader Flash match.
"""
normalized = model.lower()
if "gemini-3.1-flash-lite-image" in normalized:
return _GEMINI_31_FLASH_LITE_IMAGE_SIZES
if "gemini-3.1-flash-image" in normalized:
return _GEMINI_31_FLASH_IMAGE_SIZES
if "gemini-3-pro-image" in normalized:
return _GEMINI_3_IMAGE_SIZES
return set()
async def _aihubmix_images_from_payload(
client: httpx.AsyncClient,
payload: dict[str, Any],
*,
proxy: str | None = None,
) -> list[str]:
images: list[str] = []
candidates: list[Any] = []
@@ -764,7 +923,7 @@ async def _aihubmix_images_from_payload(
if value.startswith("data:image/"):
images.append(value)
elif value.startswith(("http://", "https://")):
images.append(await _download_image_data_url(client, value))
images.append(await _download_image_data_url(value, proxy=proxy))
return
if not isinstance(value, dict):
return
@@ -965,15 +1124,7 @@ class OpenAIImageGenerationClient(ImageGenerationProvider):
return model
async def _parse_images_response(self, payload: dict[str, Any]) -> list[str]:
client = self._client
owns_client = client is None
if owns_client:
client = httpx.AsyncClient(timeout=self.timeout)
try:
return await _openai_images_from_payload(client, payload)
finally:
if owns_client:
await client.aclose()
return await _openai_images_from_payload(payload, proxy=self.proxy)
async def _post_image_edit(
self,
@@ -1003,7 +1154,7 @@ class OpenAIImageGenerationClient(ImageGenerationProvider):
data=body,
files=files,
)
async with httpx.AsyncClient(timeout=self.timeout) as c:
async with httpx.AsyncClient(**self._http_client_kwargs()) as c:
return await c.post(
f"{self.api_base}/images/edits",
headers=headers,
@@ -1184,15 +1335,7 @@ class CustomImageGenerationClient(ImageGenerationProvider):
logger.info("Custom Images API response ({}): {}", response.status_code,
{k: v for k, v in payload.items() if k != "data"})
client = self._client
owns_client = client is None
if owns_client:
client = httpx.AsyncClient(timeout=self.timeout)
try:
images = await _openai_images_from_payload(client, payload)
finally:
if owns_client:
await client.aclose()
images = await _openai_images_from_payload(payload, proxy=self.proxy)
self._require_images(images, payload)
@@ -1385,8 +1528,9 @@ def _openai_explicit_size_supported(
async def _openai_images_from_payload(
client: httpx.AsyncClient,
payload: dict[str, Any],
*,
proxy: str | None = None,
) -> list[str]:
"""Extract images from OpenAI Images API response.
@@ -1402,7 +1546,7 @@ async def _openai_images_from_payload(
continue
url = item.get("url")
if isinstance(url, str) and url:
images.append(await _download_image_data_url(client, url))
images.append(await _download_image_data_url(url, proxy=proxy))
return images
@@ -1682,7 +1826,7 @@ class ZhipuImageGenerationClient(ImageGenerationProvider):
url = f"{self.api_base}/images/generations"
client = self._client or httpx.AsyncClient(timeout=self.timeout)
client = self._client or httpx.AsyncClient(**self._http_client_kwargs())
try:
return await self._generate_with_client(
client,
@@ -1716,7 +1860,7 @@ class ZhipuImageGenerationClient(ImageGenerationProvider):
raise ImageGenerationError(f"Zhipu image generation failed: {detail}") from exc
payload = response.json()
images = await _zhipu_images_from_payload(client, payload)
images = await _zhipu_images_from_payload(payload, proxy=self.proxy)
self._require_images(images, payload)
@@ -1740,8 +1884,9 @@ def _zhipu_size(
async def _zhipu_images_from_payload(
client: httpx.AsyncClient,
payload: dict[str, Any],
*,
proxy: str | None = None,
) -> list[str]:
"""Extract image data URLs from Zhipu API response.
@@ -1754,7 +1899,7 @@ async def _zhipu_images_from_payload(
continue
url = item.get("url")
if isinstance(url, str) and url:
images.append(await _download_image_data_url(client, url))
images.append(await _download_image_data_url(url, proxy=proxy))
return images
@@ -1840,7 +1985,7 @@ class ModelScopeImageGenerationClient(ImageGenerationProvider):
body.update(self.extra_body)
url = f"{self.api_base}/images/generations"
client = self._client or httpx.AsyncClient(timeout=self.timeout)
client = self._client or httpx.AsyncClient(**self._http_client_kwargs())
try:
return await self._generate_with_client(
client,
@@ -1917,7 +2062,7 @@ class ModelScopeImageGenerationClient(ImageGenerationProvider):
status = data.get("task_status")
if status == "SUCCEED":
return await self._collect_images(client, data)
return await self._collect_images(data)
if status == "FAILED":
raise ImageGenerationError(
f"ModelScope image generation task failed: {data}"
@@ -1930,9 +2075,8 @@ class ModelScopeImageGenerationClient(ImageGenerationProvider):
f"{_MODELSCOPE_POLL_MAX_ATTEMPTS} polls"
)
@staticmethod
async def _collect_images(
client: httpx.AsyncClient,
self,
data: dict[str, Any],
) -> list[str]:
images: list[str] = []
@@ -1941,7 +2085,9 @@ class ModelScopeImageGenerationClient(ImageGenerationProvider):
if url.startswith("data:image/"):
images.append(url)
else:
images.append(await _download_image_data_url(client, url))
images.append(
await _download_image_data_url(url, proxy=self.proxy)
)
return images
+6 -2
View File
@@ -35,7 +35,6 @@ from nanobot.providers.openai_responses import (
convert_tools,
parse_response_output,
)
from nanobot.providers.openrouter_attribution import OPENROUTER_ATTRIBUTION_HEADERS
if TYPE_CHECKING:
from openai import AsyncOpenAI as AsyncOpenAIType
@@ -55,6 +54,11 @@ _ALNUM = string.ascii_letters + string.digits
_STANDARD_TC_KEYS = frozenset({"id", "type", "index", "function"})
_STANDARD_FN_KEYS = frozenset({"name", "arguments"})
_DEFAULT_OPENROUTER_HEADERS = {
"HTTP-Referer": "https://github.com/HKUDS/nanobot",
"X-OpenRouter-Title": "nanobot",
"X-OpenRouter-Categories": "cli-agent,personal-agent",
}
_KIMI_K3_MODEL = "kimi-k3"
_KIMI_THINKING_MODELS: frozenset[str] = frozenset({
"kimi-k2.5",
@@ -446,7 +450,7 @@ class OpenAICompatProvider(LLMProvider):
self._effective_base = effective_base
self._default_headers = {"x-session-affinity": uuid.uuid4().hex}
if _uses_openrouter_attribution(spec, effective_base):
self._default_headers.update(OPENROUTER_ATTRIBUTION_HEADERS)
self._default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
if extra_headers:
self._default_headers.update(extra_headers)
self._api_key_for_client = api_key or "no-key"
@@ -1,7 +0,0 @@
"""Shared OpenRouter app attribution headers."""
OPENROUTER_ATTRIBUTION_HEADERS = {
"HTTP-Referer": "https://nanobot.wiki",
"X-OpenRouter-Title": "nanobot",
"X-OpenRouter-Categories": "cli-agent,personal-agent",
}
+64
View File
@@ -139,6 +139,70 @@ def append_runtime_context(
}
def detach_runtime_context(
content: Any,
marker: Mapping[str, Any],
) -> tuple[Any, list[str], list[dict[str, Any]]] | None:
"""Detach one validated runtime-context suffix for safe message merging."""
if marker.get("version") != 1:
return None
raw_sources = marker.get("sources")
sources = [
source
for source in raw_sources
if isinstance(source, str) and source
] if isinstance(raw_sources, list) else []
suffix = marker.get("suffix")
if isinstance(content, str) and isinstance(suffix, str) and suffix:
if content == suffix:
clean_content = ""
elif content.endswith("\n\n" + suffix):
clean_content = content[: -(len(suffix) + 2)]
else:
return None
return clean_content, sources, [{"type": "text", "text": suffix}]
expected = marker.get("blocks")
if isinstance(content, list) and isinstance(expected, list) and expected:
count = len(expected)
if content[-count:] != expected:
return None
return content[:-count], sources, deepcopy(expected)
return None
def reattach_runtime_context(
content: Any,
sources: Sequence[str],
blocks: Sequence[Mapping[str, Any]],
) -> tuple[Any, dict[str, Any]]:
"""Append detached runtime-context blocks after visible messages are merged."""
context_blocks = [deepcopy(dict(block)) for block in blocks]
if isinstance(content, str) and all(
block.get("type") == "text" and isinstance(block.get("text"), str)
for block in context_blocks
):
suffix = "\n\n".join(block["text"] for block in context_blocks)
merged = f"{content}\n\n{suffix}" if content else suffix
return merged, {
"version": 1,
"sources": list(sources),
"suffix": suffix,
}
visible_blocks = (
[*content]
if isinstance(content, list)
else ([] if content is None else [{"type": "text", "text": str(content)}])
)
return [*visible_blocks, *context_blocks], {
"version": 1,
"sources": list(sources),
"blocks": context_blocks,
}
def public_history_message(message: Mapping[str, Any]) -> dict[str, Any]:
"""Return a user-visible copy with trusted runtime context removed exactly."""
cleaned = deepcopy(dict(message))
+28 -3
View File
@@ -20,6 +20,7 @@ _BLOCKED_NETWORKS = [
ipaddress.ip_network("169.254.0.0/16"), # link-local / cloud metadata
ipaddress.ip_network("172.16.0.0/12"),
ipaddress.ip_network("192.168.0.0/16"),
ipaddress.ip_network("::/128"), # unspecified; may route to local host
ipaddress.ip_network("::1/128"),
ipaddress.ip_network("fc00::/7"), # unique local
ipaddress.ip_network("fe80::/10"), # link-local v6
@@ -73,7 +74,12 @@ def _is_private(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
return any(normalized in net for net in _BLOCKED_NETWORKS)
def resolve_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str, tuple[str, ...]]:
def resolve_url_target(
url: str,
*,
allow_loopback: bool = False,
trust_remote_dns: bool = False,
) -> tuple[bool, str, tuple[str, ...]]:
"""Validate a URL is safe to fetch: scheme, hostname, and resolved IPs.
``allow_loopback`` is intentionally narrow: it only permits literal
@@ -81,8 +87,14 @@ def resolve_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool,
loopback. It does not allow RFC1918, link-local, metadata, or public DNS
names that happen to resolve to loopback.
``trust_remote_dns`` accepts ordinary hostnames unavailable to local DNS.
This is only safe when a user-configured trusted proxy owns final DNS
resolution and network egress. Localhost names and private/internal IP
literals remain blocked.
Returns (ok, error_message, resolved_ips). When ok is True,
resolved_ips contains the public IPs that were validated for this URL.
resolved_ips contains the public IPs that were validated for this URL, or
is empty when an unresolved hostname is delegated to a trusted proxy.
"""
try:
p = urlparse(url)
@@ -101,7 +113,20 @@ def resolve_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool,
try:
infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
except socket.gaierror:
return False, f"Cannot resolve hostname: {hostname}", ()
if not trust_remote_dns:
return False, f"Cannot resolve hostname: {hostname}", ()
normalized_hostname = hostname.rstrip(".").lower()
if normalized_hostname == "localhost" or normalized_hostname.endswith(".localhost"):
return False, f"Blocked local/internal hostname: {hostname}", ()
try:
literal_addr = ipaddress.ip_address(normalized_hostname)
except ValueError:
return True, "", ()
if _is_private(literal_addr):
return False, f"Blocked private/internal address: {literal_addr}", ()
return True, "", (str(_normalize_addr(literal_addr)),)
addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
for info in infos:
+30
View File
@@ -2,7 +2,11 @@
from __future__ import annotations
from collections.abc import Mapping, MutableMapping
from typing import Any
UNIFIED_SESSION_KEY = "unified:default"
LAST_CHANNEL_METADATA_KEY = "last_channel"
def session_key_for_channel(channel: str, chat_id: str, *, unified_session: bool = False) -> str:
@@ -10,3 +14,29 @@ def session_key_for_channel(channel: str, chat_id: str, *, unified_session: bool
if unified_session:
return UNIFIED_SESSION_KEY
return f"{channel}:{chat_id}"
def remember_last_channel(
metadata: MutableMapping[str, Any],
channel: str,
chat_id: str,
) -> None:
"""Persist the latest concrete delivery route in session metadata."""
if not channel or not chat_id:
return
metadata[LAST_CHANNEL_METADATA_KEY] = f"{channel}:{chat_id}"
def last_channel_from_metadata(
metadata: Mapping[str, Any] | None,
) -> tuple[str, str] | None:
"""Return a concrete delivery route from persisted session metadata."""
if not isinstance(metadata, Mapping):
return None
route = metadata.get(LAST_CHANNEL_METADATA_KEY)
if not isinstance(route, str) or ":" not in route:
return None
channel, chat_id = route.split(":", 1)
if not channel or not chat_id:
return None
return channel, chat_id
+21 -66
View File
@@ -5,7 +5,6 @@ import errno
import json
import os
import re
import shutil
from collections import OrderedDict
from contextlib import suppress
from copy import deepcopy
@@ -138,6 +137,8 @@ class Session:
last_consolidated: int = 0 # Number of messages already consolidated to files
def __post_init__(self) -> None:
if not isinstance(self.metadata, dict):
self.metadata = {}
# An out-of-range offset (corrupt metadata) would hide all history; reset it.
if (
isinstance(self.last_consolidated, bool)
@@ -475,6 +476,14 @@ class SessionManager:
except _SESSION_DATA_ERRORS:
return None
@classmethod
def _session_key_from_path(cls, path: Path) -> str | None:
"""Decode a session key only from a canonical collision-resistant filename."""
key = cls._decode_storage_key(path.stem)
if key is None or cls._storage_key(key) != path.stem:
return None
return key
def _get_session_path(self, key: str) -> Path:
"""Get the collision-resistant workspace path for a session."""
return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
@@ -487,61 +496,6 @@ class SessionManager:
"""Legacy global session path (~/.nanobot/sessions/)."""
return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl"
@staticmethod
def _stored_key_for_path(path: Path) -> str | None:
"""Read the stored session key from a JSONL metadata row, if present."""
try:
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
data = json.loads(line)
if not isinstance(data, dict):
raise ValueError("session records must be JSON objects")
if data.get("_type") == "metadata":
stored_key = data.get("key")
return stored_key if isinstance(stored_key, str) else None
return None
except _SESSION_DATA_ERRORS:
return None
return None
def _resolve_session_path(self, key: str, *, migrate: bool = False) -> Path | None:
"""Resolve a session path, falling back to legacy storage locations."""
path = self._get_session_path(key)
if path.exists():
return path
# TODO(v0.3.1): Remove both legacy fallbacks. v0.3.0 is the final
# compatibility window for reading and lazily migrating legacy session files.
fallback_paths = [
(self._get_legacy_lossy_path(key), "legacy lossy path"),
(self._get_legacy_session_path(key), "legacy path"),
]
for fallback_path, description in fallback_paths:
if not fallback_path.exists():
continue
stored_key = self._stored_key_for_path(fallback_path)
if stored_key and stored_key != key:
logger.info(
"Skipping session {} from {} because it belongs to {}",
key,
description,
stored_key,
)
continue
if not migrate:
return fallback_path
try:
shutil.move(str(fallback_path), str(path))
logger.info("Migrated session {} from {}", key, description)
except Exception:
logger.exception("Failed to migrate session {}", key)
return None
return path
return None
def get_or_create(self, key: str) -> Session:
"""
Get an existing session or create a new one.
@@ -565,8 +519,8 @@ class SessionManager:
def _load(self, key: str) -> Session | None:
"""Load a session from disk."""
path = self._resolve_session_path(key, migrate=True)
if path is None:
path = self._get_session_path(key)
if not path.exists():
return None
try:
@@ -845,8 +799,8 @@ class SessionManager:
Returns ``{"key", "created_at", "updated_at", "metadata", "messages"}`` or
``None`` when the session file does not exist or fails to parse.
"""
path = self._resolve_session_path(key)
if path is None:
path = self._get_session_path(key)
if not path.exists():
return None
try:
messages: list[dict[str, Any]] = []
@@ -888,8 +842,8 @@ class SessionManager:
This is used by WebUI routes that need session-level metadata but not the
full conversation transcript.
"""
path = self._resolve_session_path(key)
if path is None:
path = self._get_session_path(key)
if not path.exists():
return None
try:
with open(path, encoding="utf-8") as f:
@@ -933,8 +887,9 @@ class SessionManager:
sessions = []
for path in self.sessions_dir.glob("*.jsonl"):
decoded = self._decode_storage_key(path.stem)
fallback_key = decoded or path.stem.replace("_", ":", 1)
storage_key = self._session_key_from_path(path)
if storage_key is None:
continue
try:
# Read the metadata line and a small preview for session lists.
with open(path, encoding="utf-8") as f:
@@ -944,7 +899,7 @@ class SessionManager:
if not isinstance(data, dict):
raise ValueError("session records must be JSON objects")
if data.get("_type") == "metadata":
key = data.get("key") or fallback_key
key = data.get("key") or storage_key
metadata = data.get("metadata", {})
title = _metadata_title(metadata)
preview = ""
@@ -989,7 +944,7 @@ class SessionManager:
except FileNotFoundError:
continue
except _SESSION_DATA_ERRORS:
repaired = self._repair(fallback_key, path=path)
repaired = self._repair(storage_key, path=path)
if repaired is not None:
sessions.append(
{
+10 -2
View File
@@ -16,6 +16,13 @@ def _int_or_zero(value: Any) -> int:
return 0 if value is None or value == "" else int(value)
def _optional_int(value: Any) -> int | None:
"""Coerce a stored JSON numeric; null/blank stays None."""
if value is None or value == "":
return None
return int(value)
@dataclass
class TriggerRunRecord:
"""A single local trigger delivery record."""
@@ -61,9 +68,10 @@ class LocalTrigger:
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "LocalTrigger":
raw_history = data.get("runHistory", data.get("run_history", [])) or []
history = [
record if isinstance(record, TriggerRunRecord) else TriggerRunRecord.from_dict(record)
for record in data.get("runHistory", data.get("run_history", []))
for record in raw_history
if isinstance(record, (dict, TriggerRunRecord))
]
return cls(
@@ -77,7 +85,7 @@ class LocalTrigger:
origin_metadata=dict(_get(data, "originMetadata", "origin_metadata", {}) or {}),
created_at_ms=_int_or_zero(_get(data, "createdAtMs", "created_at_ms", 0)),
updated_at_ms=_int_or_zero(_get(data, "updatedAtMs", "updated_at_ms", 0)),
last_run_at_ms=_get(data, "lastRunAtMs", "last_run_at_ms"),
last_run_at_ms=_optional_int(_get(data, "lastRunAtMs", "last_run_at_ms")),
last_status=_get(data, "lastStatus", "last_status"), # type: ignore[arg-type]
last_error=_get(data, "lastError", "last_error"),
run_history=history,
+17 -4
View File
@@ -14,6 +14,7 @@ _MAX_REPEAT_EXTERNAL_LOOKUPS = 2
# Third same-target workspace violation in a turn escalates to "stop retrying".
_MAX_REPEAT_WORKSPACE_VIOLATIONS = 2
_LENGTH_RECOVERY_TAIL_CHARS = 64
EMPTY_FINAL_RESPONSE_MESSAGE = (
"I completed the tool steps but couldn't produce a final answer. "
@@ -33,8 +34,10 @@ BUDGET_EXHAUSTED_FINALIZATION_PROMPT = (
)
LENGTH_RECOVERY_PROMPT = (
"Output limit reached. Continue exactly where you left off "
"— no recap, no apology. Break remaining work into smaller steps if needed."
"The previous assistant response was cut off. Continue the same response from its "
"exact endpoint. Output only new continuation text in the same language and style. "
"Do not acknowledge this instruction, restart the response, repeat its title or any "
"existing text, recap, or apologize."
)
SUSTAINED_GOAL_CONTINUE_PROMPT = (
@@ -79,9 +82,19 @@ def build_budget_exhausted_finalization_message() -> dict[str, str]:
return {"role": "user", "content": BUDGET_EXHAUSTED_FINALIZATION_PROMPT}
def build_length_recovery_message() -> dict[str, str]:
def build_length_recovery_message(content: str) -> dict[str, str]:
"""Prompt the model to continue after hitting output token limit."""
return {"role": "user", "content": LENGTH_RECOVERY_PROMPT}
tail = content[-_LENGTH_RECOVERY_TAIL_CHARS:]
prompt = (
f"{LENGTH_RECOVERY_PROMPT}\n\n"
"The following tail was already delivered to the user. Treat it as immutable "
"context and do not output it again:\n"
"<already_delivered_tail>\n"
f"{tail}\n"
"</already_delivered_tail>\n"
"Begin with the text that belongs immediately after this tail."
)
return {"role": "user", "content": prompt}
def build_goal_continue_message(custom: str | None = None) -> dict[str, str]:
+155
View File
@@ -0,0 +1,155 @@
"""Authenticated HTTP adapter for the extension management service."""
from __future__ import annotations
import json
from collections.abc import Callable
from typing import Any
from urllib.parse import unquote
from websockets.http11 import Request as WsRequest
from websockets.http11 import Response
from nanobot.extensions.service import ExtensionService
from nanobot.webui.http_utils import is_local_browser_request
_VALUES_HEADER = "X-Nanobot-Extension-Values"
_VALUES_MAX_BYTES = 32 * 1024
_ACTION_PATHS = {
"/api/extensions/install": "install",
"/api/extensions/enable": "enable",
"/api/extensions/disable": "disable",
"/api/extensions/trust": "trust",
"/api/extensions/untrust": "untrust",
"/api/extensions/permissions": "permissions",
"/api/extensions/uninstall": "uninstall",
}
class WebUIExtensionsRouter:
"""Keep extension policy and installation outside WebSocket transport."""
def __init__(
self,
*,
service: ExtensionService | None,
check_api_token: Callable[[WsRequest], bool],
json_response: Callable[[dict[str, Any]], Response],
error_response: Callable[[int, str | None], Response],
allow_remote_package_install: bool = False,
logger: Any,
) -> None:
self._service = service
self._check_api_token = check_api_token
self._json_response = json_response
self._error_response = error_response
self._allow_remote_package_install = allow_remote_package_install
self._logger = logger
async def dispatch(
self,
connection: Any,
request: WsRequest,
path: str,
) -> Response | None:
if not path.startswith("/api/extensions"):
return None
if not self._check_api_token(request):
return self._error_response(401, "Unauthorized")
if self._service is None:
return self._error_response(503, "Extension service is not available")
try:
if path == "/api/extensions":
if _method(request) != "GET":
return self._error_response(405, "Method not allowed")
return self._json_response(await self._service.status())
action = _ACTION_PATHS.get(path)
if action is None:
return None
if _method(request) != "POST":
return self._error_response(405, "Method not allowed")
if not self._mutation_allowed(action, connection, request):
return self._error_response(
403,
"Extension changes require a local WebUI connection",
)
values = self._values(request)
if (
action == "install"
and str(values.get("kind") or "git") == "local"
and not is_local_browser_request(connection, request.headers)
):
return self._error_response(
403,
"Local extension paths require a local WebUI connection",
)
return self._json_response(await self._run_action(action, values))
except KeyError as exc:
return self._error_response(404, str(exc))
except ValueError as exc:
return self._error_response(400, str(exc))
except RuntimeError as exc:
return self._error_response(502, str(exc))
except Exception:
self._logger.exception("extension management request failed")
return self._error_response(500, "Extension operation failed")
async def _run_action(self, action: str, values: dict[str, Any]) -> dict[str, Any]:
assert self._service is not None
extension_id = str(values.get("id") or "").strip()
if action == "install":
source = str(values.get("source") or "").strip()
if not source:
raise ValueError("Missing extension source")
return await self._service.install(
source,
kind=str(values.get("kind") or "git"),
ref=str(values.get("ref") or ""),
trusted=False,
)
if not extension_id:
raise ValueError("Missing extension ID")
if action == "enable":
return await self._service.set_enabled(extension_id, True)
if action == "disable":
return await self._service.set_enabled(extension_id, False)
if action == "trust":
return await self._service.set_trusted(extension_id, True)
if action == "untrust":
return await self._service.set_trusted(extension_id, False)
if action == "permissions":
permissions = values.get("permissions", [])
if not isinstance(permissions, list) or not all(
isinstance(permission, str) for permission in permissions
):
raise ValueError("Extension permissions must be an array of strings")
return await self._service.set_permissions(extension_id, set(permissions))
return await self._service.uninstall(extension_id)
def _values(self, request: WsRequest) -> dict[str, Any]:
raw = request.headers.get(_VALUES_HEADER)
if not raw:
return {}
if len(raw.encode("utf-8")) > _VALUES_MAX_BYTES:
raise ValueError("Extension request is too large")
try:
value = json.loads(unquote(raw))
except json.JSONDecodeError as exc:
raise ValueError("Invalid extension request") from exc
if not isinstance(value, dict):
raise ValueError("Extension request must be a JSON object")
return value
def _mutation_allowed(
self,
action: str,
connection: Any,
request: WsRequest,
) -> bool:
return is_local_browser_request(connection, request.headers) or (
action == "install" and self._allow_remote_package_install
)
def _method(request: WsRequest) -> str:
return str(getattr(request, "method", "GET")).upper()
+4
View File
@@ -51,6 +51,8 @@ def build_gateway_services(
local_trigger_pending_ids: Callable[[str], set[str]] | None = None,
channel_feature_action: Callable[..., Any] | None = None,
channel_runtime_status: Callable[[], dict[str, Any]] | None = None,
extension_service: Any | None = None,
allow_remote_package_install: bool = False,
logger: Any = default_logger,
) -> GatewayServices:
tokens = GatewayTokenStore()
@@ -94,6 +96,8 @@ def build_gateway_services(
local_trigger_pending_ids=local_trigger_pending_ids,
channel_feature_action=channel_feature_action,
channel_runtime_status=channel_runtime_status,
extension_service=extension_service,
allow_remote_package_install=allow_remote_package_install,
log=logger,
)
return GatewayServices(
+10 -5
View File
@@ -54,7 +54,11 @@ def _reconcile_index(session_manager: SessionManager) -> tuple[list[dict[str, An
for row in existing_rows or []
if isinstance(row.get("file"), str)
}
paths = sorted(session_manager.sessions_dir.glob("*.jsonl"))
paths = sorted(
path
for path in session_manager.sessions_dir.glob("*.jsonl")
if SessionManager._session_key_from_path(path) is not None
)
rows: list[dict[str, Any]] = []
changed = existing_rows is None
@@ -268,8 +272,9 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None:
storage_key = SessionManager._decode_storage_key(path.stem)
fallback_key = storage_key or path.stem.replace("_", ":", 1)
storage_key = SessionManager._session_key_from_path(path)
if storage_key is None:
return None
try:
with open(path, encoding="utf-8") as f:
first_line = f.readline().strip()
@@ -320,7 +325,7 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
fallback_time = datetime.fromtimestamp(signature["mtime_ns"] / 1e9).isoformat()
created_at_s = created_at_s or fallback_time
updated_at_s = updated_at_s or fallback_time
key = data.get("key") or fallback_key
key = data.get("key") or storage_key
activity_signature = _webui_activity_signature(key)
activity_updated_at = _webui_activity_updated_at(activity_signature)
return {
@@ -340,7 +345,7 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
**activity_signature,
}
except Exception:
repaired = session_manager._repair(fallback_key)
repaired = session_manager._repair(storage_key)
if repaired is None:
return None
return _indexed_row_for_session(repaired, path)
+6 -2
View File
@@ -1770,6 +1770,7 @@ def replay_transcript_to_ui_messages(
buffer_message_id = None
buffer_parts = []
continue
merge_next = rec.get("resuming") is True and rec.get("merge_next") is True
final_text = rec.get("text")
if isinstance(final_text, str):
if buffer_message_id is None:
@@ -1794,8 +1795,11 @@ def replay_transcript_to_ui_messages(
**_turn_fields(rec, "answer"),
}
break
buffer_message_id = None
buffer_parts = []
if merge_next:
buffer_parts = [final_text]
if not merge_next:
buffer_message_id = None
buffer_parts = []
continue
if ev == "reasoning_delta":
+14
View File
@@ -170,6 +170,8 @@ class GatewayHTTPHandler:
local_trigger_pending_ids: Callable[[str], set[str]] | None = None,
channel_feature_action: Callable[..., Any] | None = None,
channel_runtime_status: Callable[[], dict[str, Any]] | None = None,
extension_service: Any | None = None,
allow_remote_package_install: bool = False,
log: Any = logger,
) -> None:
self.config = config
@@ -190,6 +192,7 @@ class GatewayHTTPHandler:
self._log = log
self._runtime_surface = runtime_surface
from nanobot.webui.extensions_routes import WebUIExtensionsRouter
from nanobot.webui.settings_api import runtime_capabilities as _rc
from nanobot.webui.settings_routes import WebUISettingsRouter
@@ -206,6 +209,14 @@ class GatewayHTTPHandler:
channel_feature_action=channel_feature_action,
channel_runtime_status=channel_runtime_status,
)
self.extensions_routes = WebUIExtensionsRouter(
service=extension_service,
check_api_token=self.check_api_token,
json_response=_http_json_response,
error_response=_http_error,
allow_remote_package_install=allow_remote_package_install,
logger=self._log,
)
def workspace_controls_available(self, connection: Any) -> bool:
return self._runtime_surface == "native" or _is_localhost(connection)
@@ -247,6 +258,9 @@ class GatewayHTTPHandler:
# Settings routes (delegated)
response = await self.settings_routes.dispatch(connection, request, got)
if response is not None:
return response
response = await self.extensions_routes.dispatch(connection, request, got)
if response is not None:
return response
+53 -1
View File
@@ -11,7 +11,7 @@ from nanobot.agent.loop import AgentLoop
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.command import CommandContext
from nanobot.config.schema import AgentDefaults
from nanobot.config.schema import AgentDefaults, Config
from nanobot.providers.base import LLMResponse
@@ -180,12 +180,64 @@ class TestSessionTTLConfig:
assert data["idleCompactAfterMinutes"] == 30
assert "sessionTtlMinutes" not in data
def test_idle_scan_interval_defaults_to_sixty_seconds(self):
"""The config default should avoid scanning all sessions every idle tick."""
defaults = AgentDefaults()
assert defaults.idle_compact_check_interval_seconds == 60
def test_idle_scan_interval_uses_camel_case_config_key(self):
"""The JSON config should use the standard camelCase alias."""
defaults = AgentDefaults.model_validate({"idleCompactCheckIntervalSeconds": 10})
assert defaults.idle_compact_check_interval_seconds == 10
data = defaults.model_dump(mode="json", by_alias=True)
assert data["idleCompactCheckIntervalSeconds"] == 10
def test_session_file_cap_is_internal_constant(self):
"""Session file cap should remain an internal constant, not a config field."""
from nanobot.session.manager import FILE_MAX_MESSAGES
assert FILE_MAX_MESSAGES == 2000
class TestIdleScanThrottling:
"""Test scheduling of full idle-session scans."""
def test_configured_idle_scan_interval_throttles_checks(self, tmp_path, monkeypatch):
"""The configured interval should reach the loop and gate session scans."""
ticks = iter((1_000.0, 1_000.0, 1_009.999, 1_010.0))
monkeypatch.setattr("nanobot.agent.loop.time.monotonic", lambda: next(ticks))
config = Config.model_validate({
"agents": {
"defaults": {
"workspace": str(tmp_path),
"idleCompactCheckIntervalSeconds": 10,
}
}
})
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
loop = AgentLoop.from_config(config, provider=provider)
loop.auto_compact.check_expired = MagicMock()
loop._check_expired_sessions_if_due()
loop.auto_compact.check_expired.assert_called_once()
loop._check_expired_sessions_if_due()
loop.auto_compact.check_expired.assert_called_once()
loop._check_expired_sessions_if_due()
assert loop.auto_compact.check_expired.call_count == 2
def test_zero_idle_scan_interval_checks_every_tick(self, tmp_path, monkeypatch):
"""An explicit zero should leave each idle tick eligible to scan."""
monkeypatch.setattr("nanobot.agent.loop.time.monotonic", lambda: 1_000.0)
loop = _make_loop(tmp_path)
loop.auto_compact.check_expired = MagicMock()
loop._check_expired_sessions_if_due()
loop._check_expired_sessions_if_due()
assert loop.auto_compact.check_expired.call_count == 2
class TestAgentLoopTTLParam:
"""Test that AutoCompact receives and stores session_ttl_minutes."""
+9 -1
View File
@@ -353,6 +353,14 @@ class TestBuildSystemPrompt:
class TestBuildMessages:
def test_optional_arguments_are_keyword_only(self, tmp_path):
builder = _builder(tmp_path)
with pytest.raises(TypeError):
builder.build_system_prompt(["legacy-skill"])
with pytest.raises(TypeError):
builder.build_messages([], "hello", ["legacy-skill"])
def test_basic_empty_history(self, tmp_path):
builder = _builder(tmp_path)
messages = builder.build_messages([], "hello")
@@ -363,7 +371,7 @@ class TestBuildMessages:
def test_runtime_context_is_not_injected_by_default(self, tmp_path):
builder = _builder(tmp_path)
messages = builder.build_messages([], "hello", channel="cli", chat_id="direct")
messages = builder.build_messages([], "hello", channel="cli")
user_msg = str(messages[-1]["content"])
assert user_msg == "hello"
+1 -3
View File
@@ -70,7 +70,6 @@ def test_provider_context_appended_after_user_content(tmp_path) -> None:
history=[],
current_message="hello world",
channel="cli",
chat_id="direct",
runtime_context_blocks=[
RuntimeContextBlock(source="test", content="provider context"),
],
@@ -322,7 +321,7 @@ def test_build_messages_passes_channel_to_system_prompt(tmp_path) -> None:
messages = builder.build_messages(
history=[], current_message="hi",
channel="telegram", chat_id="123",
channel="telegram",
)
system = messages[0]["content"]
assert "Format Hint" in system
@@ -349,7 +348,6 @@ def test_subagent_result_does_not_create_consecutive_assistant_messages(tmp_path
history=[{"role": "assistant", "content": "previous result"}],
current_message="subagent result",
channel="cli",
chat_id="direct",
current_role="assistant",
)
@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind, TurnState
from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.config.schema import ChannelsConfig
@@ -27,7 +27,7 @@ def _make_loop(tmp_path: Path, channels_config: ChannelsConfig | None = None) ->
@pytest.mark.asyncio
async def test_state_restore_extracts_documents_by_default(
async def test_restore_turn_extracts_documents_by_default(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -52,14 +52,13 @@ async def test_state_restore_extracts_documents_by_default(
ctx = TurnContext(
msg=msg,
session_key="cli:c",
state=TurnState.RESTORE,
turn_id="turn-1",
runtime=loop.llm_runtime(),
kind=TurnKind.USER,
delivery=loop.turn_delivery_factory.create(msg, "cli:c"),
)
assert await loop._state_restore(ctx) == "ok"
await loop._restore_turn(ctx)
assert calls == [("summarize", [str(doc_path)])]
assert "Quarterly revenue" in ctx.msg.content
@@ -67,7 +66,7 @@ async def test_state_restore_extracts_documents_by_default(
@pytest.mark.asyncio
async def test_state_restore_references_documents_when_extraction_disabled(
async def test_restore_turn_references_documents_when_extraction_disabled(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -90,14 +89,13 @@ async def test_state_restore_references_documents_when_extraction_disabled(
ctx = TurnContext(
msg=msg,
session_key="cli:c",
state=TurnState.RESTORE,
turn_id="turn-1",
runtime=loop.llm_runtime(),
kind=TurnKind.USER,
delivery=loop.turn_delivery_factory.create(msg, "cli:c"),
)
assert await loop._state_restore(ctx) == "ok"
await loop._restore_turn(ctx)
assert "Quarterly revenue" not in ctx.msg.content
assert f"[Attachment: {doc_path}]" in ctx.msg.content
+4 -4
View File
@@ -414,13 +414,13 @@ class TestEphemeralDirect:
captured = {}
original_save = loop._state_save
original_save = loop._persist_turn
async def patched_save(ctx):
captured["ephemeral"] = ctx.ephemeral
return await original_save(ctx)
with patch.object(loop, "_state_save", side_effect=patched_save):
with patch.object(loop, "_persist_turn", side_effect=patched_save):
await loop.process_direct(
"test", session_key="dream:check", ephemeral=True,
)
@@ -435,13 +435,13 @@ class TestEphemeralDirect:
captured = {}
original_save = loop._state_save
original_save = loop._persist_turn
async def patched_save(ctx):
captured["ephemeral"] = ctx.ephemeral
return await original_save(ctx)
with patch.object(loop, "_state_save", side_effect=patched_save):
with patch.object(loop, "_persist_turn", side_effect=patched_save):
await loop.process_direct("test", session_key="cli:normal")
assert captured.get("ephemeral") is False
+183
View File
@@ -534,6 +534,189 @@ class TestToolEventProgress:
assert turn_end_msgs[0].content == ""
provider.chat_with_retry.assert_not_awaited()
@pytest.mark.asyncio
async def test_length_recovery_keeps_one_user_visible_stream(
self,
tmp_path: Path,
) -> None:
bus = MessageBus()
provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model"
responses = iter([
LLMResponse(content="first-", finish_reason="length"),
LLMResponse(content="second", finish_reason="stop"),
])
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
response = next(responses)
await on_content_delta(response.content or "")
return response
provider.chat_stream_with_retry = chat_stream_with_retry
provider.chat_with_retry = AsyncMock()
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
_attach_webui_runtime_events(loop, bus)
loop.tools.get_definitions = MagicMock(return_value=[])
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign]
await loop._dispatch(InboundMessage(
channel="websocket",
sender_id="u1",
chat_id="chat1",
content="give a long answer",
metadata={"_wants_stream": True},
))
outbound = []
while bus.outbound_size > 0:
outbound.append(await bus.consume_outbound())
deltas = [m.event for m in outbound if isinstance(m.event, StreamDeltaEvent)]
endings = [m.event for m in outbound if isinstance(m.event, StreamEndEvent)]
assert [event.content for event in deltas] == ["first-", "second"]
assert [event.resuming for event in endings] == [True, False]
assert [event.merge_next for event in endings] == [True, False]
assert {event.stream_id for event in [*deltas, *endings]} == {deltas[0].stream_id}
@pytest.mark.asyncio
async def test_length_recovery_streams_non_delta_terminal_segment(
self,
tmp_path: Path,
) -> None:
bus = MessageBus()
provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model"
call_count = 0
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
await on_content_delta("first-")
return LLMResponse(content="first-", finish_reason="length")
return LLMResponse(content="second", finish_reason="stop")
provider.chat_stream_with_retry = chat_stream_with_retry
provider.chat_with_retry = AsyncMock()
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
_attach_webui_runtime_events(loop, bus)
loop.tools.get_definitions = MagicMock(return_value=[])
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign]
await loop._dispatch(InboundMessage(
channel="websocket",
sender_id="u1",
chat_id="chat1",
content="give a long answer",
metadata={"_wants_stream": True},
))
outbound = []
while bus.outbound_size > 0:
outbound.append(await bus.consume_outbound())
deltas = [m.event for m in outbound if isinstance(m.event, StreamDeltaEvent)]
endings = [m.event for m in outbound if isinstance(m.event, StreamEndEvent)]
final = [m for m in outbound if m.content == "first-second"]
assert [event.content for event in deltas] == ["first-", "second"]
assert [event.merge_next for event in endings] == [True, False]
assert len(final) == 1
assert isinstance(final[0].event, StreamedResponseEvent)
@pytest.mark.asyncio
async def test_length_recovery_at_max_iterations_streams_only_missing_tail(
self,
tmp_path: Path,
) -> None:
bus = MessageBus()
provider = MagicMock()
provider.supports_progress_deltas = True
provider.get_default_model.return_value = "test-model"
async def chat_stream_with_retry(*, on_content_delta, **kwargs):
await on_content_delta("partial")
return LLMResponse(content="partial", finish_reason="length")
provider.chat_stream_with_retry = chat_stream_with_retry
provider.chat_with_retry = AsyncMock(
return_value=LLMResponse(content="summary", finish_reason="stop")
)
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
_attach_webui_runtime_events(loop, bus)
loop.max_iterations = 1
loop.tools.get_definitions = MagicMock(return_value=[])
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign]
await loop._dispatch(InboundMessage(
channel="websocket",
sender_id="u1",
chat_id="chat1",
content="give a long answer",
metadata={"_wants_stream": True},
))
outbound = []
while bus.outbound_size > 0:
outbound.append(await bus.consume_outbound())
deltas = [m.event for m in outbound if isinstance(m.event, StreamDeltaEvent)]
endings = [m.event for m in outbound if isinstance(m.event, StreamEndEvent)]
final = [m for m in outbound if isinstance(m.event, StreamedResponseEvent)]
assert [event.content for event in deltas] == ["partial", "\n\nsummary"]
assert [event.merge_next for event in endings] == [True, False]
assert {event.stream_id for event in [*deltas, *endings]} == {deltas[0].stream_id}
assert [message.content for message in final] == ["partial\n\nsummary"]
@pytest.mark.asyncio
async def test_cancelled_length_recovery_closes_merged_stream(
self,
tmp_path: Path,
) -> None:
bus = MessageBus()
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
loop = AgentLoop(bus=bus, provider=provider, workspace=tmp_path, model="test-model")
async def cancel_after_merge(
_msg: InboundMessage,
*,
on_stream,
on_stream_end,
**_kwargs,
):
assert on_stream is not None
assert on_stream_end is not None
await on_stream("partial")
await on_stream_end(resuming=True, merge_next=True)
raise asyncio.CancelledError
loop._process_message = cancel_after_merge # type: ignore[method-assign]
with pytest.raises(asyncio.CancelledError):
await loop._dispatch(InboundMessage(
channel="websocket",
sender_id="u1",
chat_id="chat1",
content="give a long answer",
metadata={"_wants_stream": True},
))
outbound = []
while bus.outbound_size > 0:
outbound.append(await bus.consume_outbound())
endings = [m.event for m in outbound if isinstance(m.event, StreamEndEvent)]
assert [(event.resuming, event.merge_next) for event in endings] == [
(True, True),
(False, False),
]
assert endings[0].stream_id == endings[1].stream_id
@pytest.mark.asyncio
async def test_non_streamed_finalization_is_delivered_as_regular_message(
self,
+125 -39
View File
@@ -7,7 +7,7 @@ import pytest
from loguru import logger
from nanobot.agent.context import ContextBuilder
from nanobot.agent.loop import AgentLoop, TurnState
from nanobot.agent.loop import AgentLoop
from nanobot.agent.tools.context import RequestContext, request_context
from nanobot.bus.events import InboundMessage
from nanobot.bus.outbound_events import (
@@ -30,6 +30,10 @@ from nanobot.runtime_context import (
)
from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
from nanobot.session.goal_state import GOAL_STATE_KEY
from nanobot.session.keys import (
LAST_CHANNEL_METADATA_KEY,
UNIFIED_SESSION_KEY,
)
from nanobot.session.manager import Session, SessionManager
from nanobot.session.turn_continuation import (
INTERNAL_CONTINUATION_META,
@@ -447,7 +451,6 @@ def test_build_and_save_preserves_user_text_containing_goal_guidance_tag(tmp_pat
[],
user_text,
channel="cli",
chat_id="direct",
)
assert "_meta" not in messages[-1]
@@ -472,7 +475,6 @@ def test_build_and_save_preserves_multimodal_user_block_starting_with_runtime_ta
user_text,
media=[str(image)],
channel="cli",
chat_id="direct",
)
loop._save_turn(session, messages, skip=1)
@@ -682,6 +684,85 @@ async def test_process_message_persists_user_message_before_turn_completes(tmp_p
assert persisted.updated_at >= persisted.created_at
@pytest.mark.asyncio
async def test_process_message_persists_unified_session_delivery_route(tmp_path: Path) -> None:
loop = _make_full_loop(tmp_path)
loop._unified_session = True
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign]
loop._run_agent_loop = AsyncMock(side_effect=RuntimeError("boom")) # type: ignore[method-assign]
msg = InboundMessage(
channel="feishu",
sender_id="u1",
chat_id="oc_123",
content="persist my route",
session_key_override=UNIFIED_SESSION_KEY,
)
with pytest.raises(RuntimeError, match="boom"):
await loop._process_message(msg)
loop.sessions.invalidate(UNIFIED_SESSION_KEY)
persisted = loop.sessions.get_or_create(UNIFIED_SESSION_KEY)
assert persisted.metadata[LAST_CHANNEL_METADATA_KEY] == "feishu:oc_123"
@pytest.mark.parametrize(
("msg", "is_user_turn"),
[
(
InboundMessage(
channel="cli",
sender_id="u1",
chat_id="direct",
content="cli input",
),
True,
),
(
InboundMessage(
channel="system",
sender_id="system",
chat_id="discord:automation",
content="system event",
),
False,
),
(
InboundMessage(
channel="discord",
sender_id="subagent",
chat_id="subagent-result",
content="subagent result",
),
True,
),
(
InboundMessage(
channel="discord",
sender_id="u1",
chat_id="automation",
content="scheduled turn",
metadata={CRON_TRIGGER_META: {"job_id": "job-1"}},
),
True,
),
],
)
def test_unified_session_route_ignores_non_user_destinations(
tmp_path: Path,
msg: InboundMessage,
is_user_turn: bool,
) -> None:
loop = _make_full_loop(tmp_path)
loop._unified_session = True
session = loop.sessions.get_or_create(UNIFIED_SESSION_KEY)
session.metadata[LAST_CHANNEL_METADATA_KEY] = "telegram:existing"
loop._remember_unified_session_route(session, msg, is_user_turn=is_user_turn)
assert session.metadata[LAST_CHANNEL_METADATA_KEY] == "telegram:existing"
# 1x1 PNG used by the media-persistence tests. ``extract_documents`` runs
# at the top of ``_process_message`` and filters ``msg.media`` down to
# paths that magic-byte-sniff as images, so the test fixture needs real
@@ -1018,7 +1099,7 @@ async def test_websocket_internal_continuation_keeps_single_visible_run(
@pytest.mark.asyncio
async def test_process_message_uses_context_chat_id_for_runtime_prompt(tmp_path: Path) -> None:
async def test_process_message_keeps_delivery_chat_for_thread_session(tmp_path: Path) -> None:
loop = _make_full_loop(tmp_path)
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign]
loop.context.build_messages = MagicMock( # type: ignore[method-assign]
@@ -1052,12 +1133,11 @@ async def test_process_message_uses_context_chat_id_for_runtime_prompt(tmp_path:
assert result is not None
assert result.chat_id == "thread-777"
assert loop.context.build_messages.call_args.kwargs["chat_id"] == "parent-456"
assert loop._run_agent_loop.call_args.kwargs["chat_id"] == "thread-777"
@pytest.mark.asyncio
async def test_process_message_uses_explicit_session_metadata_for_goal_context(
async def test_process_message_uses_explicit_session_for_goal_context(
tmp_path: Path,
) -> None:
loop = _make_full_loop(tmp_path)
@@ -1102,10 +1182,10 @@ async def test_process_message_uses_explicit_session_metadata_for_goal_context(
assert result is not None
assert result.content == "ok"
kwargs = loop.context.build_messages.call_args.kwargs
assert kwargs["chat_id"] == "chat-with-goal"
assert kwargs["session_metadata"] is system_session.metadata
assert GOAL_STATE_KEY not in kwargs["session_metadata"]
kwargs = loop._run_agent_loop.call_args.kwargs
assert kwargs["session"] is system_session
assert kwargs["session_key"] == "system"
assert GOAL_STATE_KEY not in kwargs["session"].metadata
@pytest.mark.asyncio
@@ -1487,27 +1567,26 @@ async def test_system_subagent_followup_does_not_log_content(tmp_path: Path) ->
@pytest.mark.asyncio
async def test_system_subagent_followup_uses_common_turn_state_machine(tmp_path: Path) -> None:
async def test_system_subagent_followup_uses_common_turn_lifecycle(tmp_path: Path) -> None:
loop = _make_full_loop(tmp_path)
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock( # type: ignore[method-assign]
return_value=False
)
visited: list[TurnState] = []
visited: list[str] = []
for state in (
TurnState.RESTORE,
TurnState.COMPACT,
TurnState.COMMAND,
TurnState.BUILD,
TurnState.RUN,
TurnState.SAVE,
TurnState.RESPOND,
for name in (
"_restore_turn",
"_compact_session",
"_dispatch_command",
"_build_turn",
"_run_turn",
"_persist_turn",
"_prepare_outbound",
):
name = f"_state_{state.name.lower()}"
original = getattr(loop, name)
async def record(ctx, *, _original=original, _state=state):
visited.append(_state)
async def record(ctx, *, _original=original, _name=name):
visited.append(_name)
return await _original(ctx)
setattr(loop, name, record)
@@ -1523,25 +1602,33 @@ async def test_system_subagent_followup_uses_common_turn_state_machine(tmp_path:
loop._run_agent_loop = fake_run_agent_loop # type: ignore[method-assign]
await loop._process_message(
InboundMessage(
channel="system",
sender_id="subagent",
chat_id="cli:test",
content="subagent result",
metadata={"subagent_task_id": "sub-1"},
logs: list[str] = []
sink_id = logger.add(logs.append, level="DEBUG", format="{message}")
try:
await loop._process_message(
InboundMessage(
channel="system",
sender_id="subagent",
chat_id="cli:test",
content="subagent result",
metadata={"subagent_task_id": "sub-1"},
)
)
)
finally:
logger.remove(sink_id)
assert visited == [
TurnState.RESTORE,
TurnState.COMPACT,
TurnState.COMMAND,
TurnState.BUILD,
TurnState.RUN,
TurnState.SAVE,
TurnState.RESPOND,
"_restore_turn",
"_compact_session",
"_dispatch_command",
"_build_turn",
"_run_turn",
"_persist_turn",
"_prepare_outbound",
]
logged = "".join(logs)
for stage in ("restore", "compact", "command", "build", "run", "save", "respond"):
assert f"Stage {stage} completed in" in logged
@pytest.mark.asyncio
@@ -1606,7 +1693,6 @@ def test_subagent_followup_uses_user_model_input_and_assistant_history(tmp_path:
current_message="subagent result",
current_role="user",
channel="cli",
chat_id="merge",
)
non_system = [m for m in projected if m.get("role") != "system"]
+1 -1
View File
@@ -225,7 +225,7 @@ async def test_process_message_captures_original_text_before_restore(
seen.append((ctx.original_user_text, ctx.runtime))
raise RuntimeError("captured before restore")
loop._state_restore = stop_after_capture # type: ignore[method-assign]
loop._restore_turn = stop_after_capture # type: ignore[method-assign]
with pytest.raises(RuntimeError, match="captured before restore"):
await loop._process_message(
+1 -1
View File
@@ -135,7 +135,7 @@ async def test_tool_fails_after_retry_exhausted():
assert "failed after retry" in output
assert "ClosedResourceError" in output
assert is_tool_error_result(wrapper.name, output)
assert is_tool_error_result(output)
assert session.call_tool.call_count == 2
+31
View File
@@ -2,6 +2,7 @@
import json
from datetime import datetime
from pathlib import Path
import pytest
@@ -538,3 +539,33 @@ class TestLegacyHistoryMigration:
assert entries[0]["timestamp"] == "2026-04-01 10:00"
assert "Broken" in entries[0]["content"]
assert "migration." in entries[0]["content"]
def test_history_skips_non_dict_jsonl_lines(tmp_path: Path) -> None:
"""Null/list/bool history lines must not crash reads or appends."""
memory = MemoryStore(tmp_path)
memory.history_file.parent.mkdir(parents=True, exist_ok=True)
memory.history_file.write_text(
"\n".join([
"null",
"[1, 2]",
"true",
json.dumps({
"cursor": 1,
"timestamp": "2026-01-01T00:00:00",
"content": "kept",
"session_key": "cli:t",
}),
"",
]),
encoding="utf-8",
)
entries = memory.read_unprocessed_history(since_cursor=0)
assert entries == [{
"cursor": 1,
"timestamp": "2026-01-01T00:00:00",
"content": "kept",
"session_key": "cli:t",
}]
next_cursor = memory.append_history("next", session_key="cli:t")
assert next_cursor == 2
+211 -1
View File
@@ -978,7 +978,14 @@ class TestMainMenuUpdate:
expected_provider_names = set()
seen_display_names: set[str] = set()
for spec in PROVIDERS:
if spec.name == "custom" or spec.is_oauth or spec.is_transcription_only:
if (
spec.name == "custom"
or spec.is_transcription_only
or (
spec.is_oauth
and spec.name not in onboard_wizard._QUICK_START_OAUTH_PROVIDERS
)
):
continue
if spec.display_name in seen_display_names:
continue
@@ -988,9 +995,212 @@ class TestMainMenuUpdate:
assert selected_provider_names == expected_provider_names
assert "assemblyai" not in selected_provider_names
assert choices["OpenAI Codex"] == "openai_codex"
assert "github_copilot" not in selected_provider_names
assert choices["OpenCode Zen"] == "opencode"
assert choices[onboard_wizard._QUICK_START_CUSTOM_PROVIDER_CHOICE] == "custom"
def test_quick_start_openai_codex_uses_oauth_and_default_model(self, monkeypatch):
"""Codex should authenticate without asking for an API key."""
config = Config()
oauth_calls: list[tuple[Config, str]] = []
model_prompts: list[tuple[str, str, str]] = []
monkeypatch.setattr(onboard_wizard, "_show_quick_start_progress", lambda *_args: None)
monkeypatch.setattr(
onboard_wizard,
"_select_with_back",
lambda *args, **kwargs: "OpenAI Codex",
)
def fail_api_key_prompt(*_args, **_kwargs):
raise AssertionError("OpenAI Codex Quick Start should not ask for an API key")
def fake_model_input(prompt, current, provider):
model_prompts.append((prompt, current, provider))
return current
monkeypatch.setattr(onboard_wizard, "_input_text", fail_api_key_prompt)
monkeypatch.setattr(onboard_wizard, "_input_model_with_autocomplete", fake_model_input)
monkeypatch.setattr(
onboard_wizard,
"_quick_start_oauth_login",
lambda selected_config, provider: oauth_calls.append(
(selected_config, provider)
)
or True,
)
assert onboard_wizard._configure_quick_start_provider(config) is True
assert oauth_calls == [(config, "openai_codex")]
assert model_prompts == [
("Model ID", "openai-codex/gpt-5.6-sol", "openai_codex")
]
assert config.providers.openai_codex.api_key is None
assert config.model_presets["primary"].provider == "openai_codex"
assert config.model_presets["primary"].model == "openai-codex/gpt-5.6-sol"
def test_quick_start_openai_codex_login_failure_does_not_create_preset(self, monkeypatch):
"""A failed Codex login must not leave a ready-looking model preset."""
config = Config()
monkeypatch.setattr(onboard_wizard, "_show_quick_start_progress", lambda *_args: None)
monkeypatch.setattr(
onboard_wizard,
"_select_with_back",
lambda *args, **kwargs: "OpenAI Codex",
)
monkeypatch.setattr(
onboard_wizard,
"_input_model_with_autocomplete",
lambda *args, **kwargs: "openai-codex/gpt-5.6-sol",
)
monkeypatch.setattr(onboard_wizard, "_quick_start_oauth_login", lambda *args: False)
assert onboard_wizard._configure_quick_start_provider(config) is False
assert "primary" not in config.model_presets
def test_quick_start_openai_codex_login_reuses_existing_token(self, monkeypatch):
"""Quick Start should not open a new login flow when Codex is already authenticated."""
import oauth_cli_kit
config = Config()
config.providers.openai.api_key = "${UNRELATED_MISSING_KEY}"
config.providers.openai_codex.proxy = "${CODEX_PROXY}"
token = SimpleNamespace(access="existing-token", account_id="account-123")
token_proxies: list[str | None] = []
login_calls: list[object] = []
monkeypatch.setenv("CODEX_PROXY", "http://127.0.0.1:8080")
monkeypatch.delenv("UNRELATED_MISSING_KEY", raising=False)
monkeypatch.setattr(
oauth_cli_kit,
"get_token",
lambda **kwargs: token_proxies.append(kwargs.get("proxy")) or token,
)
monkeypatch.setattr(
oauth_cli_kit,
"login_oauth_interactive",
lambda **kwargs: login_calls.append(kwargs),
)
monkeypatch.setattr(onboard_wizard.console, "print", lambda *args, **kwargs: None)
assert onboard_wizard._quick_start_oauth_login(config, "openai_codex") is True
assert token_proxies == ["http://127.0.0.1:8080"]
assert login_calls == []
assert config.providers.openai.api_key == "${UNRELATED_MISSING_KEY}"
assert config.providers.openai_codex.proxy == "${CODEX_PROXY}"
def test_quick_start_openai_codex_runs_interactive_login_for_bad_cached_token(
self, monkeypatch
):
"""A malformed cached token should fall back to the interactive OAuth flow."""
import oauth_cli_kit
config = Config()
config.providers.openai_codex.proxy = "http://127.0.0.1:8080"
prompts: list[str] = []
printed: list[tuple[tuple[object, ...], dict[str, object]]] = []
class FakePrompt:
def ask(self):
return "authorization-code"
def fake_login(**kwargs):
kwargs["print_fn"]("[bold]Open the browser[/bold]")
prompts.append(kwargs["prompt_fn"]("Paste the authorization code"))
assert kwargs["proxy"] == "http://127.0.0.1:8080"
return SimpleNamespace(
access="fresh-token",
account_id="[red]account-123[/red]",
)
monkeypatch.setattr(
oauth_cli_kit,
"get_token",
lambda **_kwargs: SimpleNamespace(account_id="missing-access"),
)
monkeypatch.setattr(oauth_cli_kit, "login_oauth_interactive", fake_login)
monkeypatch.setattr(
onboard_wizard,
"_get_questionary",
lambda: SimpleNamespace(text=lambda *_args, **_kwargs: FakePrompt()),
)
monkeypatch.setattr(
onboard_wizard.console,
"print",
lambda *args, **kwargs: printed.append((args, kwargs)),
)
assert onboard_wizard._quick_start_oauth_login(config, "openai_codex") is True
assert prompts == ["authorization-code"]
assert any(
args == ("[bold]Open the browser[/bold]",) and kwargs == {"markup": False}
for args, kwargs in printed
)
assert any(r"\[red]account-123\[/red]" in str(args[0]) for args, _kwargs in printed)
def test_quick_start_codex_auth_check_ignores_unrelated_missing_env(self, monkeypatch):
"""OAuth readiness should depend only on the Codex proxy and token."""
import oauth_cli_kit
config = Config()
config.providers.anthropic.api_key = "${UNRELATED_MISSING_KEY}"
monkeypatch.delenv("UNRELATED_MISSING_KEY", raising=False)
monkeypatch.setattr(
oauth_cli_kit,
"get_token",
lambda **kwargs: SimpleNamespace(access="existing-token"),
)
assert (
onboard_wizard._quick_start_oauth_is_authenticated(config, "openai_codex")
is True
)
def test_quick_start_codex_auth_check_rejects_malformed_token(self, monkeypatch):
"""A malformed cached token should report not-ready instead of crashing."""
import oauth_cli_kit
monkeypatch.setattr(
oauth_cli_kit,
"get_token",
lambda **_kwargs: SimpleNamespace(account_id="missing-access"),
)
assert (
onboard_wizard._quick_start_oauth_is_authenticated(Config(), "openai_codex")
is False
)
def test_quick_start_summary_reports_missing_codex_oauth(self, monkeypatch):
"""The review step should distinguish OAuth from an API-key setup."""
config = Config()
config.model_presets["primary"] = ModelPresetConfig(
model="openai-codex/gpt-5.6-sol",
provider="openai_codex",
)
captured: dict[str, list[tuple[str, str]]] = {}
monkeypatch.setattr(onboard_wizard, "_show_quick_start_progress", lambda *_args: None)
monkeypatch.setattr(
onboard_wizard,
"_quick_start_oauth_is_authenticated",
lambda *args: False,
)
monkeypatch.setattr(
onboard_wizard,
"_print_summary_panel",
lambda rows, _title: captured.setdefault("rows", rows),
)
onboard_wizard._show_quick_start_summary(config)
rows = dict(captured["rows"])
assert rows["Status"] == "OpenAI Codex OAuth login missing"
assert rows["WebSocket channel"] == "enabled"
def test_quick_start_provider_choice_skips_advanced_prompts(self, monkeypatch):
"""The beginner path should ask for provider credentials and model."""
config = Config()
+98
View File
@@ -450,6 +450,104 @@ async def test_runner_uses_specific_message_after_empty_finalization_retry():
assert result.stop_reason == "empty_final_response"
@pytest.mark.asyncio
async def test_runner_length_recovery_returns_all_segments():
"""Recovered output segments are returned together instead of only the tail."""
from nanobot.agent.runner import AgentRunner
provider = MagicMock(spec=LLMProvider)
provider.chat_with_retry = AsyncMock(side_effect=[
LLMResponse(content="first ", finish_reason="length"),
LLMResponse(content="second ", finish_reason="length"),
LLMResponse(content="third", finish_reason="stop"),
])
tools = MagicMock()
tools.get_definitions.return_value = []
runner = AgentRunner()
result = await runner.run(make_run_spec(provider,
initial_messages=[{"role": "user", "content": "give a long answer"}],
tools=tools,
model="test-model",
max_iterations=5,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
))
assert result.final_content == "first second third"
assert [
message["content"]
for message in result.messages
if message.get("role") == "assistant"
] == ["first", "second", "third"]
assert provider.chat_with_retry.await_count == 3
@pytest.mark.asyncio
async def test_runner_length_recovery_preserves_prefix_at_max_iterations():
"""Budget exhaustion must not replace output already produced by recovery."""
from nanobot.agent.runner import AgentRunner
provider = MagicMock(spec=LLMProvider)
provider.chat_with_retry = AsyncMock(
return_value=LLMResponse(content="partial answer", finish_reason="length")
)
tools = MagicMock()
tools.get_definitions.return_value = []
runner = AgentRunner()
result = await runner.run(make_run_spec(
provider,
initial_messages=[{"role": "user", "content": "give a long answer"}],
tools=tools,
model="test-model",
max_iterations=1,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
finalize_on_max_iterations=False,
max_iterations_message="limit reached",
))
assert result.stop_reason == "max_iterations"
assert result.final_content == "partial answer\n\nlimit reached"
assert result.pending_stream_content == "\n\nlimit reached"
assert [
message["content"]
for message in result.messages
if message.get("role") == "assistant"
] == ["partial answer", "limit reached"]
@pytest.mark.asyncio
async def test_runner_length_recovery_does_not_leak_across_tool_calls():
"""A recovered prefix belongs only to its contiguous response chain."""
from nanobot.agent.runner import AgentRunner
provider = MagicMock(spec=LLMProvider)
provider.chat_with_retry = AsyncMock(side_effect=[
LLMResponse(content="working", finish_reason="length"),
LLMResponse(
content=None,
tool_calls=[ToolCallRequest(id="call_1", name="read_file", arguments={"path": "x"})],
finish_reason="tool_calls",
),
LLMResponse(content="final answer", finish_reason="stop"),
])
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="file content")
runner = AgentRunner()
result = await runner.run(make_run_spec(provider,
initial_messages=[{"role": "user", "content": "inspect a file"}],
tools=tools,
model="test-model",
max_iterations=5,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
))
assert result.final_content == "final answer"
assert result.tools_used == ["read_file"]
@pytest.mark.asyncio
async def test_runner_empty_response_does_not_break_tool_chain():
"""An empty intermediate response must not kill an ongoing tool chain.

Some files were not shown because too many files have changed in this diff Show More